diff --git a/server/db.py b/server/db.py index bd3c7fc..d0db28f 100644 --- a/server/db.py +++ b/server/db.py @@ -62,6 +62,7 @@ MIGRATIONS = [ ("jobs", "user_id", "TEXT"), # nullable; legacy rows = owner-era ("assets", "user_id", "TEXT"), ("users", "session_epoch", "INTEGER NOT NULL DEFAULT 0"), # bump to revoke sessions + ("jobs", "node", "TEXT"), # which pool node ran it; NULL = pre-2026-08-05 rows ] diff --git a/server/main.py b/server/main.py index 8d7da86..0e9e8cb 100644 --- a/server/main.py +++ b/server/main.py @@ -204,9 +204,20 @@ def _guest_may_run(user, op) -> bool: @app.get("/api/assets") -def list_assets(user=Depends(auth.current_user), request: Request = None): +def list_assets(parent_job: str = None, user=Depends(auth.current_user), + request: Request = None): con = _con(request) rows = [a for a in store.list_assets(con) if _can_see(user, a)] + # ?parent_job= used to be accepted and silently ignored, so callers filtering on it + # received the whole table and typically took rows[0] — correct only by accident of + # newest-first ordering, and wrong the moment two jobs run at once. + if parent_job: + def _pj(a): + try: + return a["parent_job"] + except Exception: + return None + rows = [a for a in rows if str(_pj(a)) == str(parent_job)] return _with_usernames(con, rows) @@ -358,6 +369,13 @@ def system(user=Depends(auth.current_user), request: Request = None): return sysinfo.snapshot(_con(request), runner) +@app.get("/api/nodes") +def nodes(user=Depends(auth.current_user), request: Request = None): + """Per-node capability + live work. /api/system only reports name/remote/busy, + and its busy flag is never updated — this is the real per-machine view.""" + return sysinfo.nodes_detail(_con(request), runner) + + # -- inbox watch folder -------------------------------------------------------- async def watch_inbox(): INBOX.mkdir(parents=True, exist_ok=True) diff --git a/server/remote.py b/server/remote.py index 8ef5cb6..e7cd301 100644 --- a/server/remote.py +++ b/server/remote.py @@ -14,6 +14,7 @@ Omit "operators" to allow every gpu op. The implicit local node is always presen import asyncio import json import shlex +import socket import time from pathlib import Path @@ -26,7 +27,11 @@ _HEALTH_TTL = 30.0 def load_gpu_nodes() -> list[dict]: """Local node first, then remote workers from nodes.json.""" - nodes = [{"name": "local", "local": True, "busy": False}] + # Name the local node after the host, not "local" — job rows and the queue UI + # should say "m3ultra", not a placeholder that looks like every other box. + # Only the "local" flag is load-bearing; nothing keys off the name. + nodes = [{"name": socket.gethostname().split(".")[0] or "local", + "local": True, "busy": False}] f = db.ROOT / "nodes.json" if f.exists(): try: diff --git a/server/runner.py b/server/runner.py index 27f3b94..71c1316 100644 --- a/server/runner.py +++ b/server/runner.py @@ -171,6 +171,10 @@ class Runner: lane = self.lane_of(job["operator"]) if lane in ("gpu", "cpu"): # pooled lanes — distribute across the node pool node = await self._acquire_node(job["operator"], lane) + # Persist placement: the pool picks a node in memory, so without this the + # job row has no record of where it ran and the queue UI attributes + # everything to the primary. + await self._update(con, job_id, node=node.get("name") or node.get("ssh")) try: if job_id in self.cancelled: self.cancelled.discard(job_id) diff --git a/server/sysinfo.py b/server/sysinfo.py index fe59e5d..0f5583a 100644 --- a/server/sysinfo.py +++ b/server/sysinfo.py @@ -68,10 +68,97 @@ def _lanes(con, runner): def gpu_pool(runner) -> list[dict]: - """Per-node status for the dashboard: name, remote?, currently busy.""" - return [{"name": n.get("name", n.get("ssh", "?")), - "remote": not n.get("local", False), - "busy": n.get("busy", False)} for n in getattr(runner, "gpu_nodes", [])] + """Per-node status for the dashboard: name, remote?, live occupancy + free slots. + + Occupancy is read from the runner's `inflight` counters. It used to read a + `busy` key, but nothing ever wrote to that after load_gpu_nodes() set it False + — so every node reported idle even while running jobs. Don't reintroduce it. + """ + pool = [] + for n in getattr(runner, "gpu_nodes", []): + inflight = n.get("inflight") or {} + gpu, cpu = int(inflight.get("gpu", 0)), int(inflight.get("cpu", 0)) + gpu_cap, cpu_cap = runner._node_capacity(n, "gpu"), runner._node_capacity(n, "cpu") + ops = n.get("operators") + pool.append({ + "name": n.get("name", n.get("ssh", "?")), + "remote": not n.get("local", False), + "busy": (gpu + cpu) > 0, + "gpu": {"running": gpu, "slots": gpu_cap, "free": max(0, gpu_cap - gpu)}, + "cpu": {"running": cpu, "slots": cpu_cap, "free": max(0, cpu_cap - cpu)}, + "operators": len(ops) if ops else "all", + }) + return pool + + +def nodes_detail(con, runner) -> dict: + """Per-node capability + live work — the real per-machine view. + + /api/system's gpu_pool only carries name/remote/busy. This joins the runner's + live inflight counters with the job table so the UI can show who is running + what, and why a heavy job waits while a light one flies. + + `running` is attributed via jobs.node, written at dispatch (2026-08-05). + Before that column existed this had to infer placement, and every job was + credited to the primary while busy remotes rendered as idle. + """ + nodes = getattr(runner, "gpu_nodes", []) + local_name = next((n.get("name") for n in nodes if n.get("local")), "local") + + rows = con.execute( + "SELECT id, operator, status, node, created_at, started_at FROM jobs " + "WHERE status IN ('running','queued') ORDER BY created_at" + ).fetchall() + + running_by_node: dict[str, list] = {} + queued: list[dict] = [] + for r in rows: + rec = {"id": r["id"], "operator": r["operator"], + "created_at": r["created_at"], "started_at": r["started_at"]} + if r["status"] == "running": + # legacy rows predate jobs.node; credit them to the primary as before + running_by_node.setdefault(r["node"] or local_name, []).append(rec) + else: + queued.append(rec) + + out, by_operator = [], {} + for n in nodes: + name = n.get("name", n.get("ssh", "?")) + ops = n.get("operators") + explicit = bool(ops) + op_ids = list(ops) if explicit else sorted(runner.operators or {}) + for op in op_ids: + by_operator.setdefault(op, []).append(name) + + gpu_cap = runner._node_capacity(n, "gpu") + inflight = n.get("inflight") or {} + # a queued job this node could pick up if it had a free slot + eligible = [q for q in queued + if (not explicit or q["operator"] in op_ids) + and runner.lane_of(q["operator"]) == "gpu"] + + out.append({ + "name": name, + "remote": not n.get("local", False), + "ssh": n.get("ssh"), + "gpu_capacity": gpu_cap, + "cpu_capacity": runner._node_capacity(n, "cpu"), + "gpu_inflight": int(inflight.get("gpu", 0)), + "cpu_inflight": int(inflight.get("cpu", 0)), + "operators_explicit": explicit, + "operators": op_ids, + "operator_count": len(op_ids), + "running": running_by_node.get(name, []), + "eligible_queued": eligible, + "notes": n.get("_notes") or [], + }) + + placed = {j["id"] for v in running_by_node.values() for j in v} + gpu_ops = sorted(op for op in (runner.operators or {}) + if runner.lane_of(op) == "gpu") + return {"nodes": out, "by_operator": by_operator, + "queued_unassigned": [q for q in queued if q["id"] not in placed], + "gpu_ops": gpu_ops} def _jobs_24h(con): diff --git a/tests/test_node_reporting.py b/tests/test_node_reporting.py new file mode 100644 index 0000000..7c49022 --- /dev/null +++ b/tests/test_node_reporting.py @@ -0,0 +1,147 @@ +"""Node-placement reporting: gpu_pool must reflect live occupancy. + +The bug this guards: gpu_pool() read a `busy` key that load_gpu_nodes() set to +False once and nothing ever updated, so every node reported idle while running +jobs. Occupancy must come from the runner's `inflight` counters. + +Run: python3 tests/test_node_reporting.py +""" +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from server import sysinfo + + +class FakeRunner: + """Minimal stand-in — gpu_pool only needs gpu_nodes + _node_capacity.""" + + def __init__(self, nodes): + self.gpu_nodes = nodes + + def _node_capacity(self, node, lane): + if lane == "gpu": + return 1 + if lane == "cpu": + return int(node.get("cpu_slots", 3 if node.get("local") else 2)) + return 0 + + +def test_busy_reflects_inflight(): + runner = FakeRunner([ + {"name": "m3ultra", "local": True, "busy": False, + "inflight": {"gpu": 1, "cpu": 0}}, + {"name": "m1", "ssh": "x@y", "busy": False, + "inflight": {"gpu": 0, "cpu": 0}, "operators": ["flux_local"]}, + ]) + pool = {n["name"]: n for n in sysinfo.gpu_pool(runner)} + + # the stale `busy: False` on the node dict must NOT win + assert pool["m3ultra"]["busy"] is True, "node running a gpu job reported idle" + assert pool["m3ultra"]["gpu"] == {"running": 1, "slots": 1, "free": 0} + assert pool["m1"]["busy"] is False + assert pool["m1"]["gpu"] == {"running": 0, "slots": 1, "free": 1} + + +def test_local_is_not_named_local(): + """load_gpu_nodes names the local node after the host, so the UI can tell + the primary apart from the placeholder it used to show.""" + from server import remote + nodes = remote.load_gpu_nodes() + assert nodes[0]["local"] is True + assert nodes[0]["name"] != "local", "local node should carry the hostname" + + +def test_operator_count_reported(): + runner = FakeRunner([ + {"name": "all-ops", "local": True, "inflight": {"gpu": 0, "cpu": 0}}, + {"name": "fenced", "ssh": "x@y", "inflight": {"gpu": 0, "cpu": 0}, + "operators": ["a", "b", "c"]}, + ]) + pool = {n["name"]: n for n in sysinfo.gpu_pool(runner)} + assert pool["all-ops"]["operators"] == "all" # no allowlist == allow-all + assert pool["fenced"]["operators"] == 3 + + +def test_node_column_migrated(): + from server import db + assert ("jobs", "node", "TEXT") in db.MIGRATIONS, "jobs.node migration missing" + + + + +# --- nodes_detail: the /api/nodes contract fluxgod consumes ------------------- + +class FakeCon: + def __init__(self, rows): self._rows = rows + def execute(self, *a, **k): return self + def fetchall(self): return self._rows + + +def _runner_with_ops(): + r = FakeRunner([ + {"name": "m3ultra", "local": True, "inflight": {"gpu": 1, "cpu": 0}}, + {"name": "m1", "ssh": "u@h", "inflight": {"gpu": 1, "cpu": 0}, + "operators": ["mflux_image_edit", "flux_local"], "_notes": ["a note"]}, + {"name": "m2max", "ssh": "u@h2", "inflight": {"gpu": 0, "cpu": 0}, + "operators": ["flux_local"]}, + ]) + r.operators = {"mflux_image_edit": {"resources": "gpu"}, + "flux_local": {"resources": "gpu"}, + "ffprobe": {"resources": "cpu"}} + r.lane_of = lambda op: r.operators.get(op, {}).get("resources", "cpu") + return r + + +def test_running_attributed_by_node_column(): + """The whole point: a job that ran on m1 must show under m1, not the primary.""" + rows = [ + {"id": "j1", "operator": "mflux_image_edit", "status": "running", + "node": "m1", "created_at": 1.0, "started_at": 1.1}, + {"id": "j2", "operator": "mflux_image_edit", "status": "running", + "node": "m3ultra", "created_at": 2.0, "started_at": 2.1}, + ] + d = sysinfo.nodes_detail(FakeCon(rows), _runner_with_ops()) + byname = {n["name"]: n for n in d["nodes"]} + assert [j["id"] for j in byname["m1"]["running"]] == ["j1"], "m1's job misattributed" + assert [j["id"] for j in byname["m3ultra"]["running"]] == ["j2"] + assert byname["m2max"]["running"] == [] + + +def test_legacy_null_node_credited_to_primary(): + rows = [{"id": "old", "operator": "flux_local", "status": "running", + "node": None, "created_at": 1.0, "started_at": 1.1}] + d = sysinfo.nodes_detail(FakeCon(rows), _runner_with_ops()) + byname = {n["name"]: n for n in d["nodes"]} + assert [j["id"] for j in byname["m3ultra"]["running"]] == ["old"] + + +def test_eligible_queued_respects_allowlist(): + rows = [{"id": "q1", "operator": "mflux_image_edit", "status": "queued", + "node": None, "created_at": 3.0, "started_at": None}] + d = sysinfo.nodes_detail(FakeCon(rows), _runner_with_ops()) + byname = {n["name"]: n for n in d["nodes"]} + assert len(byname["m1"]["eligible_queued"]) == 1 # allowlists it + assert byname["m2max"]["eligible_queued"] == [] # does not + assert len(byname["m3ultra"]["eligible_queued"]) == 1 # no allowlist == all + + +def test_contract_keys_match_captured_api(): + """Field names fluxgod's server.py reads — captured from the live API.""" + d = sysinfo.nodes_detail(FakeCon([]), _runner_with_ops()) + assert set(d) == {"nodes", "by_operator", "queued_unassigned", "gpu_ops"} + assert set(d["nodes"][0]) == { + "name", "remote", "ssh", "gpu_capacity", "cpu_capacity", "gpu_inflight", + "cpu_inflight", "operators_explicit", "operators", "operator_count", + "running", "eligible_queued", "notes"} + assert d["by_operator"]["flux_local"] == ["m3ultra", "m1", "m2max"] + assert d["gpu_ops"] == ["flux_local", "mflux_image_edit"] + + +if __name__ == "__main__": + for name, fn in sorted(globals().items()): + if name.startswith("test_") and callable(fn): + fn() + print(f" ok {name}") + print("all node-reporting checks passed")