# -*- coding: utf-8 -*- """ Metriken-Aggregator für Satellites. Aggregiert Metriken über Zeit und Satellites für Analyse und Reporting. """ from __future__ import annotations import logging import statistics from collections import defaultdict from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum, auto from typing import Any, Callable from trixy_core.satellite.metrics.collector import ( SatelliteMetricsCollector, MetricValue, SatelliteMetrics, ) class AggregationType(Enum): """Aggregations-Typ.""" SUM = auto() # Summe AVG = auto() # Durchschnitt MIN = auto() # Minimum MAX = auto() # Maximum COUNT = auto() # Anzahl MEDIAN = auto() # Median P95 = auto() # 95. Perzentil P99 = auto() # 99. Perzentil STDDEV = auto() # Standardabweichung class TimeWindow(Enum): """Zeitfenster für Aggregation.""" MINUTE = 60 FIVE_MINUTES = 300 FIFTEEN_MINUTES = 900 HOUR = 3600 DAY = 86400 @dataclass class AggregatedMetric: """Aggregierte Metrik.""" name: str aggregation: AggregationType value: float window: TimeWindow sample_count: int start_time: datetime end_time: datetime satellite_ids: list[str] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: """Konvertiert zu Dictionary.""" return { "name": self.name, "aggregation": self.aggregation.name, "value": round(self.value, 4), "window": self.window.name, "sample_count": self.sample_count, "start_time": self.start_time.isoformat(), "end_time": self.end_time.isoformat(), "satellite_count": len(self.satellite_ids), } class MetricsAggregator: """ Aggregiert Metriken über Zeit und Satellites. Bietet verschiedene Aggregationsfunktionen und zeitbasierte Analysen. Beispiel: aggregator = MetricsAggregator(collector) # Durchschnittliche Latenz über alle Satellites avg_latency = aggregator.aggregate( metric_name="latency", aggregation=AggregationType.AVG, window=TimeWindow.FIVE_MINUTES, ) # Top Satellites nach Fehlerrate top_errors = aggregator.get_top_satellites( metric_name="error_count", limit=10, ) """ def __init__( self, collector: SatelliteMetricsCollector, logger: logging.Logger | None = None, ) -> None: """ Initialisiert den Aggregator. Args: collector: Metriken-Collector logger: Logger-Instanz """ self.collector = collector self.logger = logger or logging.getLogger(__name__) # Cache für berechnete Aggregationen self._cache: dict[str, tuple[datetime, AggregatedMetric]] = {} self._cache_ttl = timedelta(seconds=30) def _get_metric_values( self, metric_name: str, window: TimeWindow, satellite_ids: list[str] | None = None, ) -> tuple[list[float], list[str]]: """ Sammelt Metrik-Werte aus dem Collector. Args: metric_name: Name der Metrik window: Zeitfenster satellite_ids: Optional: Spezifische Satellites Returns: Tuple (Werte, Satellite-IDs) """ cutoff = datetime.now() - timedelta(seconds=window.value) values: list[float] = [] satellites: list[str] = [] all_metrics = self.collector.get_all_metrics() for sat_id, metrics in all_metrics.items(): if satellite_ids and sat_id not in satellite_ids: continue if metrics.last_updated < cutoff: continue # Standard-Metriken value = None if metric_name == "latency": value = metrics.latency_ms elif metric_name == "packet_loss": value = metrics.packet_loss_percent elif metric_name == "bandwidth": value = metrics.bandwidth_kbps elif metric_name == "messages_sent": value = float(metrics.messages_sent) elif metric_name == "messages_received": value = float(metrics.messages_received) elif metric_name == "error_count": value = float(metrics.error_count) elif metric_name == "reconnect_count": value = float(metrics.reconnect_count) if value is not None: values.append(value) satellites.append(sat_id) return values, satellites def _calculate_aggregation( self, values: list[float], aggregation: AggregationType, ) -> float: """Berechnet Aggregation für Werte.""" if not values: return 0.0 if aggregation == AggregationType.SUM: return sum(values) elif aggregation == AggregationType.AVG: return statistics.mean(values) elif aggregation == AggregationType.MIN: return min(values) elif aggregation == AggregationType.MAX: return max(values) elif aggregation == AggregationType.COUNT: return float(len(values)) elif aggregation == AggregationType.MEDIAN: return statistics.median(values) elif aggregation == AggregationType.P95: return self._percentile(values, 95) elif aggregation == AggregationType.P99: return self._percentile(values, 99) elif aggregation == AggregationType.STDDEV: return statistics.stdev(values) if len(values) > 1 else 0.0 return 0.0 def _percentile(self, values: list[float], percentile: int) -> float: """Berechnet Perzentil.""" if not values: return 0.0 sorted_values = sorted(values) index = (percentile / 100) * (len(sorted_values) - 1) if index.is_integer(): return sorted_values[int(index)] lower = sorted_values[int(index)] upper = sorted_values[int(index) + 1] fraction = index - int(index) return lower + (upper - lower) * fraction def aggregate( self, metric_name: str, aggregation: AggregationType = AggregationType.AVG, window: TimeWindow = TimeWindow.FIVE_MINUTES, satellite_ids: list[str] | None = None, use_cache: bool = True, ) -> AggregatedMetric: """ Aggregiert eine Metrik. Args: metric_name: Name der Metrik aggregation: Aggregations-Typ window: Zeitfenster satellite_ids: Optional: Spezifische Satellites use_cache: Cache verwenden Returns: Aggregierte Metrik """ # Cache prüfen cache_key = f"{metric_name}:{aggregation.name}:{window.name}" if use_cache and cache_key in self._cache: cached_time, cached_result = self._cache[cache_key] if datetime.now() - cached_time < self._cache_ttl: return cached_result # Werte sammeln values, satellites = self._get_metric_values( metric_name, window, satellite_ids ) # Aggregieren result_value = self._calculate_aggregation(values, aggregation) end_time = datetime.now() start_time = end_time - timedelta(seconds=window.value) result = AggregatedMetric( name=metric_name, aggregation=aggregation, value=result_value, window=window, sample_count=len(values), start_time=start_time, end_time=end_time, satellite_ids=satellites, ) # Cache aktualisieren self._cache[cache_key] = (datetime.now(), result) return result def aggregate_multiple( self, metric_names: list[str], aggregations: list[AggregationType], window: TimeWindow = TimeWindow.FIVE_MINUTES, ) -> dict[str, dict[str, AggregatedMetric]]: """ Aggregiert mehrere Metriken mit mehreren Aggregationen. Args: metric_names: Liste von Metrik-Namen aggregations: Liste von Aggregations-Typen window: Zeitfenster Returns: Dict[metric_name][aggregation_name] -> AggregatedMetric """ results: dict[str, dict[str, AggregatedMetric]] = {} for metric_name in metric_names: results[metric_name] = {} for aggregation in aggregations: result = self.aggregate(metric_name, aggregation, window) results[metric_name][aggregation.name] = result return results def get_top_satellites( self, metric_name: str, limit: int = 10, ascending: bool = False, ) -> list[tuple[str, float]]: """ Holt Top-Satellites nach Metrik. Args: metric_name: Metrik-Name limit: Maximale Anzahl ascending: Aufsteigend sortieren Returns: Liste von (satellite_id, value) Tupeln """ values, satellites = self._get_metric_values( metric_name, TimeWindow.HOUR, ) if not values: return [] # Zusammenführen satellite_values = list(zip(satellites, values)) # Sortieren satellite_values.sort(key=lambda x: x[1], reverse=not ascending) return satellite_values[:limit] def get_metric_distribution( self, metric_name: str, window: TimeWindow = TimeWindow.HOUR, buckets: int = 10, ) -> dict[str, Any]: """ Berechnet Verteilung einer Metrik. Args: metric_name: Metrik-Name window: Zeitfenster buckets: Anzahl der Buckets Returns: Verteilungs-Informationen """ values, _ = self._get_metric_values(metric_name, window) if not values: return { "buckets": [], "min": 0, "max": 0, "count": 0, } min_val = min(values) max_val = max(values) if min_val == max_val: return { "buckets": [{"start": min_val, "end": max_val, "count": len(values)}], "min": min_val, "max": max_val, "count": len(values), } bucket_size = (max_val - min_val) / buckets bucket_counts = [0] * buckets for value in values: bucket_index = min( int((value - min_val) / bucket_size), buckets - 1 ) bucket_counts[bucket_index] += 1 bucket_info = [] for i in range(buckets): start = min_val + i * bucket_size end = start + bucket_size bucket_info.append({ "start": round(start, 4), "end": round(end, 4), "count": bucket_counts[i], }) return { "buckets": bucket_info, "min": min_val, "max": max_val, "count": len(values), } def get_satellite_comparison( self, metric_name: str, satellite_ids: list[str], window: TimeWindow = TimeWindow.HOUR, ) -> dict[str, dict[str, float]]: """ Vergleicht Metrik über mehrere Satellites. Args: metric_name: Metrik-Name satellite_ids: Zu vergleichende Satellites window: Zeitfenster Returns: Dict mit Statistiken pro Satellite """ result: dict[str, dict[str, float]] = {} for sat_id in satellite_ids: values, _ = self._get_metric_values( metric_name, window, [sat_id] ) if values: result[sat_id] = { "current": values[0] if values else 0.0, "avg": statistics.mean(values) if values else 0.0, "min": min(values) if values else 0.0, "max": max(values) if values else 0.0, } else: result[sat_id] = { "current": 0.0, "avg": 0.0, "min": 0.0, "max": 0.0, } return result def get_health_summary(self) -> dict[str, Any]: """ Erstellt Gesundheits-Übersicht basierend auf Metriken. Returns: Gesundheits-Übersicht """ window = TimeWindow.FIVE_MINUTES avg_latency = self.aggregate("latency", AggregationType.AVG, window) p99_latency = self.aggregate("latency", AggregationType.P99, window) max_packet_loss = self.aggregate("packet_loss", AggregationType.MAX, window) total_errors = self.aggregate("error_count", AggregationType.SUM, window) # Gesundheitsstatus bestimmen health_status = "healthy" issues = [] if avg_latency.value > 100: health_status = "degraded" issues.append(f"Hohe durchschnittliche Latenz: {avg_latency.value:.1f}ms") if p99_latency.value > 500: health_status = "degraded" issues.append(f"Hohe P99-Latenz: {p99_latency.value:.1f}ms") if max_packet_loss.value > 5: health_status = "degraded" issues.append(f"Hoher Paketverlust: {max_packet_loss.value:.1f}%") if total_errors.value > 10: health_status = "unhealthy" issues.append(f"Viele Fehler: {int(total_errors.value)}") return { "status": health_status, "issues": issues, "metrics": { "avg_latency_ms": round(avg_latency.value, 2), "p99_latency_ms": round(p99_latency.value, 2), "max_packet_loss_percent": round(max_packet_loss.value, 2), "total_errors": int(total_errors.value), }, "satellite_count": avg_latency.sample_count, } def clear_cache(self) -> None: """Löscht den Cache.""" self._cache.clear()