spb/maclustr-dispatch

1"""Registre des déploiements (source de vérité) : ~/dispatch/registry.json.23Diffusion (2026-09-04) : chaque sauvegarde POUSSE une copie du registre aux4abonnés déclarés dans ~/dispatch/subscribers.json, pour que les consommateurs5(console admin-ka, gardiens ka2/ka4/ka6…) connaissent TOUJOURS l'emplacement6réel des apps sans rien coder en dur. Un abonné est un dict :78    {"app": "admin-ka", "path": "data/registry.json"}        # path relatif au dir de l'app9    {"app": "ka2", "path": "~/ka-guardian-spool/registry.json"}10    {"node": "M4M36", "path": "~/x/registry.json"}           # ou un nœud fixe1112« app » est résolu dans le registre lui-même (nœud + dir) : l'abonnement suit13les migrations. Le push est best-effort (les consommateurs re-tirent aussi14périodiquement avec `ssh gitsrv cat ~/dispatch/registry.json`).15"""16import json17import os18import shlex19import sys20import time21from . import config2223SUBSCRIBERS = os.path.join(config.STATE, "subscribers.json")242526def load():27    if os.path.exists(config.REGISTRY):28        return json.load(open(config.REGISTRY))29    return {"updated": None, "gateway": config.GATEWAY, "apps": {}, "history": []}303132def save(r, push=True):33    r["updated"] = time.strftime("%Y-%m-%dT%H:%M:%S")34    r["gateway"] = config.GATEWAY35    tmp = config.REGISTRY + ".tmp"36    json.dump(r, open(tmp, "w"), indent=2, ensure_ascii=False)37    os.replace(tmp, config.REGISTRY)38    if push:39        try:40            push_to_subscribers(r)41        except Exception as e:  # jamais bloquant pour un déploiement42            print("  registre : diffusion aux abonnés échouée (%s)" % e, file=sys.stderr)434445def node_of(app):46    return load()["apps"].get(app, {}).get("node")474849def apps_on(alias):50    return [a for a, v in load()["apps"].items() if v.get("node") == alias]515253def set_app(app, **fields):54    r = load()55    cur = r["apps"].get(app, {})56    cur.update(fields)57    cur["updated"] = time.strftime("%Y-%m-%dT%H:%M:%S")58    r["apps"][app] = cur59    save(r)606162def remove_app(app, note=""):63    r = load()64    if app in r["apps"]:65        r["history"].append({"ts": time.strftime("%Y-%m-%dT%H:%M:%S"), "app": app, "event": "removed", "was": r["apps"][app], "note": note})66        del r["apps"][app]67    save(r)686970def log(app, event, **kw):71    r = load()72    r["history"].append(dict(ts=time.strftime("%Y-%m-%dT%H:%M:%S"), app=app, event=event, **kw))73    r["history"] = r["history"][-500:]74    save(r)757677# ----------------------------------------------------------------- abonnés ---7879def load_subscribers():80    if os.path.exists(SUBSCRIBERS):81        return json.load(open(SUBSCRIBERS))82    return []838485def save_subscribers(subs):86    tmp = SUBSCRIBERS + ".tmp"87    json.dump(subs, open(tmp, "w"), indent=2, ensure_ascii=False)88    os.replace(tmp, SUBSCRIBERS)899091def _abs(p, ip=None):92    return p.replace("~", config.home_of(ip), 1) if p.startswith("~") else p939495def resolve_subscriber(sub, r):96    """→ (label, node, ip, chemin absolu) ou lève ValueError."""97    from . import nodes98    apps = r.get("apps", {})99    if sub.get("app"):100        entry = apps.get(sub["app"])101        if not entry or not entry.get("node"):102            raise ValueError("app %s absente du registre" % sub["app"])103        node = entry["node"]104        ip = entry.get("ip") or nodes.ip_of(node, rediscover=False)105        base = entry.get("dir") or "~"106        label = "%s@%s" % (sub["app"], node)107    else:108        node = sub["node"]109        ip = nodes.ip_of(node, rediscover=False)110        base = "~"111        label = node112    if not ip:113        raise ValueError("IP inconnue pour %s" % node)114    path = sub["path"]115    if not (path.startswith("/") or path.startswith("~")):116        path = base.rstrip("/") + "/" + path117    return label, node, ip, _abs(path, ip)118119120def push_to_subscribers(r=None, verbose=True):121    """Copie atomique du registre chez chaque abonné (ssh LAN, 12 s max chacun)."""122    from . import ssh123    r = r or load()124    subs = load_subscribers()125    results = []126    payload = json.dumps(r, indent=2, ensure_ascii=False)127    for sub in subs:128        try:129            label, node, ip, path = resolve_subscriber(sub, r)130        except ValueError as e:131            results.append((sub, False, str(e)))132            if verbose:133                print("  registre → %s : ignoré (%s)" % (sub.get("app") or sub.get("node"), e))134            continue135        q = shlex.quote(path)136        script = ("mkdir -p $(dirname %s) && cat > %s.tmp <<'__MLD_REG__'\n%s\n__MLD_REG__\n"137                  "mv -f %s.tmp %s && echo ok" % (q, q, payload, q, q))138        rc, out, err = ssh.run(ip, script, timeout=12, path=False)139        ok = rc == 0 and "ok" in out140        results.append((sub, ok, (err or out).strip()[:120]))141        if verbose:142            print("  registre → %s:%s %s" % (label, path, "ok" if ok else "ÉCHEC " + (err or out).strip()[:80]))143    return results144