File size: 3,294 Bytes
309b968 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | """P2P cluster scanner: probe each node's agent, measure latency, gather
resources + live status, compute the capacity plan and readiness verdict."""
import json
import socket
import time
import urllib.request
def _get(url, timeout=2.0):
with urllib.request.urlopen(url, timeout=timeout) as r:
return json.load(r)
def _latency_ms(host, port, timeout=2.0):
t = time.time()
try:
with socket.create_connection((host, int(port)), timeout=timeout):
return (time.time() - t) * 1000.0
except Exception:
return None
def scan_node(node):
host, ap = node["host"], node.get("agent_port", 8900)
base = f"http://{host}:{ap}"
out = {"name": node["name"], "host": host, "reachable": False,
"latency_ms": None, "resources": None, "status": None,
"gloo_port_open": None}
lat = _latency_ms(host, ap)
out["latency_ms"] = round(lat, 1) if lat is not None else None
if lat is None:
return out
try:
h = _get(f"{base}/health")
out["reachable"] = bool(h.get("ok"))
out["rank"] = h.get("rank")
out["resources"] = _get(f"{base}/resources")
out["status"] = _get(f"{base}/status")
except Exception as e:
out["error"] = str(e)
gp = node.get("gloo_port")
if gp:
out["gloo_port_open"] = _latency_ms(host, gp) is not None
return out
def capacity_plan(scans, base_batch=32):
live = [n for n in scans if n.get("resources")]
caps = [float(n["resources"].get("capacity") or n["resources"].get("cores", 1))
for n in live]
total = sum(caps) or 1
world = len(live)
gb = base_batch * max(1, world)
batches = [max(1, round(gb * c / total)) for c in caps]
tb = sum(batches) or 1
weights = [b / tb for b in batches]
rams = [n["resources"].get("ram_gb") for n in live
if n["resources"].get("ram_gb") is not None]
return {"world": world, "total_cores": sum(n["resources"].get("cores", 0) for n in live),
"total_ram_gb": round(sum(rams), 1) if rams else None,
"per_node": [{"name": n["name"], "device": n["resources"].get("device", "cpu"),
"cores": n["resources"].get("cores"),
"gpu": (n["resources"].get("gpu") or {}).get("name"),
"capacity": round(c, 1), "weight": round(w, 3), "batch": b,
"ram_gb": n["resources"].get("ram_gb")}
for n, c, w, b in zip(live, caps, weights, batches)],
"global_batch": sum(batches)}
def scan_cluster(nodes, expected_world=None, base_batch=32):
scans = [scan_node(n) for n in nodes]
plan = capacity_plan(scans, base_batch)
reachable = [s for s in scans if s["reachable"]]
want = expected_world if expected_world is not None else len(nodes)
ready = (len(reachable) == len(nodes) == want)
train = None
for s in scans:
st = s.get("status") or {}
if st.get("rank") == 0 and st:
train = st
break
return {"nodes": scans, "plan": plan, "ready": ready,
"reachable": len(reachable), "total": len(nodes),
"expected_world": want, "training": train,
"scanned_at": time.strftime("%H:%M:%S")}
|