"""
Performance Monitoring System
Comprehensive performance monitoring and metrics collection system for the
MCP Academic RAG Server. Provides real-time performance tracking, metrics
aggregation, and alerting capabilities.
"""
import asyncio
import json
import logging
import threading
import time
from collections import defaultdict, deque
from contextlib import contextmanager
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta
from enum import Enum
from functools import wraps
from typing import Any, Awaitable, Callable, Dict, Iterator, List, Optional, ParamSpec, TypeVar, Union, cast
import psutil
P = ParamSpec("P")
T = TypeVar("T")
[docs]
class MetricType(Enum):
"""Types of metrics that can be collected"""
COUNTER = "counter"
GAUGE = "gauge"
HISTOGRAM = "histogram"
TIMER = "timer"
RATE = "rate"
[docs]
class AlertLevel(Enum):
"""Alert severity levels"""
INFO = "info"
WARNING = "warning"
ERROR = "error"
CRITICAL = "critical"
[docs]
@dataclass
class SystemMetrics:
"""System-level performance metrics"""
cpu_percent: float
memory_percent: float
memory_used_mb: float
memory_available_mb: float
disk_usage_percent: float
disk_free_gb: float
network_bytes_sent: int
network_bytes_received: int
open_files: int
active_threads: int
timestamp: datetime
[docs]
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary"""
data = asdict(self)
data["timestamp"] = self.timestamp.isoformat()
return data
[docs]
@dataclass
class AlertRule:
"""Performance alert rule configuration"""
metric_name: str
condition: str # "greater_than", "less_than", "equals", "change_rate"
threshold: Union[int, float]
level: AlertLevel
duration_seconds: int = 60 # How long condition must persist
enabled: bool = True
description: str = ""
def __post_init__(self) -> None:
if isinstance(self.level, str):
self.level = AlertLevel(self.level)
[docs]
@dataclass
class Alert:
"""Performance alert instance"""
rule: AlertRule
triggered_at: datetime
current_value: Union[int, float]
message: str
resolved_at: Optional[datetime] = None
@property
def is_active(self) -> bool:
return self.resolved_at is None
[docs]
def resolve(self) -> None:
"""Mark alert as resolved"""
self.resolved_at = datetime.now()
[docs]
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary"""
return {
"rule_name": self.rule.metric_name,
"level": self.rule.level.value,
"triggered_at": self.triggered_at.isoformat(),
"current_value": self.current_value,
"message": self.message,
"resolved_at": self.resolved_at.isoformat() if self.resolved_at else None,
"is_active": self.is_active,
}
[docs]
class MetricsCollector:
"""Thread-safe metrics collection and aggregation"""
[docs]
def __init__(self, max_history: int = 10000):
self.max_history = max_history
self.metrics_history: deque[PerformanceMetric] = deque(maxlen=max_history)
self.current_values: Dict[str, Union[int, float]] = {}
self.aggregated_metrics: Dict[str, Dict[str, float]] = defaultdict(dict)
self._lock = threading.RLock()
self.logger = logging.getLogger("performance.collector")
[docs]
def record_metric(self, metric: PerformanceMetric) -> None:
"""Record a performance metric"""
with self._lock:
self.metrics_history.append(metric)
self.current_values[metric.name] = metric.value
# Update aggregated metrics
self._update_aggregations(metric)
self.logger.debug(f"Recorded metric: {metric.name}={metric.value}")
def _update_aggregations(self, metric: PerformanceMetric) -> None:
"""Update aggregated metric calculations"""
metric_name = metric.name
if metric.metric_type in [MetricType.GAUGE, MetricType.TIMER]:
# For gauges and timers, track min/max/avg
recent_values = [
m.value
for m in self.metrics_history
if m.name == metric_name and m.observed_at > datetime.now() - timedelta(minutes=5)
]
if recent_values:
self.aggregated_metrics[metric_name].update(
{
"min": min(recent_values),
"max": max(recent_values),
"avg": sum(recent_values) / len(recent_values),
"count": len(recent_values),
}
)
elif metric.metric_type == MetricType.COUNTER:
# For counters, track rate of change
recent_metrics = [
m
for m in self.metrics_history
if m.name == metric_name and m.observed_at > datetime.now() - timedelta(minutes=1)
]
if len(recent_metrics) > 1:
time_diff = (recent_metrics[-1].observed_at - recent_metrics[0].observed_at).total_seconds()
value_diff = recent_metrics[-1].value - recent_metrics[0].value
rate = value_diff / time_diff if time_diff > 0 else 0
self.aggregated_metrics[metric_name]["rate"] = rate
[docs]
def increment_counter(self, name: str, tags: Dict[str, str] | None = None) -> None:
"""Atomically record a cumulative counter without changing absolute counter inputs."""
with self._lock:
value = self.current_values.get(name, 0) + 1
self.record_metric(PerformanceMetric(name, value, MetricType.COUNTER, datetime.now(), tags))
[docs]
def get_current_value(self, metric_name: str) -> Optional[Union[int, float]]:
"""Get current value of a metric"""
with self._lock:
return self.current_values.get(metric_name)
[docs]
def get_aggregated_metrics(self) -> Dict[str, Dict[str, float]]:
"""Get all aggregated metrics"""
with self._lock:
return dict(self.aggregated_metrics)
[docs]
def get_metrics_history(
self, metric_name: str | None = None, since: datetime | None = None
) -> List[PerformanceMetric]:
"""Get metrics history with optional filtering"""
with self._lock:
metrics = list(self.metrics_history)
if metric_name:
metrics = [m for m in metrics if m.name == metric_name]
if since:
metrics = [m for m in metrics if m.observed_at >= since]
return metrics
[docs]
class SystemMonitor:
"""System resource monitoring"""
[docs]
def __init__(self, collection_interval: float = 1.0):
self.collection_interval = collection_interval
self.is_running = False
self.monitor_thread: Optional[threading.Thread] = None
self.metrics_collector: Optional[MetricsCollector] = None
self.logger = logging.getLogger("performance.system_monitor")
# Initialize psutil process for consistent monitoring
self.process = psutil.Process()
[docs]
def start(self, metrics_collector: MetricsCollector) -> None:
"""Start system monitoring"""
if self.is_running:
self.logger.warning("System monitor already running")
return
self.metrics_collector = metrics_collector
self.is_running = True
self.monitor_thread = threading.Thread(target=self._monitor_loop, name="SystemMonitor", daemon=True)
self.monitor_thread.start()
self.logger.info(f"System monitor started with {self.collection_interval}s interval")
[docs]
def stop(self) -> None:
"""Stop system monitoring"""
if not self.is_running:
return
self.is_running = False
if self.monitor_thread and self.monitor_thread.is_alive():
self.monitor_thread.join(timeout=5.0)
self.logger.info("System monitor stopped")
def _monitor_loop(self) -> None:
"""Main monitoring loop"""
while self.is_running:
try:
metrics = self._collect_system_metrics()
timestamp = datetime.now()
# Record individual metrics
metric_configs = [
("system.cpu.percent", metrics.cpu_percent, MetricType.GAUGE, "%"),
("system.memory.percent", metrics.memory_percent, MetricType.GAUGE, "%"),
("system.memory.used_mb", metrics.memory_used_mb, MetricType.GAUGE, "MB"),
("system.memory.available_mb", metrics.memory_available_mb, MetricType.GAUGE, "MB"),
("system.disk.usage_percent", metrics.disk_usage_percent, MetricType.GAUGE, "%"),
("system.disk.free_gb", metrics.disk_free_gb, MetricType.GAUGE, "GB"),
("system.network.bytes_sent", metrics.network_bytes_sent, MetricType.COUNTER, "bytes"),
("system.network.bytes_received", metrics.network_bytes_received, MetricType.COUNTER, "bytes"),
("system.files.open_count", metrics.open_files, MetricType.GAUGE, "count"),
("system.threads.active_count", metrics.active_threads, MetricType.GAUGE, "count"),
]
for name, value, metric_type, unit in metric_configs:
metric = PerformanceMetric(
name=name,
value=value,
metric_type=metric_type,
timestamp=timestamp,
unit=unit,
tags={"component": "system"},
)
collector = self.metrics_collector
if collector is not None:
collector.record_metric(metric)
time.sleep(self.collection_interval)
except Exception as e:
self.logger.error(f"Error in system monitoring loop: {e}")
time.sleep(self.collection_interval)
def _collect_system_metrics(self) -> SystemMetrics:
"""Collect current system metrics"""
# CPU usage
cpu_percent = psutil.cpu_percent(interval=None)
# Memory usage
memory = psutil.virtual_memory()
memory_percent = memory.percent
memory_used_mb = memory.used / (1024 * 1024)
memory_available_mb = memory.available / (1024 * 1024)
# Disk usage
disk = psutil.disk_usage(".")
disk_usage_percent = (disk.used / disk.total) * 100
disk_free_gb = disk.free / (1024 * 1024 * 1024)
# Network stats
network = psutil.net_io_counters()
network_bytes_sent = network.bytes_sent
network_bytes_received = network.bytes_recv
# Process stats
try:
open_files = len(self.process.open_files())
except (psutil.AccessDenied, psutil.NoSuchProcess):
open_files = 0
active_threads = threading.active_count()
return SystemMetrics(
cpu_percent=cpu_percent,
memory_percent=memory_percent,
memory_used_mb=memory_used_mb,
memory_available_mb=memory_available_mb,
disk_usage_percent=disk_usage_percent,
disk_free_gb=disk_free_gb,
network_bytes_sent=network_bytes_sent,
network_bytes_received=network_bytes_received,
open_files=open_files,
active_threads=active_threads,
timestamp=datetime.now(),
)
[docs]
class AlertManager:
"""Alert management and notification system"""
[docs]
def __init__(self) -> None:
self.alert_rules: Dict[str, AlertRule] = {}
self.active_alerts: Dict[str, Alert] = {}
self.alert_history: List[Alert] = []
self.alert_callbacks: List[Callable[[Alert], None]] = []
self._lock = threading.Lock()
self.logger = logging.getLogger("performance.alerts")
[docs]
def add_alert_rule(self, rule: AlertRule) -> None:
"""Add an alert rule"""
with self._lock:
self.alert_rules[rule.metric_name] = rule
self.logger.info(f"Added alert rule for {rule.metric_name}: {rule.condition} {rule.threshold}")
[docs]
def remove_alert_rule(self, metric_name: str) -> None:
"""Remove an alert rule"""
with self._lock:
if metric_name in self.alert_rules:
del self.alert_rules[metric_name]
self.logger.info(f"Removed alert rule for {metric_name}")
[docs]
def add_alert_callback(self, callback: Callable[[Alert], None]) -> None:
"""Add alert notification callback"""
self.alert_callbacks.append(callback)
[docs]
def check_metrics(self, metrics_collector: MetricsCollector) -> None:
"""Check metrics against alert rules"""
with self._lock:
for rule_name, rule in self.alert_rules.items():
if not rule.enabled:
continue
current_value = metrics_collector.get_current_value(rule.metric_name)
if current_value is None:
continue
should_alert = self._evaluate_condition(rule, current_value)
if should_alert and rule_name not in self.active_alerts:
# Create new alert
alert = Alert(
rule=rule,
triggered_at=datetime.now(),
current_value=current_value,
message=f"{rule.metric_name} {rule.condition} {rule.threshold} (current: {current_value})",
)
self.active_alerts[rule_name] = alert
self.alert_history.append(alert)
# Notify callbacks
for callback in self.alert_callbacks:
try:
callback(alert)
except Exception as e:
self.logger.error(f"Error in alert callback: {e}")
self.logger.warning(f"Alert triggered: {alert.message}")
elif not should_alert and rule_name in self.active_alerts:
# Resolve alert
alert = self.active_alerts[rule_name]
alert.resolve()
del self.active_alerts[rule_name]
self.logger.info(f"Alert resolved: {rule.metric_name}")
def _evaluate_condition(self, rule: AlertRule, current_value: Union[int, float]) -> bool:
"""Evaluate if alert condition is met"""
if rule.condition == "greater_than":
return current_value > rule.threshold
elif rule.condition == "less_than":
return current_value < rule.threshold
elif rule.condition == "equals":
return current_value == rule.threshold
elif rule.condition == "not_equals":
return current_value != rule.threshold
else:
self.logger.warning(f"Unknown alert condition: {rule.condition}")
return False
[docs]
def get_active_alerts(self) -> List[Alert]:
"""Get all active alerts"""
with self._lock:
return list(self.active_alerts.values())
[docs]
def get_alert_history(self, limit: int = 100) -> List[Alert]:
"""Get alert history"""
with self._lock:
return self.alert_history[-limit:]
# Global performance monitor instance
_performance_monitor: Optional[PerformanceMonitor] = None
[docs]
def initialize_monitoring(config: Dict[str, Any] | None = None) -> PerformanceMonitor:
"""Initialize and start performance monitoring"""
monitor = get_performance_monitor(config)
monitor.start()
return monitor
# Convenient decorators for performance tracking
[docs]
def count_calls(
metric_name: str | None = None, tags: Dict[str, str] | None = None
) -> Callable[[Callable[P, T]], Callable[P, T]]:
"""Count invocations with atomic cumulative values and independent default names."""
def decorator(func: Callable[P, T]) -> Callable[P, T]:
name = metric_name or f"function.{func.__module__}.{func.__name__}.calls"
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
get_performance_monitor().metrics_collector.increment_counter(name, tags)
return func(*args, **kwargs)
return wrapper
return decorator