"""Auto-réparation Vektor : webhook Grafana -> diagnostic -> action sûre. Principe : alertes critiques uniquement, actions explicitement allowlistées, anti-tempête et rapports Telegram narratifs. Les réparations média sont vérifiées côté PVE et ne modifient ni données ni configuration. """ from __future__ import annotations import asyncio import logging import os import time from datetime import datetime, timezone import httpx from . import actions TG_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "") TG_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "") _COOLDOWN_S = 1800 _LAST_RUN: dict[str, float] = {} _ACTIVE_ALERTS: dict[str, dict[str, str]] = {} logger = logging.getLogger("vektor.selfheal") _MEDIA_ACTIONS = {"selfheal_media_jellyfin", "selfheal_media_immich", "selfheal_media_authelia"} async def _tg(text: str) -> bool: if not TG_BOT_TOKEN or not TG_CHAT_ID: logger.warning("Rapport d'incident non envoyé : configuration Telegram absente") return False try: async with httpx.AsyncClient(timeout=10) as client: for offset in range(0, max(len(text), 1), 3800): response = await client.post( f"https://api.telegram.org/bot{TG_BOT_TOKEN}/sendMessage", data={"chat_id": TG_CHAT_ID, "text": text[offset : offset + 3800]}, ) response.raise_for_status() if not response.json().get("ok"): raise httpx.HTTPError("Telegram API a refusé l'envoi") return True except (httpx.HTTPError, ValueError) as exc: logger.warning("Envoi du rapport Telegram impossible: %s", exc.__class__.__name__) return False def _alert_key(alert: dict, title: str) -> str: return str(alert.get("fingerprint") or title or "alerte inconnue") def _probable_cause(title: str, annotations: dict) -> str: t = title.lower() if "thin pool lvm" in t: return "Le thin pool LVM a dépassé son seuil critique de capacité." if "qbittorrent" in t and ("unhealthy" in t or "down" in t): return "Le conteneur qBittorrent a échoué à son healthcheck ou ne répond plus." if "vektor-telegram" in t and "down" in t: return "Le contrôle de santé du bot Telegram Vektor est en échec." if "jellyfin" in t: return "Le contrôle de santé de Jellyfin (systemd et endpoint local) est en échec." if "immich" in t: return "Le contrôle de santé de l’API Immich est en échec." if "backup" in t and any(w in t for w in ("échec", "echec", "failed")): return "Le contrôle signale un échec de sauvegarde ; la cause exacte reste à diagnostiquer." if "authelia" in t or "auth.mayoraz-net.ch" in t or any( w.strip(".,:;!()[]-") in ("auth", "sso") for w in t.split() ): return "Le contrôle de santé du SSO Authelia est en échec ; le gardien sso-switch (VPS) peut déjà servir les sites depuis la réplique." detail = str(annotations.get("description") or annotations.get("summary") or "").strip() if detail: return f"Cause non déterminée automatiquement. Détail Grafana : {detail[:400]}" return "Cause non déterminée automatiquement à partir du nom de l’alerte." def _action_label(action: str) -> str: return { "pve_fstrim": "lancer fstrim sur le stockage Proxmox", "docker_restart_107_qbittorrent": "redémarrer qBittorrent (CT 107)", "bot_recreate": "déclencher le watchdog du bot Vektor Telegram", "selfheal_media_jellyfin": "vérifier puis redémarrer Jellyfin (CT 102, systemd)", "selfheal_media_immich": "vérifier puis redémarrer le serveur Immich (CT 109)", "selfheal_media_authelia": "vérifier puis redémarrer Authelia (CT 104, SSO)", }.get(action, action) def _action_failed(result: str) -> bool: lowered = result.strip().lower() return any(marker in lowered for marker in ( "refus", "indisponible", "échec", "echec", "erreur", "error", "timeout", "unhealthy", "max_attempts", "unhealthy:", )) def _duration(started_at: str | None, ended_at: datetime) -> str | None: if not started_at: return None try: start = datetime.fromisoformat(started_at) seconds = max(0, int((ended_at - start).total_seconds())) except (TypeError, ValueError): return None minutes, remainder = divmod(seconds, 60) return f"{minutes} min {remainder} s" if minutes else f"{remainder} s" def _incident_report( *, title: str, severity: str, annotations: dict, action: str | None = None, action_result: str = "", action_error: str = "", skipped_reason: str = "", detected_at: str = "", started_at: str | None = None, prior_action: str | None = None, prior_result: str = "", prior_error: str = "", prior_skipped_reason: str = "", resolved: bool = False, ) -> str: """Récit déterministe, distinguant l’action de la preuve santé.""" cause = _probable_cause(title, annotations) detail = str(annotations.get("summary") or annotations.get("description") or "").strip() lines = ["🧾 Rapport d’incident Vektor", f"Alerte : {title or 'alerte sans titre'}", f"Gravité : {severity or 'non précisée'}"] if resolved: duration = _duration(started_at, datetime.now(timezone.utc)) lines.extend([ "État : rétablissement détecté par Grafana.", f"Détecté le : {detected_at or 'heure non fournie'}" + (f" ; durée : {duration}." if duration else ". Durée indisponible."), f"Cause probable : {cause}", ]) if prior_action: lines.append(f"Action prise : {_action_label(prior_action)}.") if prior_action in _MEDIA_ACTIONS: lines.append("Garde-fous : cooldown 30 min, deux tentatives maximum ; aucune restauration de données/configuration, DB ou photos.") if prior_action == "selfheal_media_authelia": lines.append("Rollback de données : volontairement absent ; la DB (CT 106) et la configuration Authelia ne sont pas touchées.") else: lines.append("Rollback de données : volontairement absent pour protéger Immich et Jellyfin.") if prior_error: lines.extend([f"Résultat de l’action : échec ({prior_error[:300]}).", "Le retour résolu a ensuite été observé par Grafana."]) elif prior_result: lines.append(f"Résultat de l’action : {prior_result[:400]}") else: lines.append("Résultat de l’action : inconnu (Vektor a pu redémarrer avant le retour Grafana).") elif prior_skipped_reason: lines.append(f"Action prise : aucune réparation automatique ({prior_skipped_reason}).") else: lines.append("Action de Vektor : aucune nouvelle réparation au signal de rétablissement.") lines.extend([ "Résultat : Grafana signale l’alerte résolue ; vérifier Kuma/Grafana si l’incident a eu un impact utilisateur.", "Conseil : surveiller une récidive et consulter les journaux autour de l’heure de l’alerte.", ]) else: lines.extend(["État : panne/condition anormale détectée."]) if detected_at: lines.append(f"Détecté le : {detected_at}") lines.append(f"Cause probable : {cause}") if action: lines.append(f"Action prise : {_action_label(action)}.") if action_error: lines.extend([f"Résultat : échec de l’action ou escalade opérateur ({action_error[:350]}).", "Conseil : vérifier l’accès SSH/actions et intervenir manuellement si le service reste indisponible."]) elif action_result: if action_result.startswith("HEALTHY:"): lines.append(f"Résultat : santé confirmée par contrôle local. {action_result[:450]}") elif _action_failed(action_result): lines.append(f"Résultat : la commande a renvoyé une erreur ; succès fonctionnel non vérifié. Détail : {action_result[:450]}") elif action not in _MEDIA_ACTIONS: lines.append(f"Résultat : retour reçu, succès fonctionnel non vérifié. Détail : {action_result[:450]}") elif action_result.startswith("ATTEMPT ") and "HEALTHY:" in action_result: lines.append(f"Résultat : redémarrage et contrôle santé réussis. {action_result[:450]}") elif action_result.startswith("COOLDOWN:"): lines.append(f"Résultat : aucune nouvelle action, cooldown actif. {action_result[:350]}") else: lines.append(f"Résultat : {action_result[:500]}") if action in _MEDIA_ACTIONS: if action == "selfheal_media_authelia": lines.append("Garde-fous : cooldown 30 min et deux tentatives maximum ; seul le conteneur Authelia est redémarré, la DB (CT 106) et la configuration ne sont jamais touchées.") else: lines.append("Garde-fous : cooldown 30 min et deux tentatives maximum par service ; aucun rollback de données/configuration, DB ou photos.") lines.append("Vérification : le webhook ne remplace pas le suivi Grafana/Kuma.") lines.append("Conseil : si l’alerte persiste, lancer /diag puis consulter les journaux du service.") else: lines.append("Résultat : action déclenchée, retour non reçu ; contrôler Kuma/Grafana.") else: reason = skipped_reason or "aucune règle d’auto-réparation sûre ne correspond" lines.extend([ f"Action prise : aucune réparation automatique ({reason}).", "Résultat : Vektor n’a pas modifié le serveur.", "Conseil : lancer /diag (ou /diag services), puis consulter les journaux.", ]) if detail and not resolved: lines.append(f"Contexte Grafana : {detail[:400]}") return "\n".join(lines) def _match_rule(title: str) -> str | None: t = title.lower() if "thin pool lvm" in t and "92%" in t: return "pve_fstrim" if "qbittorrent" in t and ("unhealthy" in t or "down" in t): return "docker_restart_107_qbittorrent" if "vektor-telegram" in t and "down" in t: return "bot_recreate" if "backup" in t and any(w in t for w in ("échec", "echec", "failed")): return None fault = ("down", "unhealthy", "indisponible", "offline", "panne", "échec", "echec", "failed", "not up") if "jellyfin" in t and any(marker in t for marker in fault): return "selfheal_media_jellyfin" if "immich" in t and any(marker in t for marker in fault): return "selfheal_media_immich" if ("authelia" in t or "auth.mayoraz-net.ch" in t or any( w.strip(".,:;!()[]-") in ("auth", "sso") for w in t.split() )) and any(marker in t for marker in fault): return "selfheal_media_authelia" return None async def _run_media_action(action: str) -> tuple[str, str, str]: """Retourne (résultat, erreur rapportable, raison de skip).""" try: result = (await actions._execute(action)).strip() except Exception as exc: logger.exception("Auto-réparation média échouée") return f"ERROR: {exc.__class__.__name__}: {str(exc)[:250]}", f"{exc.__class__.__name__}: {str(exc)[:250]}", "" if result.startswith("COOLDOWN:"): return result, "", "cooldown média anti-tempête actif (30 min)" if result.startswith("MAX_ATTEMPTS:"): return result, result[:250], "deux tentatives déjà effectuées ; escalade opérateur nécessaire" if result.startswith(("ERROR:", "UNHEALTHY:")): return result, result[:250], "service toujours indisponible après contrôle local" if result.startswith("ATTEMPT ") and "HEALTHY:" not in result: return result, result[:250], "service toujours indisponible après redémarrage" if not result.startswith(("HEALTHY:", "ATTEMPT ")): return result, result[:250] or "réponse d’action vide/inconnue", "résultat d’action non reconnu" return result, "", "" async def handle_grafana_webhook(payload: dict) -> dict: alerts = payload.get("alerts") or [] acted: list[str] = [] skipped: list[str] = [] reports: list[str] = [] for alert in alerts: if not isinstance(alert, dict): skipped.append("?") continue labels = alert.get("labels") or {} annotations = alert.get("annotations") or {} title = str(alert.get("title") or labels.get("alertname", "")) severity = str(labels.get("severity", "")).lower() status = str(alert.get("status", "")).lower() key = _alert_key(alert, title) if status == "resolved": previous = _ACTIVE_ALERTS.pop(key, None) media_service = (previous or {}).get("media_service", "") if media_service: try: reset_action = f"selfheal_media_reset_{media_service}" reset = (await actions._execute(reset_action)).strip() if previous is not None: previous["action_result"] = (previous.get("action_result", "") + "\n" + reset).strip() except Exception as exc: logger.warning("Réinitialisation budget média impossible: %s", exc.__class__.__name__) if previous is None: skipped.append(f"{title or '?'} (résolution sans incident suivi)") continue reports.append(_incident_report( title=title, severity=severity or previous.get("severity", ""), annotations=annotations, detected_at=str(alert.get("startsAt") or previous.get("detected_at", "")), started_at=previous.get("detected_at"), prior_action=previous.get("action") or None, prior_result=previous.get("action_result", ""), prior_error=previous.get("action_error", ""), prior_skipped_reason=previous.get("skipped_reason", ""), resolved=True, )) continue if status != "firing": skipped.append(title or "?") continue action = _match_rule(title) if severity == "critical" else None previous = _ACTIVE_ALERTS.get(key) if previous is not None: if action in _MEDIA_ACTIONS and previous.get("health_confirmed"): skipped.append(f"{title or '?'} (santé locale déjà confirmée, attente de résolution Grafana)") continue if action in _MEDIA_ACTIONS and not previous.get("attempt_limit_reported"): last_call = float(previous.get("media_last_call", "0") or 0) now = time.monotonic() if last_call and now - last_call < _COOLDOWN_S: skipped.append(f"{title or '?'} (cooldown média anti-tempête)") continue previous["media_last_call"] = str(now) result, error, reason = await _run_media_action(action) if reason == "cooldown média anti-tempête actif (30 min)": skipped.append(f"{title} (cooldown)") continue if result.startswith(("HEALTHY:", "ATTEMPT ")) and not error: previous["health_confirmed"] = "1" if reason == "deux tentatives déjà effectuées ; escalade opérateur nécessaire": previous["attempt_limit_reported"] = "1" previous.update({ "action": action, "media_service": action.removeprefix("selfheal_media_"), "action_result": result, "action_error": error, "skipped_reason": reason, }) reports.append(_incident_report( title=title, severity=severity, annotations=annotations, action=action, action_result=result, action_error=error, skipped_reason=reason, detected_at=previous.get("detected_at", ""), )) if reason == "cooldown média anti-tempête actif (30 min)": skipped.append(f"{title} (cooldown)") elif error: skipped.append(f"{title} -> {reason}") else: acted.append(f"{title} -> {action} : {result[:200]}") continue skipped.append(f"{title or '?'} (rapport déjà envoyé pour cette alerte)") continue skipped_reason = "" action_result = "" action_error = "" if severity != "critical": skipped_reason = "la réparation automatique est réservée aux alertes critiques" skipped.append(title or "?") action = None elif action is None: skipped_reason = "aucune règle d’auto-réparation sûre ne correspond à cette alerte" skipped.append(title or "?") elif action in _MEDIA_ACTIONS: media_last_call = time.monotonic() action_result, action_error, skipped_reason = await _run_media_action(action) if skipped_reason: skipped.append(f"{title} -> {skipped_reason}") else: acted.append(f"{title} -> {action} : {action_result[:200]}") else: now = time.monotonic() last = _LAST_RUN.get(action) if last is not None and now - last < _COOLDOWN_S: skipped_reason = "cooldown anti-tempête actif (30 min)" skipped.append(f"{title} (cooldown)") action = None else: _LAST_RUN[action] = now try: if action == "bot_recreate": proc = await asyncio.create_subprocess_exec( "systemctl", "start", "bot-watchdog.service", stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, ) code = await proc.wait() action_result = f"bot-watchdog.service déclenché (code de sortie {code})" if code != 0: action_error = f"systemctl start a renvoyé {code}" else: action_result = (await actions._execute(action)).strip() if action_error: skipped.append(f"{title} -> {action_error}") elif _action_failed(action_result): action_error = action_result[:250] skipped.append(f"{title} -> action renvoyée en échec") else: acted.append(f"{title} -> {action} : {action_result[:200]}") except Exception as exc: logger.exception("Auto-réparation échouée pour %s", title) action_error = f"{exc.__class__.__name__}: {str(exc)[:250]}" skipped.append(f"{title} -> erreur d’action") detected_at = str(alert.get("startsAt") or datetime.now(timezone.utc).isoformat(timespec="seconds")) _ACTIVE_ALERTS[key] = { "health_confirmed": "1" if action in _MEDIA_ACTIONS and (action_result.startswith("HEALTHY:") or (action_result.startswith("ATTEMPT ") and "HEALTHY:" in action_result)) else "", "title": title, "severity": severity, "detected_at": detected_at, "started_at": detected_at, "action": action or "", "media_service": action.removeprefix("selfheal_media_") if action in _MEDIA_ACTIONS else "", "media_last_call": str(media_last_call) if action in _MEDIA_ACTIONS else "", "attempt_limit_reported": "1" if action in _MEDIA_ACTIONS and action_error.startswith("MAX_ATTEMPTS:") else "", "action_result": action_result, "action_error": action_error, "skipped_reason": skipped_reason, } reports.append(_incident_report( title=title, severity=severity, annotations=annotations, action=action, action_result=action_result, action_error=action_error, skipped_reason=skipped_reason, detected_at=detected_at, )) reports_sent = 0 for report in reports: if await _tg(report): reports_sent += 1 return {"acted": acted, "skipped": skipped, "reports_generated": len(reports), "reports_sent": reports_sent} def detect_cancel_qbit(text: str) -> bool: """« annule la pause des torrents » variation non couverte par resume.""" import re return bool(re.search( r"\bannule?[rs]?\s+(?:la\s+)?pause\s+(?:des\s+|de\s+)?(?:torrents?t?s?|t[ée]l[ée]chargements?)\b", text, re.I, ))