Trixy7/trixy_core/events/async_queue/worker.py

# -*- coding: utf-8 -*-
"""
Worker-Pool für Event-Verarbeitung.

Verwaltet Worker für parallele Event-Verarbeitung.
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Callable, Coroutine
import uuid

if TYPE_CHECKING:
    from trixy_core.events.event_data.base import EventData
    from trixy_core.events.async_queue.queue import AsyncEventQueue, QueuedEvent


class WorkerState(Enum):
    """
    Zustand eines Workers.
    """

    IDLE = auto()
    """Worker wartet auf Arbeit."""

    BUSY = auto()
    """Worker verarbeitet Event."""

    STOPPED = auto()
    """Worker gestoppt."""

    FAILED = auto()
    """Worker fehlerhaft."""


@dataclass
class WorkerConfig:
    """
    Konfiguration für Worker.
    """

    idle_timeout_ms: float = 30000.0
    """Timeout für Idle-Worker."""

    max_consecutive_errors: int = 5
    """Maximale aufeinanderfolgende Fehler."""

    error_cooldown_ms: float = 1000.0
    """Wartezeit nach Fehler."""

    enable_metrics: bool = True
    """Metriken aktivieren."""


@dataclass
class WorkerStats:
    """
    Statistiken eines Workers.
    """

    worker_id: str
    """ID des Workers."""

    state: WorkerState = WorkerState.IDLE
    """Aktueller Zustand."""

    events_processed: int = 0
    """Verarbeitete Events."""

    events_failed: int = 0
    """Fehlgeschlagene Events."""

    total_processing_ms: float = 0.0
    """Gesamte Verarbeitungszeit."""

    last_event_at: datetime | None = None
    """Zeitpunkt des letzten Events."""

    consecutive_errors: int = 0
    """Aufeinanderfolgende Fehler."""

    started_at: datetime | None = None
    """Startzeit."""

    @property
    def avg_processing_ms(self) -> float:
        """Durchschnittliche Verarbeitungszeit."""
        if self.events_processed == 0:
            return 0.0
        return self.total_processing_ms / self.events_processed

    @property
    def uptime_seconds(self) -> float:
        """Laufzeit in Sekunden."""
        if self.started_at is None:
            return 0.0
        return (datetime.now() - self.started_at).total_seconds()

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert in ein Dictionary."""
        return {
            "worker_id": self.worker_id,
            "state": self.state.name,
            "events_processed": self.events_processed,
            "events_failed": self.events_failed,
            "avg_processing_ms": self.avg_processing_ms,
            "uptime_seconds": self.uptime_seconds,
            "consecutive_errors": self.consecutive_errors,
        }


class Worker:
    """
    Ein einzelner Event-Worker.

    Verarbeitet Events aus einer Queue.

    Example:
        worker = Worker(
            worker_id="worker-1",
            queue=event_queue,
            handler=process_event,
        )

        await worker.start()
        # ...
        await worker.stop()
    """

    def __init__(
        self,
        worker_id: str,
        queue: "AsyncEventQueue",
        handler: Callable[[str, "EventData"], Coroutine[Any, Any, None]],
        config: WorkerConfig | None = None,
    ) -> None:
        """
        Initialisiert den Worker.

        Args:
            worker_id: ID des Workers.
            queue: Die Event-Queue.
            handler: Handler-Funktion.
            config: Konfiguration.
        """
        self._id = worker_id
        self._queue = queue
        self._handler = handler
        self._config = config or WorkerConfig()

        self._task: asyncio.Task | None = None
        self._running = False
        self._stats = WorkerStats(worker_id=worker_id)

    @property
    def id(self) -> str:
        """Worker-ID."""
        return self._id

    @property
    def state(self) -> WorkerState:
        """Aktueller Zustand."""
        return self._stats.state

    @property
    def is_running(self) -> bool:
        """Ob Worker läuft."""
        return self._running

    @property
    def stats(self) -> WorkerStats:
        """Statistiken."""
        return self._stats

    async def start(self) -> None:
        """Startet den Worker."""
        if self._running:
            return

        self._running = True
        self._stats.state = WorkerState.IDLE
        self._stats.started_at = datetime.now()
        self._task = asyncio.create_task(self._run())

    async def stop(self, timeout: float = 5.0) -> None:
        """
        Stoppt den Worker.

        Args:
            timeout: Timeout in Sekunden.
        """
        self._running = False
        self._stats.state = WorkerState.STOPPED

        if self._task:
            try:
                await asyncio.wait_for(self._task, timeout=timeout)
            except asyncio.TimeoutError:
                self._task.cancel()
            self._task = None

    async def _run(self) -> None:
        """Worker-Hauptschleife."""
        while self._running:
            try:
                # Event holen
                queued = await self._queue.dequeue(
                    timeout=self._config.idle_timeout_ms / 1000
                )

                if queued is None:
                    # Timeout - weiter warten
                    continue

                # Event verarbeiten
                self._stats.state = WorkerState.BUSY
                start_time = datetime.now()

                try:
                    await self._handler(queued.event_name, queued.event_data)

                    # Erfolg
                    self._stats.events_processed += 1
                    self._stats.consecutive_errors = 0

                except Exception as e:
                    # Fehler
                    self._stats.events_failed += 1
                    self._stats.consecutive_errors += 1

                    if self._stats.consecutive_errors >= self._config.max_consecutive_errors:
                        self._stats.state = WorkerState.FAILED
                        await asyncio.sleep(self._config.error_cooldown_ms / 1000)
                        self._stats.state = WorkerState.IDLE

                finally:
                    # Zeiterfassung
                    duration_ms = (datetime.now() - start_time).total_seconds() * 1000
                    self._stats.total_processing_ms += duration_ms
                    self._stats.last_event_at = datetime.now()
                    self._stats.state = WorkerState.IDLE

            except asyncio.CancelledError:
                break
            except Exception:
                await asyncio.sleep(0.1)


