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 update_many(changes, push=True):63    """Met à jour plusieurs apps ({app: {champs}}) en une seule sauvegarde/diffusion (status --live, heal)."""64    if not changes:65        return66    r = load()67    now = time.strftime("%Y-%m-%dT%H:%M:%S")68    for app, fields in changes.items():69        cur = r["apps"].get(app, {})70        cur.update(fields)71        cur["updated"] = now72        r["apps"][app] = cur73    save(r, push=push)747576def remove_app(app, note=""):77    r = load()78    if app in r["apps"]:79        r["history"].append({"ts": time.strftime("%Y-%m-%dT%H:%M:%S"), "app": app, "event": "removed", "was": r["apps"][app], "note": note})80        del r["apps"][app]81    save(r)828384def log(app, event, **kw):85    r = load()86    r["history"].append(dict(ts=time.strftime("%Y-%m-%dT%H:%M:%S"), app=app, event=event, **kw))87    r["history"] = r["history"][-500:]88    save(r)899091# ----------------------------------------------------------------- abonnés ---9293def load_subscribers():94    if os.path.exists(SUBSCRIBERS):95        return json.load(open(SUBSCRIBERS))96    return []979899def save_subscribers(subs):100    tmp = SUBSCRIBERS + ".tmp"101    json.dump(subs, open(tmp, "w"), indent=2, ensure_ascii=False)102    os.replace(tmp, SUBSCRIBERS)103104105def _abs(p, ip=None):106    return p.replace("~", config.home_of(ip), 1) if p.startswith("~") else p107108109def resolve_subscriber(sub, r):110    """→ (label, node, ip, chemin absolu) ou lève ValueError."""111    from . import nodes112    apps = r.get("apps", {})113    if sub.get("app"):114        entry = apps.get(sub["app"])115        if not entry or not entry.get("node"):116            raise ValueError("app %s absente du registre" % sub["app"])117        node = entry["node"]118        ip = entry.get("ip") or nodes.ip_of(node, rediscover=False)119        base = entry.get("dir") or "~"120        label = "%s@%s" % (sub["app"], node)121    else:122        node = sub["node"]123        ip = nodes.ip_of(node, rediscover=False)124        base = "~"125        label = node126    if not ip:127        raise ValueError("IP inconnue pour %s" % node)128    path = sub["path"]129    if not (path.startswith("/") or path.startswith("~")):130        path = base.rstrip("/") + "/" + path131    return label, node, ip, _abs(path, ip)132133134def push_to_subscribers(r=None, verbose=True):135    """Copie atomique du registre chez chaque abonné (ssh LAN, 12 s max chacun)."""136    from . import ssh137    r = r or load()138    subs = load_subscribers()139    results = []140    payload = json.dumps(r, indent=2, ensure_ascii=False)141    for sub in subs:142        try:143            label, node, ip, path = resolve_subscriber(sub, r)144        except ValueError as e:145            results.append((sub, False, str(e)))146            if verbose:147                print("  registre → %s : ignoré (%s)" % (sub.get("app") or sub.get("node"), e))148            continue149        q = shlex.quote(path)150        script = ("mkdir -p $(dirname %s) && cat > %s.tmp <<'__MLD_REG__'\n%s\n__MLD_REG__\n"151                  "mv -f %s.tmp %s && echo ok" % (q, q, payload, q, q))152        rc, out, err = ssh.run(ip, script, timeout=12, path=False)153        ok = rc == 0 and "ok" in out154        results.append((sub, ok, (err or out).strip()[:120]))155        if verbose:156            print("  registre → %s:%s %s" % (label, path, "ok" if ok else "ÉCHEC " + (err or out).strip()[:80]))157    return results158