354 lines
14 KiB
Python
354 lines
14 KiB
Python
"""
|
|
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}) |