@dataclass
class PoolStats:
    """
    Statistiken des Worker-Pools.
    """

    total_workers: int = 0
    """Gesamtzahl Worker."""

    active_workers: int = 0
    """Aktive Worker."""

    idle_workers: int = 0
    """Idle Worker."""

    failed_workers: int = 0
    """Fehlerhafte Worker."""

    total_events_processed: int = 0
    """Gesamtzahl verarbeiteter Events."""

    total_events_failed: int = 0
    """Gesamtzahl fehlgeschlagener Events."""

    avg_processing_ms: float = 0.0
    """Durchschnittliche Verarbeitungszeit."""

    workers: list[dict[str, Any]] = field(default_factory=list)
    """Statistiken aller Worker."""


class WorkerPool:
    """
    Pool von Event-Workern.

    Verwaltet mehrere Worker für parallele Verarbeitung.

    Example:
        pool = WorkerPool(
            queue=event_queue,
            handler=process_event,
            min_workers=2,
            max_workers=10,
        )

        await pool.start()

        # Dynamisch skalieren
        await pool.scale_up(2)
        await pool.scale_down(1)

        # Statistiken
        stats = pool.get_stats()
        print(f"Active workers: {stats.active_workers}")

        await pool.stop()
    """

    def __init__(
        self,
        queue: "AsyncEventQueue",
        handler: Callable[[str, "EventData"], Coroutine[Any, Any, None]],
        min_workers: int = 1,
        max_workers: int = 10,
        config: WorkerConfig | None = None,
    ) -> None:
        """
        Initialisiert den Pool.

        Args:
            queue: Die Event-Queue.
            handler: Handler-Funktion.
            min_workers: Minimale Anzahl Worker.
            max_workers: Maximale Anzahl Worker.
            config: Worker-Konfiguration.
        """
        self._queue = queue
        self._handler = handler
        self._min_workers = min_workers
        self._max_workers = max_workers
        self._config = config or WorkerConfig()

        self._workers: dict[str, Worker] = {}
        self._running = False
        self._worker_counter = 0
        self._lock = asyncio.Lock()

    @property
    def worker_count(self) -> int:
        """Anzahl der Worker."""
        return len(self._workers)

    @property
    def is_running(self) -> bool:
        """Ob Pool läuft."""
        return self._running

    def _create_worker_id(self) -> str:
        """Erstellt eine neue Worker-ID."""
        self._worker_counter += 1
        return f"worker-{self._worker_counter}"

    async def _create_worker(self) -> Worker:
        """Erstellt und startet einen neuen Worker."""
        worker_id = self._create_worker_id()
        worker = Worker(
            worker_id=worker_id,
            queue=self._queue,
            handler=self._handler,
            config=self._config,
        )
        await worker.start()
        self._workers[worker_id] = worker
        return worker

    async def _remove_worker(self, worker_id: str) -> bool:
        """Entfernt einen Worker."""
        worker = self._workers.pop(worker_id, None)
        if worker:
            await worker.stop()
            return True
        return False

    async def start(self) -> None:
        """Startet den Pool mit minimaler Worker-Anzahl."""
        if self._running:
            return

        self._running = True

        async with self._lock:
            for _ in range(self._min_workers):
                await self._create_worker()

    async def stop(self, timeout: float = 10.0) -> None:
        """
        Stoppt den Pool.

        Args:
            timeout: Timeout pro Worker.
        """
        self._running = False

        async with self._lock:
            for worker in list(self._workers.values()):
                await worker.stop(timeout=timeout)
            self._workers.clear()

    async def scale_up(self, count: int = 1) -> int:
        """
        Fügt Worker hinzu.

        Args:
            count: Anzahl neuer Worker.

        Returns:
            Anzahl tatsächlich hinzugefügter Worker.
        """
        async with self._lock:
            added = 0
            for _ in range(count):
                if len(self._workers) >= self._max_workers:
                    break
                await self._create_worker()
                added += 1
            return added

    async def scale_down(self, count: int = 1) -> int:
        """
        Entfernt Worker.

        Args:
            count: Anzahl zu entfernender Worker.

        Returns:
            Anzahl tatsächlich entfernter Worker.
        """
        async with self._lock:
            removed = 0

            # Idle Worker bevorzugt entfernen
            idle_workers = [
                w for w in self._workers.values()
                if w.state == WorkerState.IDLE
            ]

            for worker in idle_workers[:count]:
                if len(self._workers) <= self._min_workers:
                    break
                await self._remove_worker(worker.id)
                removed += 1

            return removed

    async def scale_to(self, target: int) -> int:
        """
        Skaliert auf eine bestimmte Worker-Anzahl.

        Args:
            target: Ziel-Anzahl.

        Returns:
            Differenz (positiv = hinzugefügt, negativ = entfernt).
        """
        target = max(self._min_workers, min(self._max_workers, target))
        current = len(self._workers)

        if target > current:
            return await self.scale_up(target - current)
        elif target < current:
            return -(await self.scale_down(current - target))
        return 0

    def get_worker(self, worker_id: str) -> Worker | None:
        """
        Gibt einen Worker zurück.

        Args:
            worker_id: ID des Workers.

        Returns:
            Worker oder None.
        """
        return self._workers.get(worker_id)

    def get_stats(self) -> PoolStats:
        """
        Gibt Pool-Statistiken zurück.

        Returns:
            PoolStats.
        """
        stats = PoolStats(total_workers=len(self._workers))

        total_processed = 0
        total_failed = 0
        total_processing_ms = 0.0

        for worker in self._workers.values():
            worker_stats = worker.stats

            if worker_stats.state == WorkerState.BUSY:
                stats.active_workers += 1
            elif worker_stats.state == WorkerState.IDLE:
                stats.idle_workers += 1
            elif worker_stats.state == WorkerState.FAILED:
                stats.failed_workers += 1

            total_processed += worker_stats.events_processed
            total_failed += worker_stats.events_failed
            total_processing_ms += worker_stats.total_processing_ms

            stats.workers.append(worker_stats.to_dict())

        stats.total_events_processed = total_processed
        stats.total_events_failed = total_failed

        if total_processed > 0:
            stats.avg_processing_ms = total_processing_ms / total_processed

        return stats

    async def restart_failed_workers(self) -> int:
        """
        Startet fehlerhafte Worker neu.

        Returns:
            Anzahl neu gestarteter Worker.
        """
        restarted = 0

        async with self._lock:
            failed_workers = [
                w for w in self._workers.values()
                if w.state == WorkerState.FAILED
            ]

            for worker in failed_workers:
                worker_id = worker.id
                await self._remove_worker(worker_id)
                await self._create_worker()
                restarted += 1

        return restarted


