diff --git a/Makefile b/Makefile index 6ab6e755..f8ff69b0 100644 --- a/Makefile +++ b/Makefile @@ -19,11 +19,11 @@ 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_prefetch tests.test_pool_accept tests.test_sampling_dispatcher tests.test_vllm_sampler tests.test_vllm_options tests.test_dashboard +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_prefetch tests.test_pool_accept tests.test_sampling_dispatcher tests.test_vllm_sampler tests.test_vllm_options tests.test_observability # 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_dashboard_kubernetes +UNIT_TESTS += tests.test_observability_kubernetes 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 b4ccf07e..4abed401 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -126,12 +126,19 @@ for. On one machine every pool has one replica, in process. See ## 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. +`/api/v1/inspect/inventory` and a run's logs at +`/api/v1/inspect/logs?run_id=`. See `/docs` for their fields. | Env var | Default | What it does | | --- | --- | --- | -| `REDIS_URL` | unset | The Redis that holds model records and live sessions. Unset, the inventory has no runs. | +| `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/`. | + +To read a run's client logs with `pod=client`, set the session tag +`openrl.client_log_id=`. Ship the client's logs to the same store with the +label `openrl_client=`. For a client pod, that is a pod label. + +A bucket view only shows client logs when that bucket's sink routes them in, so +plain `cloud-logging` is the simplest choice when clients run outside the cluster. ### Metrics and logs on GKE diff --git a/pyproject.toml b/pyproject.toml index 804694c5..644a9a09 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -65,6 +65,7 @@ vllm = [ ] # Kubernetes cluster mode: the k8s API client used to launch FFT worker pods. cluster = [ + "google-auth", "kubernetes>=29", # Kept in step with the snapshot-agent image deployed in # docs/setup/gke-fft-timeslice.md: both come from this upstream release. diff --git a/src/server/api_server.py b/src/server/api_server.py index 0024cb89..a677e856 100644 --- a/src/server/api_server.py +++ b/src/server/api_server.py @@ -27,7 +27,6 @@ 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, @@ -36,6 +35,7 @@ persist_model_metadata, pool_name, ) +from server.observability.router import router from server.sampling_dispatcher import SAMPLE_DEADLINE_SECONDS from server.sampling_store import SamplingStore from server.session_registry import SessionRegistry @@ -228,6 +228,11 @@ class Settings(BaseModel): description="What the job runs. rl starts the sampler together with the trainer, so its start-up overlaps the trainer's. " "sft or unset starts it at the first sample or save for sampling.", ) + client_log_id: str | None = Field( + default=None, + description="The id on the client's log lines, as the label openrl_client= in the server's log store. " + "The inspection API reads the run's client logs by it.", + ) # The rest apply to full fine-tuning only. weight_sync: Literal["delta", "full"] | None = Field( default=None, @@ -528,6 +533,7 @@ async def _extract_and_persist_model_metadata( trainer_gpus=settings.trainer_gpus, trainer_cp=settings.trainer_cp, recipe=settings.recipe, + client_log_id=settings.client_log_id, ) if meta_obj.sampler_replicas > 1 and meta_obj.shares_gpu(): raise ValueError("openrl.sampler_replicas above 1 needs samplers on GPUs of their own; set openrl.exclusive=true") diff --git a/src/server/dashboard/router.py b/src/server/dashboard/router.py deleted file mode 100644 index 35fdc739..00000000 --- a/src/server/dashboard/router.py +++ /dev/null @@ -1,13 +0,0 @@ -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/model_metadata.py b/src/server/model_metadata.py index 8ec44e56..7e7961d6 100644 --- a/src/server/model_metadata.py +++ b/src/server/model_metadata.py @@ -49,6 +49,7 @@ class TrainingModelMetadata(BaseModel): trainer_gpus: int = 1 trainer_cp: int = 1 recipe: Literal["rl", "sft"] | None = None + client_log_id: str | None = None status: str = "active" updated_at: float = 0.0 completed_at: float | None = None diff --git a/src/server/dashboard/__init__.py b/src/server/observability/__init__.py similarity index 100% rename from src/server/dashboard/__init__.py rename to src/server/observability/__init__.py diff --git a/src/server/observability/cloud_logging.py b/src/server/observability/cloud_logging.py new file mode 100644 index 00000000..f339b2ca --- /dev/null +++ b/src/server/observability/cloud_logging.py @@ -0,0 +1,131 @@ +"""A run's logs from Cloud Logging, read with the server's Google credentials.""" + +import asyncio +import json +import re +import threading +from datetime import UTC, datetime, timedelta + +import httpx + +from server.observability.log_selector import ClientLines, Selector + +# GKE files every stderr line as ERROR, though most are progress bars, INFO logs and warnings; such a line takes the level it prints. +RELABELED = ("DEBUG", "INFO", "WARNING") +LEVEL, PROGRESS = re.compile(r"\b(DEBUG|INFO|WARNING|WARN|ERROR|CRITICAL)\b"), re.compile(r"\r|\d+%(\||\s+Completed)") +MESSAGE_CHARS = 16384 +client: httpx.AsyncClient | None = None +credentials = None +project: str | None = None # the one the credentials name +credentials_lock = threading.Lock() + + +async def send(method: str, url: str, **kwargs) -> httpx.Response: + global client + client = client or httpx.AsyncClient(timeout=10) + return (await client.request(method, url, **kwargs)).raise_for_status() + + +def access_token() -> str: + import google.auth # the cluster extra installs it + from google.auth.transport.requests import Request + + global credentials, project + with credentials_lock: + if credentials is None: + credentials, project = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform.read-only"]) + if not credentials.valid: + credentials.refresh(Request()) + return credentials.token + + +async def gcp_headers() -> dict[str, str]: + import google.oauth2.credentials + + # A hung token refresh holds the lock; bound it rather than start a thread per call. + headers = {"Authorization": f"Bearer {await asyncio.wait_for(asyncio.to_thread(access_token), timeout=10)}"} + # A person's gcloud login draws on gcloud's shared quota and fails with 429 unless it bills its project. + if isinstance(credentials, google.oauth2.credentials.Credentials) and not credentials.quota_project_id and project: + headers["x-goog-user-project"] = project + return headers + + +def window(since: str | None, until: str | None) -> tuple[str, str]: + """ISO-8601 or Unix seconds; since also takes a span before until, such as 1h. Default: the 30 minutes ending now; at most seven days.""" + + def parse(value: str) -> datetime: + parsed = datetime.fromtimestamp(float(value), UTC) if re.fullmatch(r"\d+(\.\d+)?", value) else datetime.fromisoformat(value) + return parsed.astimezone(UTC) if parsed.tzinfo else parsed.replace(tzinfo=UTC) + + end = parse(until) if until else datetime.now(UTC) + if ago := re.fullmatch(r"(\d+)([mhd])", since or ""): + start = end - timedelta(**{{"m": "minutes", "h": "hours", "d": "days"}[ago[2]]: int(ago[1])}) + else: + start = parse(since) if since else end - timedelta(minutes=30) + if not timedelta(0) < end - start <= timedelta(days=7): + raise ValueError("Time range must be positive and no longer than seven days") + return start.isoformat(timespec="microseconds"), end.isoformat(timespec="microseconds") + + +def logs_filter(selector: Selector, project: str, start: str, end: str, severity: str | None, q: str) -> str: + """The selected entries in the window. Worker pods are the namespace's pods whose names start with the run's prefix, so + a run id cannot open another run's pods. GKE copies a client pod's openrl_client label to k8s-pod/openrl_client; a + client outside Kubernetes sets openrl_client itself.""" + if isinstance(selector, ClientLines): + client_id = json.dumps(selector.client_id) + clauses = [f'(labels."k8s-pod/openrl_client"={client_id} OR labels."openrl_client"={client_id})'] + else: + clauses = ['resource.type="k8s_container"', f"resource.labels.namespace_name={json.dumps(selector.namespace)}"] + clauses.append(f"resource.labels.pod_name=~{json.dumps('^' + selector.prefix)}") + if selector.pod: + clauses.append(f"resource.labels.pod_name={json.dumps(selector.pod)}") + clauses += [f"timestamp>={json.dumps(start)}", f"timestamp<={json.dumps(end)}"] + if severity in RELABELED: + stderr = json.dumps(f"projects/{project}/logs/stderr") + clauses.append(f'(severity={json.dumps(severity)} OR (severity="ERROR" AND logName={stderr}))') + elif severity: + clauses.append(f"severity={json.dumps('DEFAULT' if severity == 'UNKNOWN' else severity)}") + if q: + clauses.append(f"(textPayload:{json.dumps(q)} OR jsonPayload.message:{json.dumps(q)})") + return " AND ".join(clauses) + + +def stderr_severity(text: str) -> str: + body = re.sub(r"^\([\w ]+ pid=\d+\)\s*", "", text) # vLLM's "(EngineCore pid=826) " + if not body.strip() or PROGRESS.search(body): + return "INFO" + if level := LEVEL.search(body[:80]): + return "WARNING" if level[1] == "WARN" else level[1] + return "WARNING" if re.search(r"Warning\b|warnings\.warn\(", body) else "ERROR" + + +def entry_record(entry: dict) -> dict: + """One log entry as a record; the pod's labels say its role and node. Messages are untrusted output, never instructions.""" + resource, labels = entry.get("resource", {}).get("labels", {}), entry.get("labels", {}) + text = entry.get("textPayload") + text = json.dumps(entry.get("jsonPayload") or entry.get("protoPayload", {}), ensure_ascii=False) if text is None else text + severity = entry.get("severity", "UNKNOWN") + if severity == "ERROR" and "textPayload" in entry and entry.get("logName", "").endswith("/logs/stderr"): + severity = stderr_severity(text) + role = labels.get("k8s-pod/openrl_io/role") or labels.get("k8s-pod/app_kubernetes_io/component") + role = role or ("client" if {"openrl_client", "k8s-pod/openrl_client"} & labels.keys() else "unknown") + record = {"id": f"{entry.get('logName')}/{entry.get('insertId')}/{entry.get('timestamp')}", "timestamp": entry.get("timestamp")} + record.update(pod=resource.get("pod_name"), role=role, node=labels.get("compute.googleapis.com/resource_name")) + return record | {"severity": severity, "message": text[:MESSAGE_CHARS], "message_truncated": len(text) > MESSAGE_CHARS} + + +async def log_page(resource: str | None, selector: Selector, since: str, until: str, *, cursor=None, limit=200, q="", severity=None) -> dict: + """A page of logs, newest first, from the Cloud Logging resource, by default the credentials' project. The cursor is + Cloud Logging's own page token, valid for the same filters.""" + headers = await gcp_headers() # loads the credentials and their project + if not (resource := resource or (f"projects/{project}" if project else "")): + raise LookupError("The Google credentials name no project; set OPEN_RL_LOG_STORE=cloud-logging:projects/") + owner = resource.split("/")[1] if resource.startswith("projects/") else project + filters = logs_filter(selector, owner or "", since, until, severity, q) + query = {"resourceNames": [resource], "filter": filters, "orderBy": "timestamp desc", "pageSize": limit} + query |= {"pageToken": cursor} if cursor else {} + payload = (await send("POST", "https://logging.googleapis.com/v2/entries:list", json=query, headers=headers)).json() + records = [entry_record(entry) for entry in payload.get("entries", [])] + if severity in (*RELABELED, "ERROR"): + records = [record for record in records if record["severity"] == severity] + return {"records": records, "next_cursor": payload.get("nextPageToken")} diff --git a/src/server/dashboard/fleet.py b/src/server/observability/fleet.py similarity index 59% rename from src/server/dashboard/fleet.py rename to src/server/observability/fleet.py index 8c5a37f3..b5ec3975 100644 --- a/src/server/dashboard/fleet.py +++ b/src/server/observability/fleet.py @@ -7,10 +7,12 @@ from server import openrl_status from server.model_metadata import decode_model_metadata +from server.observability.kubernetes import READ_TIMEOUT_SECONDS, Cluster, kubernetes from server.store import get_state_store +from server.worker_manager import FAILED, STOPPED, WAITING_FOR_PLACEMENT, owner_id -CACHE_SECONDS, STALE_SECONDS, READ_TIMEOUT_SECONDS = 5, 60, 6 -Lifecycle = Literal["ended", "active"] +CACHE_SECONDS, STALE_SECONDS = 5, 60 +Lifecycle = Literal["ended", "failed", "waiting", "placed", "active"] @dataclass @@ -47,25 +49,42 @@ 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 owner(runtime: str) -> str: + """The owner label on the runtime's pods; the scheduler trims a truncated label value as well.""" + return owner_id(runtime).rstrip("-") -def join(run_id: str, raw: str, live: set[str]) -> dict: +def lifecycle(status: str, live: bool, states: set[str] | None) -> Lifecycle: + if status != "active" or (states and states <= {STOPPED}) or not (states or live): + return "ended" + if states and FAILED in states: + return "failed" + return "active" if states is None else "waiting" if not states or WAITING_FOR_PLACEMENT in states else "placed" + + +def join(run_id: str, raw: str, live: set[str], cluster: Cluster) -> dict: row, meta = json.loads(raw), decode_model_metadata(raw) + runtime, kind = meta.runtime(run_id), "lora" if meta.fine_tuning_type == "lora" else "fft" + shared = runtime != run_id + claims_runtime = meta.status == "active" or not shared + pods = [p for p in cluster["pods"] if claims_runtime and p["owner"] == owner(runtime)] + states = None + if cluster["workloads"] is not None: + states = {w["state"] for w in cluster["workloads"] if claims_runtime and w["model_id"] == runtime and w["training_kind"] == kind} 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["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)) - return run | {"shared_runtime": meta.runtime(run_id) != run_id} + run.update(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]} 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 = [], [] + cluster, (metas, sessions) = await asyncio.gather(kubernetes(), asyncio.gather(*scans, return_exceptions=True)) + errors, runs = cluster["errors"], [] 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 = {}, {} @@ -73,8 +92,10 @@ async def read() -> dict: for name, raw in metas.items(): run_id = name.removeprefix("open_rl:model_meta:") try: - runs.append(join(run_id, raw, live)) + runs.append(join(run_id, raw, live, cluster)) 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} + pods = [{"name": p["name"], "problem": p["problem"], "restarts": p["restarts"]} for p in cluster["pods"] if p["problem"]] + view = {"available": cluster["available"], "nodes": cluster["nodes"], "pods": pods} + return {"observed_at": observed_at, "errors": errors, "runs": runs, "cluster": view} diff --git a/src/server/dashboard/kubernetes.py b/src/server/observability/kubernetes.py similarity index 100% rename from src/server/dashboard/kubernetes.py rename to src/server/observability/kubernetes.py diff --git a/src/server/observability/log_selector.py b/src/server/observability/log_selector.py new file mode 100644 index 00000000..5be3dd46 --- /dev/null +++ b/src/server/observability/log_selector.py @@ -0,0 +1,22 @@ +"""What a log store reads for a run: its worker pods, or its client's lines.""" + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class WorkerPods: + """The namespace's pods whose names start with prefix, or the one of them named pod.""" + + namespace: str + prefix: str + pod: str | None = None + + +@dataclass(frozen=True) +class ClientLines: + """The lines labelled openrl_client=, from wherever the client runs.""" + + client_id: str + + +Selector = WorkerPods | ClientLines diff --git a/src/server/observability/router.py b/src/server/observability/router.py new file mode 100644 index 00000000..0fa20934 --- /dev/null +++ b/src/server/observability/router.py @@ -0,0 +1,73 @@ +import os +from typing import Annotated + +import httpx +from fastapi import APIRouter, HTTPException, Query + +from server.model_metadata import decode_model_metadata +from server.observability import cloud_logging, fleet +from server.observability.kubernetes import namespace +from server.observability.log_selector import ClientLines, WorkerPods +from server.store import get_state_store +from server.worker_manager import owner_id + +router = APIRouter(tags=["inspection"]) +BOUNDS = "ISO-8601 or Unix seconds; since also takes a span before until, e.g. 1h. Default: the 30 minutes ending now; at most 7 days." +Bound = Annotated[str | None, Query(max_length=64, description=BOUNDS)] +CURSOR_EXPIRED = "Invalid or expired log cursor; repeat the query without it" +# OPEN_RL_LOG_STORE is or :, the resource in the store's own terms. A store reads what a selector names. +LOG_STORES = {"cloud-logging": cloud_logging.log_page} + + +@router.get("/api/v1/inspect/inventory") +async def inventory(): + """Runs with their lifecycle (ended, failed, waiting, placed, active), pods and nodes; the cluster's nodes with their GPUs + and its pods with problems; 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 what the + cluster shows.""" + return await fleet.snapshot() + + +@router.get("/api/v1/inspect/logs") +async def logs( + run_id: str, + pod: Annotated[str | None, Query(description="One of the run's pods, or 'client' for the lines labelled openrl_client=")] = None, + q: str = Query("", max_length=1024), + severity: str | None = Query(None, pattern="^(DEBUG|INFO|WARNING|ERROR|CRITICAL|UNKNOWN)$"), + since: Bound = None, + until: Bound = None, + limit: int = Query(200, ge=1, le=1000), + cursor: Annotated[str | None, Query(max_length=16384, description="next_cursor of a page with the same filters and window")] = None, +): + """Newest first, from the OPEN_RL_LOG_STORE log store: the run's worker pods, or with pod=client its client's lines.""" + try: + since, until = cloud_logging.window(since, until) + except (ValueError, OverflowError, OSError) as exc: + raise HTTPException(400, "Use ISO-8601 or Unix-second since/until and a positive window of at most seven days") from exc + if (raw := await get_state_store().get_value(f"open_rl:model_meta:{run_id}")) is None: + raise HTTPException(404, f"No run {run_id}") + meta = decode_model_metadata(raw) + store = os.getenv("OPEN_RL_LOG_STORE", "").strip() + kind, _, resource = store.partition(":") + result = {"run_id": run_id, "source": store, "since": since, "until": until, "records": [], "next_cursor": None} + if not store: + return {**result, "available": False, "error": "No log store configured. Set OPEN_RL_LOG_STORE; see docs/configuration.md."} + if (page_of := LOG_STORES.get(kind)) is None: + return {**result, "available": False, "error": f"Unknown OPEN_RL_LOG_STORE {store!r}; supported: {', '.join(LOG_STORES)}"} + if pod == "client": + if not meta.client_log_id: + return {**result, "available": False, "error": "This run set no openrl.client_log_id; see docs/configuration.md."} + selector = ClientLines(meta.client_log_id) + else: + # The scheduler names a worker pod orw-, and scheduler_worker_manager.workload_name starts it --. + prefix = f"orw-{'lora' if meta.fine_tuning_type == 'lora' else 'fft'}-{owner_id(meta.runtime(run_id))}-" + selector = WorkerPods(namespace(), prefix, pod) + try: + page = await page_of(resource or None, selector, since, until, cursor=cursor, limit=limit, q=q, severity=severity) + return {**result, **page, "available": True} + except httpx.HTTPStatusError as exc: + if exc.response.status_code == 400 and cursor: + raise HTTPException(400, CURSOR_EXPIRED) from None + return {**result, "available": False, "error": f"{store} refused the read (HTTP {exc.response.status_code}); check credentials and IAM"} + except Exception as exc: + return {**result, "available": False, "error": str(exc) if isinstance(exc, LookupError) else f"{store} unavailable"} diff --git a/tests/test_api_server_paths.py b/tests/test_api_server_paths.py index 06283b11..908a9eb9 100644 --- a/tests/test_api_server_paths.py +++ b/tests/test_api_server_paths.py @@ -370,6 +370,11 @@ def test_tags_choose_full_fine_tuning_and_its_weight_sync(self) -> None: self.assertEqual(meta["weight_sync_config"]["strategy"], "full") self.assertEqual(meta["full_config"], {"seed": None, "cpu_offload": False, "weight_sync_strategy": "full"}) + def test_the_client_log_id_tag_lands_on_the_model(self) -> None: + session_id = self.post("create_session", {"tags": ["openrl.client_log_id=gsm8k-7"]}).json()["session_id"] + meta = self.metadata(self.post("create_model", {"base_model": "m", "session_id": session_id}).json()["request_id"]) + self.assertEqual(meta["client_log_id"], "gsm8k-7") + def test_lora_is_the_default(self) -> None: meta = self.metadata(self.post("create_model", {"base_model": "m"}).json()["request_id"]) self.assertEqual(meta["fine_tuning_type"], "lora") diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py deleted file mode 100644 index 03b4ab65..00000000 --- a/tests/test_dashboard.py +++ /dev/null @@ -1,48 +0,0 @@ -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_observability.py b/tests/test_observability.py new file mode 100644 index 00000000..93130e38 --- /dev/null +++ b/tests/test_observability.py @@ -0,0 +1,170 @@ +import asyncio +import copy +import os +import unittest +from unittest import mock + +import httpx +from fastapi.testclient import TestClient + +from server import api_server +from server.model_metadata import TrainingModelMetadata +from server.observability import cloud_logging, fleet +from server.store import InMemoryStateStore +from server.worker_manager import FAILED, READY, STARTING, STOPPED, WAITING_FOR_PLACEMENT + +NO_CLUSTER = {"available": True, "pods": [], "nodes": [], "workloads": None, "errors": []} + + +def meta(**fields) -> str: + return TrainingModelMetadata(base_model="Qwen/Qwen3-8B", **fields).model_dump_json() + + +def cluster(*states: str) -> dict: + return {"pods": [], "workloads": [{"model_id": "r1", "training_kind": "fft", "state": s} for s in states]} + + +class ApiTest(unittest.TestCase): + def setUp(self): + self.store = InMemoryStateStore() + fleet.cache.clear() + self.enterContext(mock.patch.object(fleet, "get_state_store", return_value=self.store)) + self.enterContext(mock.patch("server.observability.router.get_state_store", return_value=self.store)) + + def get(self, path: str, cluster: dict = NO_CLUSTER, **records: str) -> dict: + for key, value in records.items(): + asyncio.run(self.store.set_value(f"open_rl:{key}", value)) + self.enterContext(mock.patch.object(fleet, "kubernetes", mock.AsyncMock(return_value=copy.deepcopy(cluster)))) + return TestClient(api_server.app).get(path).json() + + +class InventoryTest(ApiTest): + def test_a_run_is_active_only_while_the_client_has_a_live_session(self): + body = self.get( + "/api/v1/inspect/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_runs_come_back_from_redis_when_kubernetes_is_unavailable(self): + down = NO_CLUSTER | {"available": False, "errors": ["Kubernetes unavailable: no kubeconfig"]} + crashing = {"name": "orw-x", "node": "n1", "owner": "x", "problem": "CrashLoopBackOff", "restarts": 4} + healthy = crashing | {"name": "orw-y", "problem": None, "restarts": 0} + records = {"model_meta:r1": meta(session_ids=["s1"]), "session:s1": "{}"} + body = self.get("/api/v1/inspect/inventory", down | {"pods": [crashing, healthy]}, **records) + + self.assertEqual([(r["run_id"], r["lifecycle"]) for r in body["runs"]], [("r1", "active")]) + self.assertEqual(body["cluster"], {"available": False, "nodes": [], "pods": [{"name": "orw-x", "problem": "CrashLoopBackOff", "restarts": 4}]}) + self.assertEqual(body["errors"], ["Kubernetes unavailable: no kubeconfig"]) + + def test_a_malformed_model_record_is_skipped_with_an_error(self): + body = self.get("/api/v1/inspect/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"]) + + +class JoinTest(unittest.TestCase): + def test_lifecycle_follows_the_runs_workloads(self): + live, raw = {"s1"}, meta(fine_tuning_type="full", session_ids=["s1"]) + self.assertEqual(fleet.join("r1", raw, live, cluster(STOPPED, STOPPED))["lifecycle"], "ended") + self.assertEqual(fleet.join("r1", raw, live, cluster(WAITING_FOR_PLACEMENT, READY))["lifecycle"], "waiting") + self.assertEqual(fleet.join("r1", raw, live, cluster(STARTING, READY))["lifecycle"], "placed") + self.assertEqual(fleet.join("r1", raw, live, cluster(FAILED, READY))["lifecycle"], "failed") + + def test_a_finished_run_on_a_shared_runtime_does_not_claim_its_pods(self): + pod = {"name": "orw-qwen-qwen3-8b-trainer", "node": "n1", "owner": "qwen-qwen3-8b-a82fd547", "problem": None, "restarts": 0} + shared = {"pods": [pod], "workloads": [{"model_id": "Qwen/Qwen3-8B", "training_kind": "lora", "state": READY}]} + finished = fleet.join("r2", meta(status="completed"), set(), shared) + active = fleet.join("r3", meta(), set(), shared) + self.assertEqual((finished["lifecycle"], finished["pods"], finished["nodes"]), ("ended", [], [])) + self.assertEqual((active["lifecycle"], active["pods"], active["nodes"]), ("placed", [{"name": pod["name"], "problem": None}], ["n1"])) + self.assertTrue(finished["shared_runtime"] and active["shared_runtime"]) + + +class LogsTest(ApiTest): + def setUp(self): + super().setUp() + self.sent = [] + + async def send(method: str, url: str, json: dict, headers: dict) -> httpx.Response: + self.sent.append(json) + entry = {"textPayload": "INFO loading", "resource": {"labels": {"pod_name": "orw-fft-r1-trainer"}}} + page = {"entries": [entry], "nextPageToken": "t2"} if "pageToken" not in json else {"entries": []} + return httpx.Response(200, json=page, request=httpx.Request(method, url)) + + self.enterContext(mock.patch.dict(os.environ, {"OPEN_RL_WORKER_NAMESPACE": "ns", "OPEN_RL_LOG_STORE": "cloud-logging"})) + self.enterContext(mock.patch.object(cloud_logging, "send", send)) + self.enterContext(mock.patch.object(cloud_logging, "credentials", None)) + self.enterContext(mock.patch.object(cloud_logging, "project", None)) + self.google_auth = self.enterContext(mock.patch("google.auth.default", return_value=(mock.Mock(valid=True, token="t"), "p"))) + + def test_logs_read_only_the_runs_pods_and_page_with_the_stores_cursor(self): + window = "since=2026-10-01T00:00:00Z&until=2026-10-01T01:00:00Z" + first = self.get(f"/api/v1/inspect/logs?run_id=r1&pod=orw-fft-r1-trainer&{window}", **{"model_meta:r1": meta(fine_tuning_type="full")}) + second = self.get(f"/api/v1/inspect/logs?run_id=r1&pod=orw-fft-r1-trainer&{window}&cursor=t2") + + self.assertEqual([(r["pod"], r["message"]) for r in first["records"]], [("orw-fft-r1-trainer", "INFO loading")]) + self.assertEqual((first["next_cursor"], second["records"], second["next_cursor"]), ("t2", [], None)) + self.assertEqual([s.get("pageToken") for s in self.sent], [None, "t2"]) + self.assertEqual(self.sent[0]["resourceNames"], ["projects/p"]) + clauses = self.sent[0]["filter"].split(" AND ") + self.assertIn('resource.labels.namespace_name="ns"', clauses) + self.assertIn('resource.labels.pod_name=~"^orw-fft-r1-"', clauses) + self.assertIn('resource.labels.pod_name="orw-fft-r1-trainer"', clauses) + self.assertNotIn("owner", self.sent[0]["filter"]) + + def test_a_named_resource_is_read_instead_of_the_credentials_project(self): + self.google_auth.return_value = (mock.Mock(valid=True, token="t"), None) + meta_r1 = {"model_meta:r1": meta()} + default = self.get("/api/v1/inspect/logs?run_id=r1", **meta_r1) + view = "projects/q/locations/global/buckets/b/views/v" + os.environ["OPEN_RL_LOG_STORE"] = f"cloud-logging:{view}" + named = self.get("/api/v1/inspect/logs?run_id=r1") + + self.assertFalse(default["available"]) + self.assertIn("The Google credentials name no project", default["error"]) + self.assertTrue(named["available"]) + self.assertEqual([s["resourceNames"] for s in self.sent], [[view]]) + + def test_logs_stay_off_without_a_known_log_store(self): + os.environ.pop("OPEN_RL_LOG_STORE") + unset = self.get("/api/v1/inspect/logs?run_id=r1", **{"model_meta:r1": meta()}) + os.environ["OPEN_RL_LOG_STORE"] = "splunk" + unknown = self.get("/api/v1/inspect/logs?run_id=r1") + + self.assertEqual((unset["available"], unset["error"]), (False, "No log store configured. Set OPEN_RL_LOG_STORE; see docs/configuration.md.")) + self.assertEqual((unknown["available"], unknown["error"]), (False, "Unknown OPEN_RL_LOG_STORE 'splunk'; supported: cloud-logging")) + self.assertEqual(self.sent, []) + + def test_client_logs_are_the_lines_labelled_with_its_id(self): + window = "since=2026-10-01T00:00:00Z&until=2026-10-01T01:00:00Z" + body = self.get(f"/api/v1/inspect/logs?run_id=r1&pod=client&{window}", **{"model_meta:r1": meta(client_log_id="c1")}) + + self.assertTrue(body["available"]) + self.assertEqual( + self.sent[0]["filter"].split(" AND "), + [ + '(labels."k8s-pod/openrl_client"="c1" OR labels."openrl_client"="c1")', + 'timestamp>="2026-10-01T00:00:00.000000+00:00"', + 'timestamp<="2026-10-01T01:00:00.000000+00:00"', + ], + ) + + def test_a_run_without_a_client_log_id_has_no_client_logs(self): + body = self.get("/api/v1/inspect/logs?run_id=r1&pod=client", **{"model_meta:r1": meta()}) + + self.assertEqual((body["available"], body["error"]), (False, "This run set no openrl.client_log_id; see docs/configuration.md.")) + self.assertEqual(self.sent, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_dashboard_kubernetes.py b/tests/test_observability_kubernetes.py similarity index 90% rename from tests/test_dashboard_kubernetes.py rename to tests/test_observability_kubernetes.py index 92c12dcd..6318eb79 100644 --- a/tests/test_dashboard_kubernetes.py +++ b/tests/test_observability_kubernetes.py @@ -2,7 +2,7 @@ import unittest from unittest import mock -from server.dashboard.kubernetes import gpu_capacity, kubernetes, problem +from server.observability.kubernetes import gpu_capacity, kubernetes, problem class NodeTest(unittest.TestCase): @@ -29,7 +29,7 @@ def test_a_pod_status_names_what_keeps_it_from_running(self): class KubernetesTest(unittest.TestCase): def test_a_cluster_without_a_client_is_reported_as_unavailable(self): - with mock.patch("server.dashboard.kubernetes.kubernetes_api", side_effect=RuntimeError("no kubeconfig")): + with mock.patch("server.observability.kubernetes.kubernetes_api", side_effect=RuntimeError("no kubeconfig")): cluster = asyncio.run(kubernetes()) self.assertFalse(cluster["available"]) diff --git a/uv.lock b/uv.lock index a1120239..6d094e7d 100644 --- a/uv.lock +++ b/uv.lock @@ -1963,6 +1963,7 @@ dependencies = [ [package.optional-dependencies] cluster = [ + { name = "google-auth" }, { name = "kubernetes" }, { name = "timeslice" }, ] @@ -2015,6 +2016,7 @@ requires-dist = [ { name = "fastapi" }, { name = "flash-linear-attention", marker = "extra == 'gpu'", specifier = "==0.5.1" }, { name = "flashinfer-jit-cache", marker = "sys_platform == 'linux' and extra == 'vllm'", specifier = "==0.6.18.post1", index = "https://flashinfer.ai/whl/cu129", conflict = { package = "open-rl", extra = "vllm" } }, + { name = "google-auth", marker = "extra == 'cluster'" }, { name = "httpx" }, { name = "kubernetes", marker = "extra == 'cluster'", specifier = ">=29" }, { name = "numpy", marker = "extra == 'cpu'" },