# -*- coding: utf-8 -*- """ Graceful Shutdown Manager für geordnetes Herunterfahren. Ermöglicht das kontrollierte Stoppen von Services mit Timeouts und Abhängigkeitsberücksichtigung. """ from __future__ import annotations import asyncio import signal import sys from dataclasses import dataclass, field from datetime import datetime from enum import Enum, auto from typing import TYPE_CHECKING, Any, Callable, Coroutine from trixy_core.service.enums import ServicePriority, ServiceState if TYPE_CHECKING: from trixy_core.service.iservice import IService class ShutdownPhase(Enum): """ Phasen des Shutdown-Prozesses. """ NOT_STARTED = auto() """Shutdown noch nicht gestartet.""" PRE_SHUTDOWN = auto() """Vorbereitungsphase - Hooks werden ausgeführt.""" STOPPING_SERVICES = auto() """Services werden gestoppt.""" POST_SHUTDOWN = auto() """Aufräumphase - Finale Hooks.""" COMPLETED = auto() """Shutdown abgeschlossen.""" FAILED = auto() """Shutdown mit Fehlern beendet.""" class ShutdownReason(Enum): """ Grund für den Shutdown. """ NORMAL = auto() """Normaler Shutdown durch Benutzeranforderung.""" SIGNAL = auto() """Shutdown durch Signal (SIGTERM, SIGINT).""" ERROR = auto() """Shutdown durch kritischen Fehler.""" TIMEOUT = auto() """Shutdown durch Timeout.""" MANUAL = auto() """Manueller Shutdown durch API.""" @dataclass class ShutdownResult: """ Ergebnis eines Service-Shutdowns. """ service_name: str """Name des Services.""" success: bool """Ob der Shutdown erfolgreich war.""" duration: float """Dauer in Sekunden.""" error: Exception | None = None """Fehler, falls aufgetreten.""" timed_out: bool = False """Ob ein Timeout aufgetreten ist.""" def __str__(self) -> str: """String-Repräsentation.""" status = "OK" if self.success else "FAILED" if self.timed_out: status = "TIMEOUT" return f"{self.service_name}: {status} ({self.duration:.3f}s)" @dataclass class ShutdownReport: """ Gesamtbericht des Shutdown-Prozesses. """ reason: ShutdownReason """Grund für den Shutdown.""" phase: ShutdownPhase """Endphase des Shutdowns.""" started_at: datetime """Startzeitpunkt.""" completed_at: datetime | None = None """Endzeitpunkt.""" results: list[ShutdownResult] = field(default_factory=list) """Ergebnisse pro Service.""" errors: list[str] = field(default_factory=list) """Allgemeine Fehler.""" @property def duration(self) -> float: """Gesamtdauer in Sekunden.""" if self.completed_at is None: return 0.0 return (self.completed_at - self.started_at).total_seconds() @property def success(self) -> bool: """Ob der gesamte Shutdown erfolgreich war.""" return ( self.phase == ShutdownPhase.COMPLETED and all(r.success for r in self.results) ) @property def failed_services(self) -> list[str]: """Liste der fehlgeschlagenen Services.""" return [r.service_name for r in self.results if not r.success] def to_dict(self) -> dict[str, Any]: """Konvertiert in ein Dictionary.""" return { "reason": self.reason.name, "phase": self.phase.name, "success": self.success, "duration": self.duration, "started_at": self.started_at.isoformat(), "completed_at": self.completed_at.isoformat() if self.completed_at else None, "services": [ { "name": r.service_name, "success": r.success, "duration": r.duration, "timed_out": r.timed_out, "error": str(r.error) if r.error else None, } for r in self.results ], "failed_services": self.failed_services, "errors": self.errors, } @dataclass class ShutdownConfig: """ Konfiguration für den Shutdown-Manager. """ default_timeout: float = 30.0 """Standard-Timeout pro Service in Sekunden.""" total_timeout: float = 120.0 """Maximale Gesamtzeit für Shutdown.""" force_after_timeout: bool = True """Ob nach Timeout forciert wird.""" parallel_shutdown: bool = False """Ob Services parallel gestoppt werden (wenn möglich).""" respect_dependencies: bool = True """Ob Abhängigkeiten beim Shutdown berücksichtigt werden.""" service_timeouts: dict[str, float] = field(default_factory=dict) """Individuelle Timeouts pro Service.""" register_signals: bool = True """Ob Signal-Handler registriert werden.""" class GracefulShutdownManager: """ Manager für geordnetes Herunterfahren von Services. Stoppt Services in umgekehrter Startreihenfolge unter Berücksichtigung von Abhängigkeiten und Timeouts. Example: manager = GracefulShutdownManager(service_container) # Normaler Shutdown report = await manager.shutdown() # Mit Hooks manager.add_pre_shutdown_hook(save_state) manager.add_post_shutdown_hook(cleanup) # Signal-Handler automatisch registrieren manager.register_signal_handlers() """ def __init__( self, services: dict[str, "IService"] | None = None, config: ShutdownConfig | None = None, ) -> None: """ Initialisiert den Shutdown-Manager. Args: services: Dictionary der Services (Name → Service). config: Shutdown-Konfiguration. """ self._services = services or {} self._config = config or ShutdownConfig() self._phase = ShutdownPhase.NOT_STARTED self._reason: ShutdownReason | None = None self._pre_hooks: list[Callable[[], Coroutine[Any, Any, None]]] = [] self._post_hooks: list[Callable[[ShutdownReport], Coroutine[Any, Any, None]]] = [] self._shutdown_lock = asyncio.Lock() self._shutdown_event = asyncio.Event() if self._config.register_signals: self._register_default_signals() @property def phase(self) -> ShutdownPhase: """Aktuelle Shutdown-Phase.""" return self._phase @property def is_shutting_down(self) -> bool: """Prüft, ob ein Shutdown läuft.""" return self._phase not in ( ShutdownPhase.NOT_STARTED, ShutdownPhase.COMPLETED, ShutdownPhase.FAILED, ) def set_services(self, services: dict[str, "IService"]) -> None: """ Setzt die zu verwaltenden Services. Args: services: Dictionary der Services. """ self._services = services def add_pre_shutdown_hook( self, hook: Callable[[], Coroutine[Any, Any, None]], ) -> None: """ Fügt einen Pre-Shutdown-Hook hinzu. Args: hook: Async-Funktion die vor dem Shutdown ausgeführt wird. """ self._pre_hooks.append(hook) def add_post_shutdown_hook( self, hook: Callable[[ShutdownReport], Coroutine[Any, Any, None]], ) -> None: """ Fügt einen Post-Shutdown-Hook hinzu. Args: hook: Async-Funktion die nach dem Shutdown ausgeführt wird. """ self._post_hooks.append(hook) def get_timeout_for_service(self, service_name: str) -> float: """ Gibt den Timeout für einen Service zurück. Args: service_name: Name des Services. Returns: Timeout in Sekunden. """ return self._config.service_timeouts.get( service_name, self._config.default_timeout, ) def _register_default_signals(self) -> None: """Registriert Standard-Signal-Handler.""" try: loop = asyncio.get_running_loop() except RuntimeError: # Kein Event-Loop läuft - Handler später registrieren return if sys.platform == "win32": # Windows: add_signal_handler nicht verfügbar for sig in (signal.SIGINT, signal.SIGTERM): signal.signal(sig, lambda s, f, _loop=loop: _loop.call_soon_threadsafe( lambda _s=s: asyncio.create_task(self._signal_handler(_s)) )) else: # Unix-Signale for sig in (signal.SIGTERM, signal.SIGINT): loop.add_signal_handler( sig, lambda s=sig: asyncio.create_task( self._signal_handler(s) ), ) async def _signal_handler(self, sig: signal.Signals) -> None: """ Handler für System-Signale. Args: sig: Empfangenes Signal. """ await self.shutdown(ShutdownReason.SIGNAL) def _get_shutdown_order(self) -> list[str]: """ Berechnet die Shutdown-Reihenfolge. Returns: Liste von Service-Namen in Shutdown-Reihenfolge. """ if not self._config.respect_dependencies: # Einfach nach Priorität sortieren (umgekehrt) return sorted( self._services.keys(), key=lambda n: ( -self._services[n].PRIORITY, # Höhere Priorität zuerst stoppen n, ), ) # Topologische Sortierung mit Abhängigkeiten # Services ohne Abhängige zuerst stoppen dependents: dict[str, set[str]] = {name: set() for name in self._services} for name, service in self._services.items(): for dep in service.DEPENDENCIES: if dep in dependents: dependents[dep].add(name) result: list[str] = [] remaining = set(self._services.keys()) while remaining: # Finde Services ohne laufende Abhängige can_stop = [ name for name in remaining if not (dependents[name] & remaining) ] if not can_stop: # Zyklische Abhängigkeit - Rest nach Priorität can_stop = sorted( remaining, key=lambda n: (-self._services[n].PRIORITY, n), ) # Nach Priorität sortieren (höhere zuerst) can_stop.sort(key=lambda n: (-self._services[n].PRIORITY, n)) for name in can_stop: result.append(name) remaining.discard(name) return result async def _stop_service( self, service: "IService", timeout: float, ) -> ShutdownResult: """ Stoppt einen einzelnen Service. Args: service: Der zu stoppende Service. timeout: Timeout in Sekunden. Returns: Ergebnis des Shutdowns. """ start_time = datetime.now() name = service.name try: # Pre-Stop Hook await asyncio.wait_for( service.on_pre_stop(), timeout=timeout / 3, ) # Stop await asyncio.wait_for( service.stop(), timeout=timeout / 3, ) # Post-Stop Hook await asyncio.wait_for( service.on_post_stop(), timeout=timeout / 3, ) duration = (datetime.now() - start_time).total_seconds() return ShutdownResult( service_name=name, success=True, duration=duration, ) except asyncio.TimeoutError: duration = (datetime.now() - start_time).total_seconds() return ShutdownResult( service_name=name, success=False, duration=duration, timed_out=True, ) except Exception as e: duration = (datetime.now() - start_time).total_seconds() return ShutdownResult( service_name=name, success=False, duration=duration, error=e, ) async def _run_pre_hooks(self) -> list[str]: """ Führt Pre-Shutdown-Hooks aus. Returns: Liste von Fehlermeldungen. """ errors: list[str] = [] for hook in self._pre_hooks: try: await asyncio.wait_for( hook(), timeout=self._config.default_timeout, ) except asyncio.TimeoutError: errors.append(f"Pre-Hook Timeout: {hook.__name__}") except Exception as e: errors.append(f"Pre-Hook Fehler ({hook.__name__}): {e}") return errors async def _run_post_hooks(self, report: ShutdownReport) -> list[str]: """ Führt Post-Shutdown-Hooks aus. Args: report: Der Shutdown-Report. Returns: Liste von Fehlermeldungen. """ errors: list[str] = [] for hook in self._post_hooks: try: await asyncio.wait_for( hook(report), timeout=self._config.default_timeout, ) except asyncio.TimeoutError: errors.append(f"Post-Hook Timeout: {hook.__name__}") except Exception as e: errors.append(f"Post-Hook Fehler ({hook.__name__}): {e}") return errors async def shutdown( self, reason: ShutdownReason = ShutdownReason.NORMAL, ) -> ShutdownReport: """ Führt einen geordneten Shutdown durch. Args: reason: Grund für den Shutdown. Returns: Bericht über den Shutdown. """ async with self._shutdown_lock: if self.is_shutting_down: # Bereits im Shutdown await self._shutdown_event.wait() return ShutdownReport( reason=reason, phase=self._phase, started_at=datetime.now(), completed_at=datetime.now(), errors=["Shutdown bereits in Bearbeitung"], ) self._reason = reason self._shutdown_event.clear() report = ShutdownReport( reason=reason, phase=ShutdownPhase.PRE_SHUTDOWN, started_at=datetime.now(), ) try: # Phase 1: Pre-Hooks self._phase = ShutdownPhase.PRE_SHUTDOWN hook_errors = await self._run_pre_hooks() report.errors.extend(hook_errors) # Phase 2: Services stoppen self._phase = ShutdownPhase.STOPPING_SERVICES shutdown_order = self._get_shutdown_order() if self._config.parallel_shutdown: # Paralleles Stoppen (nur wenn keine Abhängigkeiten) tasks = [] for name in shutdown_order: service = self._services.get(name) if service and service.state == ServiceState.RUNNING: timeout = self.get_timeout_for_service(name) tasks.append(self._stop_service(service, timeout)) if tasks: results = await asyncio.gather(*tasks, return_exceptions=True) for result in results: if isinstance(result, ShutdownResult): report.results.append(result) elif isinstance(result, Exception): report.errors.append(str(result)) else: # Sequentielles Stoppen for name in shutdown_order: service = self._services.get(name) if service and service.state == ServiceState.RUNNING: timeout = self.get_timeout_for_service(name) result = await self._stop_service(service, timeout) report.results.append(result) # Phase 3: Post-Hooks self._phase = ShutdownPhase.POST_SHUTDOWN post_errors = await self._run_post_hooks(report) report.errors.extend(post_errors) # Abschluss self._phase = ShutdownPhase.COMPLETED report.phase = ShutdownPhase.COMPLETED except Exception as e: self._phase = ShutdownPhase.FAILED report.phase = ShutdownPhase.FAILED report.errors.append(f"Kritischer Fehler: {e}") finally: report.completed_at = datetime.now() self._shutdown_event.set() return report async def wait_for_shutdown(self) -> None: """Wartet auf das Ende des Shutdowns.""" await self._shutdown_event.wait() def request_shutdown( self, reason: ShutdownReason = ShutdownReason.MANUAL, ) -> None: """ Fordert einen Shutdown an (nicht-blockierend). Args: reason: Grund für den Shutdown. """ asyncio.create_task(self.shutdown(reason))