poc1028a
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 load_nodes_from_json(json_file: str) -> Dict[str, List[str]]:
"""Load node configuration from JSON file
Expected JSON format:
{
"rack-1": ["node-1", "node-2", "node-3"],
"rack-2": ["node-4", "node-5", "node-6"],
"rack-3": ["node-7", "node-8", "node-9"]
}
"""
try:
with open(json_file, 'r') as f:
data = json.load(f)
# Validate the data format
if not isinstance(data, dict):
print(f"Error: JSON file must contain an object/dictionary", file=sys.stderr)
sys.exit(1)
nodes_by_label = {}
for rack_name, nodes in data.items():
if not isinstance(nodes, list):
print(f"Error: Rack '{rack_name}' must have a list of nodes", file=sys.stderr)
sys.exit(1)
# Validate that all nodes are strings
for node in nodes:
if not isinstance(node, str):
print(f"Error: All node names must be strings, found {type(node)} in rack '{rack_name}'", file=sys.stderr)
sys.exit(1)
# Sort nodes naturally and store
nodes_by_label[rack_name] = sorted(nodes, key=natural_sort_key)
print(f"Loaded {len(nodes_by_label)} racks from JSON file:", file=sys.stderr)
for rack, nodes in sorted(nodes_by_label.items(), key=lambda x: natural_sort_key(x[0])):
print(f" Rack {rack}: {len(nodes)} nodes ({', '.join(nodes)})", file=sys.stderr)
return nodes_by_label
except FileNotFoundError:
print(f"Error: JSON file '{json_file}' not found", file=sys.stderr)
sys.exit(1)
except json.JSONDecodeError as e:
print(f"Error: Invalid JSON in file '{json_file}': {e}", file=sys.stderr)
sys.exit(1)
except Exception as e:
print(f"Error loading JSON file '{json_file}': {e}", file=sys.stderr)
sys.exit(1)
def get_nodes_by_label(label_key: str, audit_mode: bool = False, json_file: str = None) -> Dict[str, List[str]]:
"""Get nodes grouped by label value or from JSON file"""
# If JSON file is provided, use it instead of kubectl
if json_file:
return load_nodes_from_json(json_file)
# Original kubectl-based logic
# 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,
ec_set_size: int = 8) -> List[Dict]:
"""Plan PVC placement across nodes with rack-aware constraint
Ensures that within any sliding window of size ec_set_size, we avoid
repeating the same rack if possible. Falls back to best-effort when
remaining eligible racks are fewer than ec_set_size.
"""
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)
# Per-rack node queues and counters
rack_to_nodes = {r: list(nodes_by_label[r]) for r in rack_names}
rack_capacity = {r: len(rack_to_nodes[r]) for r in rack_names} # desired appearances
rack_assigned = {r: 0 for r in rack_names} # assigned appearances so far
rack_next_idx = {r: 0 for r in rack_names}
total_nodes = sum(rack_capacity.values())
print(f"Total nodes available for placement: {total_nodes}")
# Calculate total servers across all pools
total_servers = 0
for pool in pools:
servers, _, _ = calculate_pool_pvcs(pool)
total_servers += servers
print(f"Total servers to place: {total_servers}")
print(f"Servers per node (average): {total_servers / total_nodes:.2f}")
server_count = 0
# sliding window of recent racks to avoid duplicates within ec set
from collections import deque
recent_racks = deque(maxlen=max(1, ec_set_size - 1))
rack_cursor = 0 # fair starting point across racks
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', {})
print(f"Pool '{pool_name}': {servers} servers, {volumes_per_server} volumes per server")
# Place each server (pod) on a single node with all its volumes
for server_idx in range(servers):
# Pick a rack using deficit scheduling with recent-window constraint
# Score = remaining_needed = capacity - assigned
selected_rack = None
best_score = -1
# First try excluding recent racks
for r in rack_names:
remaining_needed = rack_capacity[r] - rack_assigned[r]
if remaining_needed <= 0:
continue
if r in recent_racks:
continue
if remaining_needed > best_score:
best_score = remaining_needed
selected_rack = r
# If none found, allow recent racks (best-effort)
if selected_rack is None:
for r in rack_names:
remaining_needed = rack_capacity[r] - rack_assigned[r]
if remaining_needed <= 0:
continue
if remaining_needed > best_score:
best_score = remaining_needed
selected_rack = r
if selected_rack is None:
print("Error: No available racks with remaining nodes to place server", file=sys.stderr)
sys.exit(1)
# Get next node from the selected rack
node_idx = rack_next_idx[selected_rack]
selected_node = rack_to_nodes[selected_rack][node_idx]
rack_next_idx[selected_rack] += 1
rack_assigned[selected_rack] += 1
recent_racks.append(selected_rack)
# Debug info for first few placements
if server_count < 10:
print(f" Server {server_count} -> Node {selected_node} (Rack {selected_rack})")
# 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': selected_rack,
'pvc_template': pvc_template,
'namespace': namespace
}
placements.append(placement)
server_count += 1
# Analyze final placement distribution
node_usage = defaultdict(int)
rack_usage = defaultdict(int)
for placement in placements:
node_usage[placement['node']] += 1
rack_usage[placement['bucket']] += 1
# Report usage statistics
used_nodes = len([node for node, count in node_usage.items() if count > 0])
unused_nodes = total_nodes - used_nodes
print(f"\nPlacement Statistics:")
print(f" Nodes used: {used_nodes}/{total_nodes}")
print(f" Nodes unused: {unused_nodes}")
if unused_nodes > 0:
unused_node_list = []
for rack_name in rack_names:
for node in rack_to_nodes[rack_name]:
if node not in node_usage:
unused_node_list.append(f"{node} (Rack {rack_name})")
print(f" Unused nodes: {', '.join(unused_node_list[:10])}{'...' if len(unused_node_list) > 10 else ''}")
# Show per-rack distribution
print(f"\nPer-rack distribution:")
for rack_name in sorted(rack_usage.keys(), key=natural_sort_key):
rack_nodes = nodes_by_label[rack_name]
used_in_rack = len([node for node in rack_nodes if node in node_usage])
print(f" Rack {rack_name}: {rack_usage[rack_name]} PVCs, {used_in_rack}/{len(rack_nodes)} nodes used")
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']),
'target-node': placement['node'] # Add target node for reference
}
# 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_binding_pod_for_server(server_placements: List[Dict], objectstore_name: str,
pool_name: str, server_index: int) -> Dict:
"""Generate a temporary pod to bind all PVCs for a server to a specific node"""
# All placements for a server should be on the same node
node = server_placements[0]['node']
namespace = server_placements[0]['namespace']
# Pod name follows pattern similar to MinIO pod naming
pod_name = f"pvc-bind-{objectstore_name}-{pool_name}-{server_index}"
# Create volume mounts and volumes for all PVCs
volume_mounts = []
volumes = []
pvc_names = []
for placement in server_placements:
volume_name = f"data{placement['volume_index']}"
volume_mounts.append({
'name': volume_name,
'mountPath': f"/data{placement['volume_index']}"
})
volumes.append({
'name': volume_name,
'persistentVolumeClaim': {
'claimName': placement['pvc_name']
}
})
pvc_names.append(placement['pvc_name'])
# Create a command that lists all bound PVCs
command_str = f'echo "Binding {len(pvc_names)} PVCs for server {server_index} to node {node}:"; '
for pvc in pvc_names:
command_str += f'echo " - {pvc}"; '
command_str += 'sleep 1; echo "Binding complete"'
return {
'apiVersion': 'v1',
'kind': 'Pod',
'metadata': {
'name': pod_name,
'namespace': namespace,
'labels': {
'app': 'pvc-binder',
'objectstore': objectstore_name,
'pool': pool_name,
'server': str(server_index)
}
},
'spec': {
'restartPolicy': 'Never',
'nodeSelector': {
'kubernetes.io/hostname': node
},
'containers': [{
'name': 'binder',
'image': 'busybox:latest',
'command': ['sh', '-c', command_str],
'volumeMounts': volume_mounts,
'resources': {
'requests': {
'cpu': '10m',
'memory': '10Mi'
},
'limits': {
'cpu': '100m',
'memory': '50Mi'
}
}
}],
'volumes': volumes
}
}
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,
binding_mode: str = 'storage-class',
objectstore_name: str = None,
original_storage_class: str = None) -> str:
"""Generate final YAML output with StorageClasses and PVCs
Args:
binding_mode: 'pod' uses single storage class + binding pods,
'storage-class' (default) uses per-rack or per-node storage classes
"""
documents = []
storage_classes_created = {} # Maps sc_name to sc_doc
bucket_names = sorted(nodes_by_label.keys(), key=natural_sort_key)
if binding_mode == 'pod':
# Pod mode: Use single storage class from ObjectStore + binding pods
# We use the original storage class from the ObjectStore template
sc_name = original_storage_class or 'directpv-min-io'
# Generate PVCs with the single storage class
for placement in placements:
pvc = generate_pvc(placement, sc_name)
documents.append(pvc)
# Group placements by server (pool + server_index)
from collections import defaultdict
servers = defaultdict(list)
for placement in placements:
server_key = (placement['pool_name'], placement['server_index'])
servers[server_key].append(placement)
# Generate one binding pod per server (that mounts all its PVCs)
for (pool_name, server_index), server_placements in servers.items():
# Sort placements by volume index for consistent ordering
server_placements.sort(key=lambda p: p['volume_index'])
binding_pod = generate_binding_pod_for_server(
server_placements, objectstore_name, pool_name, server_index
)
documents.append(binding_pod)
elif binding_mode == 'storage-class':
# Storage-class mode: Use per-rack or per-node storage classes
if storage_class_per_node:
# Per-node storage classes
# 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 for storage-class mode)
# 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(), key=natural_sort_key):
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']
})
# Find max nodes in any rack
max_nodes = max(len(nodes) for nodes in nodes_by_label.values())
# Create deterministic pool names list for consistent coloring
pool_names = sorted({p['pool_name'] for p in placements}, key=natural_sort_key)
# 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(), key=natural_sort_key):
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-{pool_names.index(pod_pvcs[0]['pool']) % 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(), key=natural_sort_key):
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...")
nodes_by_label = get_nodes_by_label(args.node_label, audit_mode=True, json_file=args.nodes_json)
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...")
node_label_key = args.node_label or "rack"
erasure_sets = analyze_erasure_sets(mc_info, nodes_by_label, node_label_key,
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',
help='Node label for placement buckets (e.g., v1.min.io/rack)')
parser.add_argument('--nodes-json',
help='JSON file containing node configuration (rack -> [nodes])')
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)')
parser.add_argument('--binding-mode', choices=['pod', 'storage-class'], default='storage-class',
help='PVC binding mode: "pod" uses single storage class + binding pods, '
'"storage-class" (default) uses per-rack/per-node storage classes')
parser.add_argument('--ec-set-size', type=int, default=8,
help='Erasure coding set size for rack-aware placement (default: 8)')
# 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
# Validate required arguments for normal mode
if not args.node_label and not args.nodes_json:
print("Error: Either --node-label or --nodes-json is required", file=sys.stderr)
sys.exit(1)
if args.node_label and args.nodes_json:
print("Error: Cannot specify both --node-label and --nodes-json", file=sys.stderr)
sys.exit(1)
# 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 (now supports JSON file input)
print(f"Fetching nodes...")
nodes_by_label = get_nodes_by_label(args.node_label, json_file=args.nodes_json)
if not nodes_by_label:
if args.nodes_json:
print(f"Error: No nodes found in JSON file {args.nodes_json}", file=sys.stderr)
else:
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, args.ec_set_size)
print(f"Planned {len(placements)} PVCs across nodes")
# Validate node balance
node_pvc_counts = defaultdict(int)
for placement in placements:
node_pvc_counts[placement['node']] += 1
if node_pvc_counts:
min_pvcs = min(node_pvc_counts.values())
max_pvcs = max(node_pvc_counts.values())
print(f"Node PVC distribution: min={min_pvcs}, max={max_pvcs}")
if max_pvcs - min_pvcs > 1:
print(f"⚠️ Warning: Uneven PVC distribution detected (difference: {max_pvcs - min_pvcs})")
print("Nodes with PVC counts:")
for node, count in sorted(node_pvc_counts.items(), key=lambda x: x[1], reverse=True):
print(f" {node}: {count} PVCs")
else:
print("✅ PVC distribution is balanced across nodes")
# Extract original storage class from ObjectStore if using pod binding mode
original_storage_class = None
if args.binding_mode == 'pod' and pools:
# Get the storage class from the first pool's volumeClaimTemplate
first_pool = pools[0]
if 'volumeClaimTemplate' in first_pool:
original_storage_class = first_pool['volumeClaimTemplate'].get('spec', {}).get('storageClassName')
# Generate YAML output
print(f"Generating YAML output (mode: {args.binding_mode})...")
node_label_key = args.node_label or "rack" # Use provided label or default
yaml_output = generate_yaml_output(placements, node_label_key, nodes_by_label,
args.base_storage_class, args.directpv_drives_label,
args.storage_class_per_node, args.binding_mode,
objectstore_name, original_storage_class)
# 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