spb/maclustr-dispatch

1"""Découverte LAN des nœuds + sonde de ressources/runtimes (scan)."""2import json3import os4import time5from concurrent.futures import ThreadPoolExecutor6from . import config, ssh78RUNTIME_PROBE = "\n".join('if %s; then echo "RT %s 1"; else echo "RT %s 0"; fi' % (cmd, name, name) for name, cmd in config.RUNTIME_CHECKS.items())91011def _probe_script():12    # sonde 100 % shell (les minis sans Xcode CLT n'ont pas de python3 fonctionnel) : lignes "KV clé valeur"13    sh = r'''14export LC_ALL=C15echo "KV host $(hostname -s)"16echo "KV marker $(cat ~/.maclustr-node 2>/dev/null)"17echo "KV ip $(ifconfig | awk '/inet 192\.168\.2\./{print $2; exit}')"18echo "KV os $(sw_vers -productVersion)"19echo "KV cores $(sysctl -n hw.ncpu)"20echo "KV ram_gb $(( $(sysctl -n hw.memsize) / 1073741824 ))"21echo "KV load1 $(sysctl -n vm.loadavg | tr -d '{}' | awk '{print $1}' | tr ',' '.')"22echo "KV free_gb $(vm_stat | awk '/Pages (free|inactive|speculative|purgeable)/{gsub("\\.","",$NF); s+=$NF} END{printf "%.1f", s*16384/1073741824}')"23echo "KV disk_free_gb $(df -g / | awk 'NR==2{print $4}')"24echo "KV boot $(sysctl -n kern.boottime | sed -E 's/.*sec = ([0-9]+).*/\1/')"25echo "KV ports $(lsof -nP -iTCP -sTCP:LISTEN 2>/dev/null | awk 'NR>1{n=split($9,a,":"); print a[n]}' | sort -un | tr '\n' ',')"26echo "KV pm2 $(pm2 jlist 2>/dev/null | tr -d '\n' | grep -o '"name":"[^"]*"' | cut -d'"' -f4 | sort -u | tr '\n' ',')"27'''28    return sh + "\n" + RUNTIME_PROBE + "\n"293031def load_lan():32    if os.path.exists(config.LAN_CACHE):33        return json.load(open(config.LAN_CACHE))34    return {"updated": None, "ips": {}}353637def save_lan(d):38    d["updated"] = time.strftime("%Y-%m-%dT%H:%M:%S")39    json.dump(d, open(config.LAN_CACHE, "w"), indent=2)404142def _identify(ip):43    rc, out, _ = ssh.run(ip, "cat ~/.maclustr-node 2>/dev/null || hostname -s", timeout=12, path=False)44    if rc == 0 and out.strip():45        return ip, out.strip().splitlines()[0].strip()46    return ip, None474849def discover(rng=range(1, 255), verbose=True):50    """Balaye 192.168.2.0/24 en SSH (clé passerelle) et lit le marqueur ~/.maclustr-node ; ajoute les REMOTE_NODES joignables."""51    ips = [config.LAN_PREFIX + str(i) for i in rng]52    found = {}53    with ThreadPoolExecutor(max_workers=48) as ex:54        for ip, alias in ex.map(_identify, ips):55            if alias:56                # normalise l'alias (le marqueur est la source de vérité, sinon hostname)57                for a in config.NODES:58                    if a.lower() == alias.lower():59                        alias = a60                        break61                found[alias] = ip62    # nœuds hors LAN : connexion directe à l'IP publique, le marqueur doit confirmer l'alias63    for alias, r in config.REMOTE_NODES.items():64        _, seen = _identify(r["host"])65        if seen and seen.lower() == alias.lower():66            found[alias] = r["host"]67    lan = load_lan()68    lan["ips"] = found69    save_lan(lan)70    if verbose:71        for a in sorted(found, key=lambda x: list(config.NODES).index(x) if x in config.NODES else 99):72            print("  %-8s %s" % (a, found[a]))73        missing = [a for a in config.NODES if a not in found]74        if missing:75            print("  hors LAN / éteints : %s" % ", ".join(missing))76    return found777879def ip_of(alias, rediscover=True):80    lan = load_lan()81    ip = lan["ips"].get(alias)82    if ip and ssh.ok(ip, "true", timeout=15):83        return ip84    if rediscover:85        found = discover(verbose=False)86        return found.get(alias)87    return ip888990def probe(alias, ip):91    rc, out, err = ssh.run(ip, _probe_script(), timeout=90)92    info = {"alias": alias, "ip": ip, "online": False}93    if "KV cores" not in out:94        info["error"] = (err or out).strip()[:200]95        return info96    rts = {}97    kv = {}98    for line in out.splitlines():99        if line.startswith("KV "):100            parts = line.split(" ", 2)101            kv[parts[1]] = parts[2].strip() if len(parts) > 2 else ""102        elif line.startswith("RT "):103            _, name, val = line.split(" ", 2)104            rts[name] = val.strip() == "1"105    def num(k, f=float, d=0):106        try:107            return f(kv.get(k, "") or d)108        except Exception:109            return d110    info.update({"host": kv.get("host", ""), "marker": kv.get("marker", ""), "os": kv.get("os", ""), "cores": num("cores", int),111                 "ram_gb": num("ram_gb", int), "load1": num("load1"), "free_gb": num("free_gb"), "disk_free_gb": num("disk_free_gb", int)})112    if kv.get("ip"):113        info["ip"] = kv["ip"]114    try:115        info["uptime_h"] = round((time.time() - int(kv.get("boot") or 0)) / 3600, 1) if kv.get("boot") else None116    except Exception:117        info["uptime_h"] = None118    info["ports"] = sorted(set(int(p) for p in kv.get("ports", "").split(",") if p.isdigit()))119    info["pm2"] = [p for p in kv.get("pm2", "").split(",") if p]120    info["runtimes"] = rts121    info["online"] = info["cores"] > 0122    info.update({k: v for k, v in config.NODES.get(alias, {}).items() if k not in info or not info.get(k)})123    info["role"] = config.ROLES.get(alias, "worker")124    return info125126127def scan(aliases=None, verbose=True):128    lan = load_lan()129    if not lan["ips"]:130        discover(verbose=False)131        lan = load_lan()132    targets = [(a, ip) for a, ip in lan["ips"].items() if not aliases or a in aliases]133    # scan partiel (`mld scan X`) : on FUSIONNE dans le cache existant au lieu de marquer les autres nœuds hors ligne134    # (sinon `mld deploy` juste après voyait tout le cluster « hors ligne »)135    res = load_scan() if aliases else {}136    with ThreadPoolExecutor(max_workers=24) as ex:137        for info in ex.map(lambda t: probe(*t), targets):138            res[info["alias"]] = info139    for a in config.NODES:140        if a not in res:141            res[a] = {"alias": a, "online": False, "role": config.ROLES[a], "runtimes": {}, **config.NODES[a]}142    json.dump({"updated": time.strftime("%Y-%m-%dT%H:%M:%S"), "nodes": res}, open(config.SCAN_CACHE, "w"), indent=2)143    if verbose:144        print_scan(res)145    return res146147148def load_scan():149    if os.path.exists(config.SCAN_CACHE):150        return json.load(open(config.SCAN_CACHE))["nodes"]151    return {}152153154def print_scan(res):155    print("%-8s %-6s %-16s %5s %6s %6s %6s %7s %4s  %s" % ("nœud", "rôle", "ip", "cœurs", "RAM", "libre", "load", "disque", "pm2", "runtimes"))156    for a in config.NODES:157        n = res.get(a, {})158        if not n.get("online"):159            print("%-8s %-6s %-16s   — hors ligne" % (a, n.get("role", ""), n.get("ip", "")))160            continue161        rt = ",".join(k.replace("python@", "py").replace("uv-python@", "uvpy") for k, v in sorted(n["runtimes"].items()) if v)162        print("%-8s %-6s %-16s %5s %5sG %5sG %6.2f %6sG %4s  %s" % (a, n["role"], n["ip"], n["cores"], n["ram_gb"], n["free_gb"], n["load1"], n["disk_free_gb"], len(n.get("pm2", [])), rt))163