poc0925
#!/usr/bin/env python3
"""
Copyright MinIO 2025
PVC Allocator for MinIO ObjectStore
Generates PVCs with balanced placement across nodes for MinIO erasure sets
"""
import argparse
import json
import re
import subprocess
import sys
import yaml
from collections import defaultdict, Counter
from typing import Dict, List, Tuple, Any, Optional
def natural_sort_key(text):
"""
Generate a key for natural sorting that handles numbers correctly.
E.g., node-2 comes before node-10
"""
def atoi(text):
return int(text) if text.isdigit() else text
return [atoi(c) for c in re.split(r'(\d+)', text)]
def run_kubectl(cmd: str) -> str:
"""Execute kubectl command and return output"""
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
if result.returncode != 0:
print(f"Error running kubectl: {result.stderr}", file=sys.stderr)
sys.exit(1)
return result.stdout
def get_objectstore_cr(name: str, namespace: str = None, file: str = None) -> Dict:
"""Get ObjectStore CR from file or kubectl"""
if file:
with open(file, 'r') as f:
return yaml.safe_load(f)
elif name and namespace:
cmd = f"kubectl -n {namespace} get objectstore {name} -o yaml"
output = run_kubectl(cmd)
return yaml.safe_load(output)
else:
print("Error: Either provide --file or both --name and --namespace", file=sys.stderr)
sys.exit(1)
def get_nodes_by_label(label_key: str, audit_mode: bool = False) -> Dict[str, List[str]]:
"""Get nodes grouped by label value"""
# Get ALL nodes first, not just those with the label
cmd = "kubectl get nodes -o json"
output = run_kubectl(cmd)
nodes_data = json.loads(output)
nodes_by_label = defaultdict(list)
all_nodes = []
nodes_without_label = []
for node in nodes_data.get('items', []):
node_name = node['metadata']['name']
labels = node['metadata'].get('labels', {})
all_nodes.append(node_name)
if label_key in labels:
label_value = labels[label_key]
nodes_by_label[label_value].append(node_name)
else:
nodes_without_label.append(node_name)
# If no nodes have the label, handle appropriately
if not nodes_by_label:
if audit_mode and all_nodes:
# In audit mode, if no rack labels exist, treat each node as its own rack
print(f"Warning: No nodes have label '{label_key}'", file=sys.stderr)
print(f"Treating each node as its own rack for audit purposes", file=sys.stderr)
# Create synthetic rack labels (each node is its own rack)
for i, node in enumerate(sorted(all_nodes, key=natural_sort_key)):
nodes_by_label[str(i + 1)] = [node]
print(f"Created {len(nodes_by_label)} synthetic racks for {len(all_nodes)} nodes", file=sys.stderr)
elif all_nodes:
print(f"Warning: No nodes found with label '{label_key}'", file=sys.stderr)
print(f"Found {len(all_nodes)} nodes total: {', '.join(sorted(all_nodes, key=natural_sort_key))}", file=sys.stderr)
# Show example labels from first node
if nodes_data.get('items'):
first_node = nodes_data['items'][0]
node_labels = first_node['metadata'].get('labels', {})
relevant_labels = [k for k in node_labels.keys()
if 'min.io' in k or 'rack' in k.lower() or 'zone' in k.lower() or 'region' in k.lower()]
if relevant_labels:
print(f"\nPotentially relevant labels on node '{first_node['metadata']['name']}':", file=sys.stderr)
for key in sorted(relevant_labels):
print(f" {key}: {node_labels[key]}", file=sys.stderr)
elif nodes_without_label and audit_mode:
print(f"Warning: {len(nodes_without_label)} nodes don't have label '{label_key}': {', '.join(sorted(nodes_without_label, key=natural_sort_key))}", file=sys.stderr)
# Sort each bucket's nodes with natural sorting (node-2 before node-10)
for label_value in nodes_by_label:
nodes_by_label[label_value].sort(key=natural_sort_key)
return dict(nodes_by_label)
def calculate_pool_pvcs(pool: Dict) -> Tuple[int, int, int]:
"""Calculate total PVCs needed for a pool"""
servers = pool.get('servers', 1)
volumes_per_server = pool.get('volumesPerServer', 1)
total_pvcs = servers * volumes_per_server
return servers, volumes_per_server, total_pvcs
def plan_pvc_placement(pools: List[Dict], nodes_by_label: Dict[str, List[str]],
objectstore_name: str, namespace: str) -> List[Dict]:
"""Plan PVC placement across nodes - distribute servers round-robin across racks first"""
placements = []
# Sort rack names for consistent ordering (with natural sort for numeric rack names)
rack_names = sorted(nodes_by_label.keys(), key=natural_sort_key)
if not rack_names:
print("Error: No nodes found with the specified label", file=sys.stderr)
sys.exit(1)
# Create a 2D structure: racks -> nodes
# Each rack has a list of sorted nodes and a counter for round-robin within the rack
racks = []
for rack_name in rack_names:
nodes = sorted(nodes_by_label[rack_name], key=natural_sort_key)
racks.append({
'name': rack_name,
'nodes': nodes,
'next_node_index': 0 # Track which node to use next in this rack
})
server_count = 0
for pool_idx, pool in enumerate(pools):
pool_name = pool.get('name', f'pool-{pool_idx}')
servers, volumes_per_server, total_pvcs = calculate_pool_pvcs(pool)
# Get PVC template from pool - look for volumeClaimTemplate
pvc_template = pool.get('volumeClaimTemplate', {})
# Place each server (pod) on a single node with all its volumes
for server_idx in range(servers):
# Select rack using round-robin (server 0 -> rack 0, server 1 -> rack 1, etc.)
rack_idx = server_count % len(racks)
rack = racks[rack_idx]
# Select node within the rack using round-robin
node_idx = rack['next_node_index'] % len(rack['nodes'])
selected_node = rack['nodes'][node_idx]
# Increment the node counter for this rack for next server that lands here
rack['next_node_index'] += 1
# Place all volumes for this server on the same node
for volume_idx in range(volumes_per_server):
# Generate PVC name following StatefulSet naming convention
pvc_name = f"data{volume_idx}-{objectstore_name}-{pool_name}-{server_idx}"
placement = {
'pvc_name': pvc_name,
'pool_name': pool_name,
'server_index': server_idx,
'volume_index': volume_idx,
'node': selected_node,
'bucket': rack['name'],
'pvc_template': pvc_template,
'namespace': namespace
}
placements.append(placement)
server_count += 1
return placements
def generate_storage_class(name: str, base_storage_class: str, node_label_key: str,
node_label_value: str, node_names: List[str],
directpv_drives_label: str = None) -> Dict:
"""Generate DirectPV-compatible StorageClass with node affinity
Args:
node_names: List of node names for this storage class (single node for per-node,
multiple for per-rack)
"""
parameters = {
'csi.storage.k8s.io/fstype': 'xfs',
}
# Add DirectPV drive label if specified
if directpv_drives_label:
if '=' in directpv_drives_label:
key, value = directpv_drives_label.split('=', 1)
parameters[key] = value
else:
# If no value provided, assume it's a label key that should be present
parameters[directpv_drives_label] = ''
return {
'apiVersion': 'storage.k8s.io/v1',
'kind': 'StorageClass',
'metadata': {
'name': name,
'labels': {
'application-name': 'directpv.min.io',
'application-type': 'CSIDriver',
'directpv.min.io/created-by': 'pvc-allocator',
'directpv.min.io/version': 'v1beta1'
},
'finalizers': ['foregroundDeletion']
},
'provisioner': 'directpv-min-io',
'reclaimPolicy': 'Delete',
'volumeBindingMode': 'WaitForFirstConsumer',
'allowVolumeExpansion': True,
'allowedTopologies': [{
'matchLabelExpressions': [
{
'key': 'directpv.min.io/identity',
'values': ['directpv-min-io']
},
{
'key': 'directpv.min.io/node',
'values': node_names # Can be single node or multiple nodes
}
]
}],
'parameters': parameters
}
def generate_pvc(placement: Dict, storage_class_name: str) -> Dict:
"""Generate PVC from placement info"""
pvc_template = placement['pvc_template']
# Start with the entire template if it exists
pvc = {
'apiVersion': pvc_template.get('apiVersion', 'v1'),
'kind': 'PersistentVolumeClaim',
'metadata': {
'name': placement['pvc_name'],
'namespace': placement['namespace']
}
}
# Copy metadata from template and merge with our metadata
template_metadata = pvc_template.get('metadata', {}).copy()
# Preserve labels from template and add our tracking labels
template_labels = template_metadata.get('labels', {})
pvc['metadata']['labels'] = {
**template_labels, # Keep original labels
'app': 'minio',
'pool': placement['pool_name'],
'server': str(placement['server_index']),
'volume': str(placement['volume_index'])
}
# Preserve annotations from template
if 'annotations' in template_metadata:
pvc['metadata']['annotations'] = template_metadata['annotations']
# Copy the entire spec from template
pvc_spec = pvc_template.get('spec', {}).copy()
# Override storage class with our placement-specific one
pvc_spec['storageClassName'] = storage_class_name
pvc['spec'] = pvc_spec
return pvc
def generate_yaml_output(placements: List[Dict], node_label_key: str,
nodes_by_label: Dict[str, List[str]],
base_storage_class: str = None,
directpv_drives_label: str = None,
storage_class_per_node: bool = False) -> str:
"""Generate final YAML output with StorageClasses and PVCs"""
documents = []
storage_classes_created = {} # Maps sc_name to sc_doc
bucket_names = sorted(nodes_by_label.keys(), key=natural_sort_key)
if storage_class_per_node:
# Per-node storage classes (current behavior)
# Group placements by node for storage class creation
placements_by_node = defaultdict(list)
for placement in placements:
node_key = (placement['node'], placement['bucket'])
placements_by_node[node_key].append(placement)
# Generate StorageClasses per node
for (node, bucket), node_placements in placements_by_node.items():
bucket_index = bucket_names.index(bucket)
sc_name = f"minio-{bucket_index}-{node}"
if sc_name not in storage_classes_created:
sc = generate_storage_class(
name=sc_name,
base_storage_class=base_storage_class or 'standard',
node_label_key=node_label_key,
node_label_value=bucket,
node_names=[node], # Single node
directpv_drives_label=directpv_drives_label
)
documents.append(sc)
storage_classes_created[sc_name] = sc
else:
# Per-rack storage classes (default)
# Generate one storage class per rack
for bucket in bucket_names:
bucket_index = bucket_names.index(bucket)
sc_name = f"minio-rack-{bucket_index}"
nodes_in_rack = nodes_by_label[bucket]
sc = generate_storage_class(
name=sc_name,
base_storage_class=base_storage_class or 'standard',
node_label_key=node_label_key,
node_label_value=bucket,
node_names=nodes_in_rack, # All nodes in this rack
directpv_drives_label=directpv_drives_label
)
documents.append(sc)
storage_classes_created[sc_name] = sc
# Generate PVCs
for placement in placements:
# Determine storage class name based on granularity
bucket_index = bucket_names.index(placement['bucket'])
if storage_class_per_node:
sc_name = f"minio-{bucket_index}-{placement['node']}"
else:
sc_name = f"minio-rack-{bucket_index}"
pvc = generate_pvc(placement, sc_name)
documents.append(pvc)
# Convert to YAML
yaml_output = ""
for doc in documents:
yaml_output += "---\n"
yaml_output += yaml.dump(doc, default_flow_style=False, sort_keys=False)
return yaml_output
def generate_html_visualization(placements: List[Dict], nodes_by_label: Dict[str, List[str]],
objectstore_name: str) -> str:
"""Generate HTML table visualization of pod placement across racks and nodes"""
total_pvcs = len(placements)
num_racks = len(nodes_by_label)
total_nodes = sum(len(nodes) for nodes in nodes_by_label.values())
html = f"""<!DOCTYPE html>
<html>
<head>
<title>MinIO PVC Placement Visualization</title>
<style>
body {{
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
margin: 20px;
background-color: #f5f5f5;
}}
h1 {{
color: #333;
border-bottom: 2px solid #007bff;
padding-bottom: 10px;
}}
.info {{
background-color: #e7f3ff;
border-left: 4px solid #007bff;
padding: 10px;
margin-bottom: 20px;
}}
table {{
border-collapse: collapse;
width: 100%;
background-color: white;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}}
th {{
background-color: #007bff;
color: white;
padding: 12px;
text-align: left;
font-weight: 600;
position: sticky;
top: 0;
z-index: 10;
}}
.rack-header {{
background-color: #0056b3;
text-align: center;
}}
td {{
padding: 10px;
border: 1px solid #ddd;
vertical-align: top;
}}
.node-cell {{
background-color: #f8f9fa;
font-weight: 500;
border-right: 2px solid #007bff;
}}
.pod-cell {{
background-color: white;
}}
.pod-item {{
background-color: #e8f5e9;
border: 1px solid #4caf50;
border-radius: 4px;
padding: 5px;
margin: 3px 0;
font-size: 0.9em;
}}
.pool-0 {{ background-color: #e3f2fd; border-color: #2196f3; }}
.pool-1 {{ background-color: #f3e5f5; border-color: #9c27b0; }}
.pool-2 {{ background-color: #fff3e0; border-color: #ff9800; }}
.pool-3 {{ background-color: #e8f5e9; border-color: #4caf50; }}
.empty-cell {{
color: #999;
font-style: italic;
text-align: center;
}}
.stats {{
margin-top: 20px;
padding: 15px;
background-color: white;
border-radius: 5px;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}}
.stat-item {{
display: inline-block;
margin-right: 30px;
padding: 10px;
background-color: #f8f9fa;
border-radius: 4px;
}}
.stat-label {{
font-weight: 600;
color: #666;
}}
.stat-value {{
font-size: 1.2em;
color: #007bff;
font-weight: bold;
}}
</style>
</head>
<body>
<h1>MinIO PVC Placement Visualization - {objectstore_name}</h1>
<div class="info">
<strong>ObjectStore:</strong> {objectstore_name}<br>
<strong>Total PVCs:</strong> {total_pvcs}<br>
<strong>Racks:</strong> {num_racks}<br>
<strong>Total Nodes:</strong> {total_nodes}
</div>
<table>
<thead>
<tr>
<th>Node</th>
"""
# Add rack headers
for rack in sorted(nodes_by_label.keys()):
html += f' <th class="rack-header">Rack {rack}</th>\n'
html += ' </tr>\n </thead>\n <tbody>\n'
# Create placement lookup by node and organize by rack
placement_lookup = defaultdict(list)
for placement in placements:
key = (placement['node'], placement['bucket'])
# Generate the StatefulSet pod name that will use this PVC
# Format: {objectstore}-{pool}-{server_index}
pod_name = f"{objectstore_name}-{placement['pool_name']}-{placement['server_index']}"
placement_lookup[key].append({
'pod': pod_name,
'pvc': placement['pvc_name'],
'pool': placement['pool_name'],
'server_index': placement['server_index'],
'volume_index': placement['volume_index'],
'pool_idx': list(set(p['pool_name'] for p in placements)).index(placement['pool_name'])
})
# Find max nodes in any rack
max_nodes = max(len(nodes) for nodes in nodes_by_label.values())
# Generate rows for each node position
for node_idx in range(max_nodes):
html += ' <tr>\n'
html += f' <td class="node-cell">Node {node_idx + 1}</td>\n'
for rack in sorted(nodes_by_label.keys()):
nodes = nodes_by_label[rack]
if node_idx < len(nodes):
node = nodes[node_idx]
key = (node, rack)
pods = placement_lookup.get(key, [])
html += ' <td class="pod-cell">\n'
html += f' <strong>{node}</strong><br>\n'
if pods:
# Group by pod name to show each pod once with its volumes
pods_by_name = defaultdict(list)
for pod_info in pods:
pods_by_name[pod_info['pod']].append(pod_info)
for pod_name, pod_pvcs in pods_by_name.items():
pool_class = f"pool-{pod_pvcs[0]['pool_idx'] % 4}"
volume_info = f"Volumes: {', '.join(str(p['volume_index']) for p in pod_pvcs)}"
pvc_names = '\\n'.join(p['pvc'] for p in pod_pvcs)
html += f' <div class="pod-item {pool_class}" title="PVCs:\\n{pvc_names}">\n'
html += f' <strong>{pod_name}</strong><br>\n'
html += f' <small>{volume_info}</small>\n'
html += ' </div>\n'
else:
html += ' <div class="empty-cell">No pods</div>\n'
html += ' </td>\n'
else:
html += ' <td class="empty-cell">-</td>\n'
html += ' </tr>\n'
html += ' </tbody>\n </table>\n'
# Add statistics section
html += ' <div class="stats">\n'
html += ' <h2>Placement Statistics</h2>\n'
# Calculate stats per rack
for rack in sorted(nodes_by_label.keys()):
nodes = nodes_by_label[rack]
rack_pvcs = sum(len(placement_lookup.get((node, rack), [])) for node in nodes)
html += f' <div class="stat-item">\n'
html += f' <div class="stat-label">Rack {rack}</div>\n'
html += f' <div class="stat-value">{rack_pvcs} PVCs</div>\n'
html += f' </div>\n'
html += ' </div>\n'
html += '</body>\n</html>'
return html
def run_mc_command(alias: str, insecure: bool = False) -> Dict:
"""Run mc admin info command and return JSON output"""
cmd = f"mc admin info {alias} --json"
if insecure:
cmd += " --insecure"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
if result.returncode != 0:
print(f"Error running mc command: {result.stderr}", file=sys.stderr)
sys.exit(1)
try:
return json.loads(result.stdout)
except json.JSONDecodeError as e:
print(f"Error parsing mc output as JSON: {e}", file=sys.stderr)
print(f"Output was: {result.stdout}", file=sys.stderr)
sys.exit(1)
def get_pod_node_mapping(namespace: str, objectstore_name: str) -> Dict[str, str]:
"""
Get mapping of pod names to node names using kubectl
Returns dict mapping pod name to node name
"""
pod_to_node = {}
# Try different label selectors
label_selectors = [
f'v1.min.io/tenant={objectstore_name}', # MinIO Operator label
f'app=minio', # Generic MinIO label
f'app.kubernetes.io/name=minio', # Standard k8s label
f'app.kubernetes.io/instance={objectstore_name}', # Instance label
]
for selector in label_selectors:
try:
result = subprocess.run(
['kubectl', '-n', namespace, 'get', 'pods', '-l', selector, '-o', 'json'],
capture_output=True,
text=True,
check=True
)
pods_info = json.loads(result.stdout)
items = pods_info.get('items', [])
if items:
print(f" Found {len(items)} pods with selector '{selector}'")
for pod in items:
pod_name = pod['metadata']['name']
node_name = pod['spec'].get('nodeName', 'unknown')
# Only add if it looks like a MinIO pod (has 'minio' in name or matches objectstore name)
if objectstore_name in pod_name or 'minio' in pod_name.lower():
pod_to_node[pod_name] = node_name
if pod_to_node:
break # Found pods, stop trying other selectors
except subprocess.CalledProcessError:
continue # Try next selector
except json.JSONDecodeError:
continue
# If no labeled pods found, try to get all pods and filter by name
if not pod_to_node:
try:
print(f" No pods found with standard labels, trying name-based search...")
result = subprocess.run(
['kubectl', '-n', namespace, 'get', 'pods', '-o', 'json'],
capture_output=True,
text=True,
check=True
)
pods_info = json.loads(result.stdout)
for pod in pods_info.get('items', []):
pod_name = pod['metadata']['name']
# Filter for MinIO-like pod names
if (objectstore_name in pod_name or
'minio' in pod_name.lower() or
pod_name.startswith(f'{objectstore_name}-pool')):
node_name = pod['spec'].get('nodeName', 'unknown')
pod_to_node[pod_name] = node_name
if pod_to_node:
print(f" Found {len(pod_to_node)} MinIO pods by name pattern")
except (subprocess.CalledProcessError, json.JSONDecodeError) as e:
print(f"Warning: Could not get pod information: {e}", file=sys.stderr)
return pod_to_node
def analyze_erasure_sets(mc_info: Dict, nodes_by_label: Dict[str, List[str]],
node_label_key: str, namespace: str = None,
objectstore_name: str = None) -> Dict:
"""Analyze erasure set distribution from mc admin info output"""
# Extract servers and drives information
servers = mc_info.get('info', {}).get('servers', [])
if not servers:
print("Error: No servers found in mc admin info output", file=sys.stderr)
return {}
# Build erasure set mapping
erasure_sets = defaultdict(list) # erasure_set_id -> list of drive info
node_to_rack = {} # node -> rack mapping
# Build node to rack mapping
for rack, nodes in nodes_by_label.items():
for node in nodes:
node_to_rack[node] = rack
# Get pod to node mapping if namespace and objectstore are provided
pod_to_node = {}
if namespace and objectstore_name:
pod_to_node = get_pod_node_mapping(namespace, objectstore_name)
print(f"Found {len(pod_to_node)} pods in namespace {namespace}")
for pod, node in pod_to_node.items():
print(f" Pod {pod} is running on node {node}")
print(f"\nProcessing {len(servers)} servers from mc admin info...")
# Process each server
for i, server in enumerate(servers):
# Extract endpoint (e.g., "minio-pool-0-0.minio-hl.minio-tenant.svc.cluster.local:9000")
endpoint = server.get('endpoint', '')
# Debug: show raw endpoint
if i < 3 or 'unknown' in endpoint: # Show first few or problematic ones
print(f" Server {i}: endpoint = '{endpoint}'")
# Try to extract pod name from endpoint
# Common patterns:
# 1. "pod-name.service.namespace.svc.cluster.local:port"
# 2. "pod-name:port"
# 3. "node-name:port" (direct node access)
# 4. "myminio-pool-0-0" (just pod name)
# Remove port if present
pod_or_node_name = endpoint.split(':')[0] if ':' in endpoint else endpoint
# Remove domain parts if present
pod_or_node_name = pod_or_node_name.split('.')[0] if '.' in pod_or_node_name else pod_or_node_name
# Determine the actual node
actual_node = 'unknown'
# First try exact match with pod names
if pod_or_node_name in pod_to_node:
actual_node = pod_to_node[pod_or_node_name]
if i < 3: # Show first few for debugging
print(f" -> Matched pod '{pod_or_node_name}' -> Node {actual_node}")
# Then try node names
elif pod_or_node_name in node_to_rack:
actual_node = pod_or_node_name
if i < 3:
print(f" -> Direct node {actual_node}")
else:
# Try fuzzy matching with pod names
# Handle cases where endpoint might have different naming
matched = False
for pod_name, node_name in pod_to_node.items():
# Try various matching strategies
if (pod_name == pod_or_node_name or
pod_or_node_name in pod_name or
pod_name in pod_or_node_name or
# Handle pool naming differences (e.g., myminio-pool-0-0 vs minio-pool-0-0)
pod_name.split('-pool-')[-1] == pod_or_node_name.split('-pool-')[-1] if '-pool-' in pod_name and '-pool-' in pod_or_node_name else False):
actual_node = node_name
if i < 3:
print(f" -> Fuzzy matched to pod '{pod_name}' -> Node {actual_node}")
matched = True
break
if not matched:
print(f" WARNING: Could not determine node for endpoint '{endpoint}'")
print(f" Extracted name: '{pod_or_node_name}'")
if pod_to_node:
print(f" Available pods: {list(pod_to_node.keys())[:3]}...")
if node_to_rack:
print(f" Available nodes: {list(node_to_rack.keys())[:3]}...")
# Get rack for this node
rack = node_to_rack.get(actual_node, 'unknown')
# Process drives
drives = server.get('drives', [])
for drive in drives:
# Look for set_index field (this is what MinIO uses in newer versions)
erasure_set_id = drive.get('set_index')
if erasure_set_id is None:
# Fallback to 'set' field if set_index not found
erasure_set_id = drive.get('set')
if erasure_set_id is not None:
erasure_sets[erasure_set_id].append({
'node': actual_node,
'rack': rack,
'drive': drive.get('path', ''),
'state': drive.get('state', 'unknown'),
'endpoint': endpoint,
'pod_or_node': pod_or_node_name
})
print(f"\nFound {len(erasure_sets)} erasure sets with {sum(len(drives) for drives in erasure_sets.values())} total drives")
return erasure_sets
def generate_audit_report(erasure_sets: Dict, nodes_by_label: Dict[str, List[str]],
objectstore_name: str) -> str:
"""Generate a text-based audit report of erasure set placement"""
report = []
report.append("=" * 80)
report.append(f"MinIO Erasure Set Placement Audit Report")
report.append(f"ObjectStore: {objectstore_name}")
report.append("=" * 80)
report.append("")
# Summary statistics
total_sets = len(erasure_sets)
total_drives = sum(len(drives) for drives in erasure_sets.values())
total_racks = len(nodes_by_label)
total_nodes = sum(len(nodes) for nodes in nodes_by_label.values())
report.append("📊 Summary Statistics:")
report.append(f" • Total Erasure Sets: {total_sets}")
report.append(f" • Total Drives: {total_drives}")
report.append(f" • Total Racks: {total_racks}")
report.append(f" • Total Nodes: {total_nodes}")
report.append("")
# Analyze each erasure set
report.append("🔍 Erasure Set Analysis:")
report.append("-" * 40)
issues_found = []
for set_id in sorted(erasure_sets.keys()):
drives = erasure_sets[set_id]
report.append(f"\nErasure Set {set_id}:")
report.append(f" Drives: {len(drives)}")
# Count drives per rack
rack_counts = Counter(drive['rack'] for drive in drives)
report.append(f" Distribution across racks:")
for rack in sorted(rack_counts.keys(), key=natural_sort_key):
count = rack_counts[rack]
report.append(f" Rack {rack}: {count} drives")
# Check for issues: multiple drives from same erasure set on same rack
if count > 1:
issue = f"⚠️ WARNING: Erasure set {set_id} has {count} drives on rack {rack}"
issues_found.append(issue)
report.append(f" {issue}")
# Show detailed drive placement
report.append(" Drive details:")
for drive in sorted(drives, key=lambda x: (x['rack'], x['node'], x['drive'])):
state_icon = "✅" if drive['state'] == 'online' else "❌"
report.append(f" {state_icon} Rack {drive['rack']}, Node {drive['node']}: {drive['drive']}")
# Issues summary
report.append("")
report.append("=" * 80)
if issues_found:
report.append("⚠️ ISSUES FOUND:")
report.append("-" * 40)
for issue in issues_found:
report.append(issue)
report.append("")
report.append("RECOMMENDATION: Multiple drives from the same erasure set on the same rack")
report.append("reduces fault tolerance. Consider redistributing pods across racks.")
else:
report.append("✅ No placement issues found!")
report.append("All erasure sets are properly distributed across racks.")
report.append("=" * 80)
# Rack utilization
report.append("")
report.append("📈 Rack Utilization:")
report.append("-" * 40)
rack_drive_counts = defaultdict(int)
for drives in erasure_sets.values():
for drive in drives:
rack_drive_counts[drive['rack']] += 1
for rack in sorted(rack_drive_counts.keys(), key=natural_sort_key):
count = rack_drive_counts[rack]
bar = "█" * (count // 2) + ("▌" if count % 2 else "")
report.append(f" Rack {rack}: {bar} ({count} drives)")
return "\n".join(report)
def audit_mode(args):
"""Run in audit mode to analyze existing MinIO deployment"""
print("🔍 Running MinIO Placement Audit...")
# Get MinIO server info
print(f"Fetching MinIO server info from '{args.mc_alias}'...")
mc_info = run_mc_command(args.mc_alias, args.insecure)
# Get ObjectStore CR
print(f"Fetching ObjectStore CR...")
objectstore = get_objectstore_cr(args.name, args.namespace, args.file)
objectstore_name = objectstore['metadata']['name']
namespace = objectstore['metadata']['namespace']
# Get nodes grouped by label (pass audit_mode=True)
print(f"Fetching nodes with label {args.node_label}...")
nodes_by_label = get_nodes_by_label(args.node_label, audit_mode=True)
if not nodes_by_label:
print(f"Error: No nodes found at all", file=sys.stderr)
sys.exit(1)
# Analyze erasure sets
print("Analyzing erasure set distribution...")
erasure_sets = analyze_erasure_sets(mc_info, nodes_by_label, args.node_label,
namespace, objectstore_name)
# Generate report
report = generate_audit_report(erasure_sets, nodes_by_label, objectstore_name)
# Output report
print("\n" + report)
# Optionally save to file
if args.audit_output:
with open(args.audit_output, 'w') as f:
f.write(report)
print(f"\n📄 Report saved to: {args.audit_output}")
def main():
parser = argparse.ArgumentParser(
description='Generate PVCs for MinIO ObjectStore with balanced placement'
)
parser.add_argument('--name', help='Name of the ObjectStore CR')
parser.add_argument('--namespace', '-n', help='Namespace of the ObjectStore CR')
parser.add_argument('--file', '-f', help='Path to ObjectStore CR YAML file')
parser.add_argument('--node-label', required=True,
help='Node label for placement buckets (e.g., v1.min.io/rack)')
parser.add_argument('--output', '-o', default='pvcs.yaml',
help='Output YAML file path (default: pvcs.yaml)')
parser.add_argument('--html-output',
help='Generate HTML visualization of placement')
parser.add_argument('--base-storage-class',
help='Base storage class to use as template')
parser.add_argument('--directpv-drives-label',
help='DirectPV drive label to filter drives (e.g., directpv.min.io/tier=fast)')
parser.add_argument('--storage-class-per-node', action='store_true',
help='Create one storage class per node (default: one per rack)')
# Audit mode arguments
parser.add_argument('--audit', action='store_true',
help='Run in audit mode to analyze existing MinIO deployment')
parser.add_argument('--mc-alias',
help='MinIO mc alias for admin info command (required for audit mode)')
parser.add_argument('--insecure', action='store_true',
help='Use --insecure flag with mc command')
parser.add_argument('--audit-output',
help='Save audit report to file')
args = parser.parse_args()
# Check for audit mode
if args.audit:
if not args.mc_alias:
print("Error: --mc-alias is required for audit mode", file=sys.stderr)
sys.exit(1)
audit_mode(args)
return
# Normal mode: Generate PVCs
# Get ObjectStore CR
print(f"Fetching ObjectStore CR...")
objectstore = get_objectstore_cr(args.name, args.namespace, args.file)
objectstore_name = objectstore['metadata']['name']
namespace = objectstore['metadata']['namespace']
pools = objectstore['spec'].get('pools', [])
if not pools:
print("Error: No pools found in ObjectStore CR", file=sys.stderr)
sys.exit(1)
# Get nodes grouped by label
print(f"Fetching nodes with label {args.node_label}...")
nodes_by_label = get_nodes_by_label(args.node_label)
if not nodes_by_label:
print(f"Error: No nodes found with label {args.node_label}", file=sys.stderr)
sys.exit(1)
print(f"Found {len(nodes_by_label)} buckets with total {sum(len(nodes) for nodes in nodes_by_label.values())} nodes")
# Plan PVC placement
print("Planning PVC placement across nodes...")
placements = plan_pvc_placement(pools, nodes_by_label, objectstore_name, namespace)
print(f"Planned {len(placements)} PVCs across nodes")
# Generate YAML output
print("Generating YAML output...")
yaml_output = generate_yaml_output(placements, args.node_label, nodes_by_label,
args.base_storage_class, args.directpv_drives_label,
args.storage_class_per_node)
# Write to file
with open(args.output, 'w') as f:
f.write(yaml_output)
print(f"Successfully generated {args.output} with {len(placements)} PVCs")
# Generate HTML visualization if requested
if args.html_output:
print("Generating HTML visualization...")
html_output = generate_html_visualization(placements, nodes_by_label, objectstore_name)
with open(args.html_output, 'w') as f:
f.write(html_output)
print(f"Successfully generated HTML visualization: {args.html_output}")
# Print placement summary
print("\nPlacement Summary:")
placement_count = defaultdict(int)
for p in placements:
placement_count[f"{p['bucket']}/{p['node']}"] += 1
for location, count in sorted(placement_count.items()):
print(f" {location}: {count} PVCs")
if __name__ == "__main__":
main()
engineer