core/metrics.py - master - HookDev-Arch/ServerMonitor

core/metrics.py
220 строк8 KB
HookDev-Arch
upload files
09 окт 2025, 02:25
09 окт 2025, 02:258bd6873
"""
Система метрик и мониторинга для HookDev-Arch Dashboard v3.5
Сбор, хранение и анализ метрик производительности
"""
import time
import json
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, Any, List, Optional
from collections import defaultdict, deque
from dataclasses import dataclass, asdict
from core.config import settings
from core.logger import get_logger


@dataclass
class MetricPoint:
    """Точка метрики"""
    timestamp: float
    name: str
    value: float
    tags: Dict[str, str] = None
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            'timestamp': self.timestamp,
            'name': self.name,
            'value': self.value,
            'tags': self.tags or {}
        }


class MetricsCollector:
    """Сборщик метрик"""
    
    def __init__(self, max_points: int = 1000):
        self.max_points = max_points
        self.metrics: Dict[str, deque] = defaultdict(lambda: deque(maxlen=max_points))
        self.logger = get_logger("metrics")
        self.start_time = time.time()
    
    def record_metric(self, name: str, value: float, tags: Dict[str, str] = None):
        """Записывает метрику"""
        point = MetricPoint(
            timestamp=time.time(),
            name=name,
            value=value,
            tags=tags or {}
        )
        
        self.metrics[name].append(point)
        self.logger.debug(f"Recorded metric: {name}={value}")
    
    def get_metric(self, name: str, time_range: int = 300) -> List[MetricPoint]:
        """Получает метрики за указанный период"""
        if name not in self.metrics:
            return []
        
        cutoff_time = time.time() - time_range
        return [point for point in self.metrics[name] if point.timestamp >= cutoff_time]
    
    def get_metric_stats(self, name: str, time_range: int = 300) -> Dict[str, float]:
        """Получает статистику по метрике"""
        points = self.get_metric(name, time_range)
        if not points:
            return {}
        
        values = [point.value for point in points]
        return {
            'count': len(values),
            'min': min(values),
            'max': max(values),
            'avg': sum(values) / len(values),
            'latest': values[-1] if values else 0
        }
    
    def get_all_metrics(self) -> Dict[str, List[Dict[str, Any]]]:
        """Получает все метрики"""
        result = {}
        for name, points in self.metrics.items():
            result[name] = [point.to_dict() for point in points]
        return result


class PerformanceMonitor:
    """Монитор производительности"""
    
    def __init__(self):
        self.collector = MetricsCollector()
        self.logger = get_logger("performance")
    
    def measure_time(self, metric_name: str, tags: Dict[str, str] = None):
        """Декоратор для измерения времени выполнения"""
        def decorator(func):
            def wrapper(*args, **kwargs):
                start_time = time.time()
                try:
                    result = func(*args, **kwargs)
                    return result
                finally:
                    duration = time.time() - start_time
                    self.collector.record_metric(metric_name, duration, tags)
                    self.logger.debug(f"{func.__name__} took {duration:.3f}s")
            return wrapper
        return decorator
    
    def record_api_call(self, endpoint: str, method: str, duration: float, status_code: int):
        """Записывает метрику API вызова"""
        self.collector.record_metric("api_call_duration", duration, {
            'endpoint': endpoint,
            'method': method,
            'status_code': str(status_code)
        })
    
    def record_system_metric(self, metric_name: str, value: float):
        """Записывает системную метрику"""
        self.collector.record_metric(metric_name, value)
    
    def get_performance_summary(self) -> Dict[str, Any]:
        """Получает сводку по производительности"""
        return {
            'api_calls': self.collector.get_metric_stats("api_call_duration"),
            'system_metrics': {
                name: self.collector.get_metric_stats(name)
                for name in ['cpu_usage', 'memory_usage', 'disk_usage']
            },
            'uptime': time.time() - self.collector.start_time
        }


class HealthChecker:
    """Проверка здоровья системы"""
    
    def __init__(self):
        self.checks = {}
        self.logger = get_logger("health")
    
    def register_check(self, name: str, check_func, critical: bool = False):
        """Регистрирует проверку здоровья"""
        self.checks[name] = {
            'function': check_func,
            'critical': critical
        }
    
    async def run_health_checks(self) -> Dict[str, Any]:
        """Выполняет все проверки здоровья"""
        results = {}
        overall_healthy = True
        
        for name, check_info in self.checks.items():
            try:
                result = await check_info['function']()
                results[name] = {
                    'status': 'healthy' if result else 'unhealthy',
                    'result': result,
                    'critical': check_info['critical']
                }
                
                if not result and check_info['critical']:
                    overall_healthy = False
                    
            except Exception as e:
                results[name] = {
                    'status': 'error',
                    'error': str(e),
                    'critical': check_info['critical']
                }
                if check_info['critical']:
                    overall_healthy = False
        
        return {
            'overall': 'healthy' if overall_healthy else 'unhealthy',
            'checks': results,
            'timestamp': time.time()
        }


class AlertManager:
    """Менеджер алертов"""
    
    def __init__(self):
        self.thresholds = {
            'cpu_usage': 80,
            'memory_usage': 85,
            'disk_usage': 90,
            'api_response_time': 5.0
        }
        self.alerts = deque(maxlen=100)
        self.logger = get_logger("alerts")
    
    def check_thresholds(self, metrics: Dict[str, float]) -> List[Dict[str, Any]]:
        """Проверяет пороговые значения"""
        new_alerts = []
        
        for metric, value in metrics.items():
            if metric in self.thresholds and value > self.thresholds[metric]:
                alert = {
                    'type': metric,
                    'value': value,
                    'threshold': self.thresholds[metric],
                    'timestamp': time.time(),
                    'severity': 'high' if value > self.thresholds[metric] * 1.2 else 'medium'
                }
                new_alerts.append(alert)
                self.alerts.append(alert)
                self.logger.warning(f"Alert: {metric} = {value} (threshold: {self.thresholds[metric]})")
        
        return new_alerts
    
    def get_recent_alerts(self, hours: int = 24) -> List[Dict[str, Any]]:
        """Получает недавние алерты"""
        cutoff_time = time.time() - (hours * 3600)
        return [alert for alert in self.alerts if alert['timestamp'] >= cutoff_time]


# Глобальные экземпляры
performance_monitor = PerformanceMonitor()
health_checker = HealthChecker()
alert_manager = AlertManager()
metrics_collector = MetricsCollector()