front: expose active tasks on /metrics so training blocks an upgrade (v0.2.92)

The unattended upgrade measured idleness as requests in flight, and a LoRA
training run has none: the request that starts it returns immediately and the
work continues for hours, so coderai_engine_inflight is 0 throughout. An hourly
upgrade would have restarted straight through one -- the standing "never restart
during training" rule, reintroduced by automation.

Nothing was missing in the training path. loras.py already registers a
`training` task, the engine already ships active tasks in
/internal/engine-state, and the front already stores them on
EngineEntry.tasks. The gap was that /metrics never exposed them and the probe
never looked, so this adds coderai_engine_jobs_active per engine and
coderai_jobs_active across them, counting running/queued/paused -- paused on
purpose, because a thermally paused generation has not finished and restarting
loses it.

The probe now reads both numbers from one scrape and takes the MAXIMUM, never
the sum: a generation is both an in-flight request and a running task, so adding
them would double-count and the numbers in the log would match nothing an
operator can see. A front that predates the gauge reports "-" and the detail
line says "front too old", rather than claiming zero and silently restoring the
blind spot.

tests/test_autoupgrade_sees_training.py renders the exposition from a fake
engine carrying a training task, runs the probe's awk program as the script
actually has it, and pins both halves of the premise -- the registration in
loras.py and the engine's reporting of active states -- so the gauge cannot go
quiet without a test failing.
Co-Authored-By: 's avatarClaude Opus 5 <noreply@anthropic.com>
parent 50f94ca5
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
# Canonical product version for CoderAI — single source of truth. Both the API # Canonical product version for CoderAI — single source of truth. Both the API
# metadata and the admin web UI read from here. # metadata and the admin web UI read from here.
__version__ = "0.2.91" __version__ = "0.2.92"
__tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference." __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
# Configure the CUDA caching allocator BEFORE torch is imported anywhere. # Configure the CUDA caching allocator BEFORE torch is imported anywhere.
......
...@@ -22,7 +22,12 @@ What is exported, all prefixed ``coderai_``: ...@@ -22,7 +22,12 @@ What is exported, all prefixed ``coderai_``:
front completed, attributed to the API key that made it (its name on the front completed, attributed to the API key that made it (its name on the
Tokens page; ``session`` for a browser; ``anonymous`` when open); Tokens page; ``session`` for a browser; ``anonymous`` when open);
* ``engine_up{engine,backend,remote}``, ``engine_inflight``, * ``engine_up{engine,backend,remote}``, ``engine_inflight``,
``engine_vram_free_bytes`` / ``_total_bytes``, ``engine_loaded_models``; ``engine_jobs_active``, ``engine_vram_free_bytes`` / ``_total_bytes``,
``engine_loaded_models``;
* ``jobs_active`` — every running/queued task across the engines, training
included. Training is the case ``requests_active`` cannot see: the request
that starts it returns at once and the work runs for hours, so anything
deciding whether it is safe to restart has to read this one;
* ``node_up{node}``, ``node_recoveries_total``; * ``node_up{node}``, ``node_recoveries_total``;
* ``rpc_server_up{endpoint}``; ``discovery_peers`` / ``discovery_members``; * ``rpc_server_up{endpoint}``; ``discovery_peers`` / ``discovery_members``;
* ``runpod_spend_usd{model,period}`` (trailing hour/day/week/month), ``runpod_pods``; * ``runpod_spend_usd{model,period}`` (trailing hour/day/week/month), ``runpod_pods``;
...@@ -34,6 +39,28 @@ import time ...@@ -34,6 +39,28 @@ import time
from collections import defaultdict from collections import defaultdict
# A task still holding the card. "paused" is in here on purpose: a thermally
# paused generation has not finished, and anything that reads this to decide
# whether it is safe to restart must treat it as busy.
_ACTIVE_TASK_STATES = ("running", "queued", "paused")
def _active_jobs(engine) -> int:
"""Running/queued tasks the engine reported on its last state poll.
This is the only view of work that has NO request in flight: a LoRA
training run is started by a request that returns immediately and then
proceeds for hours, so ``engine_inflight`` is 0 throughout. Without this,
an unattended upgrade that waits for idleness would restart straight
through a training job.
"""
out = 0
for t in (getattr(engine, "tasks", None) or []):
if isinstance(t, dict) and str(t.get("status") or "") in _ACTIVE_TASK_STATES:
out += 1
return out
def _esc(v) -> str: def _esc(v) -> str:
return str(v).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") return str(v).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
...@@ -114,7 +141,7 @@ def render(front) -> str: ...@@ -114,7 +141,7 @@ def render(front) -> str:
f"coderai_front_start_time_seconds {m.started:.0f}"] f"coderai_front_start_time_seconds {m.started:.0f}"]
# Engines and nodes, from the registry. # Engines and nodes, from the registry.
up, infl, vfree, vtot, loaded, nup = [], [], [], [], [], [] up, infl, vfree, vtot, loaded, nup, jobs = [], [], [], [], [], [], []
for e in front.registry.all(): for e in front.registry.all():
if getattr(e, "role", "engine") == "system": if getattr(e, "role", "engine") == "system":
continue continue
...@@ -122,6 +149,7 @@ def render(front) -> str: ...@@ -122,6 +149,7 @@ def render(front) -> str:
lab = {"engine": e.name, "backend": e.backend or "", "remote": "1" if remote else "0"} lab = {"engine": e.name, "backend": e.backend or "", "remote": "1" if remote else "0"}
up.append("coderai_engine_up" + _labels(lab) + f" {1 if e.healthy else 0}") up.append("coderai_engine_up" + _labels(lab) + f" {1 if e.healthy else 0}")
infl.append("coderai_engine_inflight" + _labels(lab) + f" {int(getattr(e, 'inflight', 0) or 0)}") infl.append("coderai_engine_inflight" + _labels(lab) + f" {int(getattr(e, 'inflight', 0) or 0)}")
jobs.append("coderai_engine_jobs_active" + _labels(lab) + f" {_active_jobs(e)}")
v = e.vram or {} v = e.vram or {}
if v: if v:
vfree.append("coderai_engine_vram_free_bytes" + _labels(lab) vfree.append("coderai_engine_vram_free_bytes" + _labels(lab)
...@@ -134,6 +162,8 @@ def render(front) -> str: ...@@ -134,6 +162,8 @@ def render(front) -> str:
for help_, typ, rows in ( for help_, typ, rows in (
("Engine (or node) answering its health poll.", "gauge", up), ("Engine (or node) answering its health poll.", "gauge", up),
("Requests currently proxied to the engine.", "gauge", infl), ("Requests currently proxied to the engine.", "gauge", infl),
("Running/queued tasks on the engine, including training with no "
"request in flight.", "gauge", jobs),
("Free VRAM the engine reports.", "gauge", vfree), ("Free VRAM the engine reports.", "gauge", vfree),
("Total VRAM the engine reports.", "gauge", vtot), ("Total VRAM the engine reports.", "gauge", vtot),
("Models resident on the engine.", "gauge", loaded), ("Models resident on the engine.", "gauge", loaded),
...@@ -215,4 +245,16 @@ def render(front) -> str: ...@@ -215,4 +245,16 @@ def render(front) -> str:
"# TYPE coderai_requests_active gauge", f"coderai_requests_active {active}"] "# TYPE coderai_requests_active gauge", f"coderai_requests_active {active}"]
except Exception: except Exception:
pass pass
# Every piece of work in progress, whether or not a request is open on it.
# One number so "is it safe to restart?" is a single scrape and cannot be
# got wrong by summing the per-engine series (a generation shows up BOTH as
# an in-flight request and as a task, so these overlap by design).
try:
njobs = sum(_active_jobs(e) for e in front.registry.all())
lines += ["# HELP coderai_jobs_active Running/queued tasks across all engines "
"(training included; overlaps coderai_requests_active).",
"# TYPE coderai_jobs_active gauge", f"coderai_jobs_active {njobs}"]
except Exception:
pass
return "\n".join(lines) + "\n" return "\n".join(lines) + "\n"
...@@ -111,11 +111,18 @@ baked_version(){ ...@@ -111,11 +111,18 @@ baked_version(){
} }
_probe_metrics(){ _probe_metrics(){
# Sum of coderai_engine_inflight across engines. A request routed to a RunPod # Two numbers from one scrape, as "<inflight> <jobs>":
# pod is still in flight here, so this covers remote serving as well as local. # coderai_engine_inflight -- requests in flight, pod-served ones included
# coderai_jobs_active -- running/queued TASKS across the engines
# The second is what catches training: a LoRA run is started by a request
# that returns immediately and then works for hours, so inflight is 0 the
# whole time. An older front serves no jobs gauge; "-" then says so rather
# than claiming zero.
curl -fsS --max-time 10 -H "Authorization: Bearer $ADMIN_TOKEN" "$METRICS_URL" 2>/dev/null \ curl -fsS --max-time 10 -H "Authorization: Bearer $ADMIN_TOKEN" "$METRICS_URL" 2>/dev/null \
| awk '/^coderai_engine_inflight[ {]/ { n += $NF; seen = 1 } | awk '/^coderai_engine_inflight[ {]/ { n += $NF; seen = 1 }
END { if (seen) printf "%d\n", n; else print "unknown" }' /^coderai_jobs_active[ {]/ { j = $NF; jseen = 1 }
END { if (!seen) { print "unknown unknown"; exit }
printf "%d %s\n", n, (jseen ? sprintf("%d", j) : "-") }'
} }
_probe_runpod(){ _probe_runpod(){
...@@ -127,11 +134,17 @@ except Exception: print("unknown")' 2>/dev/null ...@@ -127,11 +134,17 @@ except Exception: print("unknown")' 2>/dev/null
} }
# Echoes "<count> <detail>", or "unknown <why>" when nothing could be read. # Echoes "<count> <detail>", or "unknown <why>" when nothing could be read.
#
# The MAXIMUM of the sources, never the sum: a generation appears as both an
# in-flight request and a running task, so adding them would double-count and
# the numbers in the log would not match anything an operator can see.
inflight(){ inflight(){
local m r best detail local out m j r best detail
[ -n "$ADMIN_TOKEN" ] || { echo "unknown no ADMIN_TOKEN configured"; return; } [ -n "$ADMIN_TOKEN" ] || { echo "unknown no ADMIN_TOKEN configured"; return; }
m="$(_probe_metrics)"; r="$(_probe_runpod)" out="$(_probe_metrics)"; m="${out%% *}"; j="${out##* }"
r="$(_probe_runpod)"
case "$m" in ''|unknown) m=unknown ;; *[!0-9]*) m=unknown ;; esac case "$m" in ''|unknown) m=unknown ;; *[!0-9]*) m=unknown ;; esac
case "$j" in ''|-|unknown) j=unknown ;; *[!0-9]*) j=unknown ;; esac
case "$r" in ''|unknown) r=unknown ;; *[!0-9]*) r=unknown ;; esac case "$r" in ''|unknown) r=unknown ;; *[!0-9]*) r=unknown ;; esac
if [ "$m" = unknown ] && [ "$r" = unknown ]; then if [ "$m" = unknown ] && [ "$r" = unknown ]; then
echo "unknown neither $METRICS_URL nor $STATUS_URL could be read" echo "unknown neither $METRICS_URL nor $STATUS_URL could be read"
...@@ -139,6 +152,14 @@ inflight(){ ...@@ -139,6 +152,14 @@ inflight(){
fi fi
best=0; detail="" best=0; detail=""
[ "$m" != unknown ] && { best="$m"; detail="engines=$m"; } [ "$m" != unknown ] && { best="$m"; detail="engines=$m"; }
if [ "$j" != unknown ]; then
detail="${detail:+$detail }jobs=$j"
[ "$j" -gt "$best" ] && best="$j"
else
# Only worth saying once the rest is known to work: it means the front
# predates coderai_jobs_active, so training is invisible to this check.
detail="${detail:+$detail }jobs=? (front too old)"
fi
if [ "$r" != unknown ]; then if [ "$r" != unknown ]; then
detail="${detail:+$detail }runpod=$r" detail="${detail:+$detail }runpod=$r"
[ "$r" -gt "$best" ] && best="$r" [ "$r" -gt "$best" ] && best="$r"
......
"""Training has to be visible to the thing that decides it is safe to restart.
The unattended upgrade waits for idleness before restarting, and it measured
idleness as requests in flight. A LoRA training run is started by a request that
returns immediately and then works for hours, so ``coderai_engine_inflight`` is
0 the whole time — an hourly upgrade would have restarted straight through one,
which is exactly the standing "never restart during training" rule.
Nothing was missing in the training path: ``codai/api/loras.py`` already
registers a ``training`` task, the engine already ships active tasks in
``/internal/engine-state``, and the front already stores them on
``EngineEntry.tasks``. The gap was that ``/metrics`` never exposed them and the
probe never looked. So this covers the new gauge and the probe's use of it.
"""
import pathlib
ROOT = pathlib.Path(__file__).resolve().parents[1]
SH = (ROOT / "packaging/linux/launcher/coderai-autoupgrade").read_text()
class _Engine:
role = "engine"
remote = False
def __init__(self, name, inflight=0, tasks=None):
self.name = name
self.backend = "nvidia"
self.healthy = True
self.inflight = inflight
self.vram = {}
self.loaded_models = set()
self.tasks = tasks or []
class _Registry:
def __init__(self, engines):
self._e = engines
def all(self):
return self._e
class _Front:
metrics = None
supervisor = None
def __init__(self, engines):
self.registry = _Registry(engines)
def _render(engines) -> str:
from codai.frontproxy.metrics import render
return render(_Front(engines))
def _gauge(text, name):
for line in text.splitlines():
if line.startswith(name + " ") or line.startswith(name + "{"):
return float(line.rsplit(" ", 1)[1])
return None
# --------------------------------------------------------------- the new gauge
def test_a_training_run_with_no_request_in_flight_is_visible():
"""The whole point: inflight 0, but the card is busy."""
out = _render([_Engine("nvidia", inflight=0,
tasks=[{"kind": "training", "status": "running"}])])
assert _gauge(out, "coderai_jobs_active") == 1
assert _gauge(out, "coderai_requests_active") == 0
def test_the_per_engine_series_is_labelled_like_the_others():
out = _render([_Engine("nvidia", tasks=[{"kind": "training", "status": "running"}])])
assert 'coderai_engine_jobs_active{engine="nvidia",backend="nvidia",remote="0"} 1' in out
def test_finished_tasks_are_not_counted():
"""The engine reports recent terminal tasks too, for the Tasks page history."""
out = _render([_Engine("nvidia", tasks=[
{"kind": "training", "status": "done"},
{"kind": "image", "status": "error"},
{"kind": "video", "status": "cancelled"},
])])
assert _gauge(out, "coderai_jobs_active") == 0
def test_queued_and_paused_count_as_busy():
"""A thermally paused generation has not finished; restarting loses it."""
for state in ("running", "queued", "paused"):
out = _render([_Engine("nvidia", tasks=[{"kind": "image", "status": state}])])
assert _gauge(out, "coderai_jobs_active") == 1, state
def test_jobs_are_summed_across_engines():
out = _render([
_Engine("nvidia", tasks=[{"kind": "training", "status": "running"}]),
_Engine("radeon", tasks=[{"kind": "image", "status": "running"},
{"kind": "image", "status": "done"}]),
])
assert _gauge(out, "coderai_jobs_active") == 2
def test_an_engine_with_no_tasks_reports_zero_not_nothing():
"""A missing series would read as "unknown" at the probe and fail closed."""
out = _render([_Engine("nvidia")])
assert _gauge(out, "coderai_jobs_active") == 0
assert "coderai_engine_jobs_active" in out
def test_junk_in_the_task_list_does_not_break_the_scrape():
"""It is whatever another process put in a status payload."""
out = _render([_Engine("nvidia", tasks=[None, "nonsense", 42,
{"status": "running"}])])
assert _gauge(out, "coderai_jobs_active") == 1
def test_the_gauge_says_it_overlaps_the_request_count():
"""A generation is BOTH a request in flight and a task; anyone summing the
two would double-count, so the exposition has to say so."""
out = _render([_Engine("nvidia", inflight=1,
tasks=[{"kind": "image", "status": "running"}])])
assert _gauge(out, "coderai_requests_active") == 1
assert _gauge(out, "coderai_jobs_active") == 1
assert "overlaps" in out
# ------------------------------------------------------------------- the probe
def _awk(metrics_text: str) -> str:
"""Run the probe's awk exactly as the script has it."""
import re
import subprocess
prog = re.search(r"\| awk '(.*?)'\n", SH, re.S).group(1)
r = subprocess.run(["awk", prog], input=metrics_text, capture_output=True, text=True)
return r.stdout.strip()
def test_the_probe_reads_both_numbers_from_one_scrape():
assert _awk('coderai_engine_inflight{engine="a"} 2\n'
'coderai_jobs_active 5\n') == "2 5"
def test_the_probe_sees_training_when_nothing_is_in_flight():
assert _awk('coderai_engine_inflight{engine="a"} 0\n'
'coderai_jobs_active 1\n') == "0 1"
def test_an_older_front_without_the_gauge_says_so_instead_of_zero():
"""Claiming 0 would silently restore the old blind spot."""
assert _awk('coderai_engine_inflight{engine="a"} 0\n') == "0 -"
def test_an_unreadable_scrape_is_unknown_on_both():
assert _awk("# nothing useful here\n") == "unknown unknown"
def test_the_probe_takes_the_max_not_the_sum():
"""Summing would double-count a generation, which is both."""
blk = SH.split("inflight(){")[1].split("guard_idle()")[0]
assert '[ "$j" -gt "$best" ] && best="$j"' in blk
assert '[ "$r" -gt "$best" ] && best="$r"' in blk
assert "MAXIMUM of the sources, never the sum" in SH
def test_a_missing_jobs_gauge_is_reported_in_the_detail():
blk = SH.split("inflight(){")[1].split("guard_idle()")[0]
assert "front too old" in blk
def test_the_training_path_already_registers_so_nothing_was_changed_there():
"""Guards the premise: if this registration ever goes away, the gauge goes
quiet and the blind spot comes back without anything failing."""
loras = (ROOT / "codai/api/loras.py").read_text()
assert 'task_registry.register("training"' in loras
assert 'status="running"' in loras.split('task_registry.register("training"')[1][:400]
def test_the_engine_still_reports_active_tasks_to_the_front():
"""The other half of the premise: the gauge is built from this payload."""
app = (ROOT / "codai/api/app.py").read_text()
assert '"tasks": tasks' in app
blk = app.split("from codai.tasks import task_registry")[1][:900]
assert '("running", "queued", "paused")' in blk
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment