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})