class AutoScalingPool(WorkerPool):
    """
    Worker-Pool mit automatischer Skalierung.

    Passt die Worker-Anzahl basierend auf Queue-Größe an.

    Example:
        pool = AutoScalingPool(
            queue=event_queue,
            handler=process_event,
            min_workers=2,
            max_workers=20,
            scale_up_threshold=100,
            scale_down_threshold=10,
        )

        await pool.start()
        # Pool skaliert automatisch
    """

    def __init__(
        self,
        queue: "AsyncEventQueue",
        handler: Callable[[str, "EventData"], Coroutine[Any, Any, None]],
        min_workers: int = 1,
        max_workers: int = 10,
        config: WorkerConfig | None = None,
        scale_up_threshold: int = 100,
        scale_down_threshold: int = 10,
        check_interval_ms: float = 5000.0,
    ) -> None:
        """
        Initialisiert den Auto-Scaling-Pool.

        Args:
            queue: Die Event-Queue.
            handler: Handler-Funktion.
            min_workers: Minimale Anzahl Worker.
            max_workers: Maximale Anzahl Worker.
            config: Worker-Konfiguration.
            scale_up_threshold: Queue-Größe für Scale-Up.
            scale_down_threshold: Queue-Größe für Scale-Down.
            check_interval_ms: Prüfintervall in Millisekunden.
        """
        super().__init__(queue, handler, min_workers, max_workers, config)

        self._scale_up_threshold = scale_up_threshold
        self._scale_down_threshold = scale_down_threshold
        self._check_interval_ms = check_interval_ms
        self._scaler_task: asyncio.Task | None = None

    async def start(self) -> None:
        """Startet den Pool mit Auto-Scaler."""
        await super().start()

        async def scaler():
            while self._running:
                await asyncio.sleep(self._check_interval_ms / 1000)
                await self._auto_scale()

        self._scaler_task = asyncio.create_task(scaler())

    async def stop(self, timeout: float = 10.0) -> None:
        """Stoppt den Pool."""
        if self._scaler_task:
            self._scaler_task.cancel()
            try:
                await self._scaler_task
            except asyncio.CancelledError:
                pass
            self._scaler_task = None

        await super().stop(timeout)

    async def _auto_scale(self) -> None:
        """Führt automatische Skalierung durch."""
        queue_size = self._queue.size
        current_workers = len(self._workers)

        if queue_size >= self._scale_up_threshold:
            # Mehr Worker benötigt
            workers_needed = min(
                queue_size // self._scale_up_threshold,
                self._max_workers - current_workers,
            )
            if workers_needed > 0:
                await self.scale_up(workers_needed)

        elif queue_size <= self._scale_down_threshold:
            # Weniger Worker benötigt
            if current_workers > self._min_workers:
                await self.scale_down(1)