hwmonDaemon/hwmonDaemon.py

#!/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()