core.performance_monitor

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.

Functions

count_calls([metric_name, tags])	Count invocations with atomic cumulative values and independent default names.
get_performance_monitor([config])	Get global performance monitor instance
initialize_monitoring([config])	Initialize and start performance monitoring
track_performance([metric_name, tags])	Track each decorated function independently while retaining its signature.

Classes

Alert(rule, triggered_at, current_value, message)	Performance alert instance
AlertLevel(value)	Alert severity levels
AlertManager()	Alert management and notification system
AlertRule(metric_name, condition, threshold, ...)	Performance alert rule configuration
MetricType(value)	Types of metrics that can be collected
MetricsCollector([max_history])	Thread-safe metrics collection and aggregation
PerformanceMetric(name, value, metric_type, ...)	Individual performance metric data point
PerformanceMonitor([config])	Main performance monitoring system
SystemMetrics(cpu_percent, memory_percent, ...)	System-level performance metrics
SystemMonitor([collection_interval])	System resource monitoring

class core.performance_monitor.MetricType(value)

Bases: Enum

Types of metrics that can be collected

COUNTER = 'counter'

GAUGE = 'gauge'

HISTOGRAM = 'histogram'

TIMER = 'timer'

RATE = 'rate'

class core.performance_monitor.AlertLevel(value)

Bases: Enum

Alert severity levels

INFO = 'info'

WARNING = 'warning'

ERROR = 'error'

CRITICAL = 'critical'

class core.performance_monitor.PerformanceMetric(name, value, metric_type, timestamp, tags=None, unit='count')

Bases: object

Individual performance metric data point

name: str

value: int | float

metric_type: MetricType

timestamp: datetime | str

tags: Dict[str, str] | None = None

unit: str = 'count'

property observed_at: datetime

to_dict()

Convert to dictionary for serialization

__init__(name, value, metric_type, timestamp, tags=None, unit='count')

class core.performance_monitor.SystemMetrics(cpu_percent, memory_percent, memory_used_mb, memory_available_mb, disk_usage_percent, disk_free_gb, network_bytes_sent, network_bytes_received, open_files, active_threads, timestamp)

Bases: object

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

to_dict()

Convert to dictionary

__init__(cpu_percent, memory_percent, memory_used_mb, memory_available_mb, disk_usage_percent, disk_free_gb, network_bytes_sent, network_bytes_received, open_files, active_threads, timestamp)

class core.performance_monitor.AlertRule(metric_name, condition, threshold, level, duration_seconds=60, enabled=True, description='')

Bases: object

Performance alert rule configuration

metric_name: str

condition: str

threshold: int | float

level: AlertLevel

duration_seconds: int = 60

enabled: bool = True

description: str = ''

__init__(metric_name, condition, threshold, level, duration_seconds=60, enabled=True, description='')

class core.performance_monitor.Alert(rule, triggered_at, current_value, message, resolved_at=None)

Bases: object

Performance alert instance

rule: AlertRule

triggered_at: datetime

current_value: int | float

message: str

resolved_at: datetime | None = None

property is_active: bool

resolve()

Mark alert as resolved

to_dict()

Convert to dictionary

__init__(rule, triggered_at, current_value, message, resolved_at=None)

class core.performance_monitor.MetricsCollector(max_history=10000)

Bases: object

Thread-safe metrics collection and aggregation

__init__(max_history=10000)

metrics_history: deque[PerformanceMetric]

current_values: Dict[str, int | float]

aggregated_metrics: Dict[str, Dict[str, float]]

record_metric(metric)

Record a performance metric

increment_counter(name, tags=None)

Atomically record a cumulative counter without changing absolute counter inputs.

get_current_value(metric_name)

Get current value of a metric

get_aggregated_metrics()

Get all aggregated metrics

get_metrics_history(metric_name=None, since=None)

Get metrics history with optional filtering

class core.performance_monitor.SystemMonitor(collection_interval=1.0)

Bases: object

System resource monitoring

__init__(collection_interval=1.0)

monitor_thread: Thread | None

metrics_collector: MetricsCollector | None

start(metrics_collector)

Start system monitoring

stop()

Stop system monitoring

class core.performance_monitor.AlertManager

Bases: object

Alert management and notification system

__init__()

alert_rules: Dict[str, AlertRule]

active_alerts: Dict[str, Alert]

alert_history: List[Alert]

alert_callbacks: List[Callable[[Alert], None]]

add_alert_rule(rule)

Add an alert rule

remove_alert_rule(metric_name)

Remove an alert rule

add_alert_callback(callback)

Add alert notification callback

check_metrics(metrics_collector)

Check metrics against alert rules

get_active_alerts()

Get all active alerts

get_alert_history(limit=100)

Get alert history

class core.performance_monitor.PerformanceMonitor(config=None)

Bases: object

Main performance monitoring system

__init__(config=None)

alert_check_thread: Thread | None

start()

Start performance monitoring

stop()

Stop performance monitoring

record_metric(name, value, metric_type=MetricType.GAUGE, tags=None, unit='count')

Record a custom metric

timer(name, tags=None)

Context manager for timing operations

get_status()

Get monitoring system status

export_metrics(format='json')

Export metrics in specified format

core.performance_monitor.get_performance_monitor(config=None)

Get global performance monitor instance

core.performance_monitor.initialize_monitoring(config=None)

Initialize and start performance monitoring

core.performance_monitor.track_performance(metric_name=None, tags=None)

Track each decorated function independently while retaining its signature.

core.performance_monitor.count_calls(metric_name=None, tags=None)

Count invocations with atomic cumulative values and independent default names.