# -*- coding: utf-8 -*- """ Failover-Manager für automatisches Satellite-Failover. Überwacht Satellites und führt automatisches Failover zu Backup-Satellites durch. """ import asyncio from dataclasses import dataclass, field from datetime import datetime from enum import IntEnum, auto from typing import Any, Callable, TYPE_CHECKING if TYPE_CHECKING: from trixy_core.satellite.satellite import Satellite from trixy_core.satellite.failover.backup import BackupRegistry class FailoverState(IntEnum): """Zustand des Failover-Prozesses.""" IDLE = auto() # Kein Failover aktiv DETECTING = auto() # Ausfall wird erkannt FAILING_OVER = auto() # Failover in Bearbeitung ACTIVE = auto() # Failover aktiv RECOVERING = auto() # Wiederherstellung FAILED = auto() # Failover fehlgeschlagen @dataclass class FailoverConfig: """ Konfiguration für Failover. Attributes: detection_timeout: Zeit ohne Heartbeat bis Ausfall erkannt wird failover_timeout: Maximale Zeit für Failover-Prozess recovery_timeout: Zeit bis Wiederherstellung versucht wird max_failover_attempts: Maximale Failover-Versuche auto_recovery: Automatische Wiederherstellung wenn Original zurück health_threshold: Health-Score unter dem Failover ausgelöst wird """ detection_timeout: float = 10.0 failover_timeout: float = 30.0 recovery_timeout: float = 60.0 max_failover_attempts: int = 3 auto_recovery: bool = True health_threshold: float = 40.0 @dataclass class FailoverEvent: """ Ein Failover-Ereignis. Attributes: primary_id: ID des ausgefallenen Satellites backup_id: ID des Backup-Satellites state: Aktueller Zustand reason: Grund für Failover started_at: Startzeitpunkt completed_at: Abschlusszeitpunkt attempts: Anzahl Versuche """ primary_id: str backup_id: str | None state: FailoverState reason: str = "" started_at: datetime = field(default_factory=datetime.now) completed_at: datetime | None = None attempts: int = 0 @property def duration(self) -> float | None: """Dauer des Failovers in Sekunden.""" if self.completed_at is None: return None return (self.completed_at - self.started_at).total_seconds() def to_dict(self) -> dict[str, Any]: """Konvertiert zu Dictionary.""" return { "primary_id": self.primary_id, "backup_id": self.backup_id, "state": self.state.name, "reason": self.reason, "started_at": self.started_at.isoformat(), "completed_at": self.completed_at.isoformat() if self.completed_at else None, "duration": self.duration, "attempts": self.attempts, } class FailoverManager: """ Verwaltet Failover-Prozesse für Satellites. Überwacht die Gesundheit von Satellites und führt automatisch Failover zu Backup-Satellites durch. Example: manager = FailoverManager( backup_registry=backup_registry, config=FailoverConfig( detection_timeout=10.0, auto_recovery=True ) ) # Callback für Failover-Events @manager.on_failover async def handle_failover(event: FailoverEvent): print(f"Failover: {event.primary_id} -> {event.backup_id}") # Überwachung starten await manager.start_monitoring() # Manuelles Failover auslösen event = await manager.trigger_failover( primary_id="satellite_1", reason="Manuelle Auslösung" ) """ def __init__( self, backup_registry: "BackupRegistry", config: FailoverConfig | None = None, get_satellite: Callable[[str], "Satellite | None"] | None = None, ) -> None: """ Initialisiert den Failover-Manager. Args: backup_registry: Registry für Backup-Zuweisungen config: Failover-Konfiguration get_satellite: Funktion zum Abrufen eines Satellites nach ID """ self._registry = backup_registry self._config = config or FailoverConfig() self._get_satellite = get_satellite self._active_failovers: dict[str, FailoverEvent] = {} self._history: list[FailoverEvent] = [] self._callbacks: list[Callable[[FailoverEvent], Any]] = [] self._monitoring_task: asyncio.Task | None = None self._monitored_satellites: set[str] = set() @property def config(self) -> FailoverConfig: """Failover-Konfiguration.""" return self._config @property def active_failovers(self) -> dict[str, FailoverEvent]: """Aktive Failover-Events.""" return self._active_failovers.copy() def on_failover( self, callback: Callable[[FailoverEvent], Any] ) -> Callable[[FailoverEvent], Any]: """ Registriert einen Callback für Failover-Events. Kann als Decorator verwendet werden. Args: callback: Die aufzurufende Funktion Returns: Der registrierte Callback """ self._callbacks.append(callback) return callback def add_monitored_satellite(self, satellite_id: str) -> None: """Fügt einen Satellite zur Überwachung hinzu.""" self._monitored_satellites.add(satellite_id) def remove_monitored_satellite(self, satellite_id: str) -> None: """Entfernt einen Satellite aus der Überwachung.""" self._monitored_satellites.discard(satellite_id) async def trigger_failover( self, primary_id: str, reason: str = "Manual trigger", preferred_backup: str | None = None, ) -> FailoverEvent | None: """ Löst manuell ein Failover aus. Args: primary_id: ID des Primär-Satellites reason: Grund für Failover preferred_backup: Optionale bevorzugte Backup-ID Returns: FailoverEvent oder None wenn kein Backup verfügbar """ # Prüfe ob bereits aktiv if primary_id in self._active_failovers: return self._active_failovers[primary_id] # Backup finden backup_id = preferred_backup if backup_id is None: backup = self._registry.get_best_backup(primary_id) backup_id = backup.backup_id if backup else None if backup_id is None: # Kein Backup verfügbar event = FailoverEvent( primary_id=primary_id, backup_id=None, state=FailoverState.FAILED, reason=f"{reason} - Kein Backup verfügbar", ) self._history.append(event) await self._notify(event) return event # Failover-Event erstellen event = FailoverEvent( primary_id=primary_id, backup_id=backup_id, state=FailoverState.FAILING_OVER, reason=reason, ) self._active_failovers[primary_id] = event try: # Failover durchführen success = await self._execute_failover(event) if success: event.state = FailoverState.ACTIVE event.completed_at = datetime.now() else: event.state = FailoverState.FAILED event.completed_at = datetime.now() except Exception as e: event.state = FailoverState.FAILED event.reason = f"{event.reason} - Fehler: {str(e)}" event.completed_at = datetime.now() self._history.append(event) await self._notify(event) return event async def _execute_failover(self, event: FailoverEvent) -> bool: """ Führt das eigentliche Failover durch. Args: event: Das Failover-Event Returns: True wenn erfolgreich """ if self._get_satellite is None: return False backup = self._get_satellite(event.backup_id) if backup is None or not backup.is_connected: return False # Hier würde die eigentliche Failover-Logik implementiert: # - Aktive Konversation übertragen # - Audio-Streams umleiten # - Clients benachrichtigen # etc. event.attempts += 1 # Für jetzt: Erfolg wenn Backup verbunden return True async def recover(self, primary_id: str) -> bool: """ Versucht, zum Primär-Satellite zurückzukehren. Args: primary_id: ID des Primär-Satellites Returns: True wenn erfolgreich """ if primary_id not in self._active_failovers: return False event = self._active_failovers[primary_id] if event.state != FailoverState.ACTIVE: return False # Prüfe ob Primary wieder verfügbar if self._get_satellite: primary = self._get_satellite(primary_id) if primary is None or not primary.is_connected: return False # Recovery durchführen event.state = FailoverState.RECOVERING try: # Recovery-Logik hier # - Konversation zurück übertragen # - Streams umleiten # etc. # Erfolgreich del self._active_failovers[primary_id] recovery_event = FailoverEvent( primary_id=primary_id, backup_id=event.backup_id, state=FailoverState.IDLE, reason="Recovery erfolgreich", ) recovery_event.completed_at = datetime.now() self._history.append(recovery_event) await self._notify(recovery_event) return True except Exception: event.state = FailoverState.ACTIVE return False async def _notify(self, event: FailoverEvent) -> None: """Benachrichtigt alle Callbacks.""" for callback in self._callbacks: try: result = callback(event) if asyncio.iscoroutine(result): await result except Exception: pass async def start_monitoring(self) -> None: """Startet die automatische Überwachung.""" if self._monitoring_task is not None: return self._monitoring_task = asyncio.create_task(self._monitoring_loop()) async def stop_monitoring(self) -> None: """Stoppt die automatische Überwachung.""" if self._monitoring_task is not None: self._monitoring_task.cancel() try: await self._monitoring_task except asyncio.CancelledError: pass self._monitoring_task = None async def _monitoring_loop(self) -> None: """Überwachungs-Loop.""" while True: try: await asyncio.sleep(1.0) # Prüfintervall for satellite_id in list(self._monitored_satellites): await self._check_satellite(satellite_id) # Auto-Recovery prüfen if self._config.auto_recovery: for primary_id in list(self._active_failovers.keys()): await self._check_recovery(primary_id) except asyncio.CancelledError: break except Exception: pass async def _check_satellite(self, satellite_id: str) -> None: """Prüft einen einzelnen Satellite.""" if satellite_id in self._active_failovers: return if self._get_satellite is None: return satellite = self._get_satellite(satellite_id) if satellite is None: return # Prüfe Verbindungsstatus if not satellite.is_connected: await self.trigger_failover( satellite_id, reason="Verbindung verloren" ) return # Prüfe Heartbeat if satellite.last_heartbeat: age = (datetime.now() - satellite.last_heartbeat).total_seconds() if age > self._config.detection_timeout: await self.trigger_failover( satellite_id, reason=f"Heartbeat-Timeout ({age:.1f}s)" ) async def _check_recovery(self, primary_id: str) -> None: """Prüft, ob Recovery möglich ist.""" event = self._active_failovers.get(primary_id) if event is None or event.state != FailoverState.ACTIVE: return if self._get_satellite is None: return primary = self._get_satellite(primary_id) if primary and primary.is_connected: # Prüfe ob stabil genug für Recovery if primary.last_heartbeat: age = (datetime.now() - primary.last_heartbeat).total_seconds() if age <= self._config.detection_timeout / 2: await self.recover(primary_id) def get_history( self, satellite_id: str | None = None, limit: int = 100 ) -> list[FailoverEvent]: """ Gibt die Failover-Historie zurück. Args: satellite_id: Optionaler Filter nach Satellite limit: Maximale Anzahl Einträge Returns: Liste der Failover-Events """ history = self._history if satellite_id: history = [e for e in history if e.primary_id == satellite_id] return history[-limit:] def get_statistics(self) -> dict[str, Any]: """Gibt Statistiken zurück.""" total = len(self._history) successful = sum(1 for e in self._history if e.state == FailoverState.ACTIVE) failed = sum(1 for e in self._history if e.state == FailoverState.FAILED) return { "total_failovers": total, "successful": successful, "failed": failed, "success_rate": successful / max(1, total), "active_failovers": len(self._active_failovers), "monitored_satellites": len(self._monitored_satellites), }