#!/usr/bin/env python3 import os, sys, json, requests, psutil, socket, subprocess, logging, argparse, urllib.request, re from typing import Dict, Any, List # Create a logger logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) # Create a console handler and set its level to DEBUG console_handler = logging.StreamHandler() console_handler.setLevel(logging.DEBUG) # Create a formatter formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') # Add the formatter to the console handler console_handler.setFormatter(formatter) # Add the console handler to the logger logger.addHandler(console_handler) class SystemHealthMonitor: def __init__(self, ticket_api_url: str = 'http://10.10.10.45/create_ticket_api.php', state_file: str = '/tmp/last_health_check.json', dry_run: bool = False): """ Initialize the system health monitor. :param ticket_api_url: URL for the ticket creation API. :param state_file: File path to track the last health check results. :param dry_run: If True, simulate API calls without sending requests. """ self.ticket_api_url = ticket_api_url self.state_file = state_file self.dry_run = dry_run def run(self): """ Perform a one-shot health check of the system. """ try: # Perform health checks and gather the report health_report = self.perform_health_checks() # Create tickets for any detected critical issues self._create_tickets_for_issues(health_report) except Exception as e: print(f"Unexpected error during health check: {e}") def perform_health_checks(self) -> Dict[str, Any]: """ Perform comprehensive system health checks and return a report. :return: Dictionary containing results of various health checks. """ health_report = { 'drives_health': self._check_drives_health(), 'memory_health': self._check_memory_usage(), 'cpu_health': self._check_cpu_usage(), 'network_health': self._check_network_status() } return health_report def _generate_detailed_description(self, issue: str, health_report: Dict[str, Any]) -> str: """ Generate a detailed description for the issue based on the health report. :param issue: The issue description. :param health_report: The comprehensive health report from the checks. :return: A detailed description for the issue. """ description = issue + "\n\n" if "Disk" in issue: for partition in health_report.get('drives_health', {}).get('drives', []): if partition.get('mountpoint') in issue: description += f"Disk Device: {partition['device']}\n" description += f"Mount Point: {partition['mountpoint']}\n" description += f"Total Space: {partition['total_space']}\n" description += f"Used Space: {partition['used_space']}\n" description += f"Free Space: {partition['free_space']}\n" description += f"Usage Percent: {partition['usage_percent']}%\n" if partition.get('smart_status') == 'UNHEALTHY': try: # Get additional disk information using smartctl result = subprocess.run( ['smartctl', '-a', partition['device']], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) output = result.stdout + result.stderr description += "\nSMART Information:\n" description += output except Exception as e: description += f"Error getting SMART information: {str(e)}\n" break elif "Memory" in issue: memory_health = health_report.get('memory_health', {}) description += f"Total Memory: {memory_health['total_memory']}\n" description += f"Used Memory: {memory_health['used_memory']}\n" description += f"Memory Usage Percent: {memory_health['memory_percent']}%\n" elif "CPU" in issue: cpu_health = health_report.get('cpu_health', {}) description += f"CPU Usage Percent: {cpu_health['cpu_usage_percent']}%\n" elif "Network" in issue: network_health = health_report.get('network_health', {}) for network in ['management_network', 'ceph_network']: if network_health[network]['issues']: description += f"{network.replace('_', ' ').title()} Issues:\n" description += "\n".join(network_health[network]['issues']) description += "\n" return description def _create_tickets_for_issues(self, health_report: Dict[str, Any]): """ Create tickets for detected issues with standardized parameters. """ issues = self._detect_issues(health_report) if not issues: logger.info("No issues detected.") return hostname = socket.gethostname() action_type = "[auto]" environment = "[production]" ticket_type = "[maintenance]" for issue in issues: # Set default values priority = "P4" category = "Hardware" issue_type = "Problem" scope = "[single-node]" if "Disk" in issue: hardware_type = "[hardware]" if "CRITICAL" in issue or "SMART failure" in issue: priority = "P2" elif "WARNING" in issue: priority = "P3" elif "Memory" in issue: hardware_type = "[hardware]" priority = "P3" elif "CPU" in issue: hardware_type = "[hardware]" priority = "P3" elif "Network" in issue: hardware_type = "[network]" priority = "P2" scope = "[cluster-wide]" # Create standardized ticket title ticket_title = f"[{hostname}]{action_type}{hardware_type} {issue} {scope}{environment}{ticket_type}" description = self._generate_detailed_description(issue, health_report) ticket_payload = { "title": ticket_title, "description": description, "priority": priority, "status": "Open", "category": category, "type": issue_type } if self.dry_run: logger.info("Dry-run mode enabled. Simulating ticket creation:") logger.info(json.dumps(ticket_payload, indent=4)) print("Dry-run: Ticket payload:") print(json.dumps(ticket_payload, indent=4)) else: try: response = requests.post( self.ticket_api_url, json=ticket_payload, headers={'Content-Type': 'application/json'} ) logger.info(f"Response status code: {response.status_code}") logger.info(f"Response body: {response.text}") if response.status_code in [200, 201]: logger.info(f"Ticket created successfully: {ticket_title}") else: logger.error(f"Failed to create ticket. Status code: {response.status_code}") logger.error(f"Response: {response.text}") except Exception as e: logger.error(f"Error creating ticket: {e}") def _detect_issues(self, health_report: Dict[str, Any]) -> List[str]: """ Detect issues in the health report including non-critical issues. :param health_report: The comprehensive health report from the checks. :return: List of issue descriptions detected during checks. """ issues = [] # Check for drive-related issues for partition in health_report.get('drives_health', {}).get('drives', []): if partition.get('usage_status') == 'CRITICAL_HIGH_USAGE': issues.append( f"Disk {partition['mountpoint']} is {partition['usage_percent']}% full" ) elif partition.get('usage_status') == 'WARNING_HIGH_USAGE': issues.append( f"Disk {partition['mountpoint']} is {partition['usage_percent']}% full (Warning)" ) if partition.get('smart_status') == 'UNHEALTHY': issues.append(f"Disk {partition['mountpoint']} has an unhealthy SMART status") # Check for memory-related issues memory_health = health_report.get('memory_health', {}) if memory_health and memory_health.get('memory_percent', 0) > 80: issues.append("Memory usage is above 80%") # Check for CPU-related issues cpu_health = health_report.get('cpu_health', {}) if cpu_health and cpu_health.get('cpu_usage_percent', 0) > 80: issues.append("CPU usage is above 80%") # Check for network-related issues network_health = health_report.get('network_health', {}) for network in ['management_network', 'ceph_network']: if network_health.get(network, {}).get('issues'): issues.extend(network_health[network]['issues']) return issues def _is_physical_disk(self, device_path): """ Check if the device is a physical SATA, NVMe, or MMC disk. :param device_path: Path to the device :return: Boolean indicating if it's a physical disk """ return bool(re.match(r'/dev/(sd[a-z]|nvme\d+n\d+|mmcblk\d+)', device_path)) def _check_smart_health(self, device: str) -> Dict[str, Any]: """ Check comprehensive SMART health metrics for a drive. """ smart_health = { 'status': 'HEALTHY', 'issues': [], 'temp': None, 'attributes': {} } try: result = subprocess.run( ['smartctl', '-A', '-H', device], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) output = result.stdout # Check overall SMART status first if 'FAILED' in output and 'PASSED' not in output: smart_health['status'] = 'UNHEALTHY' smart_health['issues'].append("SMART overall health check failed") # Parse SMART attributes for line in output.split('\n'): if 'ATTRIBUTE_NAME' in line: continue # Check for current failures only if 'WHEN_FAILED' in line and 'In_the_past' not in line and '-' not in line: smart_health['status'] = 'UNHEALTHY' smart_health['issues'].append(f"Current failure detected: {line}") # Monitor critical attributes for attr in ['Reallocated_Sector_Ct', 'Current_Pending_Sector', 'Offline_Uncorrectable', 'Reported_Uncorrect']: if attr in line: parts = line.split() if len(parts) >= 10: raw_value = int(parts[9]) if raw_value > 0: smart_health['status'] = 'UNHEALTHY' smart_health['issues'].append(f"{attr} has value {raw_value}") # Check temperature if 'Temperature_Celsius' in line or 'Airflow_Temperature_Cel' in line: parts = line.split() if len(parts) >= 10: temp = int(parts[9]) smart_health['temp'] = temp if temp > 65: smart_health['issues'].append(f"High drive temperature: {temp}°C") except Exception as e: smart_health['status'] = 'ERROR' smart_health['issues'].append(f"Error checking SMART: {str(e)}") return smart_health def _check_drives_health(self) -> Dict[str, Any]: """ Check overall health of physical SATA and NVMe drives including disk usage and SMART status. :return: Combined health report of all drives and their status. """ drives_health = {'overall_status': 'NORMAL', 'drives': []} try: partitions = [p for p in psutil.disk_partitions() if self._is_physical_disk(p.device)] overall_status = 'NORMAL' for partition in partitions: drive_report = { 'device': partition.device, 'mountpoint': partition.mountpoint } # Check disk usage usage = psutil.disk_usage(partition.mountpoint) disk_usage_status = 'NORMAL' if usage.percent > 90: disk_usage_status = 'CRITICAL_HIGH_USAGE' elif usage.percent > 80: disk_usage_status = 'WARNING_HIGH_USAGE' drive_report.update({ 'total_space': self._convert_bytes(usage.total), 'used_space': self._convert_bytes(usage.used), 'free_space': self._convert_bytes(usage.free), 'usage_percent': usage.percent, 'usage_status': disk_usage_status }) # Check SMART health smart_health = self._check_smart_health(partition.device) drive_report.update({ 'smart_status': smart_health['status'], 'smart_issues': smart_health['issues'], 'temperature': smart_health['temp'], 'smart_attributes': smart_health['attributes'] }) # Update overall status if smart_health['status'] == 'UNHEALTHY' or disk_usage_status == 'CRITICAL_HIGH_USAGE': overall_status = 'CRITICAL' elif disk_usage_status == 'WARNING_HIGH_USAGE' and overall_status != 'CRITICAL': overall_status = 'WARNING' drives_health['drives'].append(drive_report) drives_health['overall_status'] = overall_status except Exception as e: logger.error(f"Error checking drives health: {str(e)}") return drives_health @staticmethod def _convert_bytes(bytes_value: int, suffix: str = 'B') -> str: """ Convert bytes to a human-readable format. :param bytes_value: Number of bytes to convert. :param suffix: Suffix to append (default is 'B' for bytes). :return: Formatted string with the size in human-readable form. """ for unit in ['', 'K', 'M', 'G', 'T', 'P', 'E', 'Z']: if abs(bytes_value) < 1024.0: return f"{bytes_value:.1f}{unit}{suffix}" bytes_value /= 1024.0 return f"{bytes_value:.1f}Y{suffix}" def _check_memory_usage(self) -> Dict[str, Any]: """ Check memory usage and return health metrics. :return: Dictionary with memory health metrics. """ memory_info = psutil.virtual_memory() memory_health = { 'total_memory': self._convert_bytes(memory_info.total), 'used_memory': self._convert_bytes(memory_info.used), 'memory_percent': memory_info.percent, 'status': 'OK' if memory_info.percent < 90 else 'WARNING' } return memory_health def _check_cpu_usage(self) -> Dict[str, Any]: """ Check CPU usage and return health metrics. :return: Dictionary with CPU health metrics. """ cpu_usage_percent = psutil.cpu_percent(interval=1) cpu_health = { 'cpu_usage_percent': cpu_usage_percent, 'status': 'OK' if cpu_usage_percent < 90 else 'WARNING' } return cpu_health def _check_network_status(self) -> Dict[str, Any]: """ Check the status of network interfaces and report any issues. :return: Dictionary containing network health metrics and any issues found. """ network_health = { 'management_network': {'issues': []}, 'ceph_network': {'issues': []} } try: # Check management network connectivity proc = subprocess.run(["ping", "-c", "1", "10.10.10.1"], stdout=subprocess.PIPE, stderr=subprocess.PIPE) if proc.returncode != 0: network_health['management_network']['issues'].append( "Management network is unreachable." ) # Check Ceph network connectivity proc = subprocess.run(["ping", "-c", "1", "10.10.90.1"], stdout=subprocess.PIPE, stderr=subprocess.PIPE) if proc.returncode != 0: network_health['ceph_network']['issues'].append( "Ceph network is unreachable." ) return network_health except Exception as e: print(f"Network health check failed: {e}") return {'error': str(e)} def main(): try: # Argument parser for CLI options parser = argparse.ArgumentParser(description="System Health Monitor") parser.add_argument( "--dry-run", action="store_true", help="Enable dry-run mode (simulate ticket creation without actual API calls)." ) args = parser.parse_args() # Parse command-line arguments or read from configuration file ticket_api_url = "http://10.10.10.45/create_ticket_api.php" state_file = "/tmp/last_health_check.json" # Instantiate the SystemHealthMonitor class monitor = SystemHealthMonitor( ticket_api_url=ticket_api_url, state_file=state_file, dry_run=args.dry_run # Pass the dry-run flag ) # Run the health checks monitor.run() # Check network health synchronously network_health = monitor._check_network_status() logger.info(f"Network health: {network_health}") except Exception as e: logger.error(f"An unexpected error occurred: {e}") sys.exit(1) if __name__ == "__main__": # Argument parser for CLI options parser = argparse.ArgumentParser(description="System Health Monitor") parser.add_argument( "--dry-run", action="store_true", help="Enable dry-run mode (simulate ticket creation without actual API calls)." ) args = parser.parse_args() # Set dry-run mode if specified dry_run_mode = args.dry_run main()