Funk-lab/stt/client/stt/portal/health.py

"""Santé live des services du portail — sondes HTTP + Prometheus + Alertmanager.

Tout tourne côté backend Python (pas de CORS) dans un thread dédié : à intervalle
régulier, on sonde chaque service et on pousse l'état au HUD via `emit` (le même
pont thread-safe que le moteur vocal). Trois sources :

  • probe HTTP de `service.probe_url`            → en ligne / hors ligne + latence
  • Prometheus `/api/v1/query` (composants)      → état de chaque brique (PromQL → 1/0)
  • Alertmanager `/api/v2/alerts` (alertes)      → alertes actives, filtrées par service

Aucune dépendance nouvelle : `requests` (déjà utilisé) + `concurrent.futures`.
"""

from __future__ import annotations

import re
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Callable

from .registry import Service

# L'alerte « Watchdog » est volontairement toujours active (test du pipeline) :
# on l'exclut pour ne pas polluer chaque page.
_ALWAYS_ON_ALERTS = {"Watchdog"}


def probe_http(url: str, timeout: float) -> dict[str, Any]:
    """GET `url` → {state: up|down, latency_ms, code}. Toute erreur = down."""
    import requests

    t0 = time.monotonic()
    try:
        r = requests.get(url, timeout=timeout, allow_redirects=True)
        latency = int((time.monotonic() - t0) * 1000)
        ok = r.status_code < 400
        return {"state": "up" if ok else "down", "latency_ms": latency, "code": r.status_code}
    except requests.RequestException:
        return {"state": "down", "latency_ms": None, "code": None}


def query_prometheus(base: str, expr: str, timeout: float) -> list[float] | None:
    """Évalue une requête instantanée → liste des valeurs, ou None si la requête échoue.

    [] = requête OK mais aucune série (cible inconnue → état « indéterminé »).
    """
    import requests

    try:
        r = requests.get(
            f"{base.rstrip('/')}/api/v1/query",
            params={"query": expr}, timeout=timeout,
        )
        r.raise_for_status()
        data = r.json()
        if data.get("status") != "success":
            return None
        out: list[float] = []
        for s in data.get("data", {}).get("result", []):
            try:
                out.append(float(s["value"][1]))
            except (KeyError, IndexError, ValueError, TypeError):
                pass
        return out
    except (requests.RequestException, ValueError):
        return None


def _component_state(values: list[float] | None) -> str:
    """Interprète les valeurs PromQL d'un composant : up / down / unknown."""
    if values is None:
        return "unknown"        # Prometheus injoignable
    if not values:
        return "unknown"        # aucune série (cible absente)
    if any(v == 0 for v in values):
        return "down"
    return "up"                  # toutes les séries non nulles


def fetch_alerts(base: str, timeout: float) -> list[dict[str, Any]] | None:
    """Alertes actives d'Alertmanager (hors Watchdog), normalisées. None si injoignable."""
    import requests

    try:
        r = requests.get(f"{base.rstrip('/')}/api/v2/alerts",
                         params={"active": "true", "silenced": "false"}, timeout=timeout)
        r.raise_for_status()
        raw = r.json()
    except (requests.RequestException, ValueError):
        return None
    out: list[dict[str, Any]] = []
    for a in raw if isinstance(raw, list) else []:
        labels = a.get("labels", {}) or {}
        if labels.get("alertname") in _ALWAYS_ON_ALERTS:
            continue
        if (a.get("status", {}) or {}).get("state") not in (None, "active"):
            continue
        ann = a.get("annotations", {}) or {}
        out.append({
            "name": labels.get("alertname", "alerte"),
            "severity": labels.get("severity", "none"),
            "summary": ann.get("summary") or ann.get("description") or "",
            "labels": labels,
        })
    return out


