#!/usr/bin/env python3 """ VStorage Cluster Information Collector This script collects comprehensive information about VStorage clusters including: - Node configuration and system details - Storage roles and statistics - Network configuration - Disk and storage information - Cluster health and status Usage: python3 cluster-info.py [--debug] The script requires vinfra authentication and SSH access to cluster nodes. Results are written to cluster-info.json in the same directory. """ import json import subprocess import os import xml.etree.ElementTree as ET from typing import Dict, List, Any, Optional import sys import atexit from pathlib import Path import socket import getpass from datetime import datetime import argparse import time import traceback # Current script version SCRIPT_VERSION = "1.0.0" # Global debug flag DEBUG = False class CommandRunner: """Handles execution of both local and SSH commands with built-in SSH key management""" _instance = None DEFAULT_TIMEOUT = 300 # 5 minutes in seconds def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if self._initialized: return self.temp_dir = Path("/tmp/vstorage-info") self.private_key = self.temp_dir / "id_rsa" self.public_key = self.temp_dir / "id_rsa.pub" self.key_id = None self._initialize() atexit.register(self._cleanup) self._initialized = True def run(self, cmd: str, host: Optional[str] = None, timeout: Optional[int] = None) -> str: """Run a command either locally or via SSH Args: cmd: The command to run host: Optional remote host. If specified, runs command via SSH timeout: Command execution timeout in seconds. Defaults to DEFAULT_TIMEOUT """ try: if host: full_cmd = f"ssh -i {self.private_key} -o StrictHostKeyChecking=no root@{host} {cmd}" else: full_cmd = cmd if DEBUG: print(f"DEBUG: Executing command: {full_cmd}", file=sys.stderr) result = subprocess.run( full_cmd, shell=True, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=timeout or self.DEFAULT_TIMEOUT ) return result.stdout.strip() except subprocess.TimeoutExpired as e: timeout_val = timeout or self.DEFAULT_TIMEOUT error_msg = f"Command '{full_cmd}' timed out after {timeout_val} seconds: {e}" if DEBUG: print(f"DEBUG: {error_msg}", file=sys.stderr) raise RuntimeError(error_msg) from e except subprocess.CalledProcessError as e: error_msg = f"Error running command '{full_cmd}': {e}" if e.stderr: error_msg += f"\nStderr: {e.stderr}" if DEBUG: print(f"DEBUG: {error_msg}", file=sys.stderr) raise RuntimeError(error_msg) from e def _initialize(self): """Initialize SSH keys and add to cluster""" self._cleanup() self.temp_dir.mkdir(parents=True, exist_ok=True) try: if DEBUG: print(f"DEBUG: Creating SSH keys in {self.temp_dir}", file=sys.stderr) self.run(f"ssh-keygen -t rsa -b 2048 -C vstorage-info@automation -f {self.private_key} -N ''") self.private_key.chmod(0o600) self.public_key.chmod(0o644) if DEBUG: print(f"DEBUG: Adding SSH key to cluster", file=sys.stderr) result = json.loads(self.run(f"vinfra cluster sshkey add {self.public_key} --wait -f json")) self.key_id = result["id"] if DEBUG: print(f"DEBUG: SSH key added with ID: {self.key_id}", file=sys.stderr) except Exception: self._cleanup() raise def _cleanup(self): """Cleanup SSH keys and temporary files""" if self.key_id: try: if DEBUG: print(f"DEBUG: Removing SSH key {self.key_id} from cluster", file=sys.stderr) self.run(f"vinfra cluster sshkey delete {self.key_id} --wait") except Exception as e: print(f"Warning: Failed to delete SSH key from cluster: {e}", file=sys.stderr) self.key_id = None if self.temp_dir.exists(): for file in self.temp_dir.glob("*"): try: file.unlink(missing_ok=True) except Exception as e: if DEBUG: print(f"DEBUG: Failed to delete {file}: {e}", file=sys.stderr) try: self.temp_dir.rmdir() if DEBUG: print(f"DEBUG: Cleaned up temporary directory {self.temp_dir}", file=sys.stderr) except Exception as e: if DEBUG: print(f"DEBUG: Failed to remove directory {self.temp_dir}: {e}", file=sys.stderr) class Utils: """Utility functions""" @staticmethod def sizeof_fmt(num: float, suffix: str = 'B') -> str: """Convert bytes to human readable format""" for unit in ['', 'Ki', 'Mi', 'Gi', 'Ti', 'Pi', 'Ei', 'Zi']: if abs(num) < 1024.0: return f"{num:3.1f} {unit}{suffix}" num /= 1024.0 return f"{num:.1f} Yi{suffix}" @staticmethod def safe_get_text(element: Optional[ET.Element], child_name: str, default: str = "") -> str: """Safely get text from an XML element's child""" if element is None: return default child = element.find(child_name) return child.text if child is not None and child.text else default @staticmethod def safe_get_int(element: Optional[ET.Element], default: int = 0) -> int: """Safely get integer value from XML element""" if element is None or element.text is None: return default try: return int(element.text) except (ValueError, TypeError): return default @staticmethod def _is_ip_address(addr: str) -> bool: """Check if a string is an IP address""" try: socket.inet_aton(addr) return True except socket.error: return False class CPUInfoParser: """Parses CPU information""" def parse(self, cpu_info: str) -> List[Dict[str, Any]]: processors: Dict[str, Dict[str, Any]] = {} current_processor: Dict[str, Any] = {} for line in cpu_info.splitlines(): line = line.strip() if not line: if current_processor: physical_id = current_processor.get("physical id", "0") if physical_id not in processors: processors[physical_id] = current_processor processors[physical_id]["cores_count"] = 1 else: processors[physical_id]["cores_count"] += 1 current_processor = {} continue if ":" not in line: continue key, value = [part.strip() for part in line.split(":", 1)] if key in ["processor", "core id", "cpu cores"]: try: value = int(value) except ValueError: pass current_processor[key] = value if current_processor: physical_id = current_processor.get("physical id", "0") if physical_id not in processors: processors[physical_id] = current_processor processors[physical_id]["cores_count"] = 1 else: processors[physical_id]["cores_count"] += 1 return list(processors.values()) class RAMInfoParser: """Parses RAM information""" def parse(self, mem_info: str) -> Dict[str, Any]: for line in mem_info.splitlines(): if line.startswith("Mem:"): parts = line.split() return { "total": Utils.sizeof_fmt(int(parts[1])), "used": Utils.sizeof_fmt(int(parts[2])), "free": Utils.sizeof_fmt(int(parts[3])), "shared": Utils.sizeof_fmt(int(parts[4])), "buff_cache": Utils.sizeof_fmt(int(parts[5])), "available": Utils.sizeof_fmt(int(parts[6])) } return {} class DiskInfoParser: """Parses disk information with comprehensive RAID detection""" def parse(self, df_output: str, host: Optional[str] = None, cmd_runner: Optional[CommandRunner] = None) -> Dict[ str, Any]: for line in df_output.splitlines(): if line.startswith("/dev"): parts = line.split() device = parts[0] return { "device": device, "total": Utils.sizeof_fmt(int(parts[1])), "used": Utils.sizeof_fmt(int(parts[2])), "available": Utils.sizeof_fmt(int(parts[3])), "use_percentage": parts[4], "raid": self._get_raid_info(device, host, cmd_runner) } return {} def _get_raid_info(self, device: str, host: Optional[str], cmd_runner: Optional[CommandRunner]) -> Dict[str, Any]: if not cmd_runner: cmd_runner = CommandRunner() raid_info = { "is_raid": False, "raid_type": None, "raid_level": None, "controller": None, "state": None, "error_message": None, "disks": [] } try: # Get the real device (in case it's a partition) parent_device = cmd_runner.run(f"lsblk -no pkname {device} | head -1", host).strip() if parent_device: device = f"/dev/{parent_device}" # First check for software RAID (mdadm) if self._check_software_raid(device, host, cmd_runner, raid_info): return raid_info # Then check for hardware RAID self._check_hardware_raid(device, host, cmd_runner, raid_info) except Exception as e: raid_info["error_message"] = f"Failed to get RAID information: {str(e)}" if DEBUG: print(f"DEBUG: RAID detection error: {e}", file=sys.stderr) return raid_info def _check_software_raid(self, device: str, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]) -> bool: """Check for software RAID (mdadm) and populate raid_info""" try: # Check if device is part of MD RAID mdadm_output = cmd_runner.run(f"mdadm --detail {device} 2>/dev/null || true", host) if "Raid Level" in mdadm_output: raid_info["is_raid"] = True raid_info["raid_type"] = "software" raid_info["controller"] = "mdadm" # Parse mdadm output for line in mdadm_output.splitlines(): line = line.strip() if "Raid Level" in line: raid_info["raid_level"] = line.split(":")[-1].strip() elif line.startswith("State :"): raid_info["state"] = line.split(":")[-1].strip() elif "/dev/" in line and ("active" in line or "spare" in line): # Extract device info parts = line.split() if len(parts) >= 2: device_name = parts[-1] # Get detailed disk info for each device disk_info = self._get_disk_details(device_name, host, cmd_runner) raid_info["disks"].append(disk_info) return True except Exception as e: if DEBUG: print(f"DEBUG: Software RAID check failed: {e}", file=sys.stderr) return False def _check_hardware_raid(self, device: str, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Check for hardware RAID using vendor-specific tools""" try: # Get PCI devices to identify RAID controllers lspci_output = cmd_runner.run("lspci | grep -i raid", host) if not lspci_output: # Also check for storage controllers that might be RAID lspci_output = cmd_runner.run("lspci | grep -i 'storage\\|scsi\\|sas'", host) if lspci_output: controller_info = self._identify_raid_controller(lspci_output) if controller_info["vendor"]: raid_info["is_raid"] = True raid_info["raid_type"] = "hardware" raid_info["controller"] = controller_info["name"] # Try to get detailed info using vendor-specific tools self._get_hardware_raid_details(controller_info["vendor"], host, cmd_runner, raid_info) except Exception as e: if DEBUG: print(f"DEBUG: Hardware RAID check failed: {e}", file=sys.stderr) def _identify_raid_controller(self, lspci_output: str) -> Dict[str, str]: """Identify RAID controller vendor and model from lspci output""" controller_info = {"vendor": None, "name": "Unknown"} lspci_lower = lspci_output.lower() if any(keyword in lspci_lower for keyword in ["lsi", "broadcom", "avago", "megaraid"]): controller_info["vendor"] = "lsi" controller_info["name"] = "LSI/Broadcom MegaRAID" elif any(keyword in lspci_lower for keyword in ["dell", "perc"]): controller_info["vendor"] = "dell" controller_info["name"] = "Dell PERC" elif any(keyword in lspci_lower for keyword in ["hewlett-packard", "hpe", "smart array"]): controller_info["vendor"] = "hp" controller_info["name"] = "HP/HPE Smart Array" elif "adaptec" in lspci_lower: controller_info["vendor"] = "adaptec" controller_info["name"] = "Adaptec RAID" elif "3ware" in lspci_lower: controller_info["vendor"] = "3ware" controller_info["name"] = "3ware RAID" elif "intel" in lspci_lower and "raid" in lspci_lower: controller_info["vendor"] = "intel" controller_info["name"] = "Intel RAID" else: # Generic hardware RAID detected controller_info["vendor"] = "generic" controller_info["name"] = lspci_output.split(":")[-1].strip() if ":" in lspci_output else "Hardware RAID" return controller_info def _get_hardware_raid_details(self, vendor: str, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Get detailed RAID information using vendor-specific tools""" if vendor == "lsi": self._get_lsi_raid_info(host, cmd_runner, raid_info) elif vendor == "dell": self._get_dell_raid_info(host, cmd_runner, raid_info) elif vendor == "hp": self._get_hp_raid_info(host, cmd_runner, raid_info) elif vendor == "adaptec": self._get_adaptec_raid_info(host, cmd_runner, raid_info) elif vendor == "3ware": self._get_3ware_raid_info(host, cmd_runner, raid_info) elif vendor == "intel": self._get_intel_raid_info(host, cmd_runner, raid_info) else: raid_info["error_message"] = f"Unsupported RAID controller vendor: {vendor}" def _get_lsi_raid_info(self, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Get LSI/Broadcom RAID information using storcli or MegaCli""" tools = ["storcli64", "storcli", "MegaCli64", "MegaCli", "megacli"] for tool in tools: try: # Check if tool exists cmd_runner.run(f"which {tool}", host) if tool.startswith("storcli"): self._parse_storcli_output(tool, host, cmd_runner, raid_info) else: self._parse_megacli_output(tool, host, cmd_runner, raid_info) return except Exception: continue raid_info["error_message"] = "LSI RAID tools not found. Install storcli or MegaCli to get detailed RAID information." def _get_dell_raid_info(self, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Get Dell PERC RAID information using perccli or omreport""" tools = ["perccli64", "perccli", "omreport"] for tool in tools: try: cmd_runner.run(f"which {tool}", host) if tool.startswith("perccli"): self._parse_perccli_output(tool, host, cmd_runner, raid_info) else: self._parse_omreport_output(tool, host, cmd_runner, raid_info) return except Exception: continue raid_info["error_message"] = "Dell RAID tools not found. Install perccli or Dell OpenManage to get detailed RAID information." def _get_hp_raid_info(self, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Get HP Smart Array RAID information using ssacli/hpssacli/hpacucli""" tools = ["ssacli", "hpssacli", "hpacucli"] for tool in tools: try: cmd_runner.run(f"which {tool}", host) self._parse_hp_cli_output(tool, host, cmd_runner, raid_info) return except Exception: continue raid_info["error_message"] = "HP RAID tools not found. Install ssacli, hpssacli, or hpacucli to get detailed RAID information." def _get_adaptec_raid_info(self, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Get Adaptec RAID information using arcconf""" try: cmd_runner.run("which arcconf", host) self._parse_arcconf_output(host, cmd_runner, raid_info) except Exception: raid_info["error_message"] = "Adaptec RAID tool not found. Install arcconf to get detailed RAID information." def _get_3ware_raid_info(self, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Get 3ware RAID information using tw_cli""" try: cmd_runner.run("which tw_cli", host) self._parse_tw_cli_output(host, cmd_runner, raid_info) except Exception: raid_info["error_message"] = "3ware RAID tool not found. Install tw_cli to get detailed RAID information." def _get_intel_raid_info(self, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Get Intel RAID information (usually uses LSI tools)""" self._get_lsi_raid_info(host, cmd_runner, raid_info) def _parse_storcli_output(self, tool: str, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Parse storcli output for RAID information""" try: # Get controller info ctrl_output = cmd_runner.run(f"{tool} /c0 show J", host) # Get virtual drive info vd_output = cmd_runner.run(f"{tool} /c0/vall show J", host) # Parse JSON output if available, otherwise fall back to text parsing if ctrl_output.strip().startswith('{'): self._parse_storcli_json(ctrl_output, vd_output, raid_info) else: self._parse_storcli_text(tool, host, cmd_runner, raid_info) except Exception as e: raid_info["error_message"] = f"Failed to parse storcli output: {str(e)}" def _parse_storcli_json(self, ctrl_output: str, vd_output: str, raid_info: Dict[str, Any]): """Parse storcli JSON output""" try: import json ctrl_data = json.loads(ctrl_output) vd_data = json.loads(vd_output) # Extract RAID level and state from virtual drive info if 'Controllers' in vd_data and len(vd_data['Controllers']) > 0: controller = vd_data['Controllers'][0] if 'Response Data' in controller and 'Virtual Drives' in controller['Response Data']: vds = controller['Response Data']['Virtual Drives'] if isinstance(vds, list) and len(vds) > 0: vd = vds[0] raid_info["raid_level"] = vd.get("TYPE", "Unknown") state = vd.get("State", "Unknown") # Convert short state to full name if state == "Optl": raid_info["state"] = "Optimal" else: raid_info["state"] = state # Get physical drive information from controller data self._extract_storcli_disk_info(ctrl_data, raid_info) except Exception as e: if DEBUG: print(f"DEBUG: Failed to parse storcli JSON: {e}", file=sys.stderr) import traceback traceback.print_exc() def _parse_storcli_text(self, tool: str, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Parse storcli text output""" try: # Get basic RAID info output = cmd_runner.run(f"{tool} /c0/vall show", host) for line in output.splitlines(): if "RAID" in line and "/" in line: parts = line.split() if len(parts) >= 3: raid_info["raid_level"] = parts[2] if "RAID" in parts[2] else "Unknown" raid_info["state"] = parts[3] if len(parts) > 3 else "Unknown" break # Get physical disk info pd_output = cmd_runner.run(f"{tool} /c0/eall/sall show", host) self._parse_storcli_physical_disks(pd_output, raid_info) except Exception as e: if DEBUG: print(f"DEBUG: Failed to parse storcli text output: {e}", file=sys.stderr) def _extract_storcli_disk_info(self, ctrl_data: Dict, raid_info: Dict[str, Any]): """Extract disk information from storcli JSON data""" try: raid_info["disks"] = [] if 'Controllers' in ctrl_data and len(ctrl_data['Controllers']) > 0: controller = ctrl_data['Controllers'][0] if 'Response Data' in controller and 'PD LIST' in controller['Response Data']: pd_list = controller['Response Data']['PD LIST'] for pd in pd_list: disk_info = { "name": f"EID:Slot {pd.get('EID:Slt', 'Unknown')}", "model": pd.get('Model', 'Unknown').strip(), "serial_number": "Unknown", # storcli doesn't show serial in basic list "type": self._determine_storcli_disk_type(pd) } raid_info["disks"].append(disk_info) except Exception as e: if DEBUG: print(f"DEBUG: Failed to extract storcli disk info: {e}", file=sys.stderr) import traceback traceback.print_exc() def _parse_storcli_physical_disks(self, output: str, raid_info: Dict[str, Any]): """Parse physical disk information from storcli output""" raid_info["disks"] = [] for line in output.splitlines(): if "/c0/e" in line and "/s" in line: parts = line.split() if len(parts) >= 4: disk_info = { "name": f"/dev/sd{chr(97 + len(raid_info['disks']))}", # Placeholder "model": parts[3] if len(parts) > 3 else "Unknown", "serial_number": "Unknown", "type": self._determine_disk_type(parts[3] if len(parts) > 3 else "") } raid_info["disks"].append(disk_info) def _parse_megacli_output(self, tool: str, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Parse MegaCli output for RAID information""" try: # Get virtual drive info vd_output = cmd_runner.run(f"{tool} -LDInfo -Lall -aALL -NoLog", host) for line in vd_output.splitlines(): if "RAID Level" in line: raid_info["raid_level"] = line.split(":")[-1].strip() elif "State" in line: raid_info["state"] = line.split(":")[-1].strip() # Get physical drive info pd_output = cmd_runner.run(f"{tool} -PDList -aALL -NoLog", host) self._parse_megacli_physical_disks(pd_output, raid_info) except Exception as e: raid_info["error_message"] = f"Failed to parse MegaCli output: {str(e)}" def _parse_megacli_physical_disks(self, output: str, raid_info: Dict[str, Any]): """Parse physical disk information from MegaCli output""" raid_info["disks"] = [] current_disk = {} for line in output.splitlines(): line = line.strip() if "Device Id:" in line: if current_disk: raid_info["disks"].append(current_disk) device_id = line.split(":")[-1].strip() current_disk = { "name": f"Device {device_id}", "model": "Unknown", "serial_number": "Unknown", "type": "unknown" } elif "Inquiry Data:" in line: inquiry = line.split(":")[-1].strip() # Parse inquiry data: format is usually "SERIAL_NUMBER MODEL_NAME FIRMWARE" parts = inquiry.split() if len(parts) >= 2: current_disk["serial_number"] = parts[0] current_disk["model"] = parts[1] elif len(parts) >= 1: current_disk["serial_number"] = parts[0] elif "Media Type:" in line: media_type = line.split(":")[-1].strip() if "Solid State Device" in media_type: # Check if it's NVMe by looking at the model name model = current_disk.get("model", "").lower() if "nvme" in model or "pcie" in media_type.lower(): current_disk["type"] = "nvme" else: current_disk["type"] = "ssd" elif "Hard Disk Device" in media_type: current_disk["type"] = "hdd" else: current_disk["type"] = "hdd" # Default fallback if current_disk: raid_info["disks"].append(current_disk) def _parse_perccli_output(self, tool: str, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Parse perccli output for Dell RAID information""" try: # perccli uses similar syntax to storcli self._parse_storcli_text(tool, host, cmd_runner, raid_info) except Exception as e: raid_info["error_message"] = f"Failed to parse perccli output: {str(e)}" def _parse_omreport_output(self, tool: str, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Parse omreport output for Dell OpenManage RAID information""" try: # Get virtual disk info vd_output = cmd_runner.run(f"{tool} storage vdisk", host) for line in vd_output.splitlines(): if "Layout" in line: raid_info["raid_level"] = line.split(":")[-1].strip() elif "Status" in line: raid_info["state"] = line.split(":")[-1].strip() # Get physical disk info pd_output = cmd_runner.run(f"{tool} storage pdisk controller=0", host) self._parse_omreport_physical_disks(pd_output, raid_info) except Exception as e: raid_info["error_message"] = f"Failed to parse omreport output: {str(e)}" def _parse_omreport_physical_disks(self, output: str, raid_info: Dict[str, Any]): """Parse physical disk information from omreport output""" raid_info["disks"] = [] # Parse omreport physical disk output # This would need specific implementation based on omreport format pass def _parse_hp_cli_output(self, tool: str, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Parse HP CLI output for Smart Array RAID information""" try: # Get array info array_output = cmd_runner.run(f"{tool} ctrl all show config", host) for line in array_output.splitlines(): if "RAID" in line and "Array" in line: if "RAID 0" in line: raid_info["raid_level"] = "RAID 0" elif "RAID 1" in line: raid_info["raid_level"] = "RAID 1" elif "RAID 5" in line: raid_info["raid_level"] = "RAID 5" elif "RAID 6" in line: raid_info["raid_level"] = "RAID 6" elif "Status:" in line: raid_info["state"] = line.split(":")[-1].strip() # Get physical drive info pd_output = cmd_runner.run(f"{tool} ctrl all show config detail", host) self._parse_hp_physical_disks(pd_output, raid_info) except Exception as e: raid_info["error_message"] = f"Failed to parse HP CLI output: {str(e)}" def _parse_hp_physical_disks(self, output: str, raid_info: Dict[str, Any]): """Parse physical disk information from HP CLI output""" raid_info["disks"] = [] current_disk = {} for line in output.splitlines(): line = line.strip() if "physicaldrive" in line: if current_disk: raid_info["disks"].append(current_disk) current_disk = { "name": line.split()[1] if len(line.split()) > 1 else "Unknown", "model": "Unknown", "serial_number": "Unknown", "type": "unknown" } elif "Model:" in line: current_disk["model"] = line.split(":")[-1].strip() elif "Serial Number:" in line: current_disk["serial_number"] = line.split(":")[-1].strip() elif "Interface Type:" in line: interface = line.split(":")[-1].strip() current_disk["type"] = self._determine_disk_type_from_interface(interface) if current_disk: raid_info["disks"].append(current_disk) def _parse_arcconf_output(self, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Parse arcconf output for Adaptec RAID information""" try: # Get logical drive info ld_output = cmd_runner.run("arcconf getconfig 1 LD", host) for line in ld_output.splitlines(): if "RAID level" in line: raid_info["raid_level"] = line.split(":")[-1].strip() elif "Status of logical drive" in line: raid_info["state"] = line.split(":")[-1].strip() # Get physical drive info pd_output = cmd_runner.run("arcconf getconfig 1 PD", host) self._parse_arcconf_physical_disks(pd_output, raid_info) except Exception as e: raid_info["error_message"] = f"Failed to parse arcconf output: {str(e)}" def _parse_arcconf_physical_disks(self, output: str, raid_info: Dict[str, Any]): """Parse physical disk information from arcconf output""" raid_info["disks"] = [] current_disk = {} for line in output.splitlines(): line = line.strip() if "Device #" in line: if current_disk: raid_info["disks"].append(current_disk) current_disk = { "name": f"Device {line.split('#')[-1]}", "model": "Unknown", "serial_number": "Unknown", "type": "unknown" } elif "Model" in line: current_disk["model"] = line.split(":")[-1].strip() elif "Serial number" in line: current_disk["serial_number"] = line.split(":")[-1].strip() elif "Type" in line: disk_type = line.split(":")[-1].strip() if "nvme" in disk_type.lower(): current_disk["type"] = "nvme" elif "SSD" in disk_type: current_disk["type"] = "ssd" else: current_disk["type"] = "hdd" if current_disk: raid_info["disks"].append(current_disk) def _parse_tw_cli_output(self, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Parse tw_cli output for 3ware RAID information""" try: # Get unit info unit_output = cmd_runner.run("tw_cli /c0 show", host) for line in unit_output.splitlines(): if "RAID-" in line: parts = line.split() for part in parts: if "RAID-" in part: raid_info["raid_level"] = part break elif "OK" in line or "DEGRADED" in line: raid_info["state"] = "OK" if "OK" in line else "DEGRADED" # Get drive info self._parse_tw_cli_drives(host, cmd_runner, raid_info) except Exception as e: raid_info["error_message"] = f"Failed to parse tw_cli output: {str(e)}" def _parse_tw_cli_drives(self, host: Optional[str], cmd_runner: CommandRunner, raid_info: Dict[str, Any]): """Parse drive information from tw_cli""" try: drive_output = cmd_runner.run("tw_cli /c0 show drivestatus", host) raid_info["disks"] = [] for line in drive_output.splitlines(): if line.startswith("p"): parts = line.split() if len(parts) >= 2: disk_info = { "name": f"/dev/sd{parts[0][-1]}", "model": "Unknown", "serial_number": "Unknown", "type": "hdd" # Default, would need more detailed parsing } raid_info["disks"].append(disk_info) except Exception as e: if DEBUG: print(f"DEBUG: Failed to parse tw_cli drives: {e}", file=sys.stderr) def _get_disk_details(self, device_path: str, host: Optional[str], cmd_runner: CommandRunner) -> Dict[str, Any]: """Get detailed information about a specific disk""" disk_info = { "name": device_path, "model": "Unknown", "serial_number": "Unknown", "type": "unknown" } try: # Use smartctl to get disk information smart_output = cmd_runner.run(f"smartctl -i {device_path}", host) for line in smart_output.splitlines(): if "Device Model:" in line: disk_info["model"] = line.split(":")[-1].strip() elif "Serial Number:" in line: disk_info["serial_number"] = line.split(":")[-1].strip() elif "Rotation Rate:" in line: if "Solid State Device" in line: disk_info["type"] = "ssd" else: disk_info["type"] = "hdd" # Check if it's NVMe if "nvme" in device_path.lower(): disk_info["type"] = "nvme" except Exception: # Fallback: try to determine type from device name disk_info["type"] = self._determine_disk_type(device_path) return disk_info def _determine_disk_type(self, device_or_model: str) -> str: """Determine disk type from device name or model""" device_lower = device_or_model.lower() if "nvme" in device_lower: return "nvme" elif any(keyword in device_lower for keyword in ["ssd", "solid", "flash"]): return "ssd" elif any(keyword in device_lower for keyword in ["hdd", "disk", "drive"]): return "hdd" else: return "unknown" def _determine_disk_type_from_interface(self, interface: str) -> str: """Determine disk type from interface information""" interface_lower = interface.lower() if "nvme" in interface_lower or "pcie" in interface_lower: return "nvme" elif "ssd" in interface_lower: return "ssd" else: return "hdd" def _determine_storcli_disk_type(self, pd: Dict[str, Any]) -> str: """Determine disk type from storcli physical drive data""" interface = pd.get('Intf', '').lower() media = pd.get('Med', '').lower() # Check interface first - NVMe is most specific if 'nvme' in interface: return "nvme" elif 'ssd' in media: return "ssd" elif 'hdd' in media: return "hdd" else: return "unknown" class NetworkInfoParser: """Parses network information""" def __init__(self): self.cmd_runner = CommandRunner() def parse(self, lshw_output: str, host: str) -> List[Dict[str, Any]]: network_adapters = [] network_devices = ET.fromstring(lshw_output) for node in network_devices.findall("./node[@class='network']"): adapter_info = self._parse_basic_info(node) virtio_node = node.find("./node[@class='network']") if virtio_node is not None: self._parse_virtio_info(virtio_node, adapter_info) else: self._parse_physical_info(node, adapter_info) self._get_mtu(adapter_info, host) network_adapters.append(adapter_info) return network_adapters def _parse_basic_info(self, node: ET.Element) -> Dict[str, Any]: """Parse basic network adapter information""" return { 'vendor': Utils.safe_get_text(node, 'vendor', 'Unknown'), 'product': Utils.safe_get_text(node, 'product', 'Unknown'), 'logical_name': Utils.safe_get_text(node, 'logicalname', 'Unknown') } def _parse_virtio_info(self, virtio_node: ET.Element, adapter_info: Dict[str, Any]): """Parse virtio-specific network information""" virtio_logicalname = virtio_node.find('logicalname') if virtio_logicalname is not None: adapter_info['logical_name'] = virtio_logicalname.text virtio_config = virtio_node.find('configuration') if virtio_config is not None: for setting in virtio_config.findall('setting'): setting_id = setting.get('id') if setting_id in ['speed', 'firmware', 'driver', 'driverversion']: adapter_info[setting_id] = setting.get('value') def _parse_physical_info(self, node: ET.Element, adapter_info: Dict[str, Any]): """Parse physical network adapter information""" capabilities = node.find('capabilities') if capabilities is not None: speeds = [] for cap in capabilities.findall('capability'): if 'bx-fd' in cap.get('id', ''): speed_text = cap.text if speed_text: speeds.append(speed_text) if speeds: adapter_info['max_speed'] = speeds[-1] config = node.find('configuration') if config is not None: for setting in config.findall('setting'): setting_id = setting.get('id') if setting_id in ['speed', 'firmware', 'driver', 'driverversion']: adapter_info[setting_id] = setting.get('value') def _get_mtu(self, adapter_info: Dict[str, Any], host: str): """Get MTU information for network adapter""" try: mtu_output = self.cmd_runner.run( f"cat /sys/class/net/{adapter_info['logical_name']}/mtu", host ) adapter_info["mtu"] = int(mtu_output.strip()) except: adapter_info["mtu"] = "Unknown" pass class FstabParser: """Parses fstab information""" def parse(self, fstab_content: str) -> List[Dict[str, str]]: mounts = [] for line in fstab_content.splitlines(): line = line.strip() if not line or line.startswith('#'): continue fields = line.split() if len(fields) >= 4: mount = { "device": fields[0], "mountpoint": fields[1], "fstype": fields[2], "options": fields[3] } if len(fields) > 4: mount["dump"] = fields[4] if len(fields) > 5: mount["fsck"] = fields[5] mounts.append(mount) return mounts class SystemInfoCollector: """Collects system information from remote hosts""" def __init__(self, host: str, vhi_host_id: str): self.host = host self.vhi_host_id = vhi_host_id self.cmd_runner = CommandRunner() def get_system_info(self) -> Dict[str, Any]: """Get complete system information""" try: if DEBUG: start_time = time.time() print(f"DEBUG: Starting system info collection for {self.host}", file=sys.stderr) print(" Getting kernel information...", end=" ", flush=True) step_start = time.time() if DEBUG else None kernel = self._get_kernel_info() if DEBUG: print(f"DEBUG: Kernel info took {time.time() - step_start:.2f}s", file=sys.stderr) print("OK") print(" Getting CPU information...", end=" ", flush=True) step_start = time.time() if DEBUG else None processors = self._get_cpu_info() if DEBUG: print(f"DEBUG: CPU info took {time.time() - step_start:.2f}s", file=sys.stderr) print("OK") print(" Getting RAM information...", end=" ", flush=True) step_start = time.time() if DEBUG else None ram = self._get_ram_info() if DEBUG: print(f"DEBUG: RAM info took {time.time() - step_start:.2f}s", file=sys.stderr) print("OK") print(" Getting root disk information...", end=" ", flush=True) step_start = time.time() if DEBUG else None root_disk = self._get_root_disk_info() if DEBUG: print(f"DEBUG: Root disk info took {time.time() - step_start:.2f}s", file=sys.stderr) print("OK") print(" Getting network adapter information...", end=" ", flush=True) step_start = time.time() if DEBUG else None network_adapters = self._get_network_info() if DEBUG: print(f"DEBUG: Network info took {time.time() - step_start:.2f}s", file=sys.stderr) print("OK") print(" Getting mount information...", end=" ", flush=True) step_start = time.time() if DEBUG else None mounts = self._get_mount_info() if DEBUG: print(f"DEBUG: Mount info took {time.time() - step_start:.2f}s", file=sys.stderr) print("OK") print(" Getting vstorage disk information...", end=" ", flush=True) step_start = time.time() if DEBUG else None vstorage_disks = self._get_vstorage_disks_info() if DEBUG: print(f"DEBUG: VStorage disk info took {time.time() - step_start:.2f}s", file=sys.stderr) print("OK") print(" Getting VHI version...", end=" ", flush=True) step_start = time.time() if DEBUG else None vhi_version = self._get_os_version() if DEBUG: print(f"DEBUG: VHI version took {time.time() - step_start:.2f}s", file=sys.stderr) print("OK") if DEBUG: total_time = time.time() - start_time print(f"DEBUG: Total system info collection took {total_time:.2f}s", file=sys.stderr) return { "kernel": kernel, "processors": processors, "ram": ram, "root_disk": root_disk, "network_adapters": network_adapters, "mounts": mounts, "vstorage_disks": vstorage_disks, "vhi_version": vhi_version } except Exception as e: print("FAILED") if DEBUG: print(f"DEBUG: System info collection failed: {e}", file=sys.stderr) traceback.print_exc() return { "error": str(e), "processors": [], "ram": {}, "root_disk": {}, "network_adapters": [], "mounts": [], "vstorage_disks": {} } def _get_kernel_info(self) -> str: """Get kernel information""" try: kernel_info = self.cmd_runner.run("uname -r", self.host) return kernel_info.strip() except Exception as e: print(f"Error getting kernel information from {self.host}: {e}", file=sys.stderr) return "unknown" def _get_vstorage_disks_info(self) -> Dict[str, Dict[str, List[Dict[str, Any]]]]: """Get vstorage disk information""" try: disk_info = json.loads(self.cmd_runner.run("vinfra node disk list -a --long -f json")) result: Dict[str, Any] = {} for disk in disk_info: if disk.get("node_id") != self.vhi_host_id: continue role = disk.get("role", "unknown") if role == "unknown": continue if role not in result: result[role] = {"disks": []} result[role]["disks"].append(self._create_disk_entry(disk)) return result except Exception as e: print(f"Error getting vstorage disk information from {self.host}: {e}", file=sys.stderr) return {} def _create_disk_entry(self, disk: Dict[str, Any]) -> Dict[str, Any]: """Create a standardized disk entry""" disk_status = "unknown" if "chunk_servers" in disk and disk["chunk_servers"]: cs = disk["chunk_servers"][0] disk_status = cs.get("status", "unknown") for chunk_server in disk["chunk_servers"]: journal_type = chunk_server.get("journal_type", "") chunk_server["checksumming"] = "enabled" if journal_type != "no_cache" else "disabled" elif "metadata_server" in disk and disk["metadata_server"]: disk_status = disk["metadata_server"].get("status", "unknown") entry = { "id": disk.get("id", "unknown"), "device": disk.get("device", "unknown"), "type": disk.get("type", "unknown"), "disk_status": disk_status, "used": Utils.sizeof_fmt(disk.get("used", 0)), "size": Utils.sizeof_fmt(disk.get("size", 0)), "physical_size": Utils.sizeof_fmt(disk.get("physical_size", disk.get("size", 0))), "cache": disk.get("cache", "unknown"), "encryption": disk.get("encryption", "disabled"), "nvme": "nvme" in (disk.get("device", "").lower() or ""), "vendor": disk.get("vendor", "unknown"), "serial_number": disk.get("serial_number", "unknown"), } if "chunk_servers" in disk and disk["chunk_servers"]: entry["chunk_servers"] = disk["chunk_servers"] if "metadata_server" in disk and disk["metadata_server"]: entry["metadata_server"] = disk["metadata_server"] return entry def _get_cpu_info(self) -> List[Dict[str, Any]]: """Get CPU information""" cpu_info = self.cmd_runner.run("cat /proc/cpuinfo", self.host) return CPUInfoParser().parse(cpu_info) def _get_ram_info(self) -> Dict[str, Any]: """Get RAM information""" mem_info = self.cmd_runner.run("free -b", self.host) return RAMInfoParser().parse(mem_info) def _get_root_disk_info(self) -> Dict[str, Any]: """Get root disk usage information""" df_output = self.cmd_runner.run("df -B1 /", self.host) return DiskInfoParser().parse(df_output, self.host, self.cmd_runner) def _get_network_info(self) -> List[Dict[str, Any]]: """Get network adapter information""" try: lshw_output = self.cmd_runner.run("lshw -class network -xml", self.host) return NetworkInfoParser().parse(lshw_output, self.host) except Exception as e: print(f"Error getting network information from {self.host}: {e}", file=sys.stderr) return [] def _get_mount_info(self) -> List[Dict[str, str]]: """Get mount information""" fstab_info = self.cmd_runner.run("cat /etc/fstab", self.host) return FstabParser().parse(fstab_info) # method runs 'cat /etc/hci-release' and returns the OS version def _get_os_version(self) -> str: """Get OS version""" try: os_version = self.cmd_runner.run("cat /etc/hci-release", self.host) return os_version.strip() except Exception as e: print(f"Error getting OS version from {self.host}: {e}", file=sys.stderr) return "unknown" class ClusterInfoParser: """Parses cluster information""" def __init__(self, cluster_overview: Dict[str, Any], node_list: List[Dict[str, Any]], node_disk_list: List[Dict[str, Any]], reachable_hosts: List[str], ip_to_node_mapping: Dict[str, Dict[str, Any]]): self.cmd_runner = CommandRunner() self.cluster_overview = cluster_overview or {} self.node_list = node_list or [] self.node_disk_list = node_disk_list or [] self.reachable_hosts = reachable_hosts self.ip_to_node_mapping = ip_to_node_mapping or {} self.node_info = self._get_node_info() def _get_hwid(self, cluster_name: str) -> str: """Get hardware ID from mds_cache file Args: cluster_name: Name of the cluster Returns: Hardware ID string or empty string if not found """ try: mds_cache_path = f"/root/.vstorage/{cluster_name}/mds_cache" content = self.cmd_runner.run(f"cat {mds_cache_path}") # Process each line for line in content.splitlines(): line = line.strip() # Skip empty lines and lines containing IP addresses if not line or ':' in line: continue # Return first non-empty line that doesn't contain an IP return line return "" except Exception as e: if DEBUG: print(f"DEBUG: Error reading HWID from mds_cache: {e}", file=sys.stderr) return "" def parse(self, cluster_name: str) -> Dict[str, Any]: """Parse cluster information""" if DEBUG: start_time = time.time() print(f"DEBUG: Starting cluster info parsing for {cluster_name}", file=sys.stderr) current_time = datetime.now() info = { "script_version": SCRIPT_VERSION, "collected_at": { "datetime": current_time.strftime("%Y-%m-%d %H:%M:%S"), "timestamp": int(current_time.timestamp()) }, "name": cluster_name, "hwid": self._get_hwid(cluster_name), "status": self.cluster_overview.get("status", "unknown"), "rdma": self._get_rdma_status(), "fanout": self._get_client_fanout(), "license": self._get_license_info(), "roles": {}, "space": {}, "tiers": self._parse_tiers(), "vstorage_file_stats": self._get_file_stats(cluster_name), "storage_runtime_config": self._get_storage_runtime_config(cluster_name), "storage_stat": "", "hosts": {} } try: if DEBUG: print("DEBUG: Getting vstorage stat XML...", file=sys.stderr) step_start = time.time() xml_output = self.cmd_runner.run(f"vstorage -c {cluster_name} stat -A -X") root = ET.fromstring(xml_output) info["storage_stat"] = self._xml_to_dict(root) if DEBUG: print(f"DEBUG: XML parsing took {time.time() - step_start:.2f}s", file=sys.stderr) self._parse_space(root, info) self._parse_hosts(root, info, cluster_name) self._parse_roles(root, info) except Exception as e: print(f"Error getting cluster information: {e}", file=sys.stderr) if DEBUG: print(f"DEBUG: Full error details:", file=sys.stderr) traceback.print_exc() if DEBUG: total_time = time.time() - start_time print(f"DEBUG: Total cluster info parsing took {total_time:.2f}s", file=sys.stderr) return info def _xml_to_dict(self, element: ET.Element) -> Dict[str, Any]: """Convert XML element to dictionary format""" result: Dict[str, Any] = {} # Add attributes if present if element.attrib: result.update(element.attrib) # Add children elements recursively for child in element: child_data = self._xml_to_dict(child) if child.tag in result: # If the tag already exists, convert it to a list if not isinstance(result[child.tag], list): result[child.tag] = [result[child.tag]] result[child.tag].append(child_data) else: result[child.tag] = child_data # Add text content if present and no children if element.text and element.text.strip() and not result: text = element.text.strip() # Store the value in a dictionary with a special key if text.isdigit(): result["value"] = int(text) else: try: result["value"] = float(text) except ValueError: result["value"] = text return result def _get_file_stats(self, cluster_name: str) -> Dict[str, Any]: """Get vstorage file statistics""" try: xml_output = self.cmd_runner.run(f"vstorage -c {cluster_name} file-stats -p / -X") # Remove the "connected to MDS#X" line if present xml_lines = xml_output.splitlines() for i, line in enumerate(xml_lines): if line.startswith(" 0: nodes_with_mds += 1 if num_all_cses > 0: nodes_with_cs += 1 # Calculate failed counts failed_mds = total_mds - active_mds failed_cs = total_cs - active_cs # Set role statistics info["roles"] = { "nodes": total_nodes, "mds": { "total": total_mds, "active": active_mds, "failed": failed_mds }, "cs": { "total": total_cs, "active": active_cs, "failed": failed_cs }, "nodes_with_mds": nodes_with_mds, "nodes_with_cs": nodes_with_cs } def _get_client_fanout(self) -> bool: """Get client fanout mode""" try: fanout_output = self.cmd_runner.run("vinfra cluster settings client-fanout-mode show -f json") return json.loads(fanout_output).get("client_fanout", False) except Exception as e: print(f"Warning: Failed to get client fanout mode: {e}", file=sys.stderr) return False def _parse_host_roles(self, host_stat: ET.Element, host_info: Dict[str, Any]): """Parse roles for a specific host""" if not isinstance(host_info, dict): host_info = {} if "roles" not in host_info: host_info["roles"] = {} host_roles = host_stat.find("host_roles") if host_roles is not None: host_info["roles"].update({ "cs_active": Utils.safe_get_int(host_roles.find("num_act_cses")), "cs_total": Utils.safe_get_int(host_roles.find("num_all_cses")), "mds_active": Utils.safe_get_int(host_roles.find("num_act_mdses")), "mds_total": Utils.safe_get_int(host_roles.find("num_all_mdses")) }) def _parse_tiers(self) -> Dict[str, Dict[str, Any]]: """Parse tier information""" tier_info = {} external_journals = {} for tier in self.cluster_overview.get("tiers", []): if not isinstance(tier, dict): continue tier_id = str(tier.get("id", "unknown")) phys_space = tier.get("phys_space", {}) if not isinstance(phys_space, dict): continue tier_info[tier_id] = { "active_cs": self.cluster_overview.get("active_cses", {}).get(tier_id, 0), "phys_space": { "used": Utils.sizeof_fmt(phys_space.get("used", 0)), "free": Utils.sizeof_fmt(phys_space.get("free", 0)), "total": Utils.sizeof_fmt(phys_space.get("total", 0)) } } # Count external journal disks per tier for disk in self.node_disk_list: if disk.get("role") == "cs" and "chunk_servers" in disk: for cs in disk.get("chunk_servers", []): tier = str(cs.get("tier", "unknown")) if tier not in external_journals: external_journals[tier] = 0 if cs.get("journal_disk_id", None) is not None and cs.get("journal_type") == "external_cache": external_journals[tier] += 1 # Add external journal count to tier info for tier_id in tier_info: tier_info[tier_id]["external_journal_disks"] = external_journals.get(tier_id, 0) return tier_info def _get_node_info(self) -> Dict[str, Dict[str, Any]]: """Get node information mapping from IP to hostname/vhi_id Uses the interface-based mapping passed from VStorageInfo to map IPs to nodes. This ensures we use the same hostnames that were verified during SSH connectivity checks. """ # Convert ip_to_node_mapping format to node_info format # ip_to_node_mapping: IP -> {"node_id": str, "hostname": str, "vhi_id": str} # node_info: IP -> {"hostname": str, "vhi_id": str} node_info = {} for ip, node_data in self.ip_to_node_mapping.items(): node_info[ip] = { "hostname": node_data.get("hostname", "unknown"), "vhi_id": node_data.get("vhi_id", "unknown") } if DEBUG: print(f"DEBUG: Built node_info mapping for {len(node_info)} IPs", file=sys.stderr) return node_info def _get_rdma_status(self) -> bool: """Get RDMA status""" try: config = json.loads(self.cmd_runner.run("vinfra cses-config show -f json")) or {} return config.get("rdma", False) except Exception as e: print(f"Warning: Failed to get RDMA status: {e}", file=sys.stderr) return False def _get_license_info(self) -> Dict[str, Any]: """Get license information""" try: license_info = json.loads(self.cmd_runner.run("vinfra cluster license show -f json")) or {} return { "capacity": Utils.sizeof_fmt(license_info.get("capacity", 0)), "core_number": license_info.get("core_number", 0), "core_number_used": license_info.get("core_number_used", 0), "expiration_ts": license_info.get("expiration_ts", -1), "free_size": Utils.sizeof_fmt(license_info.get("free_size", 0)), "keynumber": license_info.get("keynumber"), "license_type": license_info.get("license_type"), "status": license_info.get("status", "unknown"), "total_size": Utils.sizeof_fmt(license_info.get("total_size", 0)), "used_size": Utils.sizeof_fmt(license_info.get("used_size", 0)) } except Exception as e: print(f"Warning: Failed to get license information: {e}", file=sys.stderr) return { "capacity": "0B", "core_number": 0, "core_number_used": 0, "expiration_ts": -1, "free_size": "0B", "keynumber": None, "license_type": None, "status": "unknown", "total_size": "0B", "used_size": "0B" } def _get_storage_runtime_config(self, cluster_name: str) -> Dict[str, Any]: """Get storage runtime configuration""" try: config_output = self.cmd_runner.run(f"vstorage -c {cluster_name} get-config") config = {} for line in config_output.splitlines(): if not line or line.startswith('connected to MDS'): continue key, value = line.split('=') config[key] = value return config except Exception as e: print(f"Error getting storage runtime configuration: {e}", file=sys.stderr) return {} def _read_config_file(self, filename: str, host: Optional[str] = None) -> Dict[str, Any]: """Read and parse a config file, skipping comments Args: filename: Full path to the config file host: Optional remote host. If specified, reads file via SSH """ config = {} try: config_content = self.cmd_runner.run(f"cat {filename}", host) for line in config_content.splitlines(): line = line.strip() if line and not line.startswith('#'): if '=' in line: key, value = line.split('=', 1) config[key.strip()] = value.strip().strip('"') except Exception as e: if DEBUG: print(f"DEBUG: Error reading {filename}: {e}", file=sys.stderr) return config def _read_list_file(self, filename: str, host: Optional[str] = None) -> List[str]: """Read and parse a list file, skipping comments Args: filename: Full path to the list file host: Optional remote host. If specified, reads file via SSH """ entries = [] try: content = self.cmd_runner.run(f"cat {filename}", host) for line in content.splitlines(): line = line.strip() if line and not line.startswith('#'): entries.append(line) except Exception as e: print(f"Warning: Failed to read {filename}: {e}", file=sys.stderr) return entries def _get_vstorage_config(self, host: Optional[str], cluster_name: str) -> Dict[str, Any]: """Get complete vstorage configuration including all config files and lists""" config = { "cs.config": self._read_config_file("/etc/vstorage/cs.config", host), "mds.config": self._read_config_file("/etc/vstorage/mds.config", host), "vstorage-mount.conf": self._read_config_file("/etc/vstorage/vstorage-mount.conf", host), "vstorage-mount.conf.d": {}, "clusters": { cluster_name: { "cs.config": {}, "mds.config": {}, "cs.list": [], "mds.list": [] } } } # Read vstorage-mount.conf.d files try: files = self.cmd_runner.run("ls /etc/vstorage/vstorage-mount.conf.d/", host).splitlines() for file in files: config["vstorage-mount.conf.d"][file] = self._read_config_file( f"/etc/vstorage/vstorage-mount.conf.d/{file}", host) except Exception as e: if DEBUG: print(f"DEBUG: Error reading vstorage-mount.conf.d: {e}", file=sys.stderr) # Read cluster-specific configs and lists cluster_path = f"/etc/vstorage/clusters/{cluster_name}" # Try to read cluster-specific cs.config try: cs_config = self._read_config_file(f"{cluster_path}/cs.config", host) if cs_config: # Only set if file exists and has content config["clusters"][cluster_name]["cs.config"] = cs_config except Exception as e: if DEBUG: print(f"DEBUG: Error reading cluster cs.config: {e}", file=sys.stderr) # Try to read cluster-specific mds.config try: mds_config = self._read_config_file(f"{cluster_path}/mds.config", host) if mds_config: # Only set if file exists and has content config["clusters"][cluster_name]["mds.config"] = mds_config except Exception as e: if DEBUG: print(f"DEBUG: Error reading cluster mds.config: {e}", file=sys.stderr) # Read cs.list and mds.list try: cs_list = self.cmd_runner.run(f"cat {cluster_path}/cs.list", host).splitlines() config["clusters"][cluster_name]["cs.list"] = [ line.strip() for line in cs_list if line.strip() and not line.strip().startswith('#') ] except Exception as e: if DEBUG: print(f"DEBUG: Error reading cs.list: {e}", file=sys.stderr) try: mds_list = self.cmd_runner.run(f"cat {cluster_path}/mds.list", host).splitlines() config["clusters"][cluster_name]["mds.list"] = [ line.strip() for line in mds_list if line.strip() and not line.strip().startswith('#') ] except Exception as e: if DEBUG: print(f"DEBUG: Error reading mds.list: {e}", file=sys.stderr) return config class VStorageInfo: """Main class for gathering VStorage information""" def __init__(self, skip_unreachable: bool = False): self.skip_unreachable = skip_unreachable print("Initializing vstorage-info script...") try: print("Checking vinfra authentication...", end=" ", flush=True) self._setup_vinfra_auth() print("OK") print("Checking vinfra cluster status...", end=" ", flush=True) self.cmd_runner = CommandRunner() self._cluster_overview = json.loads(self.cmd_runner.run("vinfra cluster overview -f json")) print("OK") print("Getting node list...", end=" ", flush=True) self._node_list = json.loads(self.cmd_runner.run("vinfra node list -f json")) print("OK") print("Getting node interfaces...", end=" ", flush=True) self._node_interfaces = json.loads(self.cmd_runner.run("vinfra node iface list -a -f json")) print("OK") print("Getting node disk list...", end=" ", flush=True) self._node_disk_list = json.loads(self.cmd_runner.run("vinfra node disk list -a --long -f json")) print("OK") # Build IP to node mapping from interfaces self._ip_to_node_mapping = self._build_ip_to_node_mapping() # Check SSH connectivity to all nodes and track reachable ones self.reachable_hosts = self._check_ssh_connectivity() except Exception as e: print("FAILED") print(f"Error during initialization: {e}", file=sys.stderr) sys.exit(1) def _setup_vinfra_auth(self): """Setup VINFRA authentication""" if "VINFRA_PASSWORD" not in os.environ: os.environ["VINFRA_PASSWORD"] = getpass.getpass("Enter VINFRA Admin password: ") def _check_ssh_connectivity(self): """Check SSH connectivity to all nodes and return list of reachable hosts""" reachable_hosts = [] unreachable_hosts = [] try: node_list = json.loads(self.cmd_runner.run("vinfra node list -f json")) or [] for node in node_list: if "host" not in node: continue hostname = node["host"] print(f"Checking SSH connection to {hostname}...", end="", flush=True) try: # Try a simple command to test SSH connectivity self.cmd_runner.run("hostname", hostname) print("OK") reachable_hosts.append(hostname) except Exception: print("FAILED") unreachable_hosts.append(hostname) # Print summary print("\nSSH Connectivity Summary:") print(f" Reachable hosts: {len(reachable_hosts)}") print(f" Unreachable hosts: {len(unreachable_hosts)}") if unreachable_hosts: print(f" Unreachable: {', '.join(unreachable_hosts)}") if self.skip_unreachable: print(" These hosts will be skipped during data collection.") else: raise RuntimeError( f"SSH connectivity failed for {len(unreachable_hosts)} host(s): {', '.join(unreachable_hosts)}. " "Use --skip-unreachable flag to continue with reachable hosts only." ) if not reachable_hosts: raise RuntimeError("No hosts are reachable via SSH. Cannot proceed.") return reachable_hosts except Exception as e: if "No hosts are reachable" in str(e): raise raise RuntimeError(f"Failed to check node connectivity: {e}") from e def _build_ip_to_node_mapping(self) -> Dict[str, Dict[str, Any]]: """Build mapping from IP addresses to node information Returns: Dictionary mapping IP address -> {"node_id": str, "hostname": str, "vhi_id": str} """ ip_mapping = {} # Create node_id -> hostname/vhi_id mapping node_id_to_info = { node["id"]: { "hostname": node["host"], "vhi_id": node["id"] } for node in self._node_list if "id" in node and "host" in node } # Process interfaces to extract IPs for iface in self._node_interfaces: node_id = iface.get("node_id") if not node_id or node_id not in node_id_to_info: continue node_info = node_id_to_info[node_id] ipv4_list = iface.get("ipv4", []) for ip_with_cidr in ipv4_list: # Extract IP from CIDR notation (e.g., "10.100.1.30/16" -> "10.100.1.30") ip = ip_with_cidr.split('/')[0].strip() if ip and Utils._is_ip_address(ip): ip_mapping[ip] = { "node_id": node_id, "hostname": node_info["hostname"], "vhi_id": node_info["vhi_id"] } if DEBUG: print(f"DEBUG: Built IP mapping for {len(ip_mapping)} IP addresses", file=sys.stderr) return ip_mapping def get_cluster_info(self) -> Dict[str, Dict[str, Any]]: """Get cluster information""" cluster_name = self._get_cluster_name() result = {} if cluster_name: print(f"\nGathering cluster information for '{cluster_name}'...") print("Getting cluster overview...", end=" ", flush=True) try: cluster_info = ClusterInfoParser( self._cluster_overview, self._node_list, self._node_disk_list, self.reachable_hosts, self._ip_to_node_mapping ).parse(cluster_name) print("OK") result[cluster_name] = cluster_info except Exception as e: print("FAILED") print(f"Error gathering cluster information: {e}", file=sys.stderr) sys.exit(1) else: print("Error: Could not determine cluster name", file=sys.stderr) sys.exit(1) return result def _get_cluster_name(self) -> str: """Get cluster name""" try: return self._cluster_overview.get("name", "") except Exception as e: print(f"Error reading cluster name: {e}", file=sys.stderr) return "" def main(): """Main function""" global DEBUG # Parse command line arguments parser = argparse.ArgumentParser(description='Gather VStorage cluster information') parser.add_argument('--debug', action='store_true', help='Enable debug mode with verbose output') parser.add_argument('--skip-unreachable', action='store_true', help='Skip unreachable nodes and continue with reachable ones only. ' 'By default, script fails if any node is unreachable via SSH.') args = parser.parse_args() # Set global debug flag DEBUG = args.debug if DEBUG: print("DEBUG: Debug mode enabled", file=sys.stderr) try: vstorage = VStorageInfo(args.skip_unreachable) result = vstorage.get_cluster_info() # Get the first (and only) cluster's info cluster_name = next(iter(result)) cluster_info = result[cluster_name] hwid = cluster_info.get("hwid", "unknown") # Write results to file script_dir = os.path.dirname(os.path.abspath(__file__)) output_file = os.path.join(script_dir, f"{hwid}_cluster-info.json") with open(output_file, "w", encoding="utf-8") as f: json.dump(result, f, indent=2) print(f"\nCluster information has been successfully gathered and written to: {output_file}") except Exception as e: if DEBUG: print(f"DEBUG: Full error details: {e}", file=sys.stderr) traceback.print_exc() else: print(f"Error: {e}", file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()