Gecko [ v3.cesa.co.za ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2025-09-06 08:13
mail.py	8.55 KB	-rw-r--r--	2025-09-06 08:13
server_checks.py	19.73 KB	-rw-r--r--	2025-09-06 08:13

Rename

#!/usr/bin/env python3
"""
Server Monitoring Service for IGA Core

This module provides server health monitoring including:
- Memory usage monitoring
- Disk space monitoring  
- CPU load monitoring
- Email alerts when thresholds are exceeded
"""

import os
import psutil
from typing import Dict, Any, Optional, Tuple
from datetime import datetime
from pathlib import Path
import platform

# Import our mail service and configuration
import sys
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
from config import get_server_monitor_config
from .mail import SymfonyMailService

class ServerMonitor:
    """
    Server monitoring class that checks system resources and sends alerts.
    """
    
    def __init__(self, alert_email: Optional[str] = None):
        """
        Initialize the server monitor.
        
        Args:
            alert_email: Email address to send alerts to (optional, uses config default if not provided)
        """
        # Load configuration
        config = get_server_monitor_config()
        
        self.alert_email = alert_email or config['alert_email']
        self.mail_service = SymfonyMailService()
        
        # Thresholds from configuration
        self.disk_threshold = config['disk_threshold']
        self.memory_threshold = config['memory_threshold']
        self.cpu_threshold = config['cpu_threshold']
        
        # System info
        self.cpu_count = psutil.cpu_count()
        self.hostname = platform.node()
    
    def check_disk_usage(self) -> Dict[str, Any]:
        """
        Check disk usage for all mounted filesystems.
        
        Returns:
            Dict containing disk usage information and alerts
        """
        try:
            disk_info = []
            alerts = []
            
            for partition in psutil.disk_partitions():
                try:
                    usage = psutil.disk_usage(partition.mountpoint)
                    percent_used = (usage.used / usage.total) * 100
                    percent_free = 100 - percent_used
                    
                    disk_data = {
                        'device': partition.device,
                        'mountpoint': partition.mountpoint,
                        'filesystem': partition.fstype,
                        'total_gb': round(usage.total / (1024**3), 2),
                        'used_gb': round(usage.used / (1024**3), 2),
                        'free_gb': round(usage.free / (1024**3), 2),
                        'percent_used': round(percent_used, 2),
                        'percent_free': round(percent_free, 2)
                    }
                    
                    disk_info.append(disk_data)
                    
                    # Check threshold (ignore /snap partitions)
                    if percent_used > self.disk_threshold and not partition.mountpoint.startswith('/snap'):
                        alert_msg = (
                            f"DISK ALERT: {partition.mountpoint} is {percent_used:.1f}% full "
                            f"({disk_data['free_gb']:.1f}GB free)"
                        )
                        alerts.append({
                            'type': 'disk',
                            'severity': 'high',
                            'message': alert_msg,
                            'data': disk_data
                        })
                    
                except PermissionError:
                    pass
                except Exception:
                    pass
            
            return {
                'status': 'success',
                'disk_info': disk_info,
                'alerts': alerts,
                'timestamp': datetime.now().isoformat()
            }
            
        except Exception as e:
            return {
                'status': 'error',
                'error': str(e),
                'timestamp': datetime.now().isoformat()
            }
    
    def check_memory_usage(self) -> Dict[str, Any]:
        """
        Check memory usage.
        
        Returns:
            Dict containing memory usage information and alerts
        """
        try:
            memory = psutil.virtual_memory()
            
            memory_data = {
                'total_gb': round(memory.total / (1024**3), 2),
                'available_gb': round(memory.available / (1024**3), 2),
                'used_gb': round(memory.used / (1024**3), 2),
                'percent_used': round(memory.percent, 2),
                'percent_available': round(100 - memory.percent, 2)
            }
            
            alerts = []
            
            # Check threshold
            if memory_data['percent_available'] < self.memory_threshold:
                alert_msg = (
                    f"MEMORY ALERT: Only {memory_data['percent_available']:.1f}% memory available "
                    f"({memory_data['available_gb']:.1f}GB free)"
                )
                alerts.append({
                    'type': 'memory',
                    'severity': 'high',
                    'message': alert_msg,
                    'data': memory_data
                })
            
            return {
                'status': 'success',
                'memory_info': memory_data,
                'alerts': alerts,
                'timestamp': datetime.now().isoformat()
            }
            
        except Exception as e:
            return {
                'status': 'error',
                'error': str(e),
                'timestamp': datetime.now().isoformat()
            }
    
    def check_cpu_usage(self) -> Dict[str, Any]:
        """
        Check CPU usage and load average.
        
        Returns:
            Dict containing CPU usage information and alerts
        """
        try:
            # Get CPU usage percentage
            cpu_percent = psutil.cpu_percent(interval=1)
            
            # Get load average (1, 5, 15 minute averages)
            load_avg = os.getloadavg()
            
            # Calculate load per core
            load_per_core_1min = load_avg[0] / self.cpu_count
            load_per_core_5min = load_avg[1] / self.cpu_count
            load_per_core_15min = load_avg[2] / self.cpu_count
            
            cpu_data = {
                'cpu_percent': round(cpu_percent, 2),
                'load_1min': round(load_avg[0], 2),
                'load_5min': round(load_avg[1], 2),
                'load_15min': round(load_avg[2], 2),
                'load_per_core_1min': round(load_per_core_1min, 2),
                'load_per_core_5min': round(load_per_core_5min, 2),
                'load_per_core_15min': round(load_per_core_15min, 2),
                'cpu_count': self.cpu_count
            }
            
            alerts = []
            
            # Check threshold (using 1-minute load average)
            if load_per_core_1min > self.cpu_threshold:
                alert_msg = (
                    f"CPU ALERT: Load average is {load_per_core_1min:.2f} per core "
                    f"(1min: {load_avg[0]:.2f}, 5min: {load_avg[1]:.2f}, 15min: {load_avg[2]:.2f})"
                )
                alerts.append({
                    'type': 'cpu',
                    'severity': 'medium',
                    'message': alert_msg,
                    'data': cpu_data
                })
            
            return {
                'status': 'success',
                'cpu_info': cpu_data,
                'alerts': alerts,
                'timestamp': datetime.now().isoformat()
            }
            
        except Exception as e:
            return {
                'status': 'error',
                'error': str(e),
                'timestamp': datetime.now().isoformat()
            }
    
    def run_all_checks(self) -> Dict[str, Any]:
        """
        Run all system checks.
        
        Returns:
            Dict containing all check results and alerts
        """
        results = {
            'hostname': self.hostname,
            'timestamp': datetime.now().isoformat(),
            'checks': {}
        }
        
        # Run all checks
        results['checks']['disk'] = self.check_disk_usage()
        results['checks']['memory'] = self.check_memory_usage()
        results['checks']['cpu'] = self.check_cpu_usage()
        
        # Collect all alerts
        all_alerts = []
        failed_checks = []
        
        for check_name, check_result in results['checks'].items():
            if check_result.get('status') == 'success' and 'alerts' in check_result:
                all_alerts.extend(check_result['alerts'])
            elif check_result.get('status') == 'error':
                failed_checks.append(check_name)
        
        results['total_alerts'] = len(all_alerts)
        results['alerts'] = all_alerts
        results['failed_checks'] = failed_checks
        
        # Only send email report if there are alerts or failed checks
        if all_alerts or failed_checks:
            self._send_comprehensive_report(results)
        
        return results
    
    def _send_alerts(self, alerts: list, results: Dict[str, Any]):
        """
        Send email alerts for all detected issues.
        
        Args:
            alerts: List of alert dictionaries
            results: Complete check results
        """
        try:
            # Group alerts by type
            alert_groups = {}
            for alert in alerts:
                alert_type = alert['type']
                if alert_type not in alert_groups:
                    alert_groups[alert_type] = []
                alert_groups[alert_type].append(alert)
            
            # Create email content
            subject = f"SERVER ALERT: {self.hostname} - {len(alerts)} issues detected"
            
            # Plain text body
            text_body = f"""
Server Health Alert - {self.hostname}
Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
Total Issues: {len(alerts)}

"""
            # HTML body
            html_body = f"""
<html>
<head>
    <style>
        body {{ font-family: Arial, sans-serif; margin: 20px; }}
        .header {{ background-color: #f0f0f0; padding: 15px; border-radius: 5px; }}
        .alert {{ margin: 10px 0; padding: 10px; border-left: 4px solid #ff4444; background-color: #fff5f5; }}
        .alert.high {{ border-left-color: #ff4444; }}
        .alert.medium {{ border-left-color: #ffaa00; }}
        .summary {{ background-color: #e8f5e8; padding: 15px; border-radius: 5px; margin: 20px 0; }}
        .metric {{ margin: 5px 0; }}
        .check-section {{ margin: 20px 0; padding: 15px; border: 1px solid #ddd; border-radius: 5px; }}
        .check-header {{ font-weight: bold; margin-bottom: 10px; }}
        .failed {{ border-left-color: #ff4444; background-color: #fff5f5; }}
    </style>
</head>
<body>
    <div class="header">
        <h2>🚨 Server Health Alert</h2>
        <p><strong>Server:</strong> {self.hostname}</p>
        <p><strong>Generated:</strong> {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
        <p><strong>Total Alerts:</strong> {total_alerts}</p>
        <p><strong>Failed Checks:</strong> {len(failed_checks)}</p>
    </div>
"""
            
            # Add failed checks summary
            if failed_checks:
                html_body += "<div class='summary'><h3>❌ Failed Checks</h3>"
                for check_name in failed_checks:
                    check_result = results['checks'].get(check_name, {})
                    error_msg = check_result.get('error', 'Unknown error')
                    html_body += f"<div class='metric'><strong>{check_name}:</strong> ERROR - {error_msg}</div>"
                    # print(f"Failed Check: {check_name} - {error_msg}")
                    text_body += f"Failed Check: {check_name} - {error_msg}\n"
                html_body += "</div>"
            
            # Add alerts summary
            if total_alerts > 0:
                html_body += "<h3>⚠️ Detected Alerts</h3>"
                text_body += f"\nALERTS DETECTED:\n"
                
                # Group alerts by type
                alert_groups = {}
                for alert in results.get('alerts', []):
                    alert_type = alert['type']
                    if alert_type not in alert_groups:
                        alert_groups[alert_type] = []
                    alert_groups[alert_type].append(alert)
                
                # Display alerts by type
                for alert_type, type_alerts in alert_groups.items():
                    html_body += f"<h4>{alert_type.upper()} Alerts</h4>"
                    text_body += f"\n{alert_type.upper()} ALERTS:\n"
                    
                    for alert in type_alerts:
                        text_body += f"- {alert['message']}\n"
                        html_body += f"""
    <div class="alert {alert.get('severity', 'medium')}">
        <strong>{alert['message']}</strong>
"""
                        
                        # Add detailed metrics if available
                        if 'data' in alert:
                            for key, value in alert['data'].items():
                                if isinstance(value, (int, float)):
                                    text_body += f"  {key}: {value}\n"
                                    html_body += f"<div class='metric'>{key}: {value}</div>"
                        
                        html_body += "</div>"
            
            # Add system status summary
            html_body += "<div class='summary'><h3>📊 System Status Summary</h3>"
            text_body += "\nSYSTEM STATUS:\n"
            
            for check_name, check_result in results['checks'].items():
                if check_result.get('status') == 'success':
                    # Only mention successful checks if they have alerts
                    alerts = check_result.get('alerts', [])
                    if alerts:
                        html_body += f"<div class='metric'><strong>{check_name}:</strong> OK (with {len(alerts)} alerts)</div>"
                        text_body += f"{check_name}: OK (with {len(alerts)} alerts)\n"
                    else:
                        html_body += f"<div class='metric'><strong>{check_name}:</strong> ✅ OK</div>"
                        text_body += f"{check_name}: OK\n"
                else:
                    html_body += f"<div class='metric'><strong>{check_name}:</strong> ❌ ERROR</div>"
                    text_body += f"{check_name}: ERROR\n"
            
            html_body += "</div>"
            html_body += "</body></html>"
            
            # print(f"Sending email to {self.alert_email} with subject: {subject}")
            # print(text_body)
            
            # Send email
            self.mail_service.send_email(
                to_email=self.alert_email,
                subject=subject,
                body=text_body.strip(),
                html_body=html_body
            )
                
        except Exception:
            print("Error: ", sys.exc_info()[1])
            pass
    
    def get_system_info(self) -> Dict[str, Any]:
        """
        Get general system information.
        
        Returns:
            Dict containing system information
        """
        try:
            return {
                'hostname': self.hostname,
                'platform': platform.platform(),
                'python_version': platform.python_version(),
                'cpu_count': self.cpu_count,
                'cpu_freq': psutil.cpu_freq()._asdict() if psutil.cpu_freq() else None,
                'boot_time': datetime.fromtimestamp(psutil.boot_time()).isoformat(),
                'uptime': datetime.now() - datetime.fromtimestamp(psutil.boot_time())
            }
        except Exception as e:
            return {'error': str(e)}

# Convenience function for quick monitoring
def quick_server_check(alert_email: Optional[str] = None) -> Dict[str, Any]:
    """
    Quick function to run all server checks.
    
    Args:
        alert_email: Email address to send alerts to (optional, uses config default if not provided)
    
    Returns:
        Dict containing all check results
    """
    monitor = ServerMonitor(alert_email)
    return monitor.run_all_checks()