# -*- coding: utf-8 -*- """ Timing-Middleware für Performance-Messung. Misst und protokolliert Event-Verarbeitungszeiten. """ from __future__ import annotations import statistics import time from collections import defaultdict from dataclasses import dataclass, field from datetime import datetime from typing import TYPE_CHECKING, Any, Callable from trixy_core.events.middleware.base import ( EventMiddleware, MiddlewareContext, MiddlewareResult, MiddlewareAction, ) if TYPE_CHECKING: from trixy_core.events.event_data.base import EventData @dataclass class TimingStats: """ Statistiken für Event-Verarbeitungszeiten. """ event_name: str """Name des Events.""" count: int = 0 """Anzahl der Verarbeitungen.""" total_ms: float = 0.0 """Gesamtzeit in Millisekunden.""" min_ms: float = float("inf") """Minimale Zeit.""" max_ms: float = 0.0 """Maximale Zeit.""" last_ms: float = 0.0 """Letzte Verarbeitungszeit.""" _times: list[float] = field(default_factory=list, repr=False) """Alle Zeiten für Perzentil-Berechnung.""" @property def avg_ms(self) -> float: """Durchschnittliche Zeit in Millisekunden.""" if self.count == 0: return 0.0 return self.total_ms / self.count @property def p50_ms(self) -> float: """50. Perzentil (Median).""" if not self._times: return 0.0 return statistics.median(self._times) @property def p90_ms(self) -> float: """90. Perzentil.""" if not self._times: return 0.0 return self._get_percentile(90) @property def p99_ms(self) -> float: """99. Perzentil.""" if not self._times: return 0.0 return self._get_percentile(99) def _get_percentile(self, percentile: float) -> float: """ Berechnet ein Perzentil. Args: percentile: Perzentil (0-100). Returns: Wert am Perzentil. """ if not self._times: return 0.0 sorted_times = sorted(self._times) index = (percentile / 100) * (len(sorted_times) - 1) lower = int(index) upper = min(lower + 1, len(sorted_times) - 1) fraction = index - lower return sorted_times[lower] * (1 - fraction) + sorted_times[upper] * fraction def record(self, duration_ms: float) -> None: """ Zeichnet eine Verarbeitungszeit auf. Args: duration_ms: Dauer in Millisekunden. """ self.count += 1 self.total_ms += duration_ms self.last_ms = duration_ms self.min_ms = min(self.min_ms, duration_ms) self.max_ms = max(self.max_ms, duration_ms) # Begrenzte Historie für Perzentile self._times.append(duration_ms) if len(self._times) > 1000: self._times = self._times[-500:] def to_dict(self) -> dict[str, Any]: """Konvertiert in ein Dictionary.""" return { "event_name": self.event_name, "count": self.count, "total_ms": round(self.total_ms, 2), "avg_ms": round(self.avg_ms, 2), "min_ms": round(self.min_ms, 2) if self.min_ms != float("inf") else 0, "max_ms": round(self.max_ms, 2), "p50_ms": round(self.p50_ms, 2), "p90_ms": round(self.p90_ms, 2), "p99_ms": round(self.p99_ms, 2), "last_ms": round(self.last_ms, 2), } class TimingMiddleware(EventMiddleware): """ Middleware für Event-Timing-Messungen. Misst die Verarbeitungszeit jedes Events und sammelt Statistiken für Performance-Analyse. Example: middleware = TimingMiddleware( slow_threshold_ms=100, on_slow_event=lambda name, ms: print(f"SLOW: {name} took {ms}ms"), ) pipeline.add(middleware) # Später: Statistiken abrufen stats = middleware.get_stats("my_event") print(f"Average: {stats.avg_ms}ms") """ TIMING_KEY = "_timing_start_ns" def __init__( self, name: str = "TimingMiddleware", enabled: bool = True, slow_threshold_ms: float = 100.0, on_slow_event: Callable[[str, float], None] | None = None, track_all: bool = True, max_events: int = 1000, ) -> None: """ Initialisiert die Timing-Middleware. Args: name: Name der Middleware. enabled: Ob aktiv. slow_threshold_ms: Schwellenwert für langsame Events (ms). on_slow_event: Callback für langsame Events (name, duration_ms). track_all: Ob alle Events erfasst werden. max_events: Maximale Anzahl erfasster Event-Typen. """ super().__init__(name, enabled) self._slow_threshold_ms = slow_threshold_ms self._on_slow_event = on_slow_event self._track_all = track_all self._max_events = max_events self._stats: dict[str, TimingStats] = {} self._global_stats = TimingStats(event_name="__global__") @property def slow_threshold_ms(self) -> float: """Schwellenwert für langsame Events.""" return self._slow_threshold_ms @slow_threshold_ms.setter def slow_threshold_ms(self, value: float) -> None: """Setzt den Schwellenwert.""" self._slow_threshold_ms = value def get_stats(self, event_name: str) -> TimingStats | None: """ Gibt die Statistiken für ein Event zurück. Args: event_name: Name des Events. Returns: TimingStats oder None. """ return self._stats.get(event_name) def get_all_stats(self) -> dict[str, TimingStats]: """ Gibt alle Event-Statistiken zurück. Returns: Dictionary von Event-Namen zu Statistiken. """ return dict(self._stats) def get_global_stats(self) -> TimingStats: """ Gibt die globalen Statistiken zurück. Returns: Globale TimingStats. """ return self._global_stats def get_slowest_events(self, n: int = 10) -> list[tuple[str, float]]: """ Gibt die langsamsten Events zurück. Args: n: Anzahl der Events. Returns: Liste von (event_name, avg_ms) Tupeln. """ sorted_events = sorted( self._stats.items(), key=lambda x: x[1].avg_ms, reverse=True, ) return [(name, stats.avg_ms) for name, stats in sorted_events[:n]] def reset_stats(self, event_name: str | None = None) -> None: """ Setzt Statistiken zurück. Args: event_name: Spezifisches Event oder None für alle. """ if event_name: if event_name in self._stats: del self._stats[event_name] else: self._stats.clear() self._global_stats = TimingStats(event_name="__global__") async def before( self, context: MiddlewareContext, ) -> MiddlewareResult: """ Startet die Zeitmessung. Args: context: Der Middleware-Kontext. Returns: Ergebnis. """ # Speichere Startzeit im Kontext context.set_property(self.TIMING_KEY, time.perf_counter_ns()) return MiddlewareResult() async def after( self, context: MiddlewareContext, ) -> None: """ Beendet die Zeitmessung und zeichnet Statistiken auf. Args: context: Der Middleware-Kontext. """ start_ns = context.get_property(self.TIMING_KEY) if start_ns is None: return duration_ns = time.perf_counter_ns() - start_ns duration_ms = duration_ns / 1_000_000 # Statistiken aktualisieren if self._track_all: if context.event_name not in self._stats: if len(self._stats) < self._max_events: self._stats[context.event_name] = TimingStats( event_name=context.event_name ) if context.event_name in self._stats: self._stats[context.event_name].record(duration_ms) self._global_stats.record(duration_ms) # Slow-Event-Callback if duration_ms >= self._slow_threshold_ms and self._on_slow_event: self._on_slow_event(context.event_name, duration_ms) # Speichere Dauer im Kontext für andere Middlewares context.set_property("timing_duration_ms", duration_ms) def to_dict(self) -> dict[str, Any]: """ Exportiert alle Statistiken als Dictionary. Returns: Dictionary mit allen Statistiken. """ return { "global": self._global_stats.to_dict(), "events": { name: stats.to_dict() for name, stats in self._stats.items() }, "slow_threshold_ms": self._slow_threshold_ms, "slowest_events": self.get_slowest_events(5), } class TimeoutMiddleware(EventMiddleware): """ Middleware für Event-Timeouts. Bricht Events ab, die zu lange dauern. Example: middleware = TimeoutMiddleware( default_timeout_ms=5000, per_event_timeouts={"slow_event": 10000}, ) """ def __init__( self, name: str = "TimeoutMiddleware", enabled: bool = True, default_timeout_ms: float = 5000.0, per_event_timeouts: dict[str, float] | None = None, on_timeout: Callable[[str, float], None] | None = None, ) -> None: """ Initialisiert die Timeout-Middleware. Args: name: Name der Middleware. enabled: Ob aktiv. default_timeout_ms: Standard-Timeout in Millisekunden. per_event_timeouts: Spezifische Timeouts pro Event. on_timeout: Callback bei Timeout. """ super().__init__(name, enabled) self._default_timeout_ms = default_timeout_ms self._per_event_timeouts = per_event_timeouts or {} self._on_timeout = on_timeout def get_timeout(self, event_name: str) -> float: """ Gibt den Timeout für ein Event zurück. Args: event_name: Name des Events. Returns: Timeout in Millisekunden. """ return self._per_event_timeouts.get(event_name, self._default_timeout_ms) async def before( self, context: MiddlewareContext, ) -> MiddlewareResult: """Speichert Timeout-Informationen.""" timeout_ms = self.get_timeout(context.event_name) context.set_property("_timeout_ms", timeout_ms) context.set_property("_timeout_deadline", time.perf_counter_ns() + (timeout_ms * 1_000_000)) return MiddlewareResult() async def after( self, context: MiddlewareContext, ) -> None: """Prüft auf Timeout.""" deadline = context.get_property("_timeout_deadline") if deadline is None: return if time.perf_counter_ns() > deadline: timeout_ms = context.get_property("_timeout_ms", 0) if self._on_timeout: self._on_timeout(context.event_name, context.elapsed_ms) # Markiere Event als abgebrochen wegen Timeout context.event_data.metadata["timed_out"] = True context.event_data.metadata["timeout_ms"] = timeout_ms