From bb05e64502ce45cd30e50f5f446920427c46a898 Mon Sep 17 00:00:00 2001 From: ShubyM Date: Mon, 26 Oct 2026 17:15:06 -0700 Subject: [PATCH] feat(dashboard): serve run inventory from Redis at /api/v1/inspect/inventory The API server serves a read-only list of runs at /api/v1/inspect/inventory, built from the model and session records in Redis. A run is active while the client has a live session and ended otherwise. A Redis failure or an unreadable record becomes an entry in errors. Kubernetes data comes in a separate change. --- Makefile | 2 +- docs/configuration.md | 10 ++++ src/server/api_server.py | 2 + src/server/dashboard/__init__.py | 0 src/server/dashboard/fleet.py | 80 ++++++++++++++++++++++++++++++++ src/server/dashboard/router.py | 13 ++++++ src/server/store.py | 12 +++++ tests/test_dashboard.py | 48 +++++++++++++++++++ tests/test_redis_store.py | 18 +++++++ 9 files changed, 184 insertions(+), 1 deletion(-) create mode 100644 src/server/dashboard/__init__.py create mode 100644 src/server/dashboard/fleet.py create mode 100644 src/server/dashboard/router.py create mode 100644 tests/test_dashboard.py diff --git a/Makefile b/Makefile index bd65de02d..e7bbd7e63 100644 --- a/Makefile +++ b/Makefile @@ -19,7 +19,7 @@ HOST ?= 127.0.0.1 PORT ?= 9003 # The fully qualified base URL used by local CLI tools and clients BASE_URL ?= http://$(HOST):$(PORT) -UNIT_TESTS ?= tests.test_session_lifecycle tests.test_proto_codec tests.test_forward_only tests.test_fft_batch_failure tests.test_api_server_paths tests.test_accel_timeslicer tests.test_trainer_optimizer_correctness tests.test_trainer_logprob_chunks tests.test_worker_manager tests.test_scheduler_worker_manager tests.test_estimator tests.test_redis_store tests.test_cluster_eval_script tests.test_perf_compare_script tests.test_delta_weight_sync tests.test_diffing_backends tests.test_sampler_weight_rotation tests.test_sampler_batch_limits tests.test_commands tests.test_weight_sync_config tests.test_exclusive_trainer tests.test_delete_model tests.test_automodel_worker tests.test_worker_thread_device tests.test_trainer_data_parallel tests.test_openrl_status tests.test_skill_docs tests.test_trainer_ready tests.test_sampling_store tests.test_lora_snapshots tests.test_pool_accept tests.test_sampling_dispatcher tests.test_vllm_sampler tests.test_vllm_options +UNIT_TESTS ?= tests.test_session_lifecycle tests.test_proto_codec tests.test_forward_only tests.test_fft_batch_failure tests.test_api_server_paths tests.test_accel_timeslicer tests.test_trainer_optimizer_correctness tests.test_trainer_logprob_chunks tests.test_worker_manager tests.test_scheduler_worker_manager tests.test_estimator tests.test_redis_store tests.test_cluster_eval_script tests.test_perf_compare_script tests.test_delta_weight_sync tests.test_diffing_backends tests.test_sampler_weight_rotation tests.test_sampler_batch_limits tests.test_commands tests.test_weight_sync_config tests.test_exclusive_trainer tests.test_delete_model tests.test_automodel_worker tests.test_worker_thread_device tests.test_trainer_data_parallel tests.test_openrl_status tests.test_skill_docs tests.test_trainer_ready tests.test_sampling_store tests.test_lora_snapshots tests.test_pool_accept tests.test_sampling_dispatcher tests.test_vllm_sampler tests.test_vllm_options tests.test_dashboard # 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),) diff --git a/docs/configuration.md b/docs/configuration.md index 1654ab6da..766bd964a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -122,3 +122,13 @@ for. On one machine every pool has one replica, in process. See | `OPEN_RL_DISPATCH_CONCURRENCY` | `256` | Requests one pool keeps in flight. | | `OPEN_RL_SAMPLE_DEADLINE_SECONDS` | `3600` | How long a sampling request waits for a result before it fails. Set it on the API server. Worker pods and pool dispatchers get the same value. | | `OPEN_RL_DISPATCHER_IMAGE` | the worker image | Image for each pool's dispatcher container. The release sets it to the API-server image, which carries the dispatcher without the GPU stack and pulls in seconds. | + +## Inspection API + +The API server serves a read-only inventory of runs at +`/api/v1/inspect/inventory`. It reads runs and their live sessions from the +same Redis as the rest of the server. + +| Env var | Default | What it does | +| --- | --- | --- | +| `REDIS_URL` | unset | The Redis that holds model records and live sessions. Unset, the inventory has no runs. | diff --git a/src/server/api_server.py b/src/server/api_server.py index 139aff4cd..0024cb89e 100644 --- a/src/server/api_server.py +++ b/src/server/api_server.py @@ -27,6 +27,7 @@ from starlette.exceptions import HTTPException as StarletteHTTPException from server import openrl_status, proto_codec +from server.dashboard.router import router from server.lora_snapshots import snapshot_name, snapshot_root from server.model_metadata import ( TrainingModelMetadata, @@ -809,6 +810,7 @@ def openapi_with_settings() -> dict[str, Any]: app.openapi = openapi_with_settings FastAPIInstrumentor.instrument_app(app, excluded_urls="/api/v1/retrieve_future,/api/v1/session_heartbeat") +app.include_router(router) @app.exception_handler(RequestValidationError) diff --git a/src/server/dashboard/__init__.py b/src/server/dashboard/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/server/dashboard/fleet.py b/src/server/dashboard/fleet.py new file mode 100644 index 000000000..8c5a37f3f --- /dev/null +++ b/src/server/dashboard/fleet.py @@ -0,0 +1,80 @@ +import asyncio +import json +import time +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Literal + +from server import openrl_status +from server.model_metadata import decode_model_metadata +from server.store import get_state_store + +CACHE_SECONDS, STALE_SECONDS, READ_TIMEOUT_SECONDS = 5, 60, 6 +Lifecycle = Literal["ended", "active"] + + +@dataclass +class Entry: + at: float = 0.0 + task: asyncio.Future | None = None + last_good: asyncio.Future | None = None + last_good_at: float = 0.0 + + +cache: dict[str, Entry] = {} + + +def succeeded(task: asyncio.Future | None) -> bool: + return task is not None and task.done() and not task.cancelled() and task.exception() is None + + +async def cached(name: str, seconds: float, read) -> object: + entry = cache.setdefault(name, Entry()) + task = entry.task + if succeeded(task): + entry.last_good, entry.last_good_at = task, entry.at + if task is None or (task.done() and (not succeeded(task) or time.monotonic() - entry.at > seconds)): + entry.at, entry.task = time.monotonic(), asyncio.ensure_future(read()) + entry.task.add_done_callback(lambda t: t.cancelled() or t.exception()) # retrieved, so asyncio does not log it + if succeeded(entry.task): + return entry.task.result() + if entry.last_good is not None and time.monotonic() - entry.last_good_at <= STALE_SECONDS: + return entry.last_good.result() + return await asyncio.shield(entry.task) + + +async def snapshot() -> dict: + return await cached("inventory", CACHE_SECONDS, read) + + +def lifecycle(status: str, live: bool) -> Lifecycle: + return "active" if status == "active" and live else "ended" + + +def join(run_id: str, raw: str, live: set[str]) -> dict: + row, meta = json.loads(raw), decode_model_metadata(raw) + name = row.get("name") or row.get("run_name") + run = {"run_id": run_id, "name": name or f"{(meta.base_model or 'Run').split('/')[-1]} ยท {run_id[:8]}", "display_name": name} + 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)) + run.update(created_at=meta.created_at, completed_at=meta.completed_at, session_ids=meta.session_ids, settings=openrl_status.settings_of(meta)) + return run | {"shared_runtime": meta.runtime(run_id) != run_id} + + +async def read() -> dict: + store, observed_at = get_state_store(), datetime.now(UTC).isoformat(timespec="seconds") + scans = [asyncio.wait_for(store.scan_values(prefix), READ_TIMEOUT_SECONDS) for prefix in ("open_rl:model_meta:", "open_rl:session:")] + metas, sessions = await asyncio.gather(*scans, return_exceptions=True) + errors, runs = [], [] + if failed := next((r for r in (metas, sessions) if isinstance(r, Exception)), None): + errors.append(f"Redis read failed: {f'no reply in {READ_TIMEOUT_SECONDS} s' if isinstance(failed, TimeoutError) else failed}") + metas, sessions = {}, {} + live = {name.removeprefix("open_rl:session:") for name in sessions} + for name, raw in metas.items(): + run_id = name.removeprefix("open_rl:model_meta:") + try: + runs.append(join(run_id, raw, live)) + except Exception as exc: + errors.append(f"model record {run_id} unreadable: {exc}") + runs.sort(key=lambda r: -r["created_at"]) + return {"observed_at": observed_at, "errors": errors, "runs": runs} diff --git a/src/server/dashboard/router.py b/src/server/dashboard/router.py new file mode 100644 index 000000000..35fdc7397 --- /dev/null +++ b/src/server/dashboard/router.py @@ -0,0 +1,13 @@ +from fastapi import APIRouter + +from server.dashboard import fleet + +router = APIRouter(tags=["inspection"]) + + +@router.get("/api/v1/inspect/inventory") +async def inventory(): + """Runs with their lifecycle (ended, active) and an error per unavailable source or unreadable run record. client_status is + what the client last reported, active until it deletes the model, so a crashed run stays active there; lifecycle is ended + once none of the run's sessions is live.""" + return await fleet.snapshot() diff --git a/src/server/store.py b/src/server/store.py index 7f610a972..a37b8ed86 100644 --- a/src/server/store.py +++ b/src/server/store.py @@ -269,6 +269,11 @@ async def remove_from_set(self, key: str, member: str) -> None: async def set_members(self, key: str) -> set[str]: pass + @abstractmethod + async def scan_values(self, prefix: str) -> dict[str, str]: + """Every live value whose key starts with prefix.""" + pass + class InMemoryStateStore(StateStore): def __init__(self): @@ -306,6 +311,9 @@ async def remove_from_set(self, key: str, member: str) -> None: async def set_members(self, key: str) -> set[str]: return set(self.sets.get(key, set())) + async def scan_values(self, prefix: str) -> dict[str, str]: + return {key: value for key in list(self.kv_store) if key.startswith(prefix) and (value := self.get_value_sync(key)) is not None} + class RedisStateStore(StateStore): def __init__(self, client: redis.Redis, sync_client: sync_redis.Redis): @@ -334,6 +342,10 @@ async def remove_from_set(self, key: str, member: str) -> None: async def set_members(self, key: str) -> set[str]: return set(await self.redis.smembers(key)) + async def scan_values(self, prefix: str) -> dict[str, str]: + keys = [key async for key in self.redis.scan_iter(match=prefix + "*", count=1000)] + return {key: value for key, value in zip(keys, await self.redis.mget(keys)) if value is not None} if keys else {} + @cache def redis_client(redis_url: str) -> redis.Redis: diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py new file mode 100644 index 000000000..03b4ab655 --- /dev/null +++ b/tests/test_dashboard.py @@ -0,0 +1,48 @@ +import asyncio +import unittest +from unittest import mock + +from fastapi.testclient import TestClient + +from server import api_server +from server.dashboard import fleet +from server.model_metadata import TrainingModelMetadata +from server.store import InMemoryStateStore + + +def meta(**fields) -> str: + return TrainingModelMetadata(base_model="Qwen/Qwen3-8B", **fields).model_dump_json() + + +class InventoryTest(unittest.TestCase): + def inventory(self, **records: str) -> dict: + store = InMemoryStateStore() + for key, value in records.items(): + asyncio.run(store.set_value(f"open_rl:{key}", value)) + fleet.cache.clear() + self.enterContext(mock.patch.object(fleet, "get_state_store", return_value=store)) + return TestClient(api_server.app).get("/api/v1/inspect/inventory").json() + + def test_a_run_is_active_only_while_the_client_has_a_live_session(self): + body = self.inventory( + **{ + "model_meta:live": meta(session_ids=["s1"], created_at=3), + "model_meta:idle": meta(session_ids=["s2"], created_at=2), + "model_meta:done": meta(status="completed", session_ids=["s1"], created_at=1), + "session:s1": "{}", + } + ) + + runs = [(r["run_id"], r["lifecycle"], r["client_status"]) for r in body["runs"]] + self.assertEqual(runs, [("live", "active", "active"), ("idle", "ended", "active"), ("done", "ended", "completed")]) + self.assertEqual(body["errors"], []) + + def test_a_malformed_model_record_is_skipped_with_an_error(self): + body = self.inventory(**{"model_meta:good": meta(), "model_meta:bad": "not json"}) + + self.assertEqual([r["run_id"] for r in body["runs"]], ["good"]) + self.assertTrue(any("bad" in error for error in body["errors"]), body["errors"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_redis_store.py b/tests/test_redis_store.py index 003024eaa..7c910cdfe 100644 --- a/tests/test_redis_store.py +++ b/tests/test_redis_store.py @@ -151,6 +151,15 @@ async def test_values_expire_and_sets_hold_members(self) -> None: self.assertEqual(await self.state.set_members("s"), {"b"}) self.assertEqual(await self.state.set_members("missing"), set()) + async def test_scan_values_returns_live_values_under_a_prefix(self) -> None: + await self.state.set_value("open_rl:model_meta:a", "1") + await self.state.set_value("open_rl:model_meta:b", "2") + await self.state.set_value("open_rl:model_meta:gone", "3", ttl_seconds=0.2) + await self.state.set_value("open_rl:session:a", "4") + await asyncio.sleep(0.5) + self.assertEqual(await self.state.scan_values("open_rl:model_meta:"), {"open_rl:model_meta:a": "1", "open_rl:model_meta:b": "2"}) + self.assertEqual(await self.state.scan_values("open_rl:missing:"), {}) + class InMemoryStoreTest(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: @@ -203,6 +212,15 @@ async def test_values_expire_and_sets_hold_members(self) -> None: self.assertEqual(await self.state.set_members("s"), {"b"}) self.assertEqual(await self.state.set_members("missing"), set()) + async def test_scan_values_returns_live_values_under_a_prefix(self) -> None: + await self.state.set_value("open_rl:model_meta:a", "1") + await self.state.set_value("open_rl:model_meta:b", "2") + await self.state.set_value("open_rl:model_meta:gone", "3", ttl_seconds=0.2) + await self.state.set_value("open_rl:session:a", "4") + await asyncio.sleep(0.5) + self.assertEqual(await self.state.scan_values("open_rl:model_meta:"), {"open_rl:model_meta:a": "1", "open_rl:model_meta:b": "2"}) + self.assertEqual(await self.state.scan_values("open_rl:missing:"), {}) + if __name__ == "__main__": unittest.main()