From fa56944d0fe6f89e6cee3f3a41584ddcb16037b0 Mon Sep 17 00:00:00 2001 From: ShubyM Date: Thu, 29 Oct 2026 19:20:09 -0700 Subject: [PATCH] feat(observability): summarize runs and sessions Add three inspection routes. /api/v1/inspect/runs gives each run in one line from small records. /runs//summary gives the facts recorded about one run, and /sessions//summary the facts about one client session. A summary shows a block only when it has content. The API server now keeps each session's record and the SDK's error events through the telemetry buffer. Session models come from the model records, which bind_session already writes. Ages count from the request and never go negative. An ended run shows no ages. Goodput splits the trainer's idle time into GPU wait, queued, rollout and client. The ledger reports a lost request only after the gap has held across reads more than a flush apart. Log pointers use the /logs arguments run_id, pod, since and until. --- Makefile | 2 +- docs/configuration.md | 6 + src/server/api_server.py | 6 +- src/server/observability/facts.py | 283 +++++++++++++++++++++++++++++ src/server/observability/fleet.py | 2 +- src/server/observability/router.py | 46 ++++- src/server/store.py | 34 +++- src/server/telemetry.py | 56 ++++++ tests/test_summaries.py | 160 ++++++++++++++++ 9 files changed, 585 insertions(+), 10 deletions(-) create mode 100644 src/server/observability/facts.py create mode 100644 tests/test_summaries.py diff --git a/Makefile b/Makefile index 41bec14d..500ef6d6 100644 --- a/Makefile +++ b/Makefile @@ -23,7 +23,7 @@ UNIT_TESTS ?= tests.test_session_lifecycle tests.test_proto_codec tests.test_for # Only forward BASE_URL to e2e when the user supplied it. The Makefile default # is for local CLI usage; e2e should start its own backend by default. TRAINING_TEST_BASE_URL ?= $(if $(filter environment command line,$(origin BASE_URL)),$(BASE_URL),) -UNIT_TESTS += tests.test_observability_kubernetes +UNIT_TESTS += tests.test_observability_kubernetes tests.test_summaries TRAINING_TEST_EXTRA ?= gpu TRAINING_TEST_ARGS ?= PIGLATIN_TEST_PYTHONPATH ?= examples/sft/pig-latin diff --git a/docs/configuration.md b/docs/configuration.md index 19e258ed..76b14694 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -144,6 +144,12 @@ its own phase, GPU turns and settings. `/api/v1/inspect/runs//steps?last=&fields=` returns the run's trainer ops grouped by step, with their counts, failures, time and metrics. +`/api/v1/inspect/runs` returns each live run in one line, and `?ended=true` +adds ended runs. `/api/v1/inspect/runs//summary` and +`/api/v1/inspect/sessions//summary` return the facts recorded about one run +or one client session, such as its pace, goodput, unanswered requests, failures +and SDK errors. See `/docs` for their fields. + | Env var | Default | What it does | | --- | --- | --- | | `OPEN_RL_LOG_STORE` | unset | Where it reads logs. Unset turns log reads off. `cloud-logging` reads Cloud Logging in the project of the server's Google credentials. `cloud-logging:` reads that Cloud Logging resource instead, such as `projects//locations/global/buckets//views/`. | diff --git a/src/server/api_server.py b/src/server/api_server.py index d878aed8..c29aff5b 100644 --- a/src/server/api_server.py +++ b/src/server/api_server.py @@ -64,6 +64,7 @@ async def bind_session(session_id: str | None, model_id: str) -> None: if not session_id: return + run_records.session_seen(session_id) owner = await asyncio.to_thread(owner_of, model_id) async with owner_locks[owner]: # On the model, for the model view: which sessions are this model's own. @@ -962,6 +963,7 @@ async def create_session(req: CreateSessionRequest): raise HTTPException(status_code=400, detail=str(exc)) from exc session_id = f"sess-{uuid.uuid4().hex[:12]}" await session_registry.update_metadata(session_id, user_metadata) + run_records.session_seen(session_id, user_metadata) return {"session_id": session_id, "type": "create_session"} @@ -969,6 +971,7 @@ async def create_session(req: CreateSessionRequest): async def session_heartbeat(req: SessionHeartbeatRequest): if req.session_id: await session_registry.heartbeat(req.session_id) + run_records.session_seen(req.session_id) return {"type": "session_heartbeat"} @@ -1428,5 +1431,6 @@ async def list_adapters(): @app.post("/api/v1/telemetry") -async def telemetry(_: dict): +async def telemetry(body: dict): + run_records.sdk_events(body) return {"status": "accepted"} diff --git a/src/server/observability/facts.py b/src/server/observability/facts.py new file mode 100644 index 00000000..c60b1717 --- /dev/null +++ b/src/server/observability/facts.py @@ -0,0 +1,283 @@ +"""Facts about a run or a session, joined from their records, with no verdicts. A block appears only when it has content, +so an absent block means nothing of its kind was recorded. A log pointer is a read against /api/v1/inspect to run as +given. Ages count from the request, and a run that ended shows none.""" + +import asyncio +import json +import math +import statistics +from datetime import UTC, datetime + +from server.model_metadata import pool_name +from server.observability.steps import TRAINING +from server.store import get_state_store, get_store +from server.telemetry import FLUSH_SECONDS, key + +STALE_SECONDS = 6 * FLUSH_SECONDS # a worker that has not flushed in this long no longer reports +BUCKETS = {"train": TRAINING, "checkpoint": {"save_state", "save_weights_for_sampler"}, "forward": {"forward"}} +SETTING_ENV = {"openrl.fine_tuning_type": "OPEN_RL_FINE_TUNING_TYPE", "openrl.weight_sync": "OPEN_RL_WEIGHT_SYNC_STRATEGY"} + +# Each run's ledger gaps as this API server's reads saw them, (at, gap) oldest first. A process's counts land up to a +# flush late, so a gap is a loss only once it has held across reads more than a flush apart. +gaps: dict[str, list[tuple[float, int]]] = {} + + +def iso(at: float) -> str: + return datetime.fromtimestamp(at, UTC).isoformat(timespec="seconds") + + +def read(route: str, **args) -> dict: + return {"read": route, "args": {k: v for k, v in args.items() if v is not None}} + + +def present(**blocks) -> dict: + return {k: v for k, v in blocks.items() if v not in (None, [], {})} + + +def age(now: float, at: float | None) -> int | None: + return None if at is None else max(0, round(now - at)) + + +async def workers() -> list[dict]: + return [json.loads(raw) for raw in (await get_state_store().scan_values(key("worker", ""))).values()] + + +def workers_of(found: list[dict], run: dict) -> list[dict]: + ids = {run["run_id"], run["runtime"], pool_name(run["runtime"])} + return [w for w in found if w.get("runtime_id") in ids] + + +def alive(worker: dict, now: float) -> bool: + return worker.get("phase") != "exited" and now - worker.get("heartbeat_at", 0) < STALE_SECONDS + + +def unanswered(run: dict, workers: list[dict], now: float) -> list[dict]: + """Trainer requests the client waits on, with the pod holding each and whether its worker still reports.""" + return [ + {"op": g["op"], "count": g["count"], "age_s": age(now, g["since"]), "pod": w.get("pod") or w["id"], "holder_alive": alive(w, now)} + for w in workers + for g in w.get("outstanding") or [] + if g["run_id"] == run["run_id"] + ] + + +def relaunches(run: dict, workers: list[dict]) -> list[dict]: + """Worker processes that replaced an earlier one in the same pod during the run: a container restarted in place, or a + new pod after one died holding requests. A clean exit is not a relaunch.""" + by_pod: dict[str, list[dict]] = {} + low, high = run["created_at"] - 60, run["completed_at"] or math.inf + for w in sorted(workers, key=lambda w: w["started_at"]): + if w.get("pod") and low <= w["started_at"] <= high: + by_pod.setdefault(w["pod"], []).append(w) + return [ + {"pod": b["pod"], "stopped_at": a.get("heartbeat_at", b["started_at"]), "started_at": b["started_at"]} + for p in by_pod.values() + for a, b in zip(p, p[1:]) + if a.get("phase") != "exited" and (a.get("pod_uid") == b.get("pod_uid") or a.get("outstanding")) + ] + + +def pace(times: list[float], now: float, ended: bool) -> dict: + gaps_s = [b - a for a, b in zip(times, times[1:])] + return present( + step_seconds={"median": round(statistics.median(gaps_s), 1), "last": round(gaps_s[-1], 1)} if gaps_s else None, + since_last_step_s=None if ended or not times else age(now, times[-1]), + ) + + +async def sdk_errors(session_ids: list[str]) -> list[dict]: + found = await asyncio.gather(*(get_state_store().read_samples(key("sdk_errors", s)) for s in session_ids)) + return sorted((e for errors in found for e in errors), key=lambda e: e["at"]) + + +def errors_block(errors: list[dict]) -> dict | None: + return {"count": len(errors), "last": errors[-1]} if errors else None + + +async def line(run: dict, workers: list[dict], now: float) -> dict: + """A run in one line, enough to see which run to open, from small records rather than its op history.""" + store, run_id, ended = get_state_store(), run["run_id"], run["lifecycle"] == "ended" + steps, times, ledger, last_ops, samples, errors = await asyncio.gather( + store.get_value(key("steps", run_id)), + store.read_samples(key("step_times", run_id)), + store.read_counts(key("ledger", run_id)), + store.read_samples(key("operations", run_id), last=1), + store.read_samples(key("operations", run_id, "samples")), + sdk_errors(run["session_ids"]), + ) + held, last = unanswered(run, workers, now), max((r["at"] for r in last_ops + samples[-1:]), default=None) + found = pace(times, now, ended) + return { + "lifecycle": run["lifecycle"], + "steps": int(steps or 0), + **present(since_last_op_s=None if ended else age(now, last), since_last_step_s=found.get("since_last_step_s")), + "median_step_s": found.get("step_seconds", {}).get("median"), + "trainer_requests_unanswered": sum(o["count"] for o in held), + "held_by_dead_worker": sum(o["count"] for o in held if not o["holder_alive"]), + "failures": ledger.get("failed", 0), + "sample_failures": sum(s["status"] == "failed" for s in samples), + "sdk_errors": len(errors), + "relaunches": len(relaunches(run, workers)), + } + + +def lost(run_id: str, gap: int, now: float) -> int | None: + """Requests enqueued that no worker popped and the queue no longer holds, once the gap has lasted over a flush.""" + seen, cutoff = gaps.setdefault(run_id, []), now - FLUSH_SECONDS + seen.append((now, gap)) + while len(seen) > 1 and seen[1][0] < cutoff: + seen.pop(0) + return max(0, min(g for _, g in seen)) if seen[0][0] < cutoff else None + + +def covered(spans: list[tuple[float, float]], low: float, high: float) -> float: + """Seconds of [low, high] inside at least one of the spans.""" + total, reach = 0.0, low + for a, b in sorted((max(a, low), min(b, high)) for a, b in spans): + if (top := max(a, reach)) < b: + total, reach = total + b - top, b + return total + + +def goodput(ops: list[dict], samples: list[dict], workers: list[dict], now: float, ended: bool) -> dict | None: + """Share of the trainer's span spent training (Google ML goodput), and where the rest went.""" + if not ops: + return None + start = min(s["started_at"] for s in ops) + end = max(start + 1, max(s["at"] for s in ops) if ended else now) + spent = {kind: sum(s["elapsed_seconds"] for s in ops if s["operation"] in names) for kind, names in BUCKETS.items()} + # Each idle second counts once, in the first of these that covers it; client is the rest. + layers = { + "busy": [(s["started_at"], s["at"]) for s in ops], + "gpu_wait": [tuple(w) for worker in workers if worker.get("role") == "trainer" for w in worker.get("waits") or []], + "queued": [(s["started_at"] - s["queue_seconds"], s["started_at"]) for s in ops if "queue_seconds" in s], + "rollout": [(s["started_at"], s["at"]) for s in samples if s["status"] == "succeeded"], + } + split, spans, before = {}, [], 0.0 + for name, more in layers.items(): + spans += more + reached = covered(spans, start, end) + split[name], before = reached - before, reached + del split["busy"] + steps = [s["at"] for s in ops if s["operation"] == "optim_step" and s["status"] == "succeeded"] + return present( + goodput=round(spent["train"] / (end - start), 3), + seconds={k: round(v) for k, v in ({"wall": end - start} | spent | split | {"client": end - start - before}).items()}, + first_step_after_s=round(min(steps) - start) if steps else None, + ) + + +def failures(ops: list[dict], run_id: str) -> dict | None: + if not (failed := [s for s in ops if s["status"] == "failed"]): + return None + last = failed[-1] + return { + "count": len(failed), + "ops": [{"operation": s["operation"], "error_type": s["error_type"], "at": iso(s["at"]), "pod": s.get("pod")} for s in reversed(failed[-5:])], + "logs": read("logs", run_id=run_id, pod=last.get("pod"), since=iso(last["started_at"] - 60), until=iso(last["at"] + 60)), + } + + +def saves(ops: list[dict], steps: list[dict]) -> dict | None: + if not (found := [s for s in ops if s["operation"] == "save_state"]): + return None + last = max((s["at"] for s in found if s["status"] == "succeeded"), default=None) + return present( + count=len(found), + failed=sum(s["status"] == "failed" for s in found), + last_at=iso(last) if last else None, + steps_since_last=sum(s["at"] > (last or 0) for s in steps), + ) + + +def samplers(samples: list[dict]) -> dict | None: + """A vLLM sampler keeps one record per group it served each flush, with the group's count, and one per failed sample.""" + if not samples: + return None + groups = [s for s in samples if s["status"] == "succeeded"] + failed = len(samples) - len(groups) + found = {"sends": sum(s.get("count", 1) for s in groups) + failed, "failed": failed} + return found | ({"median_s": round(statistics.median(s["elapsed_seconds"] for s in groups), 2)} if groups else {}) + + +def config(run: dict, workers: list[dict]) -> list[dict] | None: + """Workers whose settings env differs from what the run asked for.""" + found = [] + for w in workers: + env = (w.get("config") or {}).get("env") or {} + for setting, name in SETTING_ENV.items(): + asked, running = run["settings"].get(setting), env.get(name) + if asked is not None and running is not None and str(asked) != running: + found.append({"pod": w.get("pod"), "setting": setting, "asked": asked, "running": running}) + return found or None + + +async def summary(run: dict, workers: list[dict], now: float) -> dict: + store, run_id, ended = get_state_store(), run["run_id"], run["lifecycle"] == "ended" + ops, samples, count, times, ledger, queued, errors = await asyncio.gather( + store.read_samples(key("operations", run_id)), + store.read_samples(key("operations", run_id, "samples")), + store.get_value(key("steps", run_id)), + store.read_samples(key("step_times", run_id)), + store.read_counts(key("ledger", run_id)), + get_store().queued(run_id), + sdk_errors(run["session_ids"]), + ) + steps = [s for s in ops if s["operation"] == "optim_step" and s["status"] == "succeeded"] + if "enqueued" in ledger and (gone := lost(run_id, ledger["enqueued"] - ledger.get("popped", 0) - queued, now)) is not None: + ledger["lost"] = gone + poisoned = [s for s in ops if s.get("nonfinite")] + first = present(operation=poisoned[0]["operation"], step=poisoned[0].get("step"), metrics=poisoned[0]["nonfinite"]) if poisoned else None + norms = [float(f"{s['metrics']['grad_norm:mean']:.4g}") for s in steps if "grad_norm:mean" in s["metrics"]] + restarts = None + if relaunched := relaunches(run, workers): + p = relaunched[0] + restarts = { + "workers": [{"pod": p["pod"], "stopped_reporting_at": iso(p["stopped_at"]), "restarted_at": iso(p["started_at"])} for p in relaunched], + "logs": read("logs", run_id=run_id, pod=p["pod"], since=iso(p["stopped_at"] - 120), until=iso(p["started_at"] + 60)), + } + last_op = max(ops + samples[-1:], key=lambda s: s["at"], default=None) + return present( + lifecycle=run["lifecycle"], + steps=int(count) if count else len(steps), + **pace(times, now, ended), + since_last_op_s=None if ended or last_op is None else age(now, last_op["at"]), + last_op={"operation": last_op["operation"], "status": last_op["status"], "at": iso(last_op["at"])} if last_op else None, + goodput=goodput(ops, samples, workers, now, ended), + requests=present(unanswered=unanswered(run, workers, now), ledger=ledger), + failures=failures(ops, run_id), + relaunches=restarts, + saves=saves(ops, steps), + optimizer=present(nonfinite={"count": len(poisoned), "first": first} if poisoned else None, grad_norm_last=norms[-5:]), + samplers=samplers(samples), + config=config(run, workers), + placement=present(nodes=run["nodes"], pods=[p["name"] for p in run["pods"]]), + incidents=[{"pod": p["name"], "problem": p["problem"]} for p in run["pods"] if p["problem"]], + sdk_errors=errors_block(errors), + ) + + +async def session(session_id: str, runs: list[dict], workers: list[dict], now: float) -> dict | None: + """The session's record, or what its runs and live key show when it has none, with a line per run.""" + store = get_state_store() + record, seen, live, errors = await asyncio.gather( + store.get_value(key("session_record", session_id)), + store.get_value(key("session_seen", session_id)), + store.get_value(key("session", session_id)), + store.read_samples(key("sdk_errors", session_id)), + ) + mine = [r for r in runs if session_id in r["session_ids"]] + if record is None and live is None and not mine: + return None + lines = await asyncio.gather(*(line(r, workers_of(workers, r), now) for r in mine)) + saved = json.loads(record or "{}") + return present( + session_id=session_id, + created_at=saved.get("created_at"), + metadata=saved.get("metadata"), + last_seen_at=float(seen) if seen else None, + live=live is not None, + recorded=record is not None, + models=[{"model_id": r["run_id"], **found} for r, found in zip(mine, lines)], + sdk_errors=errors_block(errors), + ) diff --git a/src/server/observability/fleet.py b/src/server/observability/fleet.py index 69428b88..6b705f23 100644 --- a/src/server/observability/fleet.py +++ b/src/server/observability/fleet.py @@ -77,7 +77,7 @@ def join(run_id: str, raw: str, live: set[str], cluster: Cluster) -> dict: run.update(model=meta.base_model, recipe_name=row.get("recipe_name"), fine_tuning_type=meta.fine_tuning_type, client_status=meta.status) run["lifecycle"] = lifecycle(meta.status, any(s in live for s in meta.session_ids), states) run.update(created_at=meta.created_at, completed_at=meta.completed_at, session_ids=meta.session_ids, settings=openrl_status.settings_of(meta)) - run.update(shared_runtime=shared, nodes=sorted({p["node"] for p in pods if p["node"]})) + run.update(runtime=runtime, shared_runtime=shared, nodes=sorted({p["node"] for p in pods if p["node"]})) return run | {"pods": [{"name": p["name"], "problem": p["problem"]} for p in pods]} diff --git a/src/server/observability/router.py b/src/server/observability/router.py index 17a23943..9a7866d2 100644 --- a/src/server/observability/router.py +++ b/src/server/observability/router.py @@ -1,4 +1,6 @@ +import asyncio import os +import time from datetime import datetime from typing import Annotated, Literal @@ -6,7 +8,7 @@ from fastapi import APIRouter, HTTPException, Query from server.model_metadata import decode_model_metadata -from server.observability import cloud_logging, fleet, metrics +from server.observability import cloud_logging, facts, fleet, metrics from server.observability.kubernetes import namespace from server.observability.log_selector import ClientLines, WorkerPods from server.observability.steps import steps @@ -80,6 +82,48 @@ async def query( return {**result, "available": False, "error": f"The metrics source is unavailable: {type(exc).__name__}"} +@router.get("/api/v1/inspect/runs") +async def runs(ended: bool = False): + """Every live run in one line, longest since its last op (a trainer op or a sample) first, so a stuck run is one read + away. ended=true adds ended runs, which show no since_last_op_s or since_last_step_s. failures counts failed trainer + requests, sample_failures failed samples, and sdk_errors the SDK's error events across the run's sessions.""" + state, found, now = await fleet.snapshot(), await facts.workers(), time.time() + chosen = [r for r in state["runs"] if ended or r["lifecycle"] != "ended"] + limit = asyncio.Semaphore(64) # each read in flight holds a store connection + + async def line(run: dict) -> dict: + async with limit: + return {"run_id": run["run_id"], "name": run["name"], **await facts.line(run, facts.workers_of(found, run), now)} + + lines = await asyncio.gather(*(line(r) for r in chosen)) + return {"runs": sorted(lines, key=lambda r: -r.get("since_last_op_s", -1))} + + +@router.get("/api/v1/inspect/runs/{run_id}/summary") +async def run_summary(run_id: str): + """The run's facts, each block only when it has content: steps and pace; goodput, the share of the trainer's span spent + training, its idle seconds split into gpu_wait, queued, rollout (a sample group in flight) and client (the rest); + requests, with the ledger and unanswered requests with their holder; failures and relaunches with a logs read; + saves; optimizer; samplers; config mismatches; placement; incidents; sdk_errors. A run_id may name a session with one + run. In the ledger, lost = enqueued - popped - queued, shown once the gap has lasted over a flush across reads.""" + state, found, now = await fleet.snapshot(), await facts.workers(), time.time() + chosen = [r for r in state["runs"] if r["run_id"] == run_id] or [r for r in state["runs"] if run_id in r["session_ids"]] + if len(chosen) != 1: + raise HTTPException(404 if not chosen else 409, f"No run {run_id}" if not chosen else f"Session {run_id} has {len(chosen)} runs") + run = chosen[0] + return {"run_id": run["run_id"], **await facts.summary(run, facts.workers_of(found, run), now)} + + +@router.get("/api/v1/inspect/sessions/{session_id}/summary") +async def session_summary(session_id: str): + """The session's record (created_at and redacted metadata, or recorded false when it has none), when the API server + last heard from it, whether its live key is present, its SDK errors, and a /runs line per model.""" + state, found, now = await fleet.snapshot(), await facts.workers(), time.time() + if (body := await facts.session(session_id, state["runs"], found, now)) is None: + raise HTTPException(404, f"No session {session_id}") + return body + + @router.get("/api/v1/inspect/runs/{run_id}/steps") async def run_steps( run_id: str, diff --git a/src/server/store.py b/src/server/store.py index cece00fd..12424f0a 100644 --- a/src/server/store.py +++ b/src/server/store.py @@ -43,6 +43,11 @@ async def get_result(self, req_id: str, timeout: float) -> dict[str, Any] | None """Block until the future resolves or the timeout is reached.""" pass + @abstractmethod + async def queued(self, model_id: str) -> int: + """How many of the model's requests wait in its queue.""" + pass + class InMemoryStore(RequestStore): def __init__(self): @@ -100,6 +105,9 @@ async def get_requests(self, active_set_id: str | None = None) -> list[dict[str, await self.active_tenants_cv.wait() + async def queued(self, model_id: str) -> int: + return queue.qsize() if (queue := self.queues.get(model_id)) else 0 + async def get_requests_for_model(self, model_id: str) -> list[dict[str, Any]]: raise RuntimeError("Per-model full fine-tuning workers require REDIS_URL; in-memory queues cannot be shared across processes") @@ -183,6 +191,9 @@ async def get_requests(self, active_set_id: str | None = None) -> list[dict[str, await self.redis.lmove(active_list, active_list, "LEFT", "RIGHT") return batch + async def queued(self, model_id: str) -> int: + return await self.redis.llen(f"open_rl:queue:{model_id}") + async def get_requests_for_model(self, model_id: str) -> list[dict[str, Any]]: queue_key = f"open_rl:queue:{model_id}" try: @@ -304,8 +315,13 @@ def write_sync(self, writes: Writes, ttl_seconds: float) -> None: pass @abstractmethod - async def read_samples(self, key: str) -> list[dict[str, Any]]: - """The JSON items of a list write_sync appends to, oldest first.""" + async def read_samples(self, key: str, last: int | None = None) -> list[dict[str, Any]]: + """The JSON items of a list write_sync appends to, oldest first, or only its last ones.""" + pass + + @abstractmethod + async def read_counts(self, key: str) -> dict[str, int]: + """The counts in a hash write_sync adds to.""" pass @@ -369,8 +385,11 @@ def write_sync(self, writes: Writes, ttl_seconds: float) -> None: self.kv_store[key] = value self.expiries[key] = time.monotonic() + ttl_seconds - async def read_samples(self, key: str) -> list[dict[str, Any]]: - return [json.loads(item) for item in self.lists.get(key, [])] + async def read_samples(self, key: str, last: int | None = None) -> list[dict[str, Any]]: + return [json.loads(item) for item in self.lists.get(key, [])[-(last or 0) :]] + + async def read_counts(self, key: str) -> dict[str, int]: + return dict(self.hashes.get(key, {})) class RedisStateStore(StateStore): @@ -425,11 +444,14 @@ def write_sync(self, writes: Writes, ttl_seconds: float) -> None: pipe.set(key, value, ex=ttl) pipe.execute() - async def read_samples(self, key: str) -> list[dict[str, Any]]: - raw = await self.redis.lrange(key, 0, -1) + async def read_samples(self, key: str, last: int | None = None) -> list[dict[str, Any]]: + raw = await self.redis.lrange(key, -(last or 0), -1) # Decoded off the event loop, since a list holds up to thousands of records. return await asyncio.to_thread(lambda: [json.loads(item) for item in raw]) + async def read_counts(self, key: str) -> dict[str, int]: + return {name: int(count) for name, count in (await self.redis.hgetall(key)).items()} + @cache def redis_client(redis_url: str) -> redis.Redis: diff --git a/src/server/telemetry.py b/src/server/telemetry.py index c5fdb422..ae8de109 100644 --- a/src/server/telemetry.py +++ b/src/server/telemetry.py @@ -11,6 +11,12 @@ open_rl:ledger: a hash counting the run's trainer requests enqueued, popped, answered and failed open_rl:worker: a worker process's phase, GPU turns and the waits for them, outstanding requests and settings env + +The API server also keeps each client session's record, its last heartbeat and the SDK's error events for it: + + open_rl:session_record: when the session was created, with its metadata, secrets redacted + open_rl:session_seen: when the API server last heard from it + open_rl:sdk_errors: the SDK's ERROR and CRITICAL events, the newest SDK_ERRORS """ import asyncio @@ -27,6 +33,7 @@ from collections.abc import Awaitable, Callable, Iterable from contextlib import asynccontextmanager, suppress from dataclasses import dataclass +from datetime import datetime from functools import wraps from numbers import Real @@ -37,6 +44,8 @@ SAMPLE_LIMIT = 2000 # records kept per list STEP_TIMES = 64 MAX_TURNS = 200 +SDK_ERRORS = 200 +MESSAGE_CHARS = 500 FLUSH_SECONDS = 5 FINISH_SECONDS = 5 # how long an exiting process waits for its last flush FAILED = "RequestFailedResponse" @@ -50,6 +59,8 @@ # ("sampled", run, runtime, started_at, at) one sample a client got # ("step", run, at) an optimizer step that ended at `at` # ("count", run, event) one more of the run's trainer requests reached event +# ("session", session, at, metadata) the API server heard from a session; metadata only when it was created +# ("sdk_error", session, record) an SDK error event # While flushes fail the oldest go. buffer: deque[tuple] = deque(maxlen=100_000) @@ -139,6 +150,13 @@ def plan(entries: Iterable[tuple], step_starts: dict[str, int], pod: dict[str, s append(writes, key("step_times", run), rest[0], STEP_TIMES) elif kind == "count": writes.fields.setdefault(key("ledger", run), Counter())[rest[0]] += 1 + elif kind == "session": + at, metadata = rest + writes.values[key("session_seen", run)] = str(at) + if metadata is not None: + writes.values[key("session_record", run)] = json.dumps({"session_id": run, "created_at": at, "metadata": metadata}) + elif kind == "sdk_error": + append(writes, key("sdk_errors", run), rest[0], SDK_ERRORS) for run, group in groups.items(): append(writes, key("operations", run, "samples"), group | {"elapsed_seconds": group["at"] - group["started_at"]} | pod, SAMPLE_LIMIT) return writes @@ -382,3 +400,41 @@ def sample_answered(run_id: str, pool: str, request_id: str, result: dict, start add("sample", run_id, op_record("sample", run_id, request_id, pool, recorder.role, started_at, FAILED)) else: add("sampled", run_id, pool, started_at, time.time()) + + +# Session records, kept by the API server + + +def redact(value): + if isinstance(value, dict): + return {k: "[redacted]" if SECRET.search(str(k)) else redact(v) for k, v in value.items()} + return [redact(v) for v in value] if isinstance(value, list) else value + + +@best_effort +def session_seen(session_id: str, metadata: dict | None = None) -> None: + """The API server heard from a session: its creation with metadata, a heartbeat, or a model attached.""" + add("session", session_id, time.time(), None if metadata is None else redact(metadata)) + + +@best_effort +def sdk_events(body: dict) -> None: + """Keep the ERROR and CRITICAL events of a batch the tinker SDK posts about its session. Other events are dropped.""" + session_id, events = body.get("session_id"), body.get("events") + if not isinstance(session_id, str) or not session_id or not isinstance(events, list): + return + for event in (e for e in events if isinstance(e, dict)): + if (severity := str(event.get("severity") or "").upper()) in ("ERROR", "CRITICAL"): + add("sdk_error", session_id, sdk_error(event, severity)) + + +def sdk_error(event: dict, severity: str) -> dict: + data = event.get("event_data") if isinstance(event.get("event_data"), dict) else {} + name = str(event.get("error_type") or event.get("event_name") or event.get("event") or "event").rsplit(".", 1)[-1] + status = f"HTTP {data['status_code']}: " if data.get("status_code") else "" + message = status + str(event.get("error_message") or data.get("detail") or data.get("exception") or data.get("message") or "") + try: + at = datetime.fromisoformat(str(event.get("timestamp"))).timestamp() + except ValueError: + at = time.time() + return {"at": at, "severity": severity, "reason": f"SDK {name}", "message": message[:MESSAGE_CHARS]} diff --git a/tests/test_summaries.py b/tests/test_summaries.py new file mode 100644 index 00000000..9a248cb3 --- /dev/null +++ b/tests/test_summaries.py @@ -0,0 +1,160 @@ +"""The run and session summaries, read through their routes from records in the shapes the recorder writes.""" + +import asyncio +import json +import time +import unittest +from unittest import mock + +from fastapi.testclient import TestClient + +from server import api_server, telemetry +from server.observability import facts, fleet, router +from server.session_registry import SessionRegistry +from server.store import InMemoryStateStore, InMemoryStore, Writes +from tests.test_observability import NO_CLUSTER, meta + +POD = "orw-fft-run-a-trainer" + + +class SummaryTest(unittest.TestCase): + def setUp(self) -> None: + self.store, self.queues, self.now = InMemoryStateStore(), InMemoryStore(), time.time() + fleet.cache.clear() + facts.gaps.clear() + telemetry.buffer.clear() + for module in (fleet, facts, router, telemetry): + self.enterContext(mock.patch.object(module, "get_state_store", return_value=self.store)) + self.enterContext(mock.patch.object(facts, "get_store", return_value=self.queues)) + self.enterContext(mock.patch.object(fleet, "kubernetes", mock.AsyncMock(side_effect=lambda: json.loads(json.dumps(NO_CLUSTER))))) + self.enterContext(mock.patch.object(router, "time", mock.Mock(time=lambda: self.now))) + self.recorder = self.enterContext(mock.patch.object(telemetry, "recorder", telemetry.Recorder("api_server", None))) + + def run_a(self, live: bool = True, sessions: tuple[str, ...] = ("sess-a",), **fields) -> None: + asyncio.run(self.store.set_value("open_rl:model_meta:run-a", meta(session_ids=list(sessions), created_at=self.now - 3600, **fields))) + if live: + asyncio.run(self.store.set_value(f"open_rl:session:{sessions[0]}", "{}")) + + def record(self, *entries: tuple) -> None: + self.store.write_sync(telemetry.plan(entries, {"run-a": 0}, {"pod": POD}, "trainer"), telemetry.TTL_SECONDS) + + def op(self, operation: str, ago: float, seconds: float = 1.0, error_type: str | None = None, **fields) -> tuple: + at = self.now - ago + record = telemetry.op_record(operation, "run-a", None, "run-a", "trainer", at - seconds, error_type) + return ("op", "run-a", record | {"at": at, "elapsed_seconds": seconds, "step": 0} | fields) + + def worker(self, **fields) -> None: + w = json.loads(telemetry.Worker("trainer", "run-a").snapshot()) | {"pod": POD} | fields + self.store.write_sync(Writes(values={telemetry.key("worker", w["id"]): json.dumps(w)}), telemetry.TTL_SECONDS) + + def get(self, path: str, status: int = 200, **params) -> dict: + response = TestClient(api_server.app).get(f"/api/v1/inspect/{path}", params=params) + self.assertEqual(response.status_code, status, response.text) + return response.json() + + def summary(self) -> dict: + return self.get("runs/run-a/summary") + + def line(self) -> dict: + return next(r for r in self.get("runs", ended="true")["runs"] if r["run_id"] == "run-a") + + def test_a_request_is_lost_only_once_the_gap_outlasts_a_flush(self) -> None: + self.run_a() + self.record(*[("count", "run-a", "enqueued")] * 98, *[("count", "run-a", "popped")] * 94) + asyncio.run(self.queues.put_request({"model_id": "run-a"})) + self.assertNotIn("lost", self.summary()["requests"]["ledger"]) + self.now += telemetry.FLUSH_SECONDS + 1 + self.assertEqual(self.summary()["requests"]["ledger"], {"enqueued": 98, "popped": 94, "lost": 3}) + + def test_requests_held_by_a_dead_worker_name_its_pod(self) -> None: + self.run_a() + self.worker(heartbeat_at=self.now - 120, outstanding=[{"op": "forward_backward", "run_id": "run-a", "since": self.now - 300, "count": 2}]) + held = {"op": "forward_backward", "count": 2, "age_s": 300, "pod": POD, "holder_alive": False} + self.assertEqual(self.summary()["requests"]["unanswered"], [held]) + self.assertEqual((self.line()["trainer_requests_unanswered"], self.line()["held_by_dead_worker"]), (2, 2)) + + def test_step_pace_and_ages_count_from_the_request_and_never_go_negative(self) -> None: + self.run_a() + for i in range(30): + self.record(self.op("optim_step", 600 - 20 * i), ("step", "run-a", self.now - 600 + 20 * i)) + self.record(self.op("forward", -3)) # recorded after the request's clock read + found, line = self.summary(), self.line() + pace = [found[k] for k in ("steps", "step_seconds", "since_last_step_s", "since_last_op_s")] + self.assertEqual(pace, [30, {"median": 20, "last": 20}, 20, 0]) + self.assertEqual((line["median_step_s"], line["since_last_step_s"], line["since_last_op_s"]), (20, 20, 0)) + self.now += 100 # the inventory snapshot stays cached; ages still move + self.assertEqual([self.line()[k] for k in ("since_last_step_s", "since_last_op_s")], [120, 97]) + + def test_an_ended_run_shows_no_ages(self) -> None: + self.run_a(live=False) + self.record(self.op("optim_step", 60), ("step", "run-a", self.now - 60), self.op("optim_step", 30), ("step", "run-a", self.now - 30)) + self.assertEqual(self.get("runs")["runs"], []) + for found in (self.summary(), self.line()): + self.assertEqual((found["lifecycle"], found["steps"]), ("ended", 2)) + self.assertFalse({"since_last_op_s", "since_last_step_s"} & found.keys()) + + def test_a_worker_running_other_settings_than_asked_is_named(self) -> None: + self.run_a(fine_tuning_type="full") + self.worker(config={"env": {"OPEN_RL_FINE_TUNING_TYPE": "lora"}}) + mismatch = {"pod": POD, "setting": "openrl.fine_tuning_type", "asked": "full", "running": "lora"} + self.assertEqual(self.summary()["config"], [mismatch]) + + def test_a_failure_points_at_a_logs_read_the_logs_route_takes(self) -> None: + self.run_a() + self.record(self.op("forward_backward", 30, seconds=2, error_type="RuntimeError"), ("count", "run-a", "failed")) + failures = self.summary()["failures"] + self.assertEqual((failures["count"], failures["ops"][0]["error_type"]), (1, "RuntimeError")) + window = {"since": facts.iso(self.now - 92), "until": facts.iso(self.now + 30)} + self.assertEqual(failures["logs"], {"read": "logs", "args": {"run_id": "run-a", "pod": POD, **window}}) + self.assertFalse(self.get(failures["logs"]["read"], **failures["logs"]["args"])["available"]) # no log store here + self.assertEqual(self.line()["failures"], 1) + + def test_a_nonfinite_metric_is_named(self) -> None: + self.run_a() + self.record(self.op("optim_step", 30, step=30, nonfinite=["loss:sum"])) + first = {"operation": "optim_step", "step": 30, "metrics": ["loss:sum"]} + self.assertEqual(self.summary()["optimizer"]["nonfinite"], {"count": 1, "first": first}) + + def test_sampler_sends_failures_and_median_group_time(self) -> None: + self.run_a() + for i in range(10): + self.record(*[("sampled", "run-a", "sampler-run-a", self.now - 620 + 60 * i, self.now - 600 + 61 * i)] * 64) + failed = telemetry.op_record("sample", "run-a", "r1", "sampler-run-a", "dispatcher", self.now - 5, telemetry.FAILED) + self.record(("sample", "run-a", failed)) + self.assertEqual(self.summary()["samplers"], {"sends": 641, "failed": 1, "median_s": 24.5}) + self.assertEqual(self.line()["sample_failures"], 1) + + def test_a_session_keeps_its_record_sdk_errors_and_models(self) -> None: + self.enterContext(mock.patch.object(api_server, "session_registry", SessionRegistry(self.store))) + client = TestClient(api_server.app) + session = client.post("/api/v1/create_session", json={"user_metadata": {"recipe_name": "s8", "api_key": "k"}}).json()["session_id"] + self.run_a(live=False, sessions=(session,)) + client.post("/api/v1/session_heartbeat", json={"session_id": session}) + events = [ + {"severity": "ERROR", "event_name": "APIStatusError", "event_data": {"status_code": 400, "detail": "bad"}, "timestamp": "2020-01-01T00:00Z"}, + {"severity": "CRITICAL", "error_type": "tinker.Fatal", "error_message": "x" * 900}, + {"severity": "INFO", "event_name": "session_start"}, + ] + client.post("/api/v1/telemetry", json={"session_id": session, "events": events}) + self.recorder.flush() + found = self.get(f"sessions/{session}/summary") + self.assertEqual((found["metadata"], found["live"], found["recorded"]), ({"recipe_name": "s8", "api_key": "[redacted]"}, True, True)) + self.assertEqual([(m["model_id"], m["sdk_errors"]) for m in found["models"]], [("run-a", 2)]) + self.assertEqual((found["sdk_errors"]["count"], found["sdk_errors"]["last"]["reason"]), (2, "SDK Fatal")) + self.assertEqual(len(found["sdk_errors"]["last"]["message"]), telemetry.MESSAGE_CHARS) + self.assertEqual(self.get(f"runs/{session}/summary")["run_id"], "run-a") + + def test_a_session_without_its_record_is_summarized_from_its_runs(self) -> None: + self.run_a(live=False, sessions=("sess-old",)) + found = self.get("sessions/sess-old/summary") + self.assertEqual((found["recorded"], found["live"], [m["model_id"] for m in found["models"]]), (False, False, ["run-a"])) + self.get("sessions/sess-none/summary", status=404) + + def test_idle_trainer_time_splits_into_gpu_wait_queued_rollout_and_client(self) -> None: + self.run_a(live=False) + t0 = self.now - 1000 + self.record(self.op("forward_backward", 1000 - 10, seconds=10), self.op("optim_step", 1000 - 110, seconds=10, queue_seconds=30)) + self.worker(waits=[[t0 + 60, t0 + 80]]) + self.record(("sampled", "run-a", "sampler-run-a", t0 + 20, t0 + 50)) + seconds = {"wall": 110, "train": 20, "checkpoint": 0, "forward": 0, "gpu_wait": 20, "queued": 20, "rollout": 30, "client": 20} + self.assertEqual(self.summary()["goodput"], {"goodput": 0.182, "seconds": seconds, "first_step_after_s": 110})