Trixy7/trixy_core/service/metrics/collector.py

# -*- coding: utf-8 -*-
"""
Metriken-Collector für Service-Monitoring.

Erfasst Metriken wie Startup-Zeit, Health-Check-Dauer,
Fehler-Counts und mehr.
"""

from __future__ import annotations

import statistics
import threading
import time
from abc import ABC, abstractmethod
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any, Callable, Dict, List


class MetricType(Enum):
    """
    Typen von Metriken.
    """

    COUNTER = "counter"
    """Monoton steigender Zähler."""

    GAUGE = "gauge"
    """Wert der steigen und fallen kann."""

    HISTOGRAM = "histogram"
    """Verteilung von Werten."""

    SUMMARY = "summary"
    """Zusammenfassung mit Percentilen."""


@dataclass
class MetricValue:
    """
    Einzelner Metrik-Wert mit Metadaten.
    """

    name: str
    """Name der Metrik."""

    value: float
    """Aktueller Wert."""

    metric_type: MetricType
    """Typ der Metrik."""

    labels: dict[str, str] = field(default_factory=dict)
    """Labels/Tags für die Metrik."""

    timestamp: datetime = field(default_factory=datetime.now)
    """Zeitpunkt der Erfassung."""

    help_text: str = ""
    """Beschreibung der Metrik."""

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert in ein Dictionary."""
        return {
            "name": self.name,
            "value": self.value,
            "type": self.metric_type.value,
            "labels": self.labels,
            "timestamp": self.timestamp.isoformat(),
        }


class Metric(ABC):
    """
    Abstrakte Basisklasse für Metriken.
    """

    def __init__(
        self,
        name: str,
        help_text: str = "",
        labels: list[str] | None = None,
    ) -> None:
        """
        Initialisiert die Metrik.

        Args:
            name: Name der Metrik.
            help_text: Beschreibung.
            labels: Mögliche Label-Namen.
        """
        self.name = name
        self.help_text = help_text
        self.label_names = labels or []
        self._lock = threading.Lock()

    @abstractmethod
    def get_value(self, labels: dict[str, str] | None = None) -> float:
        """Gibt den aktuellen Wert zurück."""
        pass

    @abstractmethod
    def reset(self, labels: dict[str, str] | None = None) -> None:
        """Setzt die Metrik zurück."""
        pass


class Counter(Metric):
    """
    Counter-Metrik (monoton steigend).

    Für Ereigniszähler, Request-Counts, etc.
    """

    metric_type = MetricType.COUNTER

    def __init__(
        self,
        name: str,
        help_text: str = "",
        labels: list[str] | None = None,
    ) -> None:
        """
        Initialisiert den Counter.

        Args:
            name: Name der Metrik.
            help_text: Beschreibung.
            labels: Mögliche Label-Namen.
        """
        super().__init__(name, help_text, labels)
        self._values: dict[tuple, float] = defaultdict(float)

    def _label_key(self, labels: dict[str, str] | None) -> tuple:
        """Erstellt einen Schlüssel aus Labels."""
        if not labels:
            return ()
        return tuple(sorted(labels.items()))

    def inc(self, value: float = 1.0, labels: dict[str, str] | None = None) -> None:
        """
        Erhöht den Counter.

        Args:
            value: Wert um den erhöht wird (muss positiv sein).
            labels: Label-Werte.
        """
        if value < 0:
            raise ValueError("Counter kann nur erhöht werden")

        key = self._label_key(labels)
        with self._lock:
            self._values[key] += value

    def get_value(self, labels: dict[str, str] | None = None) -> float:
        """Gibt den aktuellen Wert zurück."""
        key = self._label_key(labels)
        with self._lock:
            return self._values.get(key, 0.0)

    def reset(self, labels: dict[str, str] | None = None) -> None:
        """Setzt den Counter zurück."""
        key = self._label_key(labels)
        with self._lock:
            if labels:
                self._values[key] = 0.0
            else:
                self._values.clear()

    def get_all_values(self) -> list[MetricValue]:
        """Gibt alle Label-Kombinationen zurück."""
        with self._lock:
            return [
                MetricValue(
                    name=self.name,
                    value=value,
                    metric_type=self.metric_type,
                    labels=dict(key),
                    help_text=self.help_text,
                )
                for key, value in self._values.items()
            ]


class Gauge(Metric):
    """
    Gauge-Metrik (kann steigen und fallen).

    Für aktuelle Werte wie Speichernutzung, aktive Verbindungen, etc.
    """

    metric_type = MetricType.GAUGE

    def __init__(
        self,
        name: str,
        help_text: str = "",
        labels: list[str] | None = None,
    ) -> None:
        """
        Initialisiert die Gauge.

        Args:
            name: Name der Metrik.
            help_text: Beschreibung.
            labels: Mögliche Label-Namen.
        """
        super().__init__(name, help_text, labels)
        self._values: dict[tuple, float] = defaultdict(float)

    def _label_key(self, labels: dict[str, str] | None) -> tuple:
        """Erstellt einen Schlüssel aus Labels."""
        if not labels:
            return ()
        return tuple(sorted(labels.items()))

    def set(self, value: float, labels: dict[str, str] | None = None) -> None:
        """
        Setzt den Gauge-Wert.

        Args:
            value: Neuer Wert.
            labels: Label-Werte.
        """
        key = self._label_key(labels)
        with self._lock:
            self._values[key] = value

    def inc(self, value: float = 1.0, labels: dict[str, str] | None = None) -> None:
        """Erhöht den Gauge."""
        key = self._label_key(labels)
        with self._lock:
            self._values[key] += value

    def dec(self, value: float = 1.0, labels: dict[str, str] | None = None) -> None:
        """Verringert den Gauge."""
        key = self._label_key(labels)
        with self._lock:
            self._values[key] -= value

    def get_value(self, labels: dict[str, str] | None = None) -> float:
        """Gibt den aktuellen Wert zurück."""
        key = self._label_key(labels)
        with self._lock:
            return self._values.get(key, 0.0)

    def reset(self, labels: dict[str, str] | None = None) -> None:
        """Setzt den Gauge zurück."""
        key = self._label_key(labels)
        with self._lock:
            if labels:
                self._values[key] = 0.0
            else:
                self._values.clear()

    def get_all_values(self) -> list[MetricValue]:
        """Gibt alle Label-Kombinationen zurück."""
        with self._lock:
            return [
                MetricValue(
                    name=self.name,
                    value=value,
                    metric_type=self.metric_type,
                    labels=dict(key),
                    help_text=self.help_text,
                )
                for key, value in self._values.items()
            ]


class Histogram(Metric):
    """
    Histogram-Metrik für Verteilungen.

    Für Latenzen, Request-Größen, etc.
    """

    metric_type = MetricType.HISTOGRAM

    DEFAULT_BUCKETS = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0)

    def __init__(
        self,
        name: str,
        help_text: str = "",
        labels: list[str] | None = None,
        buckets: tuple[float, ...] | None = None,
    ) -> None:
        """
        Initialisiert das Histogram.

        Args:
            name: Name der Metrik.
            help_text: Beschreibung.
            labels: Mögliche Label-Namen.
            buckets: Bucket-Grenzen für das Histogram.
        """
        super().__init__(name, help_text, labels)
        self.buckets = buckets or self.DEFAULT_BUCKETS
        self._observations: dict[tuple, list[float]] = defaultdict(list)

    def _label_key(self, labels: dict[str, str] | None) -> tuple:
        """Erstellt einen Schlüssel aus Labels."""
        if not labels:
            return ()
        return tuple(sorted(labels.items()))

    def observe(self, value: float, labels: dict[str, str] | None = None) -> None:
        """
        Zeichnet einen Wert auf.

        Args:
            value: Beobachteter Wert.
            labels: Label-Werte.
        """
        key = self._label_key(labels)
        with self._lock:
            self._observations[key].append(value)

    def get_value(self, labels: dict[str, str] | None = None) -> float:
        """Gibt die Summe aller Beobachtungen zurück."""
        key = self._label_key(labels)
        with self._lock:
            observations = self._observations.get(key, [])
            return sum(observations)

    def get_count(self, labels: dict[str, str] | None = None) -> int:
        """Gibt die Anzahl der Beobachtungen zurück."""
        key = self._label_key(labels)
        with self._lock:
            return len(self._observations.get(key, []))

    def get_buckets(self, labels: dict[str, str] | None = None) -> dict[float, int]:
        """Gibt die Bucket-Verteilung zurück."""
        key = self._label_key(labels)
        with self._lock:
            observations = self._observations.get(key, [])

            result = {}
            for bucket in self.buckets:
                count = sum(1 for v in observations if v <= bucket)
                result[bucket] = count

            result[float("inf")] = len(observations)
            return result

    def get_percentile(
        self,
        percentile: float,
        labels: dict[str, str] | None = None,
    ) -> float:
        """
        Berechnet ein Percentil.

        Args:
            percentile: Percentil (0-100).
            labels: Label-Werte.

        Returns:
            Percentil-Wert.
        """
        key = self._label_key(labels)
        with self._lock:
            observations = self._observations.get(key, [])
            if not observations:
                return 0.0

            sorted_obs = sorted(observations)
            index = int((percentile / 100) * len(sorted_obs))
            index = min(index, len(sorted_obs) - 1)
            return sorted_obs[index]

    def reset(self, labels: dict[str, str] | None = None) -> None:
        """Setzt das Histogram zurück."""
        key = self._label_key(labels)
        with self._lock:
            if labels:
                self._observations[key] = []
            else:
                self._observations.clear()


@dataclass
class ServiceMetrics:
    """
    Standard-Metriken für einen Service.
    """

    service_name: str
    """Name des Services."""

    startup_time: float = 0.0
    """Startup-Zeit in Sekunden."""

    startup_timestamp: datetime | None = None
    """Zeitpunkt des Starts."""

    health_check_duration: float = 0.0
    """Letzte Health-Check-Dauer."""

    health_check_count: int = 0
    """Anzahl der Health-Checks."""

    error_count: int = 0
    """Anzahl der Fehler."""

    last_error: str | None = None
    """Letzter Fehler."""

    last_error_timestamp: datetime | None = None
    """Zeitpunkt des letzten Fehlers."""

    custom_metrics: dict[str, float] = field(default_factory=dict)
    """Benutzerdefinierte Metriken."""

    def record_error(self, error: str) -> None:
        """Zeichnet einen Fehler auf."""
        self.error_count += 1
        self.last_error = error
        self.last_error_timestamp = datetime.now()

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert in ein Dictionary."""
        return {
            "service_name": self.service_name,
            "startup_time_seconds": self.startup_time,
            "startup_timestamp": (
                self.startup_timestamp.isoformat()
                if self.startup_timestamp else None
            ),
            "health_check_duration_seconds": self.health_check_duration,
            "health_check_count": self.health_check_count,
            "error_count": self.error_count,
            "last_error": self.last_error,
            "custom_metrics": self.custom_metrics,
        }


