spb/maclustr-dispatch

1"""`mld heal` : auto-réparation du cluster depuis la passerelle (toutes les 5 min via le LaunchDaemon io.maclustr.mld-heal).23Pour chaque nœud qui héberge des apps du registre, UN seul SSH ramène : uptime, démon PM2, statut de chaque processus,4LaunchAgents chargés, interface WireGuard wg1, code HTTP local de chaque app. Puis, seulement si quelque chose manque :5  - démon PM2 mort ou processus absents  → `pm2 resurrect` puis `deploy.start(app)` (régénère l'ecosystem, idempotent)6  - processus errored/stopped (hors cron) → `pm2 restart`7  - LaunchAgent absent                   → re-bootstrap du plist (ou start complet s'il n'existe plus)8  - wg1 tombé                            → kickstart du LaunchDaemon (sinon raccordement complet)9  - route Caddy absente / mauvais nœud   → `tunnelctl add` (via tunnel.ensure_route)10Garde-fous : nœud redémarré depuis < 3 min ignoré (pm2 resurrect fait son travail), au plus 6 actions par app et par heure,11registre sauvegardé UNE fois (diffusion aux abonnés) et seulement si un état change. Journal : ~/dispatch/logs/heal.log.12"""13import json14import os15import plistlib16import shlex17import time18from . import config, deploy, manifest, nodes, registry, ssh, tunnel1920GRACE_S = 18021MAX_ACTIONS_PER_HOUR = 622HEAL_LABEL = "io.maclustr.mld-heal"23HEAL_INTERVAL_S = 300242526def _now():27    return time.strftime("%Y-%m-%d %H:%M:%S")282930def _log(msg, quiet=False):31    line = "%s %s" % (_now(), msg)32    if not quiet:33        print(msg, flush=True)34    with open(os.path.join(config.LOG_DIR, "heal.log"), "a") as f:35        f.write(line + "\n")363738def _state():39    if os.path.exists(config.HEAL_STATE):40        try:41            return json.load(open(config.HEAL_STATE))42        except Exception:43            pass44    return {"actions": {}, "down": {}}454647def _save_state(st):48    tmp = config.HEAL_STATE + ".tmp"49    json.dump(st, open(tmp, "w"), indent=1)50    os.replace(tmp, config.HEAL_STATE)515253def _allowed(st, app):54    cut = time.time() - 360055    acts = [t for t in st["actions"].get(app, []) if t > cut]56    st["actions"][app] = acts57    return len(acts) < MAX_ACTIONS_PER_HOUR585960def _record(st, app):61    st["actions"].setdefault(app, []).append(time.time())626364def _node_report(ip, apps):65    """Un SSH : BOOT/NOW, PM2 up|down, PROC nom statut restarts, LA label, WG1 up|down, HUB ok|ko, HTTP app code."""66    script = r'''67echo "BOOT $(sysctl -n kern.boottime | sed -E 's/.*sec = ([0-9]+).*/\1/')"68echo "NOW $(date +%s)"69if pm2 ping >/dev/null 2>&1; then echo "PM2 up"; else echo "PM2 down"; fi70pm2 jlist 2>/dev/null | python3 -c "71import json,sys72try: L=json.load(sys.stdin)73except Exception: L=[]74for p in L: print('PROC', p['name'], p['pm2_env'].get('status'), p['pm2_env'].get('restart_time',0))" 2>/dev/null75launchctl list 2>/dev/null | awk 'NR>1{print "LA", $3}'76ifconfig 2>/dev/null | grep -q 'inet 10\.67\.0\.' && echo "WG1 up" || echo "WG1 down"77ping -c 1 -W 2000 10.67.0.1 >/dev/null 2>&1 && echo "HUB ok" || echo "HUB ko"78'''79    for app, m in apps.items():80        if m.get("port"):81            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 "/")82    rc, out, err = ssh.run(ip, script, timeout=90)83    if rc != 0 and "BOOT" not in out:84        return None85    rep = {"pm2": None, "procs": {}, "la": set(), "wg1": None, "hub": None, "http": {}, "boot": 0, "now": 0}86    for line in out.splitlines():87        p = line.split()88        if not p:89            continue90        if p[0] == "BOOT" and len(p) > 1 and p[1].isdigit():91            rep["boot"] = int(p[1])92        elif p[0] == "NOW" and len(p) > 1:93            rep["now"] = int(p[1])94        elif p[0] == "PM2":95            rep["pm2"] = p[1] == "up"96        elif p[0] == "PROC" and len(p) >= 3:97            rep["procs"][p[1]] = (p[2], int(p[3]) if len(p) > 3 and p[3].isdigit() else 0)98        elif p[0] == "LA" and len(p) > 1:99            rep["la"].add(p[1])100        elif p[0] == "WG1":101            rep["wg1"] = p[1] == "up"102        elif p[0] == "HUB":103            rep["hub"] = p[1] == "ok"104        elif p[0] == "HTTP" and len(p) >= 2:105            rep["http"][p[1]] = p[2] if len(p) > 2 else "000"106    rep["uptime"] = (rep["now"] - rep["boot"]) if rep["boot"] and rep["now"] else 10 ** 6107    return rep108109110def _http_ok(code):111    return code.isdigit() and code != "000" and int(code) < 500112113114def heal(apps_filter=None, nodes_filter=None, dry_run=False, quiet=False, public=True):115    t0 = time.time()116    st = _state()117    reg = registry.load()118    by_node = {}119    for app, v in reg["apps"].items():120        if apps_filter and app not in apps_filter:121            continue122        if not v.get("node"):123            continue124        if nodes_filter and v["node"] not in nodes_filter:125            continue126        by_node.setdefault(v["node"], {})[app] = manifest.load(app)127    changes = {}128    summary = {"ok": 0, "fixed": 0, "ko": 0, "skipped": 0}129    tun_state = None130    try:131        tun_state = tunnel.state()132    except tunnel.TunnelError as e:133        _log("tunnel : %s" % e, quiet)134    for alias in sorted(by_node):135        apps = by_node[alias]136        ip = nodes.ip_of(alias, rediscover=False)137        if not ip:138            # une seule redécouverte par passage139            ip = nodes.ip_of(alias, rediscover=True)140        if not ip:141            since = st["down"].setdefault(alias, _now())142            _log("%s : INJOIGNABLE (depuis %s) — %d app(s) : %s" % (alias, since, len(apps), ", ".join(sorted(apps))), quiet)143            for app in apps:144                if reg["apps"][app].get("status") != "node-down":145                    changes[app] = {"status": "node-down", "health": "nœud %s injoignable" % alias}146            summary["ko"] += len(apps)147            continue148        if alias in st["down"]:149            _log("%s : de retour (était injoignable depuis %s)" % (alias, st["down"].pop(alias)), quiet)150        rep = _node_report(ip, apps)151        if rep is None:152            _log("%s : sonde impossible" % alias, quiet)153            summary["skipped"] += len(apps)154            continue155        if rep["uptime"] < GRACE_S:156            _log("%s : redémarré il y a %ds — délai de grâce, on repasse plus tard" % (alias, rep["uptime"]), quiet)157            summary["skipped"] += len(apps)158            continue159        # --- WireGuard vers la passerelle (seulement si le nœud expose un site)160        exposes = any(tunnel.spec(m, alias) for m in apps.values())161        if exposes and (rep["wg1"] is False or rep["hub"] is False):162            _log("%s : wg1 %s / hub %s" % (alias, "up" if rep["wg1"] else "DOWN", "ok" if rep["hub"] else "KO"), quiet)163            if not dry_run:164                try:165                    tunnel.ensure_peer(alias, log=lambda s: _log(s, quiet), st=tun_state)166                except tunnel.TunnelError as e:167                    _log("  tunnel : %s" % e, quiet)168        resurrected = False169        for app in sorted(apps):170            m = apps[app]171            names = manifest.pm2_names(m)172            labels = manifest.launchd_labels(m)173            missing = [n for n in names if n not in rep["procs"]] if rep["pm2"] else list(names)174            bad = [n for n in names if n in rep["procs"] and rep["procs"][n][0] in ("errored", "stopped")175                   and next((p for p in m["processes"] if p["name"] == n), {}).get("autorestart", True)176                   and not next((p for p in m["processes"] if p["name"] == n), {}).get("cron_restart")]177            la_missing = [l for l in labels if l not in rep["la"]]178            code = rep["http"].get(app)179            http_ok = _http_ok(code) if m.get("port") else True180            problems = []181            if missing:182                problems.append("PM2 absent: %s" % ",".join(missing))183            if bad:184                problems.append("PM2 %s: %s" % (rep["procs"][bad[0]][0], ",".join(bad)))185            if la_missing:186                problems.append("launchd absent: %s" % ",".join(la_missing))187            if not http_ok and not missing and not bad:188                problems.append("HTTP local %s" % code)189            if not problems:190                summary["ok"] += 1191                if reg["apps"][app].get("status") != "online":192                    changes[app] = {"status": "online", "health": "local %s" % code if m.get("port") else "processus online"}193                # route publique : suit-elle bien ce nœud ?194                if tun_state and public and tunnel.spec(m, alias):195                    s = tunnel.spec(m, alias)196                    r = tunnel.routes(s["gateway"], tun_state).get(s["domain"])197                    if not tunnel._route_matches(r, s["upstreams"]):198                        _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)199                        if not dry_run and _allowed(st, app):200                            try:201                                tunnel.ensure_route(m, alias, log=lambda x: _log(x, quiet), st=tun_state)202                                _record(st, app)203                                changes[app] = dict(changes.get(app, {}), tunnel={"gateway": s["gateway"], "domain": s["domain"], "upstream": s["upstreams"][0], "url": "https://%s" % s["domain"]})204                                summary["fixed"] += 1205                            except tunnel.TunnelError as e:206                                _log("  tunnel : %s" % e, quiet)207                continue208            _log("%s/%s : %s" % (alias, app, " ; ".join(problems)), quiet)209            if dry_run:210                summary["ko"] += 1211                continue212            if not _allowed(st, app):213                _log("  %s : trop d'actions dans l'heure (%d), on laisse tranquille" % (app, MAX_ACTIONS_PER_HOUR), quiet)214                summary["ko"] += 1215                changes[app] = {"status": "unhealthy", "health": " ; ".join(problems)}216                continue217            _record(st, app)218            try:219                if rep["pm2"] is False and not resurrected:220                    rc, out, _ = ssh.run(ip, "pm2 resurrect 2>&1 | tail -1", timeout=120)221                    _log("  pm2 resurrect sur %s : %s" % (alias, out.strip()[-100:]), quiet)222                    resurrected = True223                if missing or la_missing:224                    deploy.start(app, alias, m=m, ip=ip)225                elif bad:226                    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)227                    _log("  pm2 restart %s : %s" % (",".join(bad), out.strip()[-40:]), quiet)228                elif not http_ok:229                    rc, out, _ = ssh.run(ip, "pm2 restart %s --update-env >/dev/null 2>&1; echo ok" % " ".join(names), timeout=120) if names else (0, "", "")230                    _log("  HTTP local %s → pm2 restart %s" % (code, ",".join(names)), quiet)231                ok, det = deploy.health(app, alias, ip=ip, m=m, timeout=45, public=False)232                tun = None233                if ok and tunnel.spec(m, alias):234                    tun = deploy._publish(app, alias, m, ip)235                    ok, det = deploy.health(app, alias, ip=ip, m=m, timeout=20, public=public and bool(tun))236                _log("  %s : %s (%s)" % (app, "RÉPARÉE" if ok else "toujours KO", det), quiet)237                fields = {"status": "online" if ok else "unhealthy", "health": det, "healed": _now()}238                if tun:239                    fields["tunnel"] = tun240                changes[app] = fields241                summary["fixed" if ok else "ko"] += 1242            except (deploy.DeployError, tunnel.TunnelError) as e:243                _log("  %s : ÉCHEC de réparation — %s" % (app, e), quiet)244                changes[app] = {"status": "unhealthy", "health": str(e)[:200]}245                summary["ko"] += 1246    if changes and not dry_run:247        registry.update_many(changes)248    _save_state(st)249    _log("heal terminé en %.0fs : %d ok, %d réparées, %d KO, %d ignorées (grâce/sonde)%s" % (250        time.time() - t0, summary["ok"], summary["fixed"], summary["ko"], summary["skipped"], " [dry-run]" if dry_run else ""), quiet)251    return summary252253254# ------------------------------------------------------------ LaunchDaemon sur la passerelle ---255256def plist_content():257    mld_bin = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "bin", "mld")258    d = {259        "Label": HEAL_LABEL,260        "ProgramArguments": [mld_bin, "heal", "--quiet"],261        "UserName": config.USER,262        "RunAtLoad": True,263        "StartInterval": HEAL_INTERVAL_S,264        "EnvironmentVariables": {"HOME": config.HOME, "PATH": "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin", "LANG": "fr_CA.UTF-8"},265        "WorkingDirectory": config.HOME,266        "StandardOutPath": os.path.join(config.LOG_DIR, "heal.out"),267        "StandardErrorPath": os.path.join(config.LOG_DIR, "heal.err"),268        "ProcessType": "Background",269        "LowPriorityIO": True,270    }271    return plistlib.dumps(d).decode()272273274def install(interval=None):275    """Installe (ou met à jour) le LaunchDaemon io.maclustr.mld-heal sur la passerelle (sudo via ~/dispatch/.sudo)."""276    import subprocess277    global HEAL_INTERVAL_S278    if interval:279        HEAL_INTERVAL_S = int(interval)280    pwf = config.SUDO_PW_FILE281    pw = open(pwf).read().strip() if os.path.exists(pwf) else None282    if not pw:283        raise SystemExit("mot de passe sudo absent : %s" % pwf)284    tmp = "/tmp/%s.plist" % HEAL_LABEL285    open(tmp, "w").write(plist_content())286    dest = "/Library/LaunchDaemons/%s.plist" % HEAL_LABEL287    sh = ("echo %s | sudo -S -p '' sh -c 'install -m 644 -o root -g wheel %s %s && launchctl bootout system/%s 2>/dev/null; "288          "launchctl bootstrap system %s && launchctl print system/%s | grep -E \"state|interval\" | head -2'") % (289        shlex.quote(pw), tmp, dest, HEAL_LABEL, dest, HEAL_LABEL)290    p = subprocess.run(["bash", "-c", sh], capture_output=True, text=True)291    print(p.stdout.strip() or p.stderr.strip()[-300:])292    os.remove(tmp)293    ok = p.returncode == 0294    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))295    return ok296