"""`mld heal` : auto-réparation du cluster depuis la passerelle (toutes les 5 min via le LaunchDaemon io.maclustr.mld-heal). Pour chaque nœud qui héberge des apps du registre, UN seul SSH ramène : uptime, démon PM2, statut de chaque processus, LaunchAgents chargés, interface WireGuard wg1, code HTTP local de chaque app. Puis, seulement si quelque chose manque : - démon PM2 mort ou processus absents → `pm2 resurrect` puis `deploy.start(app)` (régénère l'ecosystem, idempotent) - processus errored/stopped (hors cron) → `pm2 restart` - LaunchAgent absent → re-bootstrap du plist (ou start complet s'il n'existe plus) - wg1 tombé → kickstart du LaunchDaemon (sinon raccordement complet) - route Caddy absente / mauvais nœud → `tunnelctl add` (via tunnel.ensure_route) Garde-fous : nœud redémarré depuis < 3 min ignoré (pm2 resurrect fait son travail), au plus 6 actions par app et par heure, registre sauvegardé UNE fois (diffusion aux abonnés) et seulement si un état change. Journal : ~/dispatch/logs/heal.log. """ import json import os import plistlib import shlex import time from . import config, deploy, manifest, nodes, registry, ssh, tunnel GRACE_S = 180 MAX_ACTIONS_PER_HOUR = 6 HEAL_LABEL = "io.maclustr.mld-heal" HEAL_INTERVAL_S = 300 def _now(): return time.strftime("%Y-%m-%d %H:%M:%S") def _log(msg, quiet=False): line = "%s %s" % (_now(), msg) if not quiet: print(msg, flush=True) with open(os.path.join(config.LOG_DIR, "heal.log"), "a") as f: f.write(line + "\n") def _state(): if os.path.exists(config.HEAL_STATE): try: return json.load(open(config.HEAL_STATE)) except Exception: pass return {"actions": {}, "down": {}} def _save_state(st): tmp = config.HEAL_STATE + ".tmp" json.dump(st, open(tmp, "w"), indent=1) os.replace(tmp, config.HEAL_STATE) def _allowed(st, app): cut = time.time() - 3600 acts = [t for t in st["actions"].get(app, []) if t > cut] st["actions"][app] = acts return len(acts) < MAX_ACTIONS_PER_HOUR def _record(st, app): st["actions"].setdefault(app, []).append(time.time()) def _node_report(ip, apps): """Un SSH : BOOT/NOW, PM2 up|down, PROC nom statut restarts, LA label, WG1 up|down, HUB ok|ko, HTTP app code.""" script = r''' echo "BOOT $(sysctl -n kern.boottime | sed -E 's/.*sec = ([0-9]+).*/\1/')" echo "NOW $(date +%s)" if pm2 ping >/dev/null 2>&1; then echo "PM2 up"; else echo "PM2 down"; fi pm2 jlist 2>/dev/null | python3 -c " import json,sys try: L=json.load(sys.stdin) except Exception: L=[] for p in L: print('PROC', p['name'], p['pm2_env'].get('status'), p['pm2_env'].get('restart_time',0))" 2>/dev/null launchctl list 2>/dev/null | awk 'NR>1{print "LA", $3}' ifconfig 2>/dev/null | grep -q 'inet 10\.67\.0\.' && echo "WG1 up" || echo "WG1 down" ping -c 1 -W 2000 10.67.0.1 >/dev/null 2>&1 && echo "HUB ok" || echo "HUB ko" ''' for app, m in apps.items(): if m.get("port"): script += "echo \"HTTP %s $(curl -s -o /dev/null -m 6 -w '%%{http_code}' http://127.0.0.1:%s%s)\"\n" % (app, m["port"], m.get("health_path") or "/") rc, out, err = ssh.run(ip, script, timeout=90) if rc != 0 and "BOOT" not in out: return None rep = {"pm2": None, "procs": {}, "la": set(), "wg1": None, "hub": None, "http": {}, "boot": 0, "now": 0} for line in out.splitlines(): p = line.split() if not p: continue if p[0] == "BOOT" and len(p) > 1 and p[1].isdigit(): rep["boot"] = int(p[1]) elif p[0] == "NOW" and len(p) > 1: rep["now"] = int(p[1]) elif p[0] == "PM2": rep["pm2"] = p[1] == "up" elif p[0] == "PROC" and len(p) >= 3: rep["procs"][p[1]] = (p[2], int(p[3]) if len(p) > 3 and p[3].isdigit() else 0) elif p[0] == "LA" and len(p) > 1: rep["la"].add(p[1]) elif p[0] == "WG1": rep["wg1"] = p[1] == "up" elif p[0] == "HUB": rep["hub"] = p[1] == "ok" elif p[0] == "HTTP" and len(p) >= 2: rep["http"][p[1]] = p[2] if len(p) > 2 else "000" rep["uptime"] = (rep["now"] - rep["boot"]) if rep["boot"] and rep["now"] else 10 ** 6 return rep def _http_ok(code): return code.isdigit() and code != "000" and int(code) < 500 def heal(apps_filter=None, nodes_filter=None, dry_run=False, quiet=False, public=True): t0 = time.time() st = _state() reg = registry.load() by_node = {} suspended = [] for app, v in reg["apps"].items(): if apps_filter and app not in apps_filter: continue if not v.get("node"): continue if nodes_filter and v["node"] not in nodes_filter: continue if v.get("status") == "stopped": # arrêt volontaire (`mld stop`) : on ne ressuscite pas, `mld start` remet online suspended.append(app) continue by_node.setdefault(v["node"], {})[app] = manifest.load(app) changes = {} summary = {"ok": 0, "fixed": 0, "ko": 0, "skipped": len(suspended)} if suspended: _log("suspendues (mld stop, ignorées) : %s" % ", ".join(sorted(suspended)), quiet) tun_state = None try: tun_state = tunnel.state() except tunnel.TunnelError as e: _log("tunnel : %s" % e, quiet) for alias in sorted(by_node): apps = by_node[alias] ip = nodes.ip_of(alias, rediscover=False) if not ip: # une seule redécouverte par passage ip = nodes.ip_of(alias, rediscover=True) if not ip: since = st["down"].setdefault(alias, _now()) _log("%s : INJOIGNABLE (depuis %s) — %d app(s) : %s" % (alias, since, len(apps), ", ".join(sorted(apps))), quiet) for app in apps: if reg["apps"][app].get("status") != "node-down": changes[app] = {"status": "node-down", "health": "nœud %s injoignable" % alias} summary["ko"] += len(apps) continue if alias in st["down"]: _log("%s : de retour (était injoignable depuis %s)" % (alias, st["down"].pop(alias)), quiet) rep = _node_report(ip, apps) if rep is None: _log("%s : sonde impossible" % alias, quiet) summary["skipped"] += len(apps) continue if rep["uptime"] < GRACE_S: _log("%s : redémarré il y a %ds — délai de grâce, on repasse plus tard" % (alias, rep["uptime"]), quiet) summary["skipped"] += len(apps) continue # --- WireGuard vers la passerelle (seulement si le nœud expose un site) exposes = any(tunnel.spec(m, alias) for m in apps.values()) if exposes and (rep["wg1"] is False or rep["hub"] is False): _log("%s : wg1 %s / hub %s" % (alias, "up" if rep["wg1"] else "DOWN", "ok" if rep["hub"] else "KO"), quiet) if not dry_run: try: tunnel.ensure_peer(alias, log=lambda s: _log(s, quiet), st=tun_state) except tunnel.TunnelError as e: _log(" tunnel : %s" % e, quiet) resurrected = False for app in sorted(apps): m = apps[app] names = manifest.pm2_names(m) labels = manifest.launchd_labels(m) missing = [n for n in names if n not in rep["procs"]] if rep["pm2"] else list(names) bad = [n for n in names if n in rep["procs"] and rep["procs"][n][0] in ("errored", "stopped") and next((p for p in m["processes"] if p["name"] == n), {}).get("autorestart", True) and not next((p for p in m["processes"] if p["name"] == n), {}).get("cron_restart")] la_missing = [l for l in labels if l not in rep["la"]] code = rep["http"].get(app) http_ok = _http_ok(code) if m.get("port") else True problems = [] if missing: problems.append("PM2 absent: %s" % ",".join(missing)) if bad: problems.append("PM2 %s: %s" % (rep["procs"][bad[0]][0], ",".join(bad))) if la_missing: problems.append("launchd absent: %s" % ",".join(la_missing)) if not http_ok and not missing and not bad: problems.append("HTTP local %s" % code) if not problems: summary["ok"] += 1 if reg["apps"][app].get("status") != "online": changes[app] = {"status": "online", "health": "local %s" % code if m.get("port") else "processus online"} # route publique : suit-elle bien ce nœud ? if tun_state and public and tunnel.spec(m, alias): s = tunnel.spec(m, alias) r = tunnel.routes(s["gateway"], tun_state).get(s["domain"]) if not tunnel._route_matches(r, s["upstreams"]): _log("%s/%s : route https://%s %s" % (alias, app, s["domain"], "absente" if not r else "→ %s au lieu de %s" % (", ".join(u["addr"] for u in r.get("upstreams", [])), s["upstreams"][0])), quiet) if not dry_run and _allowed(st, app): try: tunnel.ensure_route(m, alias, log=lambda x: _log(x, quiet), st=tun_state) _record(st, app) changes[app] = dict(changes.get(app, {}), tunnel={"gateway": s["gateway"], "domain": s["domain"], "upstream": s["upstreams"][0], "url": "https://%s" % s["domain"]}) summary["fixed"] += 1 except tunnel.TunnelError as e: _log(" tunnel : %s" % e, quiet) continue _log("%s/%s : %s" % (alias, app, " ; ".join(problems)), quiet) if dry_run: summary["ko"] += 1 continue if not _allowed(st, app): _log(" %s : trop d'actions dans l'heure (%d), on laisse tranquille" % (app, MAX_ACTIONS_PER_HOUR), quiet) summary["ko"] += 1 changes[app] = {"status": "unhealthy", "health": " ; ".join(problems)} continue _record(st, app) try: if rep["pm2"] is False and not resurrected: rc, out, _ = ssh.run(ip, "pm2 resurrect 2>&1 | tail -1", timeout=120) _log(" pm2 resurrect sur %s : %s" % (alias, out.strip()[-100:]), quiet) resurrected = True if missing or la_missing: deploy.start(app, alias, m=m, ip=ip) elif bad: rc, out, _ = ssh.run(ip, "pm2 restart %s --update-env >/dev/null 2>&1; pm2 save --force >/dev/null 2>&1; echo ok" % " ".join(bad), timeout=120) _log(" pm2 restart %s : %s" % (",".join(bad), out.strip()[-40:]), quiet) elif not http_ok: rc, out, _ = ssh.run(ip, "pm2 restart %s --update-env >/dev/null 2>&1; echo ok" % " ".join(names), timeout=120) if names else (0, "", "") _log(" HTTP local %s → pm2 restart %s" % (code, ",".join(names)), quiet) ok, det = deploy.health(app, alias, ip=ip, m=m, timeout=45, public=False) tun = None if ok and tunnel.spec(m, alias): tun = deploy._publish(app, alias, m, ip) ok, det = deploy.health(app, alias, ip=ip, m=m, timeout=20, public=public and bool(tun)) _log(" %s : %s (%s)" % (app, "RÉPARÉE" if ok else "toujours KO", det), quiet) fields = {"status": "online" if ok else "unhealthy", "health": det, "healed": _now()} if tun: fields["tunnel"] = tun changes[app] = fields summary["fixed" if ok else "ko"] += 1 except (deploy.DeployError, tunnel.TunnelError) as e: _log(" %s : ÉCHEC de réparation — %s" % (app, e), quiet) changes[app] = {"status": "unhealthy", "health": str(e)[:200]} summary["ko"] += 1 if changes and not dry_run: registry.update_many(changes) _save_state(st) _log("heal terminé en %.0fs : %d ok, %d réparées, %d KO, %d ignorées (grâce/sonde)%s" % ( time.time() - t0, summary["ok"], summary["fixed"], summary["ko"], summary["skipped"], " [dry-run]" if dry_run else ""), quiet) return summary # ------------------------------------------------------------ LaunchDaemon sur la passerelle --- def plist_content(): mld_bin = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "bin", "mld") d = { "Label": HEAL_LABEL, "ProgramArguments": [mld_bin, "heal", "--quiet"], "UserName": config.USER, "RunAtLoad": True, "StartInterval": HEAL_INTERVAL_S, "EnvironmentVariables": {"HOME": config.HOME, "PATH": "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin", "LANG": "fr_CA.UTF-8"}, "WorkingDirectory": config.HOME, "StandardOutPath": os.path.join(config.LOG_DIR, "heal.out"), "StandardErrorPath": os.path.join(config.LOG_DIR, "heal.err"), "ProcessType": "Background", "LowPriorityIO": True, } return plistlib.dumps(d).decode() def install(interval=None): """Installe (ou met à jour) le LaunchDaemon io.maclustr.mld-heal sur la passerelle (sudo via ~/dispatch/.sudo).""" import subprocess global HEAL_INTERVAL_S if interval: HEAL_INTERVAL_S = int(interval) pwf = config.SUDO_PW_FILE pw = open(pwf).read().strip() if os.path.exists(pwf) else None if not pw: raise SystemExit("mot de passe sudo absent : %s" % pwf) tmp = "/tmp/%s.plist" % HEAL_LABEL open(tmp, "w").write(plist_content()) dest = "/Library/LaunchDaemons/%s.plist" % HEAL_LABEL sh = ("echo %s | sudo -S -p '' sh -c 'install -m 644 -o root -g wheel %s %s && launchctl bootout system/%s 2>/dev/null; " "launchctl bootstrap system %s && launchctl print system/%s | grep -E \"state|interval\" | head -2'") % ( shlex.quote(pw), tmp, dest, HEAL_LABEL, dest, HEAL_LABEL) p = subprocess.run(["bash", "-c", sh], capture_output=True, text=True) print(p.stdout.strip() or p.stderr.strip()[-300:]) os.remove(tmp) ok = p.returncode == 0 print("LaunchDaemon %s : %s (toutes les %d s, journal %s/heal.log)" % (HEAL_LABEL, "installé" if ok else "ÉCHEC", HEAL_INTERVAL_S, config.LOG_DIR)) return ok