spb/maclustr-dispatch
Public
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."""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 lan = load_lan()63 lan["ips"] = found64 save_lan(lan)65 if verbose:66 for a in sorted(found, key=lambda x: list(config.NODES).index(x) if x in config.NODES else 99):67 print(" %-8s %s" % (a, found[a]))68 missing = [a for a in config.NODES if a not in found]69 if missing:70 print(" hors LAN / éteints : %s" % ", ".join(missing))71 return found727374def ip_of(alias, rediscover=True):75 lan = load_lan()76 ip = lan["ips"].get(alias)77 if ip and ssh.ok(ip, "true", timeout=15):78 return ip79 if rediscover:80 found = discover(verbose=False)81 return found.get(alias)82 return ip838485def probe(alias, ip):86 rc, out, err = ssh.run(ip, _probe_script(), timeout=90)87 info = {"alias": alias, "ip": ip, "online": False}88 if "KV cores" not in out:89 info["error"] = (err or out).strip()[:200]90 return info91 rts = {}92 kv = {}93 for line in out.splitlines():94 if line.startswith("KV "):95 parts = line.split(" ", 2)96 kv[parts[1]] = parts[2].strip() if len(parts) > 2 else ""97 elif line.startswith("RT "):98 _, name, val = line.split(" ", 2)99 rts[name] = val.strip() == "1"100 def num(k, f=float, d=0):101 try:102 return f(kv.get(k, "") or d)103 except Exception:104 return d105 info.update({"host": kv.get("host", ""), "marker": kv.get("marker", ""), "os": kv.get("os", ""), "cores": num("cores", int),106 "ram_gb": num("ram_gb", int), "load1": num("load1"), "free_gb": num("free_gb"), "disk_free_gb": num("disk_free_gb", int)})107 if kv.get("ip"):108 info["ip"] = kv["ip"]109 try:110 info["uptime_h"] = round((time.time() - int(kv.get("boot") or 0)) / 3600, 1) if kv.get("boot") else None111 except Exception:112 info["uptime_h"] = None113 info["ports"] = sorted(set(int(p) for p in kv.get("ports", "").split(",") if p.isdigit()))114 info["pm2"] = [p for p in kv.get("pm2", "").split(",") if p]115 info["runtimes"] = rts116 info["online"] = info["cores"] > 0117 info.update({k: v for k, v in config.NODES.get(alias, {}).items() if k not in info or not info.get(k)})118 info["role"] = config.ROLES.get(alias, "worker")119 return info120121122def scan(aliases=None, verbose=True):123 lan = load_lan()124 if not lan["ips"]:125 discover(verbose=False)126 lan = load_lan()127 targets = [(a, ip) for a, ip in lan["ips"].items() if not aliases or a in aliases]128 res = {}129 with ThreadPoolExecutor(max_workers=24) as ex:130 for info in ex.map(lambda t: probe(*t), targets):131 res[info["alias"]] = info132 for a in config.NODES:133 if a not in res:134 res[a] = {"alias": a, "online": False, "role": config.ROLES[a], "runtimes": {}, **config.NODES[a]}135 json.dump({"updated": time.strftime("%Y-%m-%dT%H:%M:%S"), "nodes": res}, open(config.SCAN_CACHE, "w"), indent=2)136 if verbose:137 print_scan(res)138 return res139140141def load_scan():142 if os.path.exists(config.SCAN_CACHE):143 return json.load(open(config.SCAN_CACHE))["nodes"]144 return {}145146147def print_scan(res):148 print("%-8s %-6s %-16s %5s %6s %6s %6s %7s %4s %s" % ("nœud", "rôle", "ip", "cœurs", "RAM", "libre", "load", "disque", "pm2", "runtimes"))149 for a in config.NODES:150 n = res.get(a, {})151 if not n.get("online"):152 print("%-8s %-6s %-16s — hors ligne" % (a, n.get("role", ""), n.get("ip", "")))153 continue154 rt = ",".join(k.replace("python@", "py").replace("uv-python@", "uvpy") for k, v in sorted(n["runtimes"].items()) if v)155 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))156