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    suspended = []120    for app, v in reg["apps"].items():121        if apps_filter and app not in apps_filter:122            continue123        if not v.get("node"):124            continue125        if nodes_filter and v["node"] not in nodes_filter:126            continue127        if v.get("status") == "stopped":128            # arrêt volontaire (`mld stop`) : on ne ressuscite pas, `mld start` remet online129            suspended.append(app)130            continue131        by_node.setdefault(v["node"], {})[app] = manifest.load(app)132    changes = {}133    summary = {"ok": 0, "fixed": 0, "ko": 0, "skipped": len(suspended)}134    if suspended:135        _log("suspendues (mld stop, ignorées) : %s" % ", ".join(sorted(suspended)), quiet)136    tun_state = None137    try:138        tun_state = tunnel.state()139    except tunnel.TunnelError as e:140        _log("tunnel : %s" % e, quiet)141    for alias in sorted(by_node):142        apps = by_node[alias]143        ip = nodes.ip_of(alias, rediscover=False)144        if not ip:145            # une seule redécouverte par passage146            ip = nodes.ip_of(alias, rediscover=True)147        if not ip:148            since = st["down"].setdefault(alias, _now())149            _log("%s : INJOIGNABLE (depuis %s) — %d app(s) : %s" % (alias, since, len(apps), ", ".join(sorted(apps))), quiet)150            for app in apps:151                if reg["apps"][app].get("status") != "node-down":152                    changes[app] = {"status": "node-down", "health": "nœud %s injoignable" % alias}153            summary["ko"] += len(apps)154            continue155        if alias in st["down"]:156            _log("%s : de retour (était injoignable depuis %s)" % (alias, st["down"].pop(alias)), quiet)157        rep = _node_report(ip, apps)158        if rep is None:159            _log("%s : sonde impossible" % alias, quiet)160            summary["skipped"] += len(apps)161            continue162        if rep["uptime"] < GRACE_S:163            _log("%s : redémarré il y a %ds — délai de grâce, on repasse plus tard" % (alias, rep["uptime"]), quiet)164            summary["skipped"] += len(apps)165            continue166        # --- WireGuard vers la passerelle (seulement si le nœud expose un site)167        exposes = any(tunnel.spec(m, alias) for m in apps.values())168        if exposes and (rep["wg1"] is False or rep["hub"] is False):169            _log("%s : wg1 %s / hub %s" % (alias, "up" if rep["wg1"] else "DOWN", "ok" if rep["hub"] else "KO"), quiet)170            if not dry_run:171                try:172                    tunnel.ensure_peer(alias, log=lambda s: _log(s, quiet), st=tun_state)173                except tunnel.TunnelError as e:174                    _log("  tunnel : %s" % e, quiet)175        resurrected = False176        for app in sorted(apps):177            m = apps[app]178            names = manifest.pm2_names(m)179            labels = manifest.launchd_labels(m)180            missing = [n for n in names if n not in rep["procs"]] if rep["pm2"] else list(names)181            bad = [n for n in names if n in rep["procs"] and rep["procs"][n][0] in ("errored", "stopped")182                   and next((p for p in m["processes"] if p["name"] == n), {}).get("autorestart", True)183                   and not next((p for p in m["processes"] if p["name"] == n), {}).get("cron_restart")]184            la_missing = [l for l in labels if l not in rep["la"]]185            code = rep["http"].get(app)186            http_ok = _http_ok(code) if m.get("port") else True187            problems = []188            if missing:189                problems.append("PM2 absent: %s" % ",".join(missing))190            if bad:191                problems.append("PM2 %s: %s" % (rep["procs"][bad[0]][0], ",".join(bad)))192            if la_missing:193                problems.append("launchd absent: %s" % ",".join(la_missing))194            if not http_ok and not missing and not bad:195                problems.append("HTTP local %s" % code)196            if not problems:197                summary["ok"] += 1198                if reg["apps"][app].get("status") != "online":199                    changes[app] = {"status": "online", "health": "local %s" % code if m.get("port") else "processus online"}200                # route publique : suit-elle bien ce nœud ?201                if tun_state and public and tunnel.spec(m, alias):202                    s = tunnel.spec(m, alias)203                    r = tunnel.routes(s["gateway"], tun_state).get(s["domain"])204                    if not tunnel._route_matches(r, s["upstreams"]):205                        _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)206                        if not dry_run and _allowed(st, app):207                            try:208                                tunnel.ensure_route(m, alias, log=lambda x: _log(x, quiet), st=tun_state)209                                _record(st, app)210                                changes[app] = dict(changes.get(app, {}), tunnel={"gateway": s["gateway"], "domain": s["domain"], "upstream": s["upstreams"][0], "url": "https://%s" % s["domain"]})211                                summary["fixed"] += 1212                            except tunnel.TunnelError as e:213                                _log("  tunnel : %s" % e, quiet)214                continue215            _log("%s/%s : %s" % (alias, app, " ; ".join(problems)), quiet)216            if dry_run:217                summary["ko"] += 1218                continue219            if not _allowed(st, app):220                _log("  %s : trop d'actions dans l'heure (%d), on laisse tranquille" % (app, MAX_ACTIONS_PER_HOUR), quiet)221                summary["ko"] += 1222                changes[app] = {"status": "unhealthy", "health": " ; ".join(problems)}223                continue224            _record(st, app)225            try:226                if rep["pm2"] is False and not resurrected:227                    rc, out, _ = ssh.run(ip, "pm2 resurrect 2>&1 | tail -1", timeout=120)228                    _log("  pm2 resurrect sur %s : %s" % (alias, out.strip()[-100:]), quiet)229                    resurrected = True230                if missing or la_missing:231                    deploy.start(app, alias, m=m, ip=ip)232                elif bad:233                    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)234                    _log("  pm2 restart %s : %s" % (",".join(bad), out.strip()[-40:]), quiet)235                elif not http_ok:236                    rc, out, _ = ssh.run(ip, "pm2 restart %s --update-env >/dev/null 2>&1; echo ok" % " ".join(names), timeout=120) if names else (0, "", "")237                    _log("  HTTP local %s → pm2 restart %s" % (code, ",".join(names)), quiet)238                ok, det = deploy.health(app, alias, ip=ip, m=m, timeout=45, public=False)239                tun = None240                if ok and tunnel.spec(m, alias):241                    tun = deploy._publish(app, alias, m, ip)242                    ok, det = deploy.health(app, alias, ip=ip, m=m, timeout=20, public=public and bool(tun))243                _log("  %s : %s (%s)" % (app, "RÉPARÉE" if ok else "toujours KO", det), quiet)244                fields = {"status": "online" if ok else "unhealthy", "health": det, "healed": _now()}245                if tun:246                    fields["tunnel"] = tun247                changes[app] = fields248                summary["fixed" if ok else "ko"] += 1249            except (deploy.DeployError, tunnel.TunnelError) as e:250                _log("  %s : ÉCHEC de réparation — %s" % (app, e), quiet)251                changes[app] = {"status": "unhealthy", "health": str(e)[:200]}252                summary["ko"] += 1253    if changes and not dry_run:254        registry.update_many(changes)255    _save_state(st)256    _log("heal terminé en %.0fs : %d ok, %d réparées, %d KO, %d ignorées (grâce/sonde)%s" % (257        time.time() - t0, summary["ok"], summary["fixed"], summary["ko"], summary["skipped"], " [dry-run]" if dry_run else ""), quiet)258    return summary259260261# ------------------------------------------------------------ LaunchDaemon sur la passerelle ---262263def plist_content():264    mld_bin = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "bin", "mld")265    d = {266        "Label": HEAL_LABEL,267        "ProgramArguments": [mld_bin, "heal", "--quiet"],268        "UserName": config.USER,269        "RunAtLoad": True,270        "StartInterval": HEAL_INTERVAL_S,271        "EnvironmentVariables": {"HOME": config.HOME, "PATH": "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin", "LANG": "fr_CA.UTF-8"},272        "WorkingDirectory": config.HOME,273        "StandardOutPath": os.path.join(config.LOG_DIR, "heal.out"),274        "StandardErrorPath": os.path.join(config.LOG_DIR, "heal.err"),275        "ProcessType": "Background",276        "LowPriorityIO": True,277    }278    return plistlib.dumps(d).decode()279280281def install(interval=None):282    """Installe (ou met à jour) le LaunchDaemon io.maclustr.mld-heal sur la passerelle (sudo via ~/dispatch/.sudo)."""283    import subprocess284    global HEAL_INTERVAL_S285    if interval:286        HEAL_INTERVAL_S = int(interval)287    pwf = config.SUDO_PW_FILE288    pw = open(pwf).read().strip() if os.path.exists(pwf) else None289    if not pw:290        raise SystemExit("mot de passe sudo absent : %s" % pwf)291    tmp = "/tmp/%s.plist" % HEAL_LABEL292    open(tmp, "w").write(plist_content())293    dest = "/Library/LaunchDaemons/%s.plist" % HEAL_LABEL294    sh = ("echo %s | sudo -S -p '' sh -c 'install -m 644 -o root -g wheel %s %s && launchctl bootout system/%s 2>/dev/null; "295          "launchctl bootstrap system %s && launchctl print system/%s | grep -E \"state|interval\" | head -2'") % (296        shlex.quote(pw), tmp, dest, HEAL_LABEL, dest, HEAL_LABEL)297    p = subprocess.run(["bash", "-c", sh], capture_output=True, text=True)298    print(p.stdout.strip() or p.stderr.strip()[-300:])299    os.remove(tmp)300    ok = p.returncode == 0301    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))302    return ok303