metalfrom.eu/libs/bm_health.py

"""
Contrôle de santé périodique des services de fond — module PARTAGÉ.

Vit dans libs/ et non dans une app : le crawler et le geocoder en avaient
deux copies rigoureusement identiques, dont une seule était couverte par des
tests. Rien ne signalait une divergence, et une correction appliquée d'un
seul côté serait passée inaperçue.

Embarqué dans les images par les Dockerfile (contexte de build = racine du
dépôt) et rendu importable en local par les conftest.py de chaque app.

Ces services ne sont pas exposés par Traefik : aucune sonde HTTP externe ne peut
les atteindre. Sans ce module, un crawler dont FlareSolverr est injoignable ou
un worker à court de quota reste muet, et le seul symptôme est l'absence de
données nouvelles — qu'il faut remarquer soi-même.

Chaque sonde renvoie (nom, ok, détail). Le résultat agrégé est écrit dans
`service_health` (une ligne par service, écrasée), lu par le dashboard admin.

Aucune sonde ne peut interrompre le service : toute exception est convertie en
échec de sonde. Un contrôle de santé qui fait tomber ce qu'il surveille serait
pire que pas de contrôle du tout.
"""
import logging
import time

log = logging.getLogger(__name__)

# Intervalle minimal entre deux contrôles complets.
DEFAULT_INTERVAL_S = 3600

_last_run = 0.0


def probe(name, fn):
    """Exécute une sonde et convertit toute exception en échec."""
    try:
        ok, detail = fn()
        return name, bool(ok), (detail or "")
    except Exception as e:  # noqa: BLE001 - une sonde ne doit jamais propager
        return name, False, f"{type(e).__name__}: {e}"


def database_probe(get_conn):
    """La base répond-elle ?"""
    def _run():
        with get_conn() as conn:
            with conn.cursor() as cur:
                cur.execute("SELECT 1")
                cur.fetchone()
        return True, "connexion établie"
    return _run


def http_probe(session_get, url, expect_below=500, timeout=10):
    """Une dépendance HTTP répond-elle sans erreur serveur ?"""
    def _run():
        r = session_get(url, timeout=timeout)
        code = getattr(r, "status_code", 0)
        return code < expect_below, f"HTTP {code}"
    return _run


def freshness_probe(get_conn, sql, max_age_hours, label):
    """Les données progressent-elles ?

    `sql` doit renvoyer un unique timestamp (le plus récent). Une base vide
    (NULL) n'est pas un échec : c'est un état légitime au premier démarrage.
    """
    def _run():
        with get_conn() as conn:
            with conn.cursor() as cur:
                cur.execute(sql)
                row = cur.fetchone()
        latest = row[0] if row else None
        if latest is None:
            return True, f"{label}: aucune donnée pour l'instant"
        with get_conn() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT EXTRACT(EPOCH FROM (now() - %s)) / 3600.0", (latest,)
                )
                age_h = float(cur.fetchone()[0])
        return age_h <= max_age_hours, f"{label}: {age_h:.1f} h (seuil {max_age_hours} h)"
    return _run


def write_health(get_conn, service, results):
    """Enregistre le résultat agrégé. N'échoue jamais bruyamment."""
    import json

    checks = {name: ok for name, ok, _ in results}
    failed = [f"{name} — {detail}" for name, ok, detail in results if not ok]
    ok = not failed
    try:
        with get_conn() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    INSERT INTO service_health (service, ok, checks, error, checked_at)
                    VALUES (%s, %s, %s::jsonb, %s, now())
                    ON CONFLICT (service) DO UPDATE SET
                      ok = EXCLUDED.ok,
                      checks = EXCLUDED.checks,
                      error = EXCLUDED.error,
                      checked_at = EXCLUDED.checked_at
                    """,
                    (service, ok, json.dumps(checks), failed[0] if failed else None),
                )
    except Exception as e:  # noqa: BLE001
        log.warning(f"[health] écriture impossible: {e}")
    return ok, failed


def run_checks(get_conn, service, probes, log_event=None):
    """Exécute toutes les sondes, enregistre le résultat, journalise les échecs."""
    results = [probe(name, fn) for name, fn in probes]
    ok, failed = write_health(get_conn, service, results)

    if failed:
        msg = f"[health:{service}] DÉGRADÉ — " + " | ".join(failed)
        log.warning(msg)
        if log_event:
            try:
                log_event("warning", msg)
            except Exception:  # noqa: BLE001, S110 - journaliser un échec de
                # journalisation n'apporte rien et risquerait une récursion.
                pass
    else:
        log.info(f"[health:{service}] toutes les sondes sont au vert")
    return ok


def maybe_run(get_conn, service, probes, interval_s=DEFAULT_INTERVAL_S, log_event=None, force=False):
    """Point d'entrée depuis une boucle de service : ne contrôle qu'une fois par intervalle."""
    global _last_run
    now = time.monotonic()
    if not force and _last_run and (now - _last_run) < interval_s:
        return None
    _last_run = now
    return run_checks(get_conn, service, probes, log_event=log_event)