""" SAAP System Metrics API - Real-time Resource Monitoring For thesis evaluation and live dashboard monitoring Includes GPU Scheduler for optimized local inference (C4) """ from fastapi import APIRouter, Depends, HTTPException from typing import Dict, List, Optional from datetime import datetime, timedelta import logging import psutil import os import asyncio from dataclasses import dataclass, field from enum import Enum import threading import time logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/metrics", tags=["System Metrics"]) # ============================================================ # GPU SCHEDULER FOR LOCAL INFERENCE OPTIMIZATION (C4) # ============================================================ class GPUPriority(Enum): HIGH = 1 # Real-time chat requests MEDIUM = 2 # Background processing LOW = 3 # Batch operations @dataclass class GPUTask: """Represents a task in the GPU queue""" task_id: str priority: GPUPriority model_name: str agent_id: str input_tokens: int created_at: float = field(default_factory=time.time) started_at: Optional[float] = None completed_at: Optional[float] = None status: str = "queued" # queued, processing, completed, failed class GPUScheduler: """ Intelligent GPU Scheduler for SAAP Local Inference Implements priority queue with fair scheduling for thesis C4 criterion """ def __init__(self): self._queue: List[GPUTask] = [] self._active_tasks: Dict[str, GPUTask] = {} self._completed_tasks: List[GPUTask] = [] self._max_completed_history = 100 self._lock = threading.Lock() self._gpu_info = None self._last_gpu_check = 0 self._gpu_check_interval = 5 # seconds # GPU detection flags self._has_nvidia = False self._has_amd = False self._has_apple_silicon = False # Statistics self.stats = { "total_tasks": 0, "completed_tasks": 0, "failed_tasks": 0, "total_queue_time_ms": 0, "total_processing_time_ms": 0, "avg_latency_reduction_pct": 0 } # Initialize GPU detection self._detect_gpu() def _detect_gpu(self): """Detect available GPU resources""" try: # Try NVIDIA (pynvml) try: import pynvml pynvml.nvmlInit() self._has_nvidia = True logger.info("✅ NVIDIA GPU detected") except: pass # Try AMD (rocm-smi) if os.path.exists("/opt/rocm/bin/rocm-smi"): self._has_amd = True logger.info("✅ AMD GPU (ROCm) detected") # Apple Silicon detection import platform if platform.system() == "Darwin" and platform.machine() == "arm64": self._has_apple_silicon = True logger.info("✅ Apple Silicon detected") except Exception as e: logger.warning(f"GPU detection error: {e}") def get_gpu_metrics(self) -> Dict: """Get current GPU metrics""" current_time = time.time() # Cache GPU info for performance if self._gpu_info and (current_time - self._last_gpu_check) < self._gpu_check_interval: return self._gpu_info gpu_info = { "available": False, "type": "none", "name": "No GPU", "utilization_percent": 0, "memory_used_gb": 0, "memory_total_gb": 0, "memory_percent": 0, "temperature_c": 0, "power_draw_w": 0, "scheduler": { "queue_length": len(self._queue), "active_tasks": len(self._active_tasks), "completed_tasks": self.stats["completed_tasks"], "avg_queue_time_ms": self._calculate_avg_queue_time() } } # NVIDIA GPU Metrics if self._has_nvidia: try: import pynvml pynvml.nvmlInit() handle = pynvml.nvmlDeviceGetHandleByIndex(0) # Get GPU name name = pynvml.nvmlDeviceGetName(handle) if isinstance(name, bytes): name = name.decode('utf-8') # Get utilization util = pynvml.nvmlDeviceGetUtilizationRates(handle) # Get memory mem = pynvml.nvmlDeviceGetMemoryInfo(handle) # Get temperature temp = pynvml.nvmlDeviceGetTemperature(handle, pynvml.NVML_TEMPERATURE_GPU) # Get power draw try: power = pynvml.nvmlDeviceGetPowerUsage(handle) / 1000 # mW to W except: power = 0 gpu_info.update({ "available": True, "type": "nvidia", "name": name, "utilization_percent": util.gpu, "memory_used_gb": round(mem.used / (1024**3), 2), "memory_total_gb": round(mem.total / (1024**3), 2), "memory_percent": round(mem.used / mem.total * 100, 1), "temperature_c": temp, "power_draw_w": round(power, 1) }) pynvml.nvmlShutdown() except Exception as e: logger.debug(f"NVIDIA metrics error: {e}") # Apple Silicon metrics (via ps/top) elif self._has_apple_silicon: gpu_info.update({ "available": True, "type": "apple_silicon", "name": "Apple Silicon (Unified Memory)", "utilization_percent": 0, # Would need powermetrics "memory_used_gb": 0, # Shared with system "memory_total_gb": 0, "memory_percent": 0, "temperature_c": 0, "power_draw_w": 0, "note": "Apple Silicon uses unified memory architecture" }) # AMD ROCm metrics elif self._has_amd: gpu_info.update({ "available": True, "type": "amd_rocm", "name": "AMD GPU (ROCm)", "note": "ROCm metrics available via rocm-smi" }) self._gpu_info = gpu_info self._last_gpu_check = current_time return gpu_info def _calculate_avg_queue_time(self) -> float: """Calculate average queue time in ms""" if self.stats["completed_tasks"] == 0: return 0 return round(self.stats["total_queue_time_ms"] / self.stats["completed_tasks"], 2) async def enqueue_task(self, task_id: str, model_name: str, agent_id: str, input_tokens: int, priority: GPUPriority = GPUPriority.MEDIUM) -> GPUTask: """Add a task to the GPU queue""" with self._lock: task = GPUTask( task_id=task_id, priority=priority, model_name=model_name, agent_id=agent_id, input_tokens=input_tokens ) # Insert by priority (lower number = higher priority) inserted = False for i, existing in enumerate(self._queue): if task.priority.value < existing.priority.value: self._queue.insert(i, task) inserted = True break if not inserted: self._queue.append(task) self.stats["total_tasks"] += 1 logger.debug(f"📋 GPU task queued: {task_id} (priority: {priority.name})") return task async def start_task(self, task_id: str) -> Optional[GPUTask]: """Mark a task as started""" with self._lock: for i, task in enumerate(self._queue): if task.task_id == task_id: task.started_at = time.time() task.status = "processing" queue_time = (task.started_at - task.created_at) * 1000 self.stats["total_queue_time_ms"] += queue_time self._active_tasks[task_id] = task self._queue.pop(i) return task return None async def complete_task(self, task_id: str, success: bool = True) -> Optional[GPUTask]: """Mark a task as completed""" with self._lock: if task_id in self._active_tasks: task = self._active_tasks.pop(task_id) task.completed_at = time.time() task.status = "completed" if success else "failed" processing_time = (task.completed_at - task.started_at) * 1000 self.stats["total_processing_time_ms"] += processing_time if success: self.stats["completed_tasks"] += 1 else: self.stats["failed_tasks"] += 1 # Keep history self._completed_tasks.append(task) if len(self._completed_tasks) > self._max_completed_history: self._completed_tasks.pop(0) return task return None def get_scheduler_stats(self) -> Dict: """Get comprehensive scheduler statistics""" with self._lock: avg_queue_time = self._calculate_avg_queue_time() avg_processing_time = ( self.stats["total_processing_time_ms"] / self.stats["completed_tasks"] if self.stats["completed_tasks"] > 0 else 0 ) return { "queue": { "current_length": len(self._queue), "active_tasks": len(self._active_tasks), "tasks_by_priority": { "high": len([t for t in self._queue if t.priority == GPUPriority.HIGH]), "medium": len([t for t in self._queue if t.priority == GPUPriority.MEDIUM]), "low": len([t for t in self._queue if t.priority == GPUPriority.LOW]) } }, "totals": { "total_tasks": self.stats["total_tasks"], "completed_tasks": self.stats["completed_tasks"], "failed_tasks": self.stats["failed_tasks"], "success_rate": round( self.stats["completed_tasks"] / self.stats["total_tasks"] * 100, 1 ) if self.stats["total_tasks"] > 0 else 100 }, "latency": { "avg_queue_time_ms": round(avg_queue_time, 2), "avg_processing_time_ms": round(avg_processing_time, 2), "estimated_wait_ms": round(len(self._queue) * avg_processing_time, 2) }, "timestamp": datetime.utcnow().isoformat() } # Global GPU Scheduler instance gpu_scheduler = GPUScheduler() @router.get("/system") async def get_system_metrics(): """ Get current system resource metrics CPU, Memory, Disk, Network for live monitoring """ try: # CPU metrics cpu_percent = psutil.cpu_percent(interval=0.1) cpu_count = psutil.cpu_count() cpu_freq = psutil.cpu_freq() # Memory metrics memory = psutil.virtual_memory() swap = psutil.swap_memory() # Disk metrics disk = psutil.disk_usage('/') # Network metrics net_io = psutil.net_io_counters() # Process-specific metrics (SAAP backend) process = psutil.Process() process_memory = process.memory_info() return { "cpu": { "percent": cpu_percent, "cores": cpu_count, "frequency_mhz": cpu_freq.current if cpu_freq else 0 }, "memory": { "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": memory.percent, "swap_percent": swap.percent }, "disk": { "total_gb": round(disk.total / (1024**3), 2), "used_gb": round(disk.used / (1024**3), 2), "free_gb": round(disk.free / (1024**3), 2), "percent": round(disk.percent, 1) }, "network": { "bytes_sent_mb": round(net_io.bytes_sent / (1024**2), 2), "bytes_recv_mb": round(net_io.bytes_recv / (1024**2), 2), "packets_sent": net_io.packets_sent, "packets_recv": net_io.packets_recv }, "process": { "memory_mb": round(process_memory.rss / (1024**2), 2), "cpu_percent": process.cpu_percent(), "threads": process.num_threads(), "pid": process.pid }, "timestamp": datetime.utcnow().isoformat() } except Exception as e: logger.error(f"❌ System metrics error: {e}") raise HTTPException(status_code=500, detail=str(e)) @router.get("/latency") async def get_latency_metrics(): """ Get latency statistics from recent requests For thesis evaluation (H4) - Performance monitoring """ try: from database.connection import db_manager from database.models import DBChatMessage from sqlalchemy import select, func async with db_manager.get_async_session() as db: # Get last 100 requests for latency analysis result = await db.execute( select(DBChatMessage.response_time) .where(DBChatMessage.response_time.isnot(None)) .order_by(DBChatMessage.created_at.desc()) .limit(100) ) response_times = [row[0] for row in result.fetchall()] if not response_times: return { "sample_size": 0, "avg_ms": 0, "min_ms": 0, "max_ms": 0, "p50_ms": 0, "p95_ms": 0, "p99_ms": 0, "timestamp": datetime.utcnow().isoformat() } # Calculate percentiles sorted_times = sorted(response_times) n = len(sorted_times) p50_idx = int(n * 0.5) p95_idx = int(n * 0.95) p99_idx = int(n * 0.99) return { "sample_size": n, "avg_ms": round(sum(response_times) / n * 1000, 2), "min_ms": round(min(response_times) * 1000, 2), "max_ms": round(max(response_times) * 1000, 2), "p50_ms": round(sorted_times[p50_idx] * 1000, 2), "p95_ms": round(sorted_times[min(p95_idx, n-1)] * 1000, 2), "p99_ms": round(sorted_times[min(p99_idx, n-1)] * 1000, 2), "timestamp": datetime.utcnow().isoformat() } except Exception as e: logger.error(f"❌ Latency metrics error: {e}") return { "sample_size": 0, "error": str(e), "timestamp": datetime.utcnow().isoformat() } @router.get("/overview") async def get_metrics_overview(): """ Get comprehensive metrics overview for dashboard Combines system, latency, and cost metrics """ try: # System metrics cpu_percent = psutil.cpu_percent(interval=0.1) memory = psutil.virtual_memory() process = psutil.Process() # Get uptime import time boot_time = psutil.boot_time() uptime_seconds = time.time() - boot_time uptime_hours = round(uptime_seconds / 3600, 1) # Get request count from database from database.connection import db_manager from database.models import DBChatMessage, DBLLMCost from sqlalchemy import select, func total_requests = 0 today_requests = 0 today_cost = 0.0 avg_response_time = 0.0 try: async with db_manager.get_async_session() as db: # Total requests result = await db.execute(select(func.count(DBChatMessage.message_id))) total_requests = result.scalar() or 0 # Today's requests today_start = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0) result = await db.execute( select(func.count(DBChatMessage.message_id)) .where(DBChatMessage.created_at >= today_start) ) today_requests = result.scalar() or 0 # Today's cost result = await db.execute( select(func.sum(DBLLMCost.cost_usd)) .where(DBLLMCost.created_at >= today_start) ) today_cost = result.scalar() or 0.0 # Average response time (last 100) result = await db.execute( select(func.avg(DBChatMessage.response_time)) .where(DBChatMessage.response_time.isnot(None)) .limit(100) ) avg_response_time = result.scalar() or 0.0 except Exception as db_error: logger.warning(f"DB metrics error: {db_error}") return { "system": { "cpu_percent": cpu_percent, "memory_percent": memory.percent, "memory_used_gb": round(memory.used / (1024**3), 2), "uptime_hours": uptime_hours }, "performance": { "avg_response_time_s": round(avg_response_time, 2), "total_requests": total_requests, "today_requests": today_requests }, "costs": { "today_usd": round(today_cost, 4) }, "process": { "memory_mb": round(process.memory_info().rss / (1024**2), 2), "threads": process.num_threads() }, "status": "healthy", "timestamp": datetime.utcnow().isoformat() } except Exception as e: logger.error(f"❌ Metrics overview error: {e}") raise HTTPException(status_code=500, detail=str(e)) @router.get("/gpu") async def get_gpu_metrics(): """ Get GPU metrics and scheduler status For thesis evaluation (C4) - Local resource optimization """ try: return gpu_scheduler.get_gpu_metrics() except Exception as e: logger.error(f"❌ GPU metrics error: {e}") return { "available": False, "error": str(e), "timestamp": datetime.utcnow().isoformat() } @router.get("/gpu/scheduler") async def get_gpu_scheduler_stats(): """ Get detailed GPU scheduler statistics """ try: return gpu_scheduler.get_scheduler_stats() except Exception as e: logger.error(f"❌ GPU scheduler stats error: {e}") raise HTTPException(status_code=500, detail=str(e)) @router.post("/gpu/scheduler/task") async def enqueue_gpu_task( task_id: str, model_name: str, agent_id: str, input_tokens: int, priority: str = "medium" ): """ Enqueue a task to the GPU scheduler Used by agent_manager for local inference """ try: priority_map = { "high": GPUPriority.HIGH, "medium": GPUPriority.MEDIUM, "low": GPUPriority.LOW } prio = priority_map.get(priority.lower(), GPUPriority.MEDIUM) task = await gpu_scheduler.enqueue_task( task_id=task_id, model_name=model_name, agent_id=agent_id, input_tokens=input_tokens, priority=prio ) return { "success": True, "task_id": task.task_id, "status": task.status, "queue_position": gpu_scheduler._queue.index(task) if task in gpu_scheduler._queue else 0 } except Exception as e: logger.error(f"❌ GPU task enqueue error: {e}") raise HTTPException(status_code=500, detail=str(e)) @router.get("/health") async def get_health_check(): """ Health check endpoint for monitoring """ try: # Check database connection from database.connection import db_manager from sqlalchemy import text db_healthy = False try: async with db_manager.get_async_session() as db: await db.execute(text("SELECT 1")) db_healthy = True except: pass # Check Colossus availability colossus_healthy = False colossus_url = os.getenv("COLOSSUS_URL", "http://localhost:11434") try: import httpx async with httpx.AsyncClient(timeout=2.0) as client: response = await client.get(f"{colossus_url}/api/tags") colossus_healthy = response.status_code == 200 except: pass # Check OpenRouter openrouter_healthy = bool(os.getenv("OPENROUTER_API_KEY")) status = "healthy" if db_healthy else "degraded" return { "status": status, "components": { "database": "healthy" if db_healthy else "unhealthy", "colossus": "healthy" if colossus_healthy else "unavailable", "openrouter": "configured" if openrouter_healthy else "not_configured" }, "timestamp": datetime.utcnow().isoformat() } except Exception as e: return { "status": "unhealthy", "error": str(e), "timestamp": datetime.utcnow().isoformat() }