Trixy7/trixy_core/satellite/health/thresholds.py

# -*- coding: utf-8 -*-
"""
Health-Schwellenwerte und Alarme.

Definiert Schwellenwerte für Metriken und löst
Alarme bei Überschreitung aus.
"""

from dataclasses import dataclass, field
from datetime import datetime
from enum import IntEnum, auto
from typing import Any, Callable

from trixy_core.satellite.health.scorer import HealthMetrics


class ViolationSeverity(IntEnum):
    """Schweregrad einer Schwellenwert-Verletzung."""
    INFO = auto()       # Informativ
    WARNING = auto()    # Warnung
    ERROR = auto()      # Fehler
    CRITICAL = auto()   # Kritisch


@dataclass
class ThresholdConfig:
    """
    Konfiguration eines Schwellenwerts.

    Attributes:
        metric_name: Name der Metrik
        warning_threshold: Schwelle für Warnung
        error_threshold: Schwelle für Fehler
        critical_threshold: Schwelle für kritisch
        comparison: "gt" (größer), "lt" (kleiner), "eq" (gleich)
        cooldown: Mindestzeit zwischen Alarmen in Sekunden
        description: Beschreibung des Schwellenwerts
    """
    metric_name: str
    warning_threshold: float | None = None
    error_threshold: float | None = None
    critical_threshold: float | None = None
    comparison: str = "gt"  # gt, lt, eq, ge, le
    cooldown: float = 60.0
    description: str = ""


