feat: metrics
This commit is contained in:
@@ -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})
|
||||
Reference in New Issue
Block a user