orchestrata_ultimate.providers.health_monitor

Reference › orchestrata_ultimate › providers › health_monitor

health_monitor derives provider health passively from real request outcomes, with no synthetic probes. HealthMonitor keeps a sliding window of RequestMetric records (timestamp, latency, success flag, error type) per provider, sized by window_size. provider_manager calls record after each request and passes get_all_health to the router, which uses the latency percentiles for latency-optimised ranking. get_health returns latency percentiles (computed by _percentile), error rate and status for one provider. The module has no internal imports.

At a glance

Class diagram

Dependency diagram

API

Classes

RequestMetric (dataclass)

Decorators: dataclass · Line 19

Single request measurement.

HealthMonitor

Line 28

Passive health monitor using sliding window metrics.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/providers/health_monitor.py

# mypy: ignore-errors
"""Passive provider health monitoring.

SPEC-011 FR-007/US-007: Tracks latency, error rate, and availability
using a sliding window of recent requests per provider. No active probes.
"""

import logging
import time
from collections import deque
from dataclasses import dataclass

logger = logging.getLogger(__name__)

WINDOW_SIZE = 100  # Last N requests per provider


@dataclass
class RequestMetric:
    """Single request measurement."""

    timestamp: float = 0.0
    latency_ms: float = 0.0
    success: bool = True
    error_type: str = ""


class HealthMonitor:
    """Passive health monitor using sliding window metrics.

    Computes P50/P95/P99 latency, error rate, and availability
    from the last WINDOW_SIZE requests per provider.
    """

    def __init__(self, window_size: int = WINDOW_SIZE):
        self._window_size = window_size
        self._metrics: dict[str, deque] = {}

    def record(
        self,
        provider_name: str,
        latency_ms: float,
        success: bool,
        error_type: str = "",
    ) -> None:
        """Record a request metric for a provider."""
        if provider_name not in self._metrics:
            self._metrics[provider_name] = deque(maxlen=self._window_size)
        self._metrics[provider_name].append(
            RequestMetric(
                timestamp=time.monotonic(),
                latency_ms=latency_ms,
                success=success,
                error_type=error_type,
            )
        )

    def get_health(self, provider_name: str) -> dict:
        """Get health metrics for a provider."""
        metrics = self._metrics.get(provider_name)
        if not metrics or len(metrics) == 0:
            return {
                "provider": provider_name,
                "status": "unknown",
                "request_count": 0,
                "error_rate": 0.0,
                "latency_p50_ms": 0.0,
                "latency_p95_ms": 0.0,
                "latency_p99_ms": 0.0,
            }
        successes = sum(1 for m in metrics if m.success)
        total = len(metrics)
        error_rate = 1.0 - (successes / total)
        latencies = sorted(m.latency_ms for m in metrics if m.success)
        p50 = self._percentile(latencies, 50)
        p95 = self._percentile(latencies, 95)
        p99 = self._percentile(latencies, 99)
        status = "healthy"
        if error_rate > 0.5:
            status = "degraded"
        elif error_rate > 0.2:
            status = "warning"
        return {
            "provider": provider_name,
            "status": status,
            "request_count": total,
            "error_rate": round(error_rate, 4),
            "latency_p50_ms": round(p50, 2),
            "latency_p95_ms": round(p95, 2),
            "latency_p99_ms": round(p99, 2),
        }

    def get_all_health(self) -> dict[str, dict]:
        """Get health metrics for all tracked providers."""
        return {name: self.get_health(name) for name in self._metrics}

    @staticmethod
    def _percentile(sorted_values: list, pct: float) -> float:
        """Calculate percentile from sorted values."""
        if not sorted_values:
            return 0.0
        idx = int((len(sorted_values) - 1) * pct / 100.0)
        return sorted_values[idx]