class MetricsCollector:
    """
    Zentraler Collector für Service-Metriken.

    Sammelt und aggregiert Metriken aus verschiedenen Services.

    Example:
        collector = MetricsCollector()

        # Metrik registrieren
        requests = collector.counter("http_requests_total")
        latency = collector.histogram("request_latency_seconds")

        # Metriken erfassen
        requests.inc(labels={"method": "GET"})
        latency.observe(0.5, labels={"endpoint": "/api"})

        # Alle Metriken abrufen
        metrics = collector.get_all_metrics()
    """

    def __init__(self, prefix: str = "") -> None:
        """
        Initialisiert den Collector.

        Args:
            prefix: Präfix für alle Metrik-Namen.
        """
        self._prefix = prefix
        self._metrics: dict[str, Metric] = {}
        self._service_metrics: dict[str, ServiceMetrics] = {}
        self._lock = threading.Lock()

    def _make_name(self, name: str) -> str:
        """Erstellt den vollständigen Metrik-Namen."""
        if self._prefix:
            return f"{self._prefix}_{name}"
        return name

    def counter(
        self,
        name: str,
        help_text: str = "",
        labels: list[str] | None = None,
    ) -> Counter:
        """
        Erstellt oder holt einen Counter.

        Args:
            name: Name der Metrik.
            help_text: Beschreibung.
            labels: Mögliche Label-Namen.

        Returns:
            Counter-Instanz.
        """
        full_name = self._make_name(name)

        with self._lock:
            if full_name not in self._metrics:
                self._metrics[full_name] = Counter(full_name, help_text, labels)
            return self._metrics[full_name]  # type: ignore

    def gauge(
        self,
        name: str,
        help_text: str = "",
        labels: list[str] | None = None,
    ) -> Gauge:
        """
        Erstellt oder holt eine Gauge.

        Args:
            name: Name der Metrik.
            help_text: Beschreibung.
            labels: Mögliche Label-Namen.

        Returns:
            Gauge-Instanz.
        """
        full_name = self._make_name(name)

        with self._lock:
            if full_name not in self._metrics:
                self._metrics[full_name] = Gauge(full_name, help_text, labels)
            return self._metrics[full_name]  # type: ignore

    def histogram(
        self,
        name: str,
        help_text: str = "",
        labels: list[str] | None = None,
        buckets: tuple[float, ...] | None = None,
    ) -> Histogram:
        """
        Erstellt oder holt ein Histogram.

        Args:
            name: Name der Metrik.
            help_text: Beschreibung.
            labels: Mögliche Label-Namen.
            buckets: Bucket-Grenzen.

        Returns:
            Histogram-Instanz.
        """
        full_name = self._make_name(name)

        with self._lock:
            if full_name not in self._metrics:
                self._metrics[full_name] = Histogram(
                    full_name, help_text, labels, buckets
                )
            return self._metrics[full_name]  # type: ignore

    def register_service(self, service_name: str) -> ServiceMetrics:
        """
        Registriert einen Service für Metriken-Erfassung.

        Args:
            service_name: Name des Services.

        Returns:
            ServiceMetrics für den Service.
        """
        with self._lock:
            if service_name not in self._service_metrics:
                self._service_metrics[service_name] = ServiceMetrics(
                    service_name=service_name
                )
            return self._service_metrics[service_name]

    def get_service_metrics(self, service_name: str) -> ServiceMetrics | None:
        """Gibt die Metriken eines Services zurück."""
        return self._service_metrics.get(service_name)

    def get_all_metrics(self) -> dict[str, Metric]:
        """Gibt alle Metriken zurück."""
        with self._lock:
            return dict(self._metrics)

    def get_all_service_metrics(self) -> dict[str, ServiceMetrics]:
        """Gibt alle Service-Metriken zurück."""
        with self._lock:
            return dict(self._service_metrics)

    def reset_all(self) -> None:
        """Setzt alle Metriken zurück."""
        with self._lock:
            for metric in self._metrics.values():
                metric.reset()


# Globaler Default-Collector
_default_collector: MetricsCollector | None = None


def get_default_collector() -> MetricsCollector:
    """Gibt den Default-Collector zurück."""
    global _default_collector
    if _default_collector is None:
        _default_collector = MetricsCollector()
    return _default_collector


def set_default_collector(collector: MetricsCollector) -> None:
    """Setzt den Default-Collector."""
    global _default_collector
    _default_collector = collector