#!/usr/bin/env python3 """ Cluster Configuration Validator Validates cluster, node, and service configuration files to ensure: - Valid YAML syntax - Required fields are present - No duplicate IPs, MACs, or node names - Service dependencies are met - At least one master node exists - Role definitions are valid """ import sys import yaml import os from pathlib import Path from typing import Dict, List, Set, Any from collections import defaultdict class ValidationError(Exception): """Raised when validation fails""" pass class ConfigValidator: def __init__(self, config_dir: str): self.config_dir = Path(config_dir) self.errors = [] self.warnings = [] self.cluster_config = None self.node_configs = {} self.service_configs = {} def load_yaml(self, file_path: Path) -> Dict[str, Any]: """Load and parse a YAML file""" try: with open(file_path, 'r') as f: return yaml.safe_load(f) except yaml.YAMLError as e: raise ValidationError(f"YAML syntax error in {file_path}: {e}") except FileNotFoundError: raise ValidationError(f"File not found: {file_path}") def add_error(self, message: str): """Add a validation error""" self.errors.append(f"ERROR: {message}") def add_warning(self, message: str): """Add a validation warning""" self.warnings.append(f"WARNING: {message}") def validate_cluster_config(self): """Validate cluster.yaml""" cluster_file = self.config_dir / "cluster.yaml" if not cluster_file.exists(): self.add_error("cluster.yaml not found") return self.cluster_config = self.load_yaml(cluster_file) # Check required top-level keys required_keys = ['cluster', 'network', 'nodes', 'services'] for key in required_keys: if key not in self.cluster_config: self.add_error(f"cluster.yaml missing required key: {key}") if 'cluster' in self.cluster_config: cluster = self.cluster_config['cluster'] if 'name' not in cluster: self.add_error("cluster.name is required") if 'domain' not in cluster: self.add_error("cluster.domain is required") # Validate nodes list if 'nodes' in self.cluster_config: self.validate_cluster_nodes() # Validate services if 'services' in self.cluster_config: self.validate_cluster_services() def validate_cluster_nodes(self): """Validate nodes list in cluster.yaml""" nodes = self.cluster_config.get('nodes', []) if not nodes: self.add_error("No nodes defined in cluster.yaml") return seen_ips = set() seen_names = set() has_master = False for idx, node in enumerate(nodes): # Check required fields if 'name' not in node: self.add_error(f"Node at index {idx} missing 'name'") continue name = node['name'] # Check for duplicates if name in seen_names: self.add_error(f"Duplicate node name: {name}") seen_names.add(name) if 'ip' not in node: self.add_error(f"Node {name} missing 'ip'") else: ip = node['ip'] if ip in seen_ips: self.add_error(f"Duplicate IP address: {ip}") seen_ips.add(ip) if 'roles' not in node: self.add_error(f"Node {name} missing 'roles'") else: roles = node['roles'] if not roles: self.add_error(f"Node {name} has empty roles list") if 'master' in roles or 'control-plane' in roles: has_master = True if not has_master: self.add_error("Cluster must have at least one master/control-plane node") def validate_cluster_services(self): """Validate services in cluster.yaml""" services = self.cluster_config.get('services', {}) if 'enabled' not in services: self.add_warning("No enabled services defined in cluster.yaml") return enabled = services['enabled'] if not enabled: self.add_warning("No services enabled") # Check that service configs exist for enabled services services_dir = self.config_dir / "services" for service_name in enabled: service_file = services_dir / f"{service_name}.yaml" if not service_file.exists(): self.add_error(f"Service config not found: {service_file}") def validate_node_configs(self): """Validate all node configuration files""" nodes_dir = self.config_dir / "nodes" if not nodes_dir.exists(): self.add_error("nodes/ directory not found") return node_files = list(nodes_dir.glob("*.yaml")) if not node_files: self.add_warning("No node configuration files found in nodes/") return seen_macs = defaultdict(list) for node_file in node_files: node_config = self.load_yaml(node_file) node_name = node_file.stem self.node_configs[node_name] = node_config if 'node' not in node_config: self.add_error(f"{node_file.name}: missing 'node' section") continue node = node_config['node'] # Validate required fields if 'name' not in node: self.add_error(f"{node_file.name}: missing node.name") elif node['name'] != node_name: self.add_warning(f"{node_file.name}: node.name '{node['name']}' doesn't match filename") if 'roles' not in node: self.add_error(f"{node_file.name}: missing node.roles") elif not node['roles']: self.add_error(f"{node_file.name}: node.roles is empty") # Check MAC addresses for duplicates if 'hardware' in node and 'mac_addresses' in node['hardware']: for mac in node['hardware']['mac_addresses']: seen_macs[mac].append(node_name) # Report duplicate MACs for mac, nodes in seen_macs.items(): if len(nodes) > 1: self.add_error(f"Duplicate MAC address {mac} in nodes: {', '.join(nodes)}") # Check that cluster.yaml nodes have corresponding node configs if self.cluster_config and 'nodes' in self.cluster_config: cluster_nodes = {n['name'] for n in self.cluster_config['nodes']} config_nodes = set(self.node_configs.keys()) missing = cluster_nodes - config_nodes extra = config_nodes - cluster_nodes if missing: self.add_error(f"Nodes in cluster.yaml missing node configs: {', '.join(missing)}") if extra: self.add_warning(f"Node configs not referenced in cluster.yaml: {', '.join(extra)}") def validate_service_configs(self): """Validate service configuration files""" services_dir = self.config_dir / "services" if not services_dir.exists(): self.add_error("services/ directory not found") return service_files = list(services_dir.glob("*.yaml")) if not service_files: self.add_warning("No service configuration files found") return for service_file in service_files: service_config = self.load_yaml(service_file) service_name = service_file.stem self.service_configs[service_name] = service_config if 'service' not in service_config: self.add_error(f"{service_file.name}: missing 'service' section") continue service = service_config['service'] if 'name' not in service: self.add_error(f"{service_file.name}: missing service.name") elif service['name'] != service_name: self.add_warning(f"{service_file.name}: service.name '{service['name']}' doesn't match filename") if 'version' not in service: self.add_warning(f"{service_file.name}: missing service.version") def validate_all(self) -> bool: """Run all validations""" print(f"Validating configuration in: {self.config_dir}") print("=" * 60) try: self.validate_cluster_config() self.validate_node_configs() self.validate_service_configs() except ValidationError as e: self.add_error(str(e)) # Print results print() if self.warnings: print("Warnings:") for warning in self.warnings: print(f" {warning}") print() if self.errors: print("Errors:") for error in self.errors: print(f" {error}") print() print(f"Validation FAILED with {len(self.errors)} error(s)") return False else: print("✓ Validation PASSED") if self.warnings: print(f" ({len(self.warnings)} warning(s))") return True def main(): if len(sys.argv) > 1: config_dir = sys.argv[1] else: # Default to configs/ in the same directory as this script script_dir = Path(__file__).parent.parent config_dir = script_dir / "configs" validator = ConfigValidator(config_dir) success = validator.validate_all() sys.exit(0 if success else 1) if __name__ == "__main__": main()