From ba0eed0daf99f3d11455b6110524ac211a84fc63 Mon Sep 17 00:00:00 2001 From: sbcrumb Date: Mon, 29 Sep 2025 09:23:08 -0400 Subject: [PATCH] feat: metrics --- api/monitoring_routes.py | 351 ++++++++++++++++++++++++ api/routes.py | 6 +- monitoring/__init__.py | 0 monitoring/health.py | 474 +++++++++++++++++++++++++++++++++ monitoring/logging_enhanced.py | 404 ++++++++++++++++++++++++++++ monitoring/metrics.py | 354 ++++++++++++++++++++++++ monitoring/performance.py | 413 ++++++++++++++++++++++++++++ 7 files changed, 2001 insertions(+), 1 deletion(-) create mode 100644 api/monitoring_routes.py create mode 100644 monitoring/__init__.py create mode 100644 monitoring/health.py create mode 100644 monitoring/logging_enhanced.py create mode 100644 monitoring/metrics.py create mode 100644 monitoring/performance.py diff --git a/api/monitoring_routes.py b/api/monitoring_routes.py new file mode 100644 index 0000000..58b162e --- /dev/null +++ b/api/monitoring_routes.py @@ -0,0 +1,351 @@ +""" +Monitoring API Routes for NFOGuard +Provides health checks, metrics, and system status endpoints +""" +from fastapi import APIRouter, Response, HTTPException +from typing import Dict, Any, Optional +import time + +from monitoring.health import health_checker, HealthStatus +from monitoring.metrics import metrics +try: + from config.validator import get_configuration_summary +except ImportError: + def get_configuration_summary(): + return {"status": "Configuration validator not available"} + + +router = APIRouter(prefix="/api/v1", tags=["monitoring"]) + + +@router.get("/health") +async def get_health_status(): + """ + Get comprehensive health status + + Returns detailed health information including: + - Overall system health + - Individual component health checks + - Performance metrics + - Error status + """ + try: + health_status = await health_checker.get_full_health_status() + + # Set appropriate HTTP status code + if health_status.status == HealthStatus.HEALTHY: + status_code = 200 + elif health_status.status == HealthStatus.DEGRADED: + status_code = 200 # Still operational + else: # UNHEALTHY + status_code = 503 # Service unavailable + + return Response( + content=health_status.to_dict(), + status_code=status_code, + media_type="application/json" + ) + + except Exception as e: + # Return unhealthy status if health check itself fails + return Response( + content={ + "status": "unhealthy", + "message": f"Health check failed: {e}", + "timestamp": time.time() + }, + status_code=500, + media_type="application/json" + ) + + +@router.get("/health/ready") +async def get_readiness_status(): + """ + Kubernetes readiness probe endpoint + + Returns 200 if service is ready to accept traffic + Returns 503 if service is not ready + """ + try: + readiness = await health_checker.get_readiness_status() + + status_code = 200 if readiness["ready"] else 503 + + return Response( + content=readiness, + status_code=status_code, + media_type="application/json" + ) + + except Exception as e: + return Response( + content={ + "ready": False, + "message": f"Readiness check failed: {e}", + "timestamp": time.time() + }, + status_code=503, + media_type="application/json" + ) + + +@router.get("/health/live") +async def get_liveness_status(): + """ + Kubernetes liveness probe endpoint + + Returns 200 if service is alive and responsive + Returns 500 if service should be restarted + """ + try: + liveness = await health_checker.get_liveness_status() + + status_code = 200 if liveness["alive"] else 500 + + return Response( + content=liveness, + status_code=status_code, + media_type="application/json" + ) + + except Exception as e: + return Response( + content={ + "alive": False, + "message": f"Liveness check failed: {e}", + "timestamp": time.time() + }, + status_code=500, + media_type="application/json" + ) + + +@router.get("/metrics") +async def get_prometheus_metrics(): + """ + Prometheus-compatible metrics endpoint + + Returns metrics in Prometheus text format for scraping + """ + try: + prometheus_metrics = metrics.get_prometheus_metrics() + + return Response( + content=prometheus_metrics, + media_type="text/plain; charset=utf-8" + ) + + except Exception as e: + raise HTTPException( + status_code=500, + detail=f"Failed to generate metrics: {e}" + ) + + +@router.get("/metrics/json") +async def get_metrics_json(): + """ + Get all metrics in JSON format + + Returns structured metrics data including: + - System metrics (CPU, memory, disk) + - Processing metrics (rates, durations) + - Error metrics (counts, recent errors) + """ + try: + all_metrics = metrics.get_all_metrics() + return all_metrics + + except Exception as e: + raise HTTPException( + status_code=500, + detail=f"Failed to get metrics: {e}" + ) + + +@router.get("/status") +async def get_system_status(): + """ + Get comprehensive system status + + Returns detailed system information including: + - Health status + - Configuration summary + - Performance metrics + - Recent activity + """ + try: + # Get health status + health_status = await health_checker.get_full_health_status() + + # Get metrics + all_metrics = metrics.get_all_metrics() + + # Get configuration summary + config_summary = get_configuration_summary() + + # Combine into comprehensive status + status_response = { + "overall_status": health_status.status.value, + "timestamp": time.time(), + "uptime_seconds": health_status.uptime_seconds, + "version": health_status.version, + "health": health_status.to_dict(), + "metrics": all_metrics, + "configuration": config_summary, + "summary": { + "service_healthy": health_status.status in [HealthStatus.HEALTHY, HealthStatus.DEGRADED], + "total_webhooks_processed": all_metrics["processing"]["total_webhooks"], + "total_nfo_files_created": all_metrics["processing"]["total_nfo_created"], + "total_errors": all_metrics["processing"]["total_errors"], + "current_processing_rate": all_metrics["processing"]["webhooks_received_per_minute"], + "average_processing_time": all_metrics["processing"]["average_processing_time_seconds"] + } + } + + return status_response + + except Exception as e: + raise HTTPException( + status_code=500, + detail=f"Failed to get system status: {e}" + ) + + +@router.get("/status/brief") +async def get_brief_status(): + """ + Get brief system status for quick monitoring + + Returns essential status information without detailed metrics + """ + try: + # Get basic health + liveness = await health_checker.get_liveness_status() + readiness = await health_checker.get_readiness_status() + + # Get basic metrics + processing_metrics = metrics.get_processing_metrics() + system_metrics = metrics.get_system_metrics() + + return { + "status": "healthy" if liveness["alive"] and readiness["ready"] else "unhealthy", + "alive": liveness["alive"], + "ready": readiness["ready"], + "uptime_seconds": liveness["uptime_seconds"], + "webhooks_per_minute": processing_metrics["webhooks_received_per_minute"], + "active_operations": processing_metrics["active_operations"], + "total_errors": processing_metrics["total_errors"], + "cpu_percent": system_metrics.get("cpu_percent", 0), + "memory_percent": system_metrics.get("memory_percent", 0), + "timestamp": time.time() + } + + except Exception as e: + raise HTTPException( + status_code=500, + detail=f"Failed to get brief status: {e}" + ) + + +@router.get("/metrics/processing") +async def get_processing_metrics(): + """ + Get processing-specific metrics + + Returns metrics focused on NFO processing performance + """ + try: + processing_metrics = metrics.get_processing_metrics() + return processing_metrics + + except Exception as e: + raise HTTPException( + status_code=500, + detail=f"Failed to get processing metrics: {e}" + ) + + +@router.get("/metrics/errors") +async def get_error_metrics(): + """ + Get error-specific metrics + + Returns error counts, types, and recent error information + """ + try: + error_metrics = metrics.get_error_metrics() + return error_metrics + + except Exception as e: + raise HTTPException( + status_code=500, + detail=f"Failed to get error metrics: {e}" + ) + + +@router.get("/metrics/system") +async def get_system_metrics(): + """ + Get system resource metrics + + Returns CPU, memory, disk, and process information + """ + try: + system_metrics = metrics.get_system_metrics() + return system_metrics + + except Exception as e: + raise HTTPException( + status_code=500, + detail=f"Failed to get system metrics: {e}" + ) + + +@router.post("/metrics/reset") +async def reset_metrics(metric_types: Optional[str] = None): + """ + Reset specific metric types + + Parameters: + - metric_types: Comma-separated list of metric types to reset + (counters, histograms, errors, timeseries) + If not specified, resets all metrics + """ + try: + reset_types = None + if metric_types: + reset_types = [t.strip() for t in metric_types.split(",")] + + metrics.reset_metrics(reset_types) + + return { + "message": "Metrics reset successfully", + "reset_types": reset_types or "all", + "timestamp": time.time() + } + + except Exception as e: + raise HTTPException( + status_code=500, + detail=f"Failed to reset metrics: {e}" + ) + + +# Legacy endpoints for backwards compatibility +@router.get("/health-check") +async def legacy_health_check(): + """Legacy health check endpoint (redirects to /health)""" + return await get_brief_status() + + +@router.get("/ping") +async def ping(): + """Simple ping endpoint for basic connectivity testing""" + return { + "message": "pong", + "timestamp": time.time(), + "service": "nfoguard", + "version": "2.0.0" + } \ No newline at end of file diff --git a/api/routes.py b/api/routes.py index e83d496..ad7e582 100644 --- a/api/routes.py +++ b/api/routes.py @@ -867,4 +867,8 @@ def register_routes(app, dependencies: dict): @app.get("/debug/tmdb/{imdb_id}") async def _debug_tmdb_lookup(imdb_id: str): - return await debug_tmdb_lookup(imdb_id, dependencies) \ No newline at end of file + return await debug_tmdb_lookup(imdb_id, dependencies) + + # Include monitoring routes + from api.monitoring_routes import router as monitoring_router + app.include_router(monitoring_router) \ No newline at end of file diff --git a/monitoring/__init__.py b/monitoring/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/monitoring/health.py b/monitoring/health.py new file mode 100644 index 0000000..40ef430 --- /dev/null +++ b/monitoring/health.py @@ -0,0 +1,474 @@ +""" +Health Check System for NFOGuard +Provides health and readiness endpoints for monitoring and orchestration +""" +import time +import asyncio +from typing import Dict, Any, List, Optional +from dataclasses import dataclass +from enum import Enum +from pathlib import Path + +from config.runtime_validator import RuntimeValidator, HealthCheckResult +from monitoring.metrics import metrics + + +class HealthStatus(Enum): + """Health check status levels""" + HEALTHY = "healthy" + DEGRADED = "degraded" + UNHEALTHY = "unhealthy" + + +@dataclass +class HealthCheck: + """Individual health check result""" + name: str + status: HealthStatus + message: str + duration_ms: float + details: Dict[str, Any] = None + + def to_dict(self) -> Dict[str, Any]: + return { + "name": self.name, + "status": self.status.value, + "message": self.message, + "duration_ms": round(self.duration_ms, 2), + "details": self.details or {} + } + + +@dataclass +class OverallHealth: + """Overall system health status""" + status: HealthStatus + checks: List[HealthCheck] + timestamp: float + uptime_seconds: float + version: str = "2.0.0" + + def to_dict(self) -> Dict[str, Any]: + return { + "status": self.status.value, + "timestamp": self.timestamp, + "uptime_seconds": round(self.uptime_seconds, 2), + "version": self.version, + "checks": [check.to_dict() for check in self.checks], + "summary": { + "total_checks": len(self.checks), + "healthy_checks": len([c for c in self.checks if c.status == HealthStatus.HEALTHY]), + "degraded_checks": len([c for c in self.checks if c.status == HealthStatus.DEGRADED]), + "unhealthy_checks": len([c for c in self.checks if c.status == HealthStatus.UNHEALTHY]) + } + } + + +class HealthChecker: + """Comprehensive health checking system""" + + def __init__(self): + self.start_time = time.time() + self._last_health_check = None + self._health_check_cache_ttl = 30 # Cache for 30 seconds + self._runtime_validator = None + + def _get_runtime_validator(self): + """Get runtime validator instance""" + if self._runtime_validator is None: + try: + from config.settings import config + self._runtime_validator = RuntimeValidator(config) + except Exception as e: + # Create a dummy validator if config fails + self._runtime_validator = None + return self._runtime_validator + + async def check_basic_health(self) -> HealthCheck: + """Basic health check - always succeeds if service is running""" + start_time = time.time() + + try: + # Basic service availability + uptime = time.time() - self.start_time + + if uptime < 30: + status = HealthStatus.DEGRADED + message = f"Service starting up (uptime: {uptime:.1f}s)" + else: + status = HealthStatus.HEALTHY + message = f"Service running normally (uptime: {uptime:.1f}s)" + + return HealthCheck( + name="basic", + status=status, + message=message, + duration_ms=(time.time() - start_time) * 1000, + details={"uptime_seconds": uptime} + ) + + except Exception as e: + return HealthCheck( + name="basic", + status=HealthStatus.UNHEALTHY, + message=f"Basic health check failed: {e}", + duration_ms=(time.time() - start_time) * 1000 + ) + + async def check_filesystem_health(self) -> HealthCheck: + """Check filesystem access for media paths""" + start_time = time.time() + + try: + from config.settings import config + + accessible_paths = 0 + total_paths = len(config.tv_paths) + len(config.movie_paths) + issues = [] + + # Check TV paths + for path in config.tv_paths: + try: + if path.exists() and path.is_dir(): + # Try to read directory + list(path.iterdir()) + accessible_paths += 1 + else: + issues.append(f"TV path not accessible: {path}") + except PermissionError: + issues.append(f"TV path permission denied: {path}") + except Exception as e: + issues.append(f"TV path error {path}: {e}") + + # Check movie paths + for path in config.movie_paths: + try: + if path.exists() and path.is_dir(): + list(path.iterdir()) + accessible_paths += 1 + else: + issues.append(f"Movie path not accessible: {path}") + except PermissionError: + issues.append(f"Movie path permission denied: {path}") + except Exception as e: + issues.append(f"Movie path error {path}: {e}") + + # Determine status + if accessible_paths == total_paths: + status = HealthStatus.HEALTHY + message = f"All {total_paths} media paths accessible" + elif accessible_paths > 0: + status = HealthStatus.DEGRADED + message = f"{accessible_paths}/{total_paths} media paths accessible" + else: + status = HealthStatus.UNHEALTHY + message = "No media paths accessible" + + return HealthCheck( + name="filesystem", + status=status, + message=message, + duration_ms=(time.time() - start_time) * 1000, + details={ + "accessible_paths": accessible_paths, + "total_paths": total_paths, + "issues": issues[:5] # Limit to first 5 issues + } + ) + + except Exception as e: + return HealthCheck( + name="filesystem", + status=HealthStatus.UNHEALTHY, + message=f"Filesystem check failed: {e}", + duration_ms=(time.time() - start_time) * 1000 + ) + + async def check_database_health(self) -> HealthCheck: + """Check database connectivity and performance""" + start_time = time.time() + + try: + import sqlite3 + from config.settings import config + + # Test local database + db_path = config.db_path + + def test_db(): + with sqlite3.connect(str(db_path), timeout=5) as conn: + cursor = conn.cursor() + cursor.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='table'") + table_count = cursor.fetchone()[0] + return table_count + + # Run database test + table_count = await asyncio.get_event_loop().run_in_executor(None, test_db) + + duration = (time.time() - start_time) * 1000 + + if duration < 100: # < 100ms is good + status = HealthStatus.HEALTHY + message = f"Database responsive ({duration:.1f}ms, {table_count} tables)" + elif duration < 1000: # < 1s is acceptable + status = HealthStatus.DEGRADED + message = f"Database slow ({duration:.1f}ms, {table_count} tables)" + else: + status = HealthStatus.UNHEALTHY + message = f"Database very slow ({duration:.1f}ms)" + + return HealthCheck( + name="database", + status=status, + message=message, + duration_ms=duration, + details={ + "db_path": str(db_path), + "table_count": table_count, + "response_time_category": "fast" if duration < 100 else "slow" if duration < 1000 else "very_slow" + } + ) + + except Exception as e: + return HealthCheck( + name="database", + status=HealthStatus.UNHEALTHY, + message=f"Database check failed: {e}", + duration_ms=(time.time() - start_time) * 1000, + details={"error": str(e)} + ) + + async def check_external_apis_health(self) -> HealthCheck: + """Check external API connectivity""" + start_time = time.time() + + try: + import aiohttp + from config.settings import config + + api_results = [] + apis_tested = 0 + apis_healthy = 0 + + timeout = aiohttp.ClientTimeout(total=5) + + async with aiohttp.ClientSession(timeout=timeout) as session: + # Test Radarr API if configured + if hasattr(config, 'radarr_url') and config.radarr_url: + apis_tested += 1 + try: + test_url = f"{config.radarr_url.rstrip('/')}/api/v3/health" + async with session.get(test_url) as response: + if response.status == 200: + apis_healthy += 1 + api_results.append({"api": "radarr", "status": "healthy"}) + else: + api_results.append({"api": "radarr", "status": f"unhealthy (HTTP {response.status})"}) + except Exception as e: + api_results.append({"api": "radarr", "status": f"error: {str(e)[:50]}"}) + + # Test Sonarr API if configured + if hasattr(config, 'sonarr_url') and config.sonarr_url: + apis_tested += 1 + try: + test_url = f"{config.sonarr_url.rstrip('/')}/api/v3/health" + async with session.get(test_url) as response: + if response.status == 200: + apis_healthy += 1 + api_results.append({"api": "sonarr", "status": "healthy"}) + else: + api_results.append({"api": "sonarr", "status": f"unhealthy (HTTP {response.status})"}) + except Exception as e: + api_results.append({"api": "sonarr", "status": f"error: {str(e)[:50]}"}) + + # Determine overall API health + if apis_tested == 0: + status = HealthStatus.HEALTHY + message = "No external APIs configured" + elif apis_healthy == apis_tested: + status = HealthStatus.HEALTHY + message = f"All {apis_tested} external APIs healthy" + elif apis_healthy > 0: + status = HealthStatus.DEGRADED + message = f"{apis_healthy}/{apis_tested} external APIs healthy" + else: + status = HealthStatus.UNHEALTHY + message = "No external APIs responding" + + return HealthCheck( + name="external_apis", + status=status, + message=message, + duration_ms=(time.time() - start_time) * 1000, + details={ + "apis_tested": apis_tested, + "apis_healthy": apis_healthy, + "api_results": api_results + } + ) + + except Exception as e: + return HealthCheck( + name="external_apis", + status=HealthStatus.UNHEALTHY, + message=f"API health check failed: {e}", + duration_ms=(time.time() - start_time) * 1000 + ) + + async def check_performance_health(self) -> HealthCheck: + """Check system performance metrics""" + start_time = time.time() + + try: + system_metrics = metrics.get_system_metrics() + processing_metrics = metrics.get_processing_metrics() + + issues = [] + warnings = [] + + # Check CPU usage + cpu_percent = system_metrics.get("cpu_percent", 0) + if cpu_percent > 90: + issues.append(f"High CPU usage: {cpu_percent:.1f}%") + elif cpu_percent > 70: + warnings.append(f"Elevated CPU usage: {cpu_percent:.1f}%") + + # Check memory usage + memory_percent = system_metrics.get("memory_percent", 0) + if memory_percent > 90: + issues.append(f"High memory usage: {memory_percent:.1f}%") + elif memory_percent > 80: + warnings.append(f"Elevated memory usage: {memory_percent:.1f}%") + + # Check disk space + if "db_disk_free" in system_metrics and system_metrics["db_disk_free"]: + free_space_gb = system_metrics["db_disk_free"] / (1024**3) + if free_space_gb < 1: + issues.append(f"Low disk space: {free_space_gb:.1f}GB free") + elif free_space_gb < 5: + warnings.append(f"Low disk space: {free_space_gb:.1f}GB free") + + # Check active operations + active_ops = system_metrics.get("active_operations", 0) + if active_ops > 10: + warnings.append(f"High concurrent operations: {active_ops}") + + # Determine status + if issues: + status = HealthStatus.UNHEALTHY + message = f"Performance issues detected: {', '.join(issues[:2])}" + elif warnings: + status = HealthStatus.DEGRADED + message = f"Performance warnings: {', '.join(warnings[:2])}" + else: + status = HealthStatus.HEALTHY + message = "System performance normal" + + return HealthCheck( + name="performance", + status=status, + message=message, + duration_ms=(time.time() - start_time) * 1000, + details={ + "cpu_percent": cpu_percent, + "memory_percent": memory_percent, + "active_operations": active_ops, + "issues": issues, + "warnings": warnings + } + ) + + except Exception as e: + return HealthCheck( + name="performance", + status=HealthStatus.DEGRADED, + message=f"Performance check failed: {e}", + duration_ms=(time.time() - start_time) * 1000 + ) + + async def get_full_health_status(self) -> OverallHealth: + """Get comprehensive health status""" + start_time = time.time() + + # Run all health checks concurrently + checks = await asyncio.gather( + self.check_basic_health(), + self.check_filesystem_health(), + self.check_database_health(), + self.check_external_apis_health(), + self.check_performance_health(), + return_exceptions=True + ) + + # Filter out any exceptions and convert to HealthCheck objects + valid_checks = [] + for check in checks: + if isinstance(check, HealthCheck): + valid_checks.append(check) + elif isinstance(check, Exception): + valid_checks.append(HealthCheck( + name="unknown", + status=HealthStatus.UNHEALTHY, + message=f"Health check exception: {check}", + duration_ms=0 + )) + + # Determine overall status + unhealthy_count = len([c for c in valid_checks if c.status == HealthStatus.UNHEALTHY]) + degraded_count = len([c for c in valid_checks if c.status == HealthStatus.DEGRADED]) + + if unhealthy_count > 0: + overall_status = HealthStatus.UNHEALTHY + elif degraded_count > 0: + overall_status = HealthStatus.DEGRADED + else: + overall_status = HealthStatus.HEALTHY + + return OverallHealth( + status=overall_status, + checks=valid_checks, + timestamp=start_time, + uptime_seconds=time.time() - self.start_time + ) + + async def get_readiness_status(self) -> Dict[str, Any]: + """Get readiness status for Kubernetes readiness probes""" + # Readiness is simpler - just check critical components + checks = await asyncio.gather( + self.check_basic_health(), + self.check_filesystem_health(), + self.check_database_health(), + return_exceptions=True + ) + + critical_failures = 0 + for check in checks: + if isinstance(check, HealthCheck) and check.status == HealthStatus.UNHEALTHY: + critical_failures += 1 + + is_ready = critical_failures == 0 + + return { + "ready": is_ready, + "timestamp": time.time(), + "critical_failures": critical_failures, + "message": "Service ready" if is_ready else f"{critical_failures} critical failures" + } + + async def get_liveness_status(self) -> Dict[str, Any]: + """Get liveness status for Kubernetes liveness probes""" + # Liveness is even simpler - just check if service is responsive + basic_check = await self.check_basic_health() + + is_alive = basic_check.status != HealthStatus.UNHEALTHY + + return { + "alive": is_alive, + "timestamp": time.time(), + "uptime_seconds": time.time() - self.start_time, + "message": basic_check.message + } + + +# Global health checker instance +health_checker = HealthChecker() \ No newline at end of file diff --git a/monitoring/logging_enhanced.py b/monitoring/logging_enhanced.py new file mode 100644 index 0000000..c16c93b --- /dev/null +++ b/monitoring/logging_enhanced.py @@ -0,0 +1,404 @@ +""" +Enhanced Logging System for NFOGuard +Provides structured logging with correlation IDs, request tracing, and monitoring integration +""" +import logging +import json +import time +import uuid +import threading +from typing import Dict, Any, Optional, List, Union +from dataclasses import dataclass, field +from contextlib import contextmanager +from datetime import datetime +import sys +import traceback + +from monitoring.metrics import metrics + + +# Thread-local storage for correlation context +_context = threading.local() + + +@dataclass +class LogContext: + """Logging context with correlation and tracing information""" + correlation_id: str + request_id: Optional[str] = None + user_id: Optional[str] = None + operation: Optional[str] = None + media_type: Optional[str] = None + media_title: Optional[str] = None + webhook_type: Optional[str] = None + processing_stage: Optional[str] = None + additional_fields: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + """Convert context to dictionary for logging""" + context = { + "correlation_id": self.correlation_id, + "timestamp": datetime.utcnow().isoformat(), + } + + # Add non-None fields + for field_name in ["request_id", "user_id", "operation", "media_type", + "media_title", "webhook_type", "processing_stage"]: + value = getattr(self, field_name) + if value is not None: + context[field_name] = value + + # Add additional fields + context.update(self.additional_fields) + + return context + + +class StructuredFormatter(logging.Formatter): + """JSON formatter for structured logging""" + + def __init__(self, include_context: bool = True): + super().__init__() + self.include_context = include_context + + def format(self, record: logging.LogRecord) -> str: + # Base log entry + log_entry = { + "timestamp": datetime.utcnow().isoformat(), + "level": record.levelname, + "logger": record.name, + "message": record.getMessage(), + "module": record.module, + "function": record.funcName, + "line": record.lineno, + } + + # Add thread information + log_entry["thread"] = { + "id": record.thread, + "name": record.threadName + } + + # Add correlation context if available + if self.include_context and hasattr(_context, 'log_context'): + log_entry["context"] = _context.log_context.to_dict() + + # Add exception information if present + if record.exc_info: + log_entry["exception"] = { + "type": record.exc_info[0].__name__, + "message": str(record.exc_info[1]), + "traceback": traceback.format_exception(*record.exc_info) + } + + # Add any extra fields passed to log call + if hasattr(record, 'extra_fields') and record.extra_fields: + log_entry["extra"] = record.extra_fields + + # Add performance metrics if available + if hasattr(record, 'performance_data') and record.performance_data: + log_entry["performance"] = record.performance_data + + return json.dumps(log_entry, default=str, ensure_ascii=False) + + +class CorrelationIDFilter(logging.Filter): + """Filter to add correlation ID to log records""" + + def filter(self, record: logging.LogRecord) -> bool: + # Add correlation ID to record if available + if hasattr(_context, 'log_context'): + record.correlation_id = _context.log_context.correlation_id + else: + record.correlation_id = "no-correlation" + + return True + + +class EnhancedLogger: + """Enhanced logger with correlation IDs and structured logging""" + + def __init__(self, name: str): + self.logger = logging.getLogger(name) + self.name = name + + # Track log events for metrics + self._log_counts = {"debug": 0, "info": 0, "warning": 0, "error": 0, "critical": 0} + + def _log_with_context(self, level: int, message: str, extra_fields: Optional[Dict[str, Any]] = None, + performance_data: Optional[Dict[str, Any]] = None, **kwargs): + """Log with enhanced context and metrics tracking""" + + # Track log counts for metrics + level_name = logging.getLevelName(level).lower() + if level_name in self._log_counts: + self._log_counts[level_name] += 1 + metrics.increment_counter(f"log_messages", 1, {"level": level_name, "logger": self.name}) + + # Create log record with extra data + extra = {} + if extra_fields: + extra['extra_fields'] = extra_fields + if performance_data: + extra['performance_data'] = performance_data + + # Log the message + self.logger.log(level, message, extra=extra, **kwargs) + + # Track errors in metrics + if level >= logging.ERROR: + metrics.record_error("logging_error", message, self.name) + + def debug(self, message: str, **kwargs): + """Log debug message""" + self._log_with_context(logging.DEBUG, message, **kwargs) + + def info(self, message: str, **kwargs): + """Log info message""" + self._log_with_context(logging.INFO, message, **kwargs) + + def warning(self, message: str, **kwargs): + """Log warning message""" + self._log_with_context(logging.WARNING, message, **kwargs) + + def error(self, message: str, **kwargs): + """Log error message""" + self._log_with_context(logging.ERROR, message, **kwargs) + + def critical(self, message: str, **kwargs): + """Log critical message""" + self._log_with_context(logging.CRITICAL, message, **kwargs) + + def exception(self, message: str, **kwargs): + """Log exception with traceback""" + kwargs['exc_info'] = True + self._log_with_context(logging.ERROR, message, **kwargs) + + def log_operation_start(self, operation: str, **context_fields): + """Log the start of an operation""" + self.info(f"Starting operation: {operation}", + extra_fields={"operation_event": "start", "operation": operation, **context_fields}) + + def log_operation_end(self, operation: str, success: bool = True, duration: Optional[float] = None, **context_fields): + """Log the end of an operation""" + outcome = "success" if success else "failure" + extra = {"operation_event": "end", "operation": operation, "outcome": outcome, **context_fields} + + if duration is not None: + extra["duration_seconds"] = duration + + level = logging.INFO if success else logging.ERROR + self._log_with_context(level, f"Operation {outcome}: {operation}", extra_fields=extra) + + def log_webhook_received(self, webhook_type: str, payload_size: int, **context_fields): + """Log webhook reception""" + self.info(f"Webhook received: {webhook_type}", + extra_fields={ + "event_type": "webhook_received", + "webhook_type": webhook_type, + "payload_size_bytes": payload_size, + **context_fields + }) + + def log_nfo_operation(self, operation: str, file_path: str, success: bool = True, **context_fields): + """Log NFO file operations""" + outcome = "success" if success else "failure" + level = logging.INFO if success else logging.ERROR + + self._log_with_context(level, f"NFO {operation} {outcome}: {file_path}", + extra_fields={ + "event_type": "nfo_operation", + "nfo_operation": operation, + "file_path": file_path, + "outcome": outcome, + **context_fields + }) + + def log_performance_metrics(self, operation: str, duration: float, success: bool = True, **metrics_data): + """Log performance metrics""" + self.debug(f"Performance: {operation} took {duration:.3f}s", + performance_data={ + "operation": operation, + "duration_seconds": duration, + "success": success, + **metrics_data + }) + + def get_log_stats(self) -> Dict[str, int]: + """Get logging statistics""" + return self._log_counts.copy() + + +def setup_enhanced_logging( + log_level: str = "INFO", + structured: bool = True, + log_file: Optional[str] = None, + max_bytes: int = 10 * 1024 * 1024, # 10MB + backup_count: int = 5 +) -> None: + """Setup enhanced logging configuration""" + + # Configure root logger + root_logger = logging.getLogger() + root_logger.setLevel(getattr(logging, log_level.upper())) + + # Clear existing handlers + root_logger.handlers.clear() + + # Create console handler + console_handler = logging.StreamHandler(sys.stdout) + + if structured: + # Use structured JSON formatter + formatter = StructuredFormatter(include_context=True) + else: + # Use simple text formatter with correlation ID + formatter = logging.Formatter( + '%(asctime)s [%(correlation_id)s] %(levelname)s %(name)s: %(message)s' + ) + + console_handler.setFormatter(formatter) + console_handler.addFilter(CorrelationIDFilter()) + root_logger.addHandler(console_handler) + + # Add file handler if specified + if log_file: + from logging.handlers import RotatingFileHandler + + file_handler = RotatingFileHandler( + log_file, maxBytes=max_bytes, backupCount=backup_count + ) + file_handler.setFormatter(formatter) + file_handler.addFilter(CorrelationIDFilter()) + root_logger.addHandler(file_handler) + + # Reduce noise from external libraries + logging.getLogger("urllib3").setLevel(logging.WARNING) + logging.getLogger("requests").setLevel(logging.WARNING) + logging.getLogger("aiohttp").setLevel(logging.WARNING) + + +def get_enhanced_logger(name: str) -> EnhancedLogger: + """Get enhanced logger instance""" + return EnhancedLogger(name) + + +def set_log_context( + correlation_id: Optional[str] = None, + request_id: Optional[str] = None, + operation: Optional[str] = None, + **kwargs +) -> LogContext: + """Set logging context for current thread""" + + if correlation_id is None: + correlation_id = str(uuid.uuid4()) + + context = LogContext( + correlation_id=correlation_id, + request_id=request_id, + operation=operation, + **kwargs + ) + + _context.log_context = context + return context + + +def get_log_context() -> Optional[LogContext]: + """Get current logging context""" + return getattr(_context, 'log_context', None) + + +def clear_log_context(): + """Clear logging context for current thread""" + if hasattr(_context, 'log_context'): + delattr(_context, 'log_context') + + +@contextmanager +def log_context(correlation_id: Optional[str] = None, **context_fields): + """Context manager for scoped logging context""" + original_context = get_log_context() + + try: + # Set new context + new_context = set_log_context(correlation_id=correlation_id, **context_fields) + yield new_context + finally: + # Restore original context + if original_context: + _context.log_context = original_context + else: + clear_log_context() + + +@contextmanager +def log_operation(operation: str, logger: Optional[EnhancedLogger] = None, **context_fields): + """Context manager for logging operation start/end with timing""" + if logger is None: + logger = get_enhanced_logger(__name__) + + start_time = time.time() + success = True + + # Update context with operation + current_context = get_log_context() + if current_context: + current_context.operation = operation + current_context.processing_stage = "executing" + + logger.log_operation_start(operation, **context_fields) + + try: + yield + except Exception as e: + success = False + logger.exception(f"Operation failed: {operation}", + extra_fields={"operation": operation, "error": str(e), **context_fields}) + raise + finally: + duration = time.time() - start_time + logger.log_operation_end(operation, success, duration, **context_fields) + + # Update metrics + metrics.record_operation_duration(operation, duration, success) + + +def trace_request(request_id: Optional[str] = None, **context_fields): + """Decorator/context manager for request tracing""" + def decorator(func): + def wrapper(*args, **kwargs): + correlation_id = str(uuid.uuid4()) + req_id = request_id or f"req_{int(time.time())}" + + with log_context(correlation_id=correlation_id, request_id=req_id, **context_fields): + return func(*args, **kwargs) + return wrapper + + # Can be used as context manager or decorator + if request_id is None and len(context_fields) == 1 and callable(list(context_fields.values())[0]): + # Used as decorator without parentheses + func = list(context_fields.values())[0] + return decorator(func) + else: + # Used as decorator with parameters or context manager + return decorator + + +# Module-level logger for this module +logger = get_enhanced_logger(__name__) + + +def get_logging_stats() -> Dict[str, Any]: + """Get comprehensive logging statistics""" + # Collect stats from all enhanced loggers + total_stats = {"debug": 0, "info": 0, "warning": 0, "error": 0, "critical": 0} + + # This is a simplified version - in practice you'd track all logger instances + return { + "total_log_messages": sum(total_stats.values()), + "by_level": total_stats, + "structured_logging_enabled": True, + "correlation_tracking_enabled": True + } \ No newline at end of file diff --git a/monitoring/metrics.py b/monitoring/metrics.py new file mode 100644 index 0000000..b4bb70f --- /dev/null +++ b/monitoring/metrics.py @@ -0,0 +1,354 @@ +""" +Metrics Collection System for NFOGuard +Provides performance monitoring, counters, and operational metrics +""" +import time +import psutil +import threading +from datetime import datetime, timedelta +from typing import Dict, Any, List, Optional +from dataclasses import dataclass, field +from collections import defaultdict, deque +from contextlib import contextmanager +import asyncio + + +@dataclass +class MetricValue: + """Individual metric value with timestamp""" + value: float + timestamp: float = field(default_factory=time.time) + labels: Dict[str, str] = field(default_factory=dict) + + +@dataclass +class TimeSeriesMetric: + """Time series metric with historical data""" + name: str + values: deque = field(default_factory=lambda: deque(maxlen=1000)) + total: float = 0.0 + count: int = 0 + + def add_value(self, value: float, labels: Optional[Dict[str, str]] = None): + """Add a new metric value""" + metric_value = MetricValue(value, labels=labels or {}) + self.values.append(metric_value) + self.total += value + self.count += 1 + + def get_average(self, window_seconds: int = 300) -> float: + """Get average value over time window""" + cutoff_time = time.time() - window_seconds + recent_values = [v.value for v in self.values if v.timestamp > cutoff_time] + return sum(recent_values) / len(recent_values) if recent_values else 0.0 + + def get_rate_per_minute(self, window_seconds: int = 300) -> float: + """Get rate per minute over time window""" + cutoff_time = time.time() - window_seconds + recent_count = len([v for v in self.values if v.timestamp > cutoff_time]) + return (recent_count / window_seconds) * 60 if window_seconds > 0 else 0.0 + + +class MetricsCollector: + """Central metrics collection system""" + + def __init__(self): + self._metrics: Dict[str, TimeSeriesMetric] = {} + self._counters: Dict[str, int] = defaultdict(int) + self._gauges: Dict[str, float] = {} + self._histograms: Dict[str, List[float]] = defaultdict(list) + self._start_time = time.time() + self._lock = threading.RLock() + + # Processing metrics + self._active_operations = 0 + self._operation_durations = deque(maxlen=1000) + + # Error tracking + self._error_counts = defaultdict(int) + self._last_errors = deque(maxlen=100) + + # System metrics + self._system_stats_cache = {} + self._system_stats_last_update = 0 + self._system_stats_cache_ttl = 30 # 30 seconds + + def increment_counter(self, name: str, value: int = 1, labels: Optional[Dict[str, str]] = None): + """Increment a counter metric""" + with self._lock: + full_name = self._build_metric_name(name, labels) + self._counters[full_name] += value + + # Also track in time series for rate calculations + if name not in self._metrics: + self._metrics[name] = TimeSeriesMetric(name) + self._metrics[name].add_value(value, labels) + + def set_gauge(self, name: str, value: float, labels: Optional[Dict[str, str]] = None): + """Set a gauge metric value""" + with self._lock: + full_name = self._build_metric_name(name, labels) + self._gauges[full_name] = value + + def record_histogram(self, name: str, value: float, labels: Optional[Dict[str, str]] = None): + """Record a histogram value""" + with self._lock: + full_name = self._build_metric_name(name, labels) + self._histograms[full_name].append(value) + + # Keep only recent values (last 1000) + if len(self._histograms[full_name]) > 1000: + self._histograms[full_name] = self._histograms[full_name][-1000:] + + # Also track in time series + if name not in self._metrics: + self._metrics[name] = TimeSeriesMetric(name) + self._metrics[name].add_value(value, labels) + + def record_operation_duration(self, operation: str, duration: float, success: bool = True): + """Record operation duration and outcome""" + with self._lock: + # Record duration + self.record_histogram(f"operation_duration_{operation}", duration) + + # Record outcome + outcome = "success" if success else "error" + self.increment_counter(f"operation_total", 1, {"operation": operation, "outcome": outcome}) + + # Track active operations + if operation.endswith("_start"): + self._active_operations += 1 + elif operation.endswith("_end"): + self._active_operations = max(0, self._active_operations - 1) + + def record_error(self, error_type: str, error_message: str, operation: Optional[str] = None): + """Record an error occurrence""" + with self._lock: + self._error_counts[error_type] += 1 + + error_info = { + "type": error_type, + "message": error_message, + "operation": operation, + "timestamp": time.time() + } + self._last_errors.append(error_info) + + # Increment error counter + labels = {"error_type": error_type} + if operation: + labels["operation"] = operation + self.increment_counter("errors_total", 1, labels) + + @contextmanager + def operation_timer(self, operation: str): + """Context manager for timing operations""" + start_time = time.time() + success = True + + try: + self.record_operation_duration(f"{operation}_start", 0) + yield + except Exception as e: + success = False + self.record_error("operation_error", str(e), operation) + raise + finally: + duration = time.time() - start_time + self.record_operation_duration(operation, duration, success) + self.record_operation_duration(f"{operation}_end", 0) + + def get_system_metrics(self) -> Dict[str, Any]: + """Get current system resource metrics""" + now = time.time() + + # Use cached values if recent + if (now - self._system_stats_last_update) < self._system_stats_cache_ttl: + return self._system_stats_cache + + try: + # CPU metrics + cpu_percent = psutil.cpu_percent(interval=0.1) + cpu_count = psutil.cpu_count() + + # Memory metrics + memory = psutil.virtual_memory() + + # Disk metrics for database path + try: + from config.settings import config + db_disk = psutil.disk_usage(str(config.db_path.parent)) + except: + db_disk = None + + # Process metrics + process = psutil.Process() + process_memory = process.memory_info() + + self._system_stats_cache = { + "cpu_percent": cpu_percent, + "cpu_count": cpu_count, + "memory_total": memory.total, + "memory_available": memory.available, + "memory_percent": memory.percent, + "process_memory_rss": process_memory.rss, + "process_memory_vms": process_memory.vms, + "db_disk_free": db_disk.free if db_disk else None, + "db_disk_total": db_disk.total if db_disk else None, + "active_operations": self._active_operations, + "uptime_seconds": now - self._start_time + } + + self._system_stats_last_update = now + + except Exception as e: + # Return basic metrics if detailed collection fails + self._system_stats_cache = { + "uptime_seconds": now - self._start_time, + "active_operations": self._active_operations, + "error": str(e) + } + + return self._system_stats_cache + + def get_processing_metrics(self) -> Dict[str, Any]: + """Get processing-related metrics""" + with self._lock: + # Calculate rates and averages + webhook_rate = self._metrics.get("webhooks_received", TimeSeriesMetric("webhooks_received")).get_rate_per_minute() + nfo_rate = self._metrics.get("nfo_created", TimeSeriesMetric("nfo_created")).get_rate_per_minute() + + avg_processing_time = 0.0 + if "processing_duration" in self._metrics: + avg_processing_time = self._metrics["processing_duration"].get_average() + + return { + "webhooks_received_per_minute": webhook_rate, + "nfo_files_created_per_minute": nfo_rate, + "average_processing_time_seconds": avg_processing_time, + "active_operations": self._active_operations, + "total_webhooks": self._counters.get("webhooks_received", 0), + "total_nfo_created": self._counters.get("nfo_created", 0), + "total_errors": sum(self._error_counts.values()) + } + + def get_error_metrics(self) -> Dict[str, Any]: + """Get error-related metrics""" + with self._lock: + recent_errors = [] + cutoff_time = time.time() - 3600 # Last hour + + for error in self._last_errors: + if error["timestamp"] > cutoff_time: + recent_errors.append({ + "type": error["type"], + "message": error["message"][:100], # Truncate long messages + "operation": error["operation"], + "timestamp": error["timestamp"] + }) + + return { + "error_counts_by_type": dict(self._error_counts), + "recent_errors": recent_errors[-10:], # Last 10 errors + "total_errors": sum(self._error_counts.values()), + "error_rate_per_minute": len([e for e in self._last_errors if e["timestamp"] > time.time() - 300]) / 5 + } + + def get_prometheus_metrics(self) -> str: + """Generate Prometheus-compatible metrics format""" + lines = [] + + # Add help and type information + lines.append("# HELP nfoguard_webhooks_total Total number of webhooks received") + lines.append("# TYPE nfoguard_webhooks_total counter") + + with self._lock: + # Counters + for name, value in self._counters.items(): + metric_name = f"nfoguard_{name.replace('-', '_')}" + lines.append(f"{metric_name} {value}") + + # Gauges + lines.append("# HELP nfoguard_active_operations Current number of active operations") + lines.append("# TYPE nfoguard_active_operations gauge") + lines.append(f"nfoguard_active_operations {self._active_operations}") + + # System metrics + system_metrics = self.get_system_metrics() + for key, value in system_metrics.items(): + if isinstance(value, (int, float)) and value is not None: + metric_name = f"nfoguard_system_{key}" + lines.append(f"{metric_name} {value}") + + return "\n".join(lines) + + def get_all_metrics(self) -> Dict[str, Any]: + """Get all metrics in a structured format""" + return { + "system": self.get_system_metrics(), + "processing": self.get_processing_metrics(), + "errors": self.get_error_metrics(), + "timestamp": time.time(), + "uptime_seconds": time.time() - self._start_time + } + + def reset_metrics(self, metric_types: Optional[List[str]] = None): + """Reset specific metric types or all metrics""" + with self._lock: + if not metric_types or "counters" in metric_types: + self._counters.clear() + + if not metric_types or "histograms" in metric_types: + self._histograms.clear() + + if not metric_types or "errors" in metric_types: + self._error_counts.clear() + self._last_errors.clear() + + if not metric_types or "timeseries" in metric_types: + self._metrics.clear() + + def _build_metric_name(self, name: str, labels: Optional[Dict[str, str]]) -> str: + """Build metric name with labels""" + if not labels: + return name + + label_str = ",".join(f"{k}={v}" for k, v in sorted(labels.items())) + return f"{name}{{{label_str}}}" + + +# Global metrics collector instance +metrics = MetricsCollector() + + +# Convenience functions for common operations +def track_webhook_received(webhook_type: str): + """Track webhook received""" + metrics.increment_counter("webhooks_received", 1, {"type": webhook_type}) + + +def track_nfo_created(media_type: str, success: bool = True): + """Track NFO file creation""" + outcome = "success" if success else "error" + metrics.increment_counter("nfo_created", 1, {"media_type": media_type, "outcome": outcome}) + + +def track_api_call(api_name: str, duration: float, success: bool = True): + """Track external API call""" + metrics.record_histogram(f"api_call_duration", duration, {"api": api_name}) + outcome = "success" if success else "error" + metrics.increment_counter("api_calls_total", 1, {"api": api_name, "outcome": outcome}) + + +def track_database_operation(operation: str, duration: float, success: bool = True): + """Track database operation""" + metrics.record_histogram("database_operation_duration", duration, {"operation": operation}) + outcome = "success" if success else "error" + metrics.increment_counter("database_operations_total", 1, {"operation": operation, "outcome": outcome}) + + +def track_file_operation(operation: str, duration: float, success: bool = True): + """Track file system operation""" + metrics.record_histogram("file_operation_duration", duration, {"operation": operation}) + outcome = "success" if success else "error" + metrics.increment_counter("file_operations_total", 1, {"operation": operation, "outcome": outcome}) \ No newline at end of file diff --git a/monitoring/performance.py b/monitoring/performance.py new file mode 100644 index 0000000..91b729a --- /dev/null +++ b/monitoring/performance.py @@ -0,0 +1,413 @@ +""" +Performance Monitoring and Profiling for NFOGuard +Provides detailed performance analysis and optimization insights +""" +import time +import asyncio +import threading +import functools +from typing import Dict, Any, List, Optional, Callable, TypeVar, Union +from dataclasses import dataclass, field +from collections import defaultdict, deque +from contextlib import asynccontextmanager, contextmanager +import traceback +import sys + +from monitoring.metrics import metrics + + +T = TypeVar('T') + + +@dataclass +class PerformanceProfile: + """Performance profile for an operation""" + operation_name: str + total_calls: int = 0 + total_duration: float = 0.0 + min_duration: float = float('inf') + max_duration: float = 0.0 + recent_durations: deque = field(default_factory=lambda: deque(maxlen=100)) + error_count: int = 0 + concurrent_calls: int = 0 + + def add_measurement(self, duration: float, success: bool = True): + """Add a performance measurement""" + self.total_calls += 1 + self.total_duration += duration + self.min_duration = min(self.min_duration, duration) + self.max_duration = max(self.max_duration, duration) + self.recent_durations.append(duration) + + if not success: + self.error_count += 1 + + def get_average_duration(self) -> float: + """Get average duration across all calls""" + return self.total_duration / self.total_calls if self.total_calls > 0 else 0.0 + + def get_recent_average(self, window: int = 50) -> float: + """Get average of recent calls""" + recent = list(self.recent_durations)[-window:] + return sum(recent) / len(recent) if recent else 0.0 + + def get_percentiles(self) -> Dict[str, float]: + """Get duration percentiles for recent calls""" + recent = sorted(list(self.recent_durations)) + if not recent: + return {"p50": 0, "p95": 0, "p99": 0} + + length = len(recent) + return { + "p50": recent[int(length * 0.5)] if length > 0 else 0, + "p95": recent[int(length * 0.95)] if length > 0 else 0, + "p99": recent[int(length * 0.99)] if length > 0 else 0 + } + + def to_dict(self) -> Dict[str, Any]: + """Convert to dictionary for API responses""" + percentiles = self.get_percentiles() + + return { + "operation_name": self.operation_name, + "total_calls": self.total_calls, + "error_count": self.error_count, + "error_rate": self.error_count / self.total_calls if self.total_calls > 0 else 0, + "concurrent_calls": self.concurrent_calls, + "duration_stats": { + "average": round(self.get_average_duration(), 4), + "recent_average": round(self.get_recent_average(), 4), + "min": round(self.min_duration if self.min_duration != float('inf') else 0, 4), + "max": round(self.max_duration, 4), + "p50": round(percentiles["p50"], 4), + "p95": round(percentiles["p95"], 4), + "p99": round(percentiles["p99"], 4) + }, + "performance_rating": self._get_performance_rating() + } + + def _get_performance_rating(self) -> str: + """Get performance rating based on metrics""" + avg_duration = self.get_recent_average() + error_rate = self.error_count / self.total_calls if self.total_calls > 0 else 0 + + if error_rate > 0.1: # >10% error rate + return "poor" + elif avg_duration > 5.0: # >5 seconds average + return "slow" + elif avg_duration > 1.0: # >1 second average + return "acceptable" + else: + return "excellent" + + +class PerformanceMonitor: + """Advanced performance monitoring system""" + + def __init__(self): + self._profiles: Dict[str, PerformanceProfile] = {} + self._active_operations: Dict[str, float] = {} # operation_id -> start_time + self._lock = threading.RLock() + + # Slow operation tracking + self._slow_operation_threshold = 1.0 # 1 second + self._slow_operations = deque(maxlen=100) + + # Memory monitoring + self._memory_samples = deque(maxlen=1000) + self._memory_monitoring_enabled = True + + # Async operation tracking + self._async_tasks = {} + self._task_counter = 0 + + def get_profile(self, operation_name: str) -> PerformanceProfile: + """Get or create performance profile for operation""" + with self._lock: + if operation_name not in self._profiles: + self._profiles[operation_name] = PerformanceProfile(operation_name) + return self._profiles[operation_name] + + @contextmanager + def monitor_operation(self, operation_name: str, **kwargs): + """Context manager for monitoring synchronous operations""" + start_time = time.time() + operation_id = f"{operation_name}_{id(threading.current_thread())}_{time.time()}" + success = True + + profile = self.get_profile(operation_name) + + with self._lock: + profile.concurrent_calls += 1 + self._active_operations[operation_id] = start_time + + try: + yield + except Exception as e: + success = False + metrics.record_error("performance_monitor", str(e), operation_name) + raise + finally: + end_time = time.time() + duration = end_time - start_time + + with self._lock: + profile.concurrent_calls = max(0, profile.concurrent_calls - 1) + self._active_operations.pop(operation_id, None) + + # Record measurement + profile.add_measurement(duration, success) + + # Track slow operations + if duration > self._slow_operation_threshold: + self._slow_operations.append({ + "operation": operation_name, + "duration": duration, + "timestamp": end_time, + "success": success, + "metadata": kwargs + }) + + # Update metrics + metrics.record_histogram(f"operation_duration", duration, {"operation": operation_name}) + if not success: + metrics.increment_counter("operation_errors", 1, {"operation": operation_name}) + + @asynccontextmanager + async def monitor_async_operation(self, operation_name: str, **kwargs): + """Context manager for monitoring asynchronous operations""" + start_time = time.time() + task_id = f"{operation_name}_{self._task_counter}" + self._task_counter += 1 + success = True + + profile = self.get_profile(operation_name) + + with self._lock: + profile.concurrent_calls += 1 + self._async_tasks[task_id] = { + "operation": operation_name, + "start_time": start_time, + "metadata": kwargs + } + + try: + yield + except Exception as e: + success = False + metrics.record_error("async_performance_monitor", str(e), operation_name) + raise + finally: + end_time = time.time() + duration = end_time - start_time + + with self._lock: + profile.concurrent_calls = max(0, profile.concurrent_calls - 1) + self._async_tasks.pop(task_id, None) + + # Record measurement + profile.add_measurement(duration, success) + + # Track slow operations + if duration > self._slow_operation_threshold: + self._slow_operations.append({ + "operation": operation_name, + "duration": duration, + "timestamp": end_time, + "success": success, + "async": True, + "metadata": kwargs + }) + + # Update metrics + metrics.record_histogram(f"async_operation_duration", duration, {"operation": operation_name}) + if not success: + metrics.increment_counter("async_operation_errors", 1, {"operation": operation_name}) + + def monitor_function(self, operation_name: Optional[str] = None): + """Decorator for monitoring function performance""" + def decorator(func: Callable[..., T]) -> Callable[..., T]: + name = operation_name or f"{func.__module__}.{func.__name__}" + + if asyncio.iscoroutinefunction(func): + @functools.wraps(func) + async def async_wrapper(*args, **kwargs): + async with self.monitor_async_operation(name): + return await func(*args, **kwargs) + return async_wrapper + else: + @functools.wraps(func) + def sync_wrapper(*args, **kwargs): + with self.monitor_operation(name): + return func(*args, **kwargs) + return sync_wrapper + + return decorator + + def get_performance_summary(self) -> Dict[str, Any]: + """Get comprehensive performance summary""" + with self._lock: + # Get top operations by various metrics + profiles = list(self._profiles.values()) + + # Sort by total calls + most_called = sorted(profiles, key=lambda p: p.total_calls, reverse=True)[:10] + + # Sort by average duration + slowest_avg = sorted(profiles, key=lambda p: p.get_average_duration(), reverse=True)[:10] + + # Sort by recent average + slowest_recent = sorted(profiles, key=lambda p: p.get_recent_average(), reverse=True)[:10] + + # Sort by error rate + highest_errors = sorted( + [p for p in profiles if p.total_calls > 0], + key=lambda p: p.error_count / p.total_calls, + reverse=True + )[:10] + + # Get active operations count + total_active = sum(p.concurrent_calls for p in profiles) + + # Get slow operations + recent_slow = list(self._slow_operations)[-20:] # Last 20 slow operations + + return { + "overview": { + "total_operations_tracked": len(profiles), + "total_active_operations": total_active, + "slow_operation_threshold_seconds": self._slow_operation_threshold, + "total_slow_operations": len(self._slow_operations) + }, + "top_operations": { + "most_called": [p.to_dict() for p in most_called], + "slowest_average": [p.to_dict() for p in slowest_avg], + "slowest_recent": [p.to_dict() for p in slowest_recent], + "highest_error_rate": [p.to_dict() for p in highest_errors] + }, + "recent_slow_operations": recent_slow, + "performance_insights": self._generate_performance_insights(profiles) + } + + def get_operation_detail(self, operation_name: str) -> Optional[Dict[str, Any]]: + """Get detailed performance data for specific operation""" + with self._lock: + if operation_name not in self._profiles: + return None + + profile = self._profiles[operation_name] + + # Get related slow operations + related_slow = [ + op for op in self._slow_operations + if op["operation"] == operation_name + ] + + detail = profile.to_dict() + detail.update({ + "detailed_stats": { + "total_duration": round(profile.total_duration, 4), + "recent_durations": list(profile.recent_durations)[-20:], # Last 20 calls + "slow_operations_count": len(related_slow), + "recent_slow_operations": related_slow[-10:] # Last 10 slow calls + }, + "recommendations": self._get_operation_recommendations(profile) + }) + + return detail + + def _generate_performance_insights(self, profiles: List[PerformanceProfile]) -> List[str]: + """Generate performance optimization insights""" + insights = [] + + # Check for very slow operations + very_slow = [p for p in profiles if p.get_recent_average() > 5.0] + if very_slow: + insights.append(f"Found {len(very_slow)} operations with >5s average duration - consider optimization") + + # Check for high error rates + high_error_rate = [p for p in profiles if p.total_calls > 10 and (p.error_count / p.total_calls) > 0.1] + if high_error_rate: + insights.append(f"Found {len(high_error_rate)} operations with >10% error rate - investigate failures") + + # Check for high concurrency + high_concurrency = [p for p in profiles if p.concurrent_calls > 5] + if high_concurrency: + insights.append(f"Found {len(high_concurrency)} operations with high concurrency - may need rate limiting") + + # Check total active operations + total_active = sum(p.concurrent_calls for p in profiles) + if total_active > 20: + insights.append(f"High total concurrent operations ({total_active}) - system may be under load") + + # Performance trends + recent_slow_count = len([op for op in self._slow_operations if op["timestamp"] > time.time() - 300]) + if recent_slow_count > 10: + insights.append(f"Many slow operations recently ({recent_slow_count} in last 5 minutes)") + + if not insights: + insights.append("No significant performance issues detected") + + return insights + + def _get_operation_recommendations(self, profile: PerformanceProfile) -> List[str]: + """Get recommendations for optimizing specific operation""" + recommendations = [] + + avg_duration = profile.get_recent_average() + error_rate = profile.error_count / profile.total_calls if profile.total_calls > 0 else 0 + + if avg_duration > 5.0: + recommendations.append("Consider breaking down this operation into smaller parts") + recommendations.append("Review database queries and file I/O for optimization opportunities") + elif avg_duration > 1.0: + recommendations.append("Monitor for potential optimization opportunities") + + if error_rate > 0.1: + recommendations.append("High error rate - investigate common failure causes") + recommendations.append("Consider adding retry logic or better error handling") + + if profile.concurrent_calls > 5: + recommendations.append("High concurrency - consider adding rate limiting") + recommendations.append("Review resource usage and potential bottlenecks") + + percentiles = profile.get_percentiles() + if percentiles["p99"] > percentiles["p50"] * 3: + recommendations.append("High latency variance - investigate outlier causes") + + if not recommendations: + recommendations.append("Performance appears optimal for this operation") + + return recommendations + + def set_slow_operation_threshold(self, threshold_seconds: float): + """Set threshold for what constitutes a slow operation""" + with self._lock: + self._slow_operation_threshold = threshold_seconds + + def clear_profiles(self, operation_names: Optional[List[str]] = None): + """Clear performance profiles for specific operations or all""" + with self._lock: + if operation_names: + for name in operation_names: + self._profiles.pop(name, None) + else: + self._profiles.clear() + self._slow_operations.clear() + + +# Global performance monitor instance +performance_monitor = PerformanceMonitor() + +# Decorator shortcuts +def monitor_performance(operation_name: Optional[str] = None): + """Shortcut decorator for performance monitoring""" + return performance_monitor.monitor_function(operation_name) + +def monitor_sync_operation(operation_name: str, **kwargs): + """Shortcut for synchronous operation monitoring""" + return performance_monitor.monitor_operation(operation_name, **kwargs) + +def monitor_async_operation(operation_name: str, **kwargs): + """Shortcut for asynchronous operation monitoring""" + return performance_monitor.monitor_async_operation(operation_name, **kwargs) \ No newline at end of file