"""Health monitoring for AI service""" import asyncio import time import logging from typing import Dict, Any, List from dataclasses import dataclass from collections import deque logger = logging.getLogger(__name__) @dataclass class HealthMetrics: """Health metrics for the service""" requests_total: int = 0 requests_success: int = 0 requests_failed: int = 0 latency_samples: deque = None last_request_time: float = 0 start_time: float = 0 def __post_init__(self): if self.latency_samples is None: self.latency_samples = deque(maxlen=100) if self.start_time == 0: self.start_time = time.time() class HealthMonitor: """Monitors service health and performance""" def __init__(self): self.metrics = HealthMetrics() self.is_healthy = True self.last_health_check = 0 self.health_check_interval = 2.0 # seconds def record_request_start(self) -> float: """Record start of request processing""" self.metrics.requests_total += 1 self.metrics.last_request_time = time.time() return self.metrics.last_request_time def record_request_success(self, start_time: float) -> None: """Record successful request""" latency_ms = int((time.time() - start_time) * 1000) self.metrics.requests_success += 1 self.metrics.latency_samples.append(latency_ms) def record_request_failure(self, start_time: float) -> None: """Record failed request""" latency_ms = int((time.time() - start_time) * 1000) self.metrics.requests_failed += 1 self.metrics.latency_samples.append(latency_ms) def get_health_status(self) -> Dict[str, Any]: """Get current health status""" current_time = time.time() uptime_seconds = current_time - self.metrics.start_time # Calculate latency percentiles latencies = list(self.metrics.latency_samples) latency_p50 = 0 latency_p95 = 0 if latencies: latencies.sort() n = len(latencies) latency_p50 = latencies[int(n * 0.5)] if n > 0 else 0 latency_p95 = latencies[int(n * 0.95)] if n > 0 else 0 # Calculate success rate success_rate = 0.0 if self.metrics.requests_total > 0: success_rate = self.metrics.requests_success / self.metrics.requests_total # Determine health status self.is_healthy = self._assess_health(success_rate, latency_p95, current_time) return { "ok": self.is_healthy, "uptime_seconds": int(uptime_seconds), "requests_total": self.metrics.requests_total, "requests_success": self.metrics.requests_success, "requests_failed": self.metrics.requests_failed, "success_rate": round(success_rate, 3), "latency_p50": latency_p50, "latency_p95": latency_p95, "last_request_age": ( int(current_time - self.metrics.last_request_time) if self.metrics.last_request_time > 0 else 0 ), } def _assess_health( self, success_rate: float, latency_p95: int, current_time: float ) -> bool: """Assess overall service health""" # Check success rate (should be > 80%) if self.metrics.requests_total >= 5 and success_rate < 0.8: logger.warning(f"Low success rate: {success_rate:.2%}") return False # Check latency (p95 should be < 5000ms) if latency_p95 > 5000: logger.warning(f"High latency p95: {latency_p95}ms") return False # Service is healthy return True def should_publish_health(self) -> bool: """Check if health status should be published""" current_time = time.time() return (current_time - self.last_health_check) >= self.health_check_interval def mark_health_published(self) -> None: """Mark health status as published""" self.last_health_check = time.time() def reset_metrics(self) -> None: """Reset all metrics (for testing)""" self.metrics = HealthMetrics() class ResourceMonitor: """Monitors system resources""" def __init__(self): self.cpu_usage = 0.0 self.memory_usage = 0.0 self.last_resource_check = 0 def get_resource_usage(self) -> Dict[str, Any]: """Get current resource usage (simplified)""" current_time = time.time() # Update resources every 5 seconds if current_time - self.last_resource_check > 5: self._update_resource_usage() self.last_resource_check = current_time return { "cpu_usage": self.cpu_usage, "memory_usage": self.memory_usage, "timestamp": current_time, } def _update_resource_usage(self) -> None: """Update resource usage metrics""" try: # Try to get actual system metrics import psutil process = psutil.Process() self.cpu_usage = process.cpu_percent() self.memory_usage = process.memory_info().rss / 1024 / 1024 # MB except ImportError: # Fallback to mock values self.cpu_usage = 5.0 self.memory_usage = 50.0 class ServiceHealth: """Main service health coordinator""" def __init__(self): self.health_monitor = HealthMonitor() self.resource_monitor = ResourceMonitor() self.running = False self.health_task: asyncio.Task = None async def start_monitoring(self, mqtt_handler) -> None: """Start health monitoring loop""" self.running = True self.health_task = asyncio.create_task(self._health_loop(mqtt_handler)) async def stop_monitoring(self) -> None: """Stop health monitoring""" self.running = False if self.health_task: self.health_task.cancel() try: await self.health_task except asyncio.CancelledError: pass async def _health_loop(self, mqtt_handler) -> None: """Health monitoring loop""" while self.running: try: if self.health_monitor.should_publish_health(): # Get health status health_status = self.health_monitor.get_health_status() # Add resource usage resources = self.resource_monitor.get_resource_usage() health_status.update(resources) # Publish health status mqtt_handler.publish_json("ai/health", health_status) self.health_monitor.mark_health_published() await asyncio.sleep(1.0) # Check every second except asyncio.CancelledError: break except Exception as e: logger.error(f"Health monitoring error: {e}") await asyncio.sleep(5.0) # Wait longer on error def get_current_status(self) -> Dict[str, Any]: """Get current comprehensive status""" health_status = self.health_monitor.get_health_status() resource_status = self.resource_monitor.get_resource_usage() return { "health": health_status, "resources": resource_status, "monitoring": { "running": self.running, "health_check_interval": self.health_monitor.health_check_interval, }, } # Global health service instance service_health = ServiceHealth()