spb/maclustr-dispatch
Public
1"""Cycle de vie d'une app : fetch (nœud -> staging), deploy (staging -> nœud), stop, start, retire, move, health."""2import json3import os4import shlex5import time6from . import config, manifest, nodes, registry, render, ssh78SYNC_BASE_EXCLUDES = [".DS_Store", "*.log", "__pycache__/", ".pytest_cache/", ".mypy_cache/"]91011class DeployError(Exception):12 pass131415def _ctx(alias, ip):16 lan = nodes.load_lan()["ips"]17 return {"HOME": "/Users/%s" % config.USER, "NODE": alias, "LAN_IP": ip, "ips": lan, "USER": config.USER}181920def _abs(p):21 return p.replace("~", "/Users/%s" % config.USER, 1) if p.startswith("~") else p222324def _log(app, msg):25 line = "%s %s" % (time.strftime("%H:%M:%S"), msg)26 print(line, flush=True)27 with open(os.path.join(config.LOG_DIR, app + ".log"), "a") as f:28 f.write(time.strftime("%Y-%m-%d ") + line + "\n")293031def _ip(alias):32 ip = nodes.ip_of(alias)33 if not ip:34 raise DeployError("nœud %s injoignable sur le LAN" % alias)35 return ip363738def _ensure_dir(ip, d):39 """Crée le répertoire cible ; sous /opt (racine root) passe par sudo -S avec le mot de passe de ~/dispatch/.sudo."""40 if ssh.ok(ip, "mkdir -p %s 2>/dev/null && test -w %s" % (shlex.quote(d), shlex.quote(d))):41 return42 pw = open(config.SUDO_PW_FILE).read().strip() if os.path.exists(config.SUDO_PW_FILE) else None43 if not pw:44 raise DeployError("impossible de créer %s (sudo requis, %s absent)" % (d, config.SUDO_PW_FILE))45 rc, out, err = ssh.run(ip, "echo %s | sudo -S -p '' mkdir -p %s && echo %s | sudo -S -p '' chown %s %s" % (shlex.quote(pw), shlex.quote(d), shlex.quote(pw), config.USER, shlex.quote(d)))46 if rc != 0:47 raise DeployError("création de %s impossible : %s" % (d, err.strip()[-200:]))484950def stage_dir(app):51 return os.path.join(config.STAGE_DIR, app)525354# ---------------------------------------------------------------- fetch : nœud -> staging55def fetch(app, from_alias, final=False):56 m = manifest.load(app)57 ip = _ip(from_alias)58 _log(app, "fetch %s depuis %s (%s)%s" % (m["dir"], from_alias, ip, " [final]" if final else ""))59 excl = SYNC_BASE_EXCLUDES + list(m.get("sync_excludes") or [])60 rc, out, err = ssh.rsync_pull(ip, _abs(m["dir"]), os.path.join(stage_dir(app), "dir"), excludes=excl)61 if rc == 23 and not final:62 _log(app, " avertissement pré-copie à chaud (fichiers en cours d'écriture, repris au delta final) : %s" % err.strip().splitlines()[0][:120])63 elif rc not in (0, 24):64 raise DeployError("rsync fetch a échoué (%s): %s" % (rc, err.strip()[-300:]))65 for extra in m.get("extra_paths") or []:66 dest = os.path.join(stage_dir(app), "extra", extra.strip("~/").replace("/", "__"))67 rc2, out2, err2 = ssh.run(ip, "test -e %s" % shlex.quote(_abs(extra)))68 if rc2 != 0:69 _log(app, " extra absent sur la source, ignoré : %s" % extra)70 continue71 isdir = ssh.ok(ip, "test -d %s" % shlex.quote(_abs(extra)))72 if isdir:73 rc2, out2, err2 = ssh.rsync_pull(ip, _abs(extra), dest, delete=True)74 else:75 os.makedirs(dest, exist_ok=True)76 import subprocess77 subprocess.run(["scp", "-p"] + ssh.SSH_OPTS + ["%s:%s" % (ssh.target(ip), _abs(extra)), os.path.join(dest, os.path.basename(extra))], capture_output=True)78 json.dump({"app": app, "from": from_alias, "ip": ip, "ts": time.strftime("%Y-%m-%dT%H:%M:%S"), "final": final},79 open(os.path.join(stage_dir(app), "fetch.json"), "w"), indent=2)80 stat = [l for l in out.splitlines() if "Total transferred file size" in l or "Number of regular files transferred" in l or "Number of files transferred" in l or "Total bytes sent" in l]81 _log(app, " ok — " + " ; ".join(s.strip() for s in stat))82 return True838485# ---------------------------------------------------------------- stop / start / status sur un nœud86def stop(app, alias, keep_dir=True):87 m = manifest.load(app)88 ip = _ip(alias)89 _log(app, "stop sur %s" % alias)90 names = manifest.pm2_names(m)91 script = ""92 if names:93 script += "for n in %s; do pm2 delete $n >/dev/null 2>&1 && echo \" pm2 delete $n\"; done; pm2 save --force >/dev/null 2>&1\n" % " ".join(names)94 for lbl in manifest.launchd_labels(m):95 script += "launchctl bootout gui/$(id -u)/%s >/dev/null 2>&1 && echo ' bootout %s'; rm -f ~/Library/LaunchAgents/%s.plist\n" % (lbl, lbl, lbl)96 if m.get("ngrok"):97 script += "pkill -f -- 'url=%s' 2>/dev/null; pkill -f -- 'url=https://%s' 2>/dev/null; pkill -f -- 'domain=%s' 2>/dev/null; true\n" % (m["ngrok"]["url"], m["ngrok"]["url"], m["ngrok"]["url"])98 for hook in (m.get("hooks") or {}).get("pre_stop") or []:99 script += hook + "\n"100 # tue ce qui écoute encore sur les ports de l'app (orphelins)101 for p in m["requires"].get("ports") or []:102 script += "sleep 1; for pid in $(lsof -nP -iTCP:%s -sTCP:LISTEN -t 2>/dev/null); do kill $pid 2>/dev/null && echo ' kill port %s pid '$pid; done\n" % (p, p)103 if m.get("ngrok"):104 script += "sleep 4\n" # laisse ngrok libérer le domaine réservé côté service105 rc, out, err = ssh.run(ip, script, timeout=180)106 for l in out.splitlines():107 _log(app, l)108 registry.log(app, "stopped", node=alias)109 return rc == 0110111112def _install_hooks(m, ip, ctx):113 """Commandes post-sync : réécritures .env, hooks, inscription pousseur."""114 script = ""115 d = _abs(m["dir"])116 for fname, kv in (m.get("env_overrides") or {}).items():117 target = os.path.join(d, fname)118 kv = manifest.render(kv, ctx)119 py = "import re,sys,os\np=%r\nkv=%r\ns=open(p).read() if os.path.exists(p) else ''\nfor k,v in kv.items():\n line=k+'='+v\n if re.search(r'^'+re.escape(k)+r'=.*$', s, re.M): s=re.sub(r'^'+re.escape(k)+r'=.*$', line.replace('\\\\','\\\\\\\\'), s, flags=re.M)\n else: s=s.rstrip('\\n')+'\\n'+line+'\\n'\nopen(p,'w').write(s)\nprint(' .env réécrit: '+', '.join(kv))" % (target, kv)120 script += "python3 - <<'PYEOF'\n%s\nPYEOF\n" % py121 if m.get("ka_repo"):122 script += "grep -qx %s ~/.ka-pousseur-repos 2>/dev/null || echo %s >> ~/.ka-pousseur-repos; echo ' pousseur: %s inscrit'\n" % (shlex.quote(m["dir"]), shlex.quote(m["dir"]), m["dir"])123 for hook in (m.get("hooks") or {}).get("post_sync") or []:124 script += "cd %s && (%s)\n" % (shlex.quote(d), manifest.render(hook, ctx))125 return script126127128def deploy(app, alias, reinstall=False, skip_sync=False):129 """staging -> nœud, puis démarrage (PM2 + launchd) et healthcheck."""130 m = manifest.load(app)131 ip = _ip(alias)132 ctx = _ctx(alias, ip)133 sd = os.path.join(stage_dir(app), "dir")134 if not skip_sync and not os.path.isdir(sd):135 raise DeployError("staging vide pour %s — faire `mld fetch %s --from <nœud>` ou `mld stage`" % (app, app))136 _log(app, "deploy → %s (%s)" % (alias, ip))137 # 1. runtimes138 scan = nodes.load_scan().get(alias) or nodes.probe(alias, ip)139 missing = [r for r in m["requires"]["runtimes"] if not scan.get("runtimes", {}).get(r)]140 if missing:141 raise DeployError("runtimes manquants sur %s : %s (→ `mld prepare %s`)" % (alias, ", ".join(missing), alias))142 # 2. sync143 d = _abs(m["dir"])144 if not skip_sync:145 _ensure_dir(ip, d)146 excl = SYNC_BASE_EXCLUDES + list(m.get("sync_excludes") or [])147 rc, out, err = ssh.rsync_push(sd, ip, d, excludes=excl, delete=True)148 if rc not in (0, 24):149 raise DeployError("rsync deploy a échoué (%s): %s" % (rc, err.strip()[-300:]))150 stat = [l.strip() for l in out.splitlines() if "Total transferred file size" in l or "Total bytes sent" in l]151 _log(app, " sync ok — %s" % (stat[0] if stat else ""))152 for extra in m.get("extra_paths") or []:153 src = os.path.join(stage_dir(app), "extra", extra.strip("~/").replace("/", "__"))154 if not os.path.exists(src):155 continue156 if os.path.isdir(src) and not os.path.exists(os.path.join(src, os.path.basename(extra))):157 ssh.rsync_push(src, ip, _abs(extra), delete=False)158 else:159 import subprocess160 ssh.run(ip, "mkdir -p %s" % shlex.quote(os.path.dirname(_abs(extra))))161 subprocess.run(["scp", "-p"] + ssh.SSH_OPTS + [os.path.join(src, os.path.basename(extra)), "%s:%s" % (ssh.target(ip), _abs(extra))], capture_output=True)162 _log(app, " extra: %s" % extra)163 # 3. install (optionnel) + hooks164 script = "cd %s || exit 3\n" % shlex.quote(d)165 if reinstall:166 script += "if [ -f pnpm-lock.yaml ]; then pnpm install --frozen-lockfile; elif [ -f package-lock.json ]; then npm ci; elif [ -f package.json ]; then npm install; fi\n"167 script += "if [ -f requirements.txt ] && [ -x .venv/bin/pip ]; then .venv/bin/pip install -q -r requirements.txt; fi\n"168 elif m.get("rebuild", True) and os.path.exists(os.path.join(sd, "package.json")) and os.path.isdir(os.path.join(sd, "node_modules")):169 script += "npm rebuild >/dev/null 2>&1 && echo ' npm rebuild ok' || echo ' npm rebuild: avertissement'\n"170 script += _install_hooks(m, ip, ctx)171 rc, out, err = ssh.run(ip, script, timeout=1800)172 for l in out.splitlines()[-15:]:173 _log(app, l)174 if rc != 0:175 raise DeployError("post-sync a échoué (%s): %s" % (rc, (err or out).strip()[-400:]))176 # 4. lancement177 start(app, alias, m=m, ctx=ctx, ip=ip)178 # 5. santé179 healthy, detail = health(app, alias, ip=ip, m=m)180 registry.set_app(app, node=alias, ip=ip, port=m.get("port"), domain=m.get("domain"), dir=m["dir"], label=m.get("label") or app,181 processes=manifest.pm2_names(m), launchd=manifest.launchd_labels(m), deployed=time.strftime("%Y-%m-%d %H:%M"),182 status="online" if healthy else "unhealthy", health=detail)183 registry.log(app, "deployed", node=alias, healthy=healthy)184 _log(app, " santé : %s" % detail)185 if not healthy:186 names = manifest.pm2_names(m)187 rc, out, _ = ssh.run(ip, "pm2 logs %s --nostream --lines 6 2>/dev/null | grep -v '^\\[TAILING' | tail -12" % " ".join(names), timeout=40)188 for l in out.splitlines():189 _log(app, " log: " + l[:160])190 raise DeployError("healthcheck KO sur %s : %s" % (alias, detail))191 return True192193194def start(app, alias, m=None, ctx=None, ip=None):195 m = m or manifest.load(app)196 ip = ip or _ip(alias)197 ctx = ctx or _ctx(alias, ip)198 _log(app, "start sur %s" % alias)199 run_dir = _abs(config.RUN_DIR)200 eco = render.ecosystem(m, ctx)201 eco_path = "%s/%s.config.cjs" % (run_dir, app)202 ssh.write_remote_file(ip, eco_path, eco, mode="600")203 script = "mkdir -p %s\n" % run_dir204 names = manifest.pm2_names(m)205 if names:206 script += "for n in %s; do pm2 delete $n >/dev/null 2>&1; done\n" % " ".join(names)207 script += "pm2 start %s --update-env 2>&1 | grep -E 'error|Error|\\[PM2\\]' | head -5; pm2 save --force >/dev/null 2>&1; pm2 jlist | python3 -c \"import json,sys; L=json.load(sys.stdin); print(' pm2: '+', '.join(p['name']+'='+p['pm2_env']['status'] for p in L if p['name'] in %s))\" 2>/dev/null || pm2 ls | grep -E '%s'\n" % (shlex.quote(eco_path), json.dumps(names), "|".join(names))208 for item in m.get("launchd") or []:209 pl = render.plist(item, ctx)210 ppath = "%s/Library/LaunchAgents/%s.plist" % (ctx["HOME"], item["label"])211 ssh.write_remote_file(ip, ppath, pl, mode="644")212 script += "launchctl bootout gui/$(id -u)/%s >/dev/null 2>&1; launchctl bootstrap gui/$(id -u) %s && echo ' launchd: %s chargé' || echo ' launchd: %s ÉCHEC'\n" % (item["label"], shlex.quote(ppath), item["label"], item["label"])213 for hook in (m.get("hooks") or {}).get("post_start") or []:214 script += "cd %s && (%s)\n" % (shlex.quote(_abs(m["dir"])), manifest.render(hook, ctx))215 rc, out, err = ssh.run(ip, script, timeout=300)216 for l in out.splitlines():217 _log(app, l)218 if err.strip():219 _log(app, " stderr: " + err.strip()[-200:])220 return rc == 0221222223def health(app, alias, ip=None, m=None, timeout=90):224 m = m or manifest.load(app)225 ip = ip or _ip(alias)226 if not m.get("port"):227 # pas de port HTTP : vérifie que les processus PM2 sont online228 names = manifest.pm2_names(m)229 if not names:230 return True, "aucun port/processus à vérifier"231 rc, out, _ = ssh.run(ip, "pm2 jlist | python3 -c \"import json,sys; L=json.load(sys.stdin); print(sum(1 for p in L if p['name'] in %s and p['pm2_env']['status']=='online'))\"" % json.dumps(names))232 n = int((out.strip() or "0").splitlines()[-1] or 0)233 return n == len(names), "%d/%d processus PM2 online" % (n, len(names))234 url = "http://127.0.0.1:%s%s" % (m["port"], m.get("health_path") or "/")235 deadline = time.time() + timeout236 code = "000"237 while time.time() < deadline:238 rc, out, _ = ssh.run(ip, "curl -s -o /dev/null -m 8 -w '%%{http_code}' %s" % shlex.quote(url), timeout=20)239 code = out.strip()[-3:] if out.strip() else "000"240 if code.isdigit() and code not in ("000",) and int(code) < 500:241 break242 time.sleep(4)243 local_ok = code.isdigit() and code != "000" and int(code) < 500244 pub = ""245 if m.get("domain") and local_ok and m.get("ngrok"):246 import subprocess247 for _ in range(20):248 p = subprocess.run(["curl", "-s", "-o", "/dev/null", "-m", "12", "-w", "%{http_code}", "https://%s%s" % (m["domain"], m.get("health_path") or "/")], capture_output=True, text=True)249 pub = p.stdout.strip()250 if pub.isdigit() and int(pub) < 500 and pub != "404":251 break252 time.sleep(5)253 detail = "local %s" % code + (" / public https://%s %s" % (m["domain"], pub) if pub else "")254 ok = local_ok and (not pub or (pub.isdigit() and int(pub) < 500 and pub != "404"))255 return ok, detail256257258def retire(app, alias, remove_dir=True):259 """Arrête et efface la copie d'un nœud (après migration réussie)."""260 m = manifest.load(app)261 ip = _ip(alias)262 stop(app, alias)263 if remove_dir:264 d = _abs(m["dir"])265 rc, out, err = ssh.run(ip, "rm -rf %s && echo ' rm -rf %s'; rm -f %s/%s.config.cjs; grep -v -x %s ~/.ka-pousseur-repos > ~/.ka-pousseur-repos.tmp 2>/dev/null && mv ~/.ka-pousseur-repos.tmp ~/.ka-pousseur-repos; true" % (shlex.quote(d), d, _abs(config.RUN_DIR), app, shlex.quote(m["dir"])), timeout=600)266 for l in out.splitlines():267 _log(app, l)268 registry.log(app, "retired", node=alias)269 return True270271272def move(app, to_alias, from_alias=None, keep_source=False, reinstall=False):273 """Migration complète : pré-sync à chaud, arrêt, sync final, déploiement, santé, retrait de la source (ou rollback)."""274 from_alias = from_alias or registry.node_of(app) or manifest.load(app).get("source_node")275 if not from_alias:276 raise DeployError("nœud source inconnu pour %s (préciser --from)" % app)277 if from_alias == to_alias:278 _log(app, "déjà sur %s → redéploiement en place" % to_alias)279 fetch(app, from_alias)280 stop(app, from_alias)281 fetch(app, from_alias, final=True)282 return deploy(app, to_alias, reinstall=reinstall)283 _log(app, "MOVE %s : %s → %s" % (app, from_alias, to_alias))284 fetch(app, from_alias) # 1. copie à chaud (gros volume)285 stop(app, from_alias) # 2. arrêt source286 fetch(app, from_alias, final=True) # 3. delta final (cohérence des bases)287 try:288 deploy(app, to_alias, reinstall=reinstall) # 4. cible289 except DeployError as e:290 _log(app, "ÉCHEC sur %s : %s — rollback sur %s" % (to_alias, e, from_alias))291 try:292 stop(app, to_alias)293 except Exception:294 pass295 start(app, from_alias)296 ok, det = health(app, from_alias)297 registry.set_app(app, node=from_alias, status="online" if ok else "unhealthy", health=det)298 raise299 if not keep_source:300 retire(app, from_alias) # 5. nettoyage source301 registry.log(app, "moved", **{"from": from_alias, "to": to_alias})302 _log(app, "MOVE terminé : %s tourne sur %s" % (app, to_alias))303 return True304305306def status(app=None):307 r = registry.load()308 apps = [app] if app else sorted(r["apps"])309 rows = []310 for a in apps:311 v = r["apps"].get(a)312 if not v:313 rows.append((a, "—", "", "", "non déployée", ""))314 continue315 rows.append((a, v.get("node"), v.get("port"), v.get("domain") or "", v.get("status"), v.get("deployed")))316 return rows317318319def live_status(app=None, public=True):320 """Vérifie en live (PM2 + port + domaine) chaque app du registre."""321 r = registry.load()322 apps = [app] if app else sorted(r["apps"])323 out = []324 for a in apps:325 v = r["apps"].get(a)326 if not v:327 continue328 m = manifest.load(a)329 try:330 ip = _ip(v["node"])331 ok, det = health(a, v["node"], ip=ip, m=m, timeout=10) if public else (None, "")332 except DeployError as e:333 ok, det = False, str(e)334 registry.set_app(a, status="online" if ok else "unhealthy", health=det)335 out.append((a, v["node"], v.get("domain") or "", "OK" if ok else "KO", det))336 return out337