def _alert_matches(alert: dict[str, Any], selectors: tuple[tuple[str, str], ...]) -> bool:
    """Vrai si TOUS les (label, regex) du service matchent les labels de l'alerte."""
    if not selectors:
        return False
    labels = alert.get("labels", {}) or {}
    for key, pattern in selectors:
        val = str(labels.get(key, ""))
        try:
            if not re.search(pattern, val):
                return False
        except re.error:
            if pattern != val:
                return False
    return True


class StatusPoller:
    """Boucle de sondage en thread : construit l'état de tous les services et l'émet."""

    def __init__(self, services: list[Service], cfg: dict, emit: Callable[[dict], None]):
        self.services = services
        self.emit = emit
        self.prom_url = cfg.get("prometheus_url", "http://prometheus.lab.local")
        self.alert_url = cfg.get("alertmanager_url", "http://alertmanager.lab.local")
        self.interval = float(cfg.get("status_interval_sec", 15) or 15)
        self.timeout = float(cfg.get("probe_timeout_sec", 4) or 4)
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None

    # ── cycle de mesure ──────────────────────────────────────────────────
    def build_status(self) -> dict[str, Any]:
        with ThreadPoolExecutor(max_workers=8) as pool:
            # 1) probes HTTP (un par service, en parallèle)
            probes = {
                s.id: pool.submit(probe_http, s.probe_url, self.timeout)
                for s in self.services
            }
            # 2) requêtes Prometheus distinctes (déduplique les exprs identiques) —
            #    composants (1/0) ET métriques chiffrées de page détail.
            exprs = {c.prom for s in self.services for c in s.components if c.prom}
            exprs |= {m.prom for s in self.services for m in s.metrics if m.prom}
            prom_futs = {
                e: pool.submit(query_prometheus, self.prom_url, e, self.timeout)
                for e in exprs
            }
            # 3) alertes (une seule requête)
            alerts_fut = pool.submit(fetch_alerts, self.alert_url, self.timeout)

            prom = {e: f.result() for e, f in prom_futs.items()}
            alerts = alerts_fut.result()

        services_out: dict[str, Any] = {}
        for s in self.services:
            probe = probes[s.id].result()
            comps = [
                {"name": c.name,
                 "state": _component_state(prom.get(c.prom)) if c.prom else "info"}
                for c in s.components
            ]
            # Métriques chiffrées : 1re valeur de la série agrégée (None si absente).
            metrics_out = []
            for m in s.metrics:
                vals = prom.get(m.prom)
                value = vals[0] if vals else None
                if value is None and m.hide_if_empty:
                    continue  # ex. cAdvisor non scrappé → on n'affiche pas la carte
                metrics_out.append({"label": m.label, "value": value, "unit": m.unit})
            svc_alerts = [a for a in (alerts or []) if _alert_matches(a, s.alerts_match)]
            degraded = probe["state"] == "up" and any(c["state"] == "down" for c in comps)
            if probe["state"] == "down":
                health = "down"
            elif degraded:
                health = "degraded"
            elif probe["state"] == "unknown":
                health = "unknown"
            else:
                health = "ok"
            services_out[s.id] = {
                "state": probe["state"],
                "latency_ms": probe["latency_ms"],
                "health": health,
                "components": comps,
                "metrics": metrics_out,
                "alerts": svc_alerts,
            }

        total = None if alerts is None else len(alerts)
        # `all_alerts` : liste complète (déjà récupérée) → alimente la page détail
        # d'Alertmanager (aperçu de TOUTES les alertes, pas seulement celles d'un service).
        return {
            "type": "portal-status",
            "services": services_out,
            "alerts_total": total,
            "all_alerts": alerts or [],
        }

    # ── thread ───────────────────────────────────────────────────────────
    def _run(self) -> None:
        while not self._stop.is_set():
            try:
                self.emit(self.build_status())
            except Exception:  # noqa: BLE001 — un cycle raté ne tue pas le poller
                pass
            self._stop.wait(self.interval)

    def start(self) -> None:
        if self._thread is not None:
            return
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()