@dataclass
class ThresholdViolation:
    """
    Eine erkannte Schwellenwert-Verletzung.

    Attributes:
        satellite_id: Betroffener Satellite
        config: Der verletzte Schwellenwert
        severity: Schweregrad
        current_value: Aktueller Wert
        threshold_value: Überschrittener Schwellenwert
        message: Beschreibende Nachricht
        timestamp: Zeitpunkt der Verletzung
    """
    satellite_id: str
    config: ThresholdConfig
    severity: ViolationSeverity
    current_value: float
    threshold_value: float
    message: str = ""
    timestamp: datetime = field(default_factory=datetime.now)

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert zu Dictionary."""
        return {
            "satellite_id": self.satellite_id,
            "metric": self.config.metric_name,
            "severity": self.severity.name,
            "current_value": self.current_value,
            "threshold_value": self.threshold_value,
            "message": self.message,
            "timestamp": self.timestamp.isoformat(),
        }


class HealthThresholds:
    """
    Verwaltet Schwellenwerte und prüft auf Verletzungen.

    Ermöglicht die Konfiguration von Alarmschwellen für
    verschiedene Metriken und benachrichtigt bei Überschreitung.

    Example:
        thresholds = HealthThresholds()

        # Schwellenwerte konfigurieren
        thresholds.add(ThresholdConfig(
            metric_name="latency_ms",
            warning_threshold=100,
            error_threshold=200,
            critical_threshold=500,
            comparison="gt"
        ))

        thresholds.add(ThresholdConfig(
            metric_name="packet_loss",
            warning_threshold=0.01,
            error_threshold=0.05,
            critical_threshold=0.10,
        ))

        # Callback für Verletzungen
        @thresholds.on_violation
        def handle_violation(violation: ThresholdViolation):
            print(f"ALARM: {violation.message}")

        # Metriken prüfen
        violations = thresholds.check("satellite_1", metrics)
    """

    # Vordefinierte Schwellenwerte
    DEFAULT_THRESHOLDS = [
        ThresholdConfig(
            metric_name="latency_ms",
            warning_threshold=100.0,
            error_threshold=200.0,
            critical_threshold=500.0,
            comparison="gt",
            description="Netzwerklatenz"
        ),
        ThresholdConfig(
            metric_name="packet_loss",
            warning_threshold=0.01,
            error_threshold=0.05,
            critical_threshold=0.10,
            comparison="gt",
            description="Paketverlustrate"
        ),
        ThresholdConfig(
            metric_name="error_rate",
            warning_threshold=0.01,
            error_threshold=0.05,
            critical_threshold=0.10,
            comparison="gt",
            description="Fehlerrate"
        ),
        ThresholdConfig(
            metric_name="last_heartbeat_age",
            warning_threshold=10.0,
            error_threshold=30.0,
            critical_threshold=60.0,
            comparison="gt",
            description="Heartbeat-Alter"
        ),
        ThresholdConfig(
            metric_name="cpu_usage",
            warning_threshold=0.70,
            error_threshold=0.85,
            critical_threshold=0.95,
            comparison="gt",
            description="CPU-Auslastung"
        ),
        ThresholdConfig(
            metric_name="memory_usage",
            warning_threshold=0.75,
            error_threshold=0.90,
            critical_threshold=0.95,
            comparison="gt",
            description="Speicherauslastung"
        ),
    ]

    def __init__(self, use_defaults: bool = True) -> None:
        """
        Initialisiert die Schwellenwert-Verwaltung.

        Args:
            use_defaults: Standard-Schwellenwerte laden
        """
        self._configs: dict[str, ThresholdConfig] = {}
        self._last_violations: dict[str, dict[str, datetime]] = {}
        self._callbacks: list[Callable[[ThresholdViolation], None]] = []

        if use_defaults:
            for config in self.DEFAULT_THRESHOLDS:
                self.add(config)

    def add(self, config: ThresholdConfig) -> None:
        """
        Fügt einen Schwellenwert hinzu.

        Args:
            config: Schwellenwert-Konfiguration
        """
        self._configs[config.metric_name] = config

    def remove(self, metric_name: str) -> ThresholdConfig | None:
        """
        Entfernt einen Schwellenwert.

        Args:
            metric_name: Name der Metrik

        Returns:
            Die entfernte Konfiguration oder None
        """
        return self._configs.pop(metric_name, None)

    def get(self, metric_name: str) -> ThresholdConfig | None:
        """Gibt eine Konfiguration zurück."""
        return self._configs.get(metric_name)

    def on_violation(
        self,
        callback: Callable[[ThresholdViolation], None]
    ) -> Callable[[ThresholdViolation], None]:
        """
        Registriert einen Callback für Verletzungen.

        Kann als Decorator verwendet werden.

        Args:
            callback: Die aufzurufende Funktion

        Returns:
            Der registrierte Callback
        """
        self._callbacks.append(callback)
        return callback

    def check(
        self,
        satellite_id: str,
        metrics: HealthMetrics
    ) -> list[ThresholdViolation]:
        """
        Prüft Metriken gegen alle Schwellenwerte.

        Args:
            satellite_id: Satellite-ID
            metrics: Zu prüfende Metriken

        Returns:
            Liste der Verletzungen
        """
        violations = []

        for config in self._configs.values():
            violation = self._check_threshold(satellite_id, metrics, config)
            if violation:
                # Cooldown prüfen
                if self._is_in_cooldown(satellite_id, config.metric_name, config.cooldown):
                    continue

                violations.append(violation)
                self._record_violation(satellite_id, config.metric_name)
                self._notify_violation(violation)

        return violations

    def _check_threshold(
        self,
        satellite_id: str,
        metrics: HealthMetrics,
        config: ThresholdConfig
    ) -> ThresholdViolation | None:
        """Prüft einen einzelnen Schwellenwert."""
        # Metrik-Wert holen
        if hasattr(metrics, config.metric_name):
            value = getattr(metrics, config.metric_name)
        elif config.metric_name in metrics.custom_metrics:
            value = metrics.custom_metrics[config.metric_name]
        else:
            return None

        # Schweregrad bestimmen
        severity = None
        threshold_value = None

        # Vom kritischsten zum harmlosesten prüfen
        if config.critical_threshold is not None:
            if self._compare(value, config.critical_threshold, config.comparison):
                severity = ViolationSeverity.CRITICAL
                threshold_value = config.critical_threshold

        if severity is None and config.error_threshold is not None:
            if self._compare(value, config.error_threshold, config.comparison):
                severity = ViolationSeverity.ERROR
                threshold_value = config.error_threshold

        if severity is None and config.warning_threshold is not None:
            if self._compare(value, config.warning_threshold, config.comparison):
                severity = ViolationSeverity.WARNING
                threshold_value = config.warning_threshold

        if severity is None:
            return None

        # Verletzung erstellen
        message = self._format_message(config, value, threshold_value, severity)

        return ThresholdViolation(
            satellite_id=satellite_id,
            config=config,
            severity=severity,
            current_value=value,
            threshold_value=threshold_value,
            message=message,
        )

    def _compare(self, value: float, threshold: float, comparison: str) -> bool:
        """Vergleicht einen Wert mit einem Schwellenwert."""
        if comparison == "gt":
            return value > threshold
        elif comparison == "lt":
            return value < threshold
        elif comparison == "ge":
            return value >= threshold
        elif comparison == "le":
            return value <= threshold
        elif comparison == "eq":
            return value == threshold
        return False

    def _format_message(
        self,
        config: ThresholdConfig,
        value: float,
        threshold: float,
        severity: ViolationSeverity
    ) -> str:
        """Formatiert die Verletzungs-Nachricht."""
        comparison_text = {
            "gt": "überschreitet",
            "lt": "unterschreitet",
            "ge": "erreicht oder überschreitet",
            "le": "erreicht oder unterschreitet",
            "eq": "entspricht",
        }.get(config.comparison, "verletzt")

        name = config.description or config.metric_name

        return (
            f"{severity.name}: {name} {comparison_text} Schwellenwert "
            f"(Wert: {value:.3f}, Schwelle: {threshold:.3f})"
        )

    def _is_in_cooldown(
        self,
        satellite_id: str,
        metric_name: str,
        cooldown: float
    ) -> bool:
        """Prüft, ob noch im Cooldown."""
        if satellite_id not in self._last_violations:
            return False

        last_time = self._last_violations[satellite_id].get(metric_name)
        if last_time is None:
            return False

        elapsed = (datetime.now() - last_time).total_seconds()
        return elapsed < cooldown

    def _record_violation(self, satellite_id: str, metric_name: str) -> None:
        """Zeichnet eine Verletzung auf (für Cooldown)."""
        if satellite_id not in self._last_violations:
            self._last_violations[satellite_id] = {}
        self._last_violations[satellite_id][metric_name] = datetime.now()

    def _notify_violation(self, violation: ThresholdViolation) -> None:
        """Benachrichtigt alle Callbacks."""
        for callback in self._callbacks:
            try:
                callback(violation)
            except Exception:
                pass

    def get_all_configs(self) -> list[ThresholdConfig]:
        """Gibt alle Konfigurationen zurück."""
        return list(self._configs.values())

    def clear_cooldowns(self, satellite_id: str | None = None) -> None:
        """
        Löscht Cooldowns.

        Args:
            satellite_id: Optionaler spezifischer Satellite
        """
        if satellite_id:
            self._last_violations.pop(satellite_id, None)
        else:
            self._last_violations.clear()