From 43498ae78d94a109f93b6adac73ebf550446c681 Mon Sep 17 00:00:00 2001 From: ShubyM Date: Thu, 29 Oct 2026 13:24:27 -0700 Subject: [PATCH 1/4] refactor: rename server.dashboard to server.observability The inspection API serves agents and the CLI as well as a dashboard, so the package is named for what it does. This moves the package and its two test modules and updates the import and the Makefile test lists. Nothing else changes. --- Makefile | 4 ++-- src/server/api_server.py | 2 +- src/server/{dashboard => observability}/__init__.py | 0 src/server/{dashboard => observability}/fleet.py | 0 src/server/{dashboard => observability}/kubernetes.py | 0 src/server/{dashboard => observability}/router.py | 2 +- tests/{test_dashboard.py => test_observability.py} | 2 +- ...shboard_kubernetes.py => test_observability_kubernetes.py} | 4 ++-- 8 files changed, 7 insertions(+), 7 deletions(-) rename src/server/{dashboard => observability}/__init__.py (100%) rename src/server/{dashboard => observability}/fleet.py (100%) rename src/server/{dashboard => observability}/kubernetes.py (100%) rename src/server/{dashboard => observability}/router.py (92%) rename tests/{test_dashboard.py => test_observability.py} (97%) rename tests/{test_dashboard_kubernetes.py => test_observability_kubernetes.py} (90%) diff --git a/Makefile b/Makefile index 6ab6e7557..f8ff69b08 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/src/server/api_server.py b/src/server/api_server.py index 0024cb89e..0f4d45df2 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 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/dashboard/fleet.py b/src/server/observability/fleet.py similarity index 100% rename from src/server/dashboard/fleet.py rename to src/server/observability/fleet.py 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/dashboard/router.py b/src/server/observability/router.py similarity index 92% rename from src/server/dashboard/router.py rename to src/server/observability/router.py index 35fdc7397..d2242952c 100644 --- a/src/server/dashboard/router.py +++ b/src/server/observability/router.py @@ -1,6 +1,6 @@ from fastapi import APIRouter -from server.dashboard import fleet +from server.observability import fleet router = APIRouter(tags=["inspection"]) diff --git a/tests/test_dashboard.py b/tests/test_observability.py similarity index 97% rename from tests/test_dashboard.py rename to tests/test_observability.py index 03b4ab655..37000fd62 100644 --- a/tests/test_dashboard.py +++ b/tests/test_observability.py @@ -5,8 +5,8 @@ from fastapi.testclient import TestClient from server import api_server -from server.dashboard import fleet from server.model_metadata import TrainingModelMetadata +from server.observability import fleet from server.store import InMemoryStateStore 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 92c12dcd4..6318eb792 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"]) From abb60ad7d59066b9b7f56950f01bb6174ac01edf Mon Sep 17 00:00:00 2001 From: ShubyM Date: Thu, 29 Oct 2026 13:36:50 -0700 Subject: [PATCH 2/4] feat(observability): show runs' pods and nodes and serve their logs The inventory now joins each run with its pods, nodes and Workloads from Kubernetes, gives it a lifecycle of ended, failed, waiting, placed or active, and reports the cluster's nodes and problem pods. /api/v1/inspect/logs pages a run's worker logs from Cloud Logging, reading only pods with the run's owner label, and with pod=client reads the client's logs.log from the shared volume. --- docs/configuration.md | 27 +++- pyproject.toml | 1 + src/server/observability/client_runs.py | 107 +++++++++++++++ src/server/observability/cloud_logging.py | 156 ++++++++++++++++++++++ src/server/observability/fleet.py | 43 ++++-- src/server/observability/router.py | 68 +++++++++- tests/test_observability.py | 119 +++++++++++++++-- uv.lock | 2 + 8 files changed, 495 insertions(+), 28 deletions(-) create mode 100644 src/server/observability/client_runs.py create mode 100644 src/server/observability/cloud_logging.py diff --git a/docs/configuration.md b/docs/configuration.md index b4ccf07e2..278b23d13 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -127,11 +127,36 @@ for. On one machine every pool has one replica, in process. See 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. +same Redis as the rest of the server. It reads pods, nodes and `Workload`s from +Kubernetes. + +Each run has a `lifecycle`. It is `ended`, `failed`, `waiting`, `placed` or +`active`, from the run's `Workload`s. Without `Workload`s it is `active` while a +client session is live. Each run also lists its `pods`, by name with a +`problem` when one is not simply running, and the `nodes` they are on. Runs +that share a runtime show its pods only while they are active. `cluster` lists +the nodes with their GPUs and the pods with problems. A source that cannot be +read adds a line to `errors`. + +`/api/v1/inspect/logs?run_id=` returns a page of the run's worker logs from +Cloud Logging, newest first, with the server's Google application default +credentials. It reads only pods with the run's owner label. +`pod` narrows it to one of them. `since` and `until` take ISO-8601 or Unix +seconds, and `since` also takes a span such as `1h`. The window defaults to the +last 30 minutes and is at most seven days. Pass `next_cursor` back as `cursor` +for the next older page. With `pod=client`, it reads the client's `logs.log` +from the run's client directory instead. That directory is the +`tinker-cookbook` `log_path` on a volume the server mounts, up to five levels +under `OPEN_RL_CLIENT_RUNS_ROOT`. | 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_WORKER_NAMESPACE` | `POD_NAMESPACE`, else `openrl-system` | The namespace whose pods, `Workload`s and logs it reads. | +| `OPEN_RL_GKE_PROJECT`, `OPEN_RL_GKE_CLUSTER`, `OPEN_RL_GKE_LOCATION` | the GKE metadata server | The cluster whose logs it reads. Outside GKE, set all three to read logs. | +| `OPEN_RL_GKE_TELEMETRY` | `auto` | `0` turns off log reads. | +| `OPEN_RL_LOG_VIEWS` | `projects/` | Comma-separated Cloud Logging resources to read, such as a log bucket view. | +| `OPEN_RL_CLIENT_RUNS_ROOT` | `/mnt/shared` | Where it looks for client directories. | ### Metrics and logs on GKE diff --git a/pyproject.toml b/pyproject.toml index 804694c58..644a9a099 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/observability/client_runs.py b/src/server/observability/client_runs.py new file mode 100644 index 000000000..65e0acc75 --- /dev/null +++ b/src/server/observability/client_runs.py @@ -0,0 +1,107 @@ +"""A run's client directory: the tinker-cookbook's log_path on a volume the server also mounts, where the client writes +config.json, checkpoints.jsonl and logs.log. Every read is blocking, so callers run it in a thread.""" + +import json +import os +import re + +# Run directories as deep as the e2e client's /mnt/shared/open-rl/runs//open-rl-tmp/. +DEPTH = 5 +MATCH_SECONDS = 60 # config.json is written just before the client creates its model +SCAN_BYTES = 4 << 20 # of logs.log per page +MESSAGE_CHARS = 16384 +TINKER_PATH = re.compile(r'tinker://([^/"]+)/') +# The cookbook's file format, "%(name)s:%(lineno)d [%(levelname)s] %(message)s"; other lines continue the record above. +HEADER = re.compile(r"\S+:\d+ \[(DEBUG|INFO|WARNING|ERROR|CRITICAL)\] ") +listings: dict[str, tuple[int, list[str], bool]] = {} # directory -> (mtime, subdirectories, is a run directory) +parsed: dict[str, tuple[tuple[int, int], object]] = {} # file -> ((mtime, size), parsed contents) + + +def root() -> str: + return os.getenv("OPEN_RL_CLIENT_RUNS_ROOT", "/mnt/shared") + + +def directories(path: str | None = None, depth: int = DEPTH) -> list[str]: + """Every directory under the root, to DEPTH, that holds config.json or checkpoints.jsonl. A run directory is not + descended into, and a directory is listed again only when its mtime changes.""" + path = path or root() + try: + mtime = os.stat(path).st_mtime_ns + if (listing := listings.get(path)) is None or listing[0] != mtime: + with os.scandir(path) as scan: + entries = list(scan) + run = any(e.name in ("config.json", "checkpoints.jsonl") for e in entries) + below = [] if run else [e.path for e in entries if not e.name.startswith(".") and e.name != "lost+found" and e.is_dir(follow_symlinks=False)] + listing = listings[path] = (mtime, below, run) + except OSError: # no root here, or a directory removed while it was scanned + return [] + if listing[2]: + return [path] + return [found for below in listing[1] for found in directories(below, depth - 1)] if depth > 0 else [] + + +def read(path: str, parse) -> tuple[float, object] | None: + """The file's mtime and parsed contents, read again only when its mtime or size changes. None when it is absent, as + checkpoints.jsonl is until the first save, or is not what the cookbook writes.""" + try: + stat = os.stat(path) + if (entry := parsed.get(path)) is None or entry[0] != (stat.st_mtime_ns, stat.st_size): + with open(path, encoding="utf-8", errors="replace") as f: + entry = parsed[path] = ((stat.st_mtime_ns, stat.st_size), parse(f.read())) + except (OSError, ValueError): + return None + return entry[0][0] / 1e9, entry[1] + + +def started(directory: str, run: dict) -> bool: + """Whether config.json was written within MATCH_SECONDS of the run's creation, for its base model.""" + if not (found := read(os.path.join(directory, "config.json"), json.loads)) or not isinstance(config := found[1], dict): + return False + return abs(found[0] - run["created_at"]) <= MATCH_SECONDS and config.get("model_name") == run["model"] + + +def checkpoint_runs(directory: str) -> set[str]: + found = read(os.path.join(directory, "checkpoints.jsonl"), lambda text: set(TINKER_PATH.findall(text))) + return found[1] if found else set() + + +def match(run: dict, candidates: list[str]) -> str | None: + """The run's directory: the one whose checkpoints.jsonl names tinker:///, else the one started() for it. Of + several, only one that was also started() for it is taken; otherwise none, never a guess.""" + exact = [d for d in candidates if run["run_id"] in checkpoint_runs(d)] + if len(exact) == 1: + return exact[0] + timed = [d for d in exact or candidates if started(d, run)] + return timed[0] if len(timed) == 1 else None + + +def log_page(directory: str, before: int | None, limit: int, q: str = "", severity: str | None = None) -> dict: + """logs.log, newest first, in the shape of a Cloud Logging page. Its lines carry no time, so a record is a header line + and the lines under it, such as a traceback, and next_cursor is the byte offset the next older page ends at.""" + path = os.path.join(directory, "logs.log") + with open(path, "rb") as f: + size = f.seek(0, os.SEEK_END) + end = size if before is None else min(before, size) + start = max(0, end - SCAN_BYTES) + f.seek(start) + block = f.read(end - start) + if start and (cut := block.find(b"\n") + 1): # the first line began before the block + block, start = block[cut:], start + cut + lines, at = [], start + for raw in block.splitlines(keepends=True): + lines.append((at, raw.decode("utf-8", "replace").rstrip("\r\n"))) + at += len(raw) + records, under, cursor = [], [], None + for at, text in reversed(lines): + under.append(text) + # The file's first line starts a record, and so does a block's first line when the block has no header at all. + if not (header := HEADER.match(text)) and at and not (at == start and cursor is None): + continue + message, under, cursor = "\n".join(reversed(under)), [], at + level = header[1] if header else "UNKNOWN" + if (not severity or level == severity) and q.lower() in message.lower(): + record = {"id": f"logs.log:{at}", "timestamp": None, "pod": "client", "role": "client", "node": None, "severity": level} + records.append(record | {"message": message[:MESSAGE_CHARS], "message_truncated": len(message) > MESSAGE_CHARS}) + if len(records) == limit: + break + return {"records": records, "next_cursor": str(cursor) if cursor else None} diff --git a/src/server/observability/cloud_logging.py b/src/server/observability/cloud_logging.py new file mode 100644 index 000000000..f4c0bf3c7 --- /dev/null +++ b/src/server/observability/cloud_logging.py @@ -0,0 +1,156 @@ +"""A run's worker logs from Cloud Logging, read with the server's Google credentials.""" + +import asyncio +import json +import os +import re +import threading +import time +from datetime import UTC, datetime, timedelta + +import httpx + +from server.observability.kubernetes import namespace + +METADATA = {"project": "project/project-id", "cluster": "instance/attributes/cluster-name", "location": "instance/attributes/cluster-location"} +# GKE copies each pod label onto its log entries as k8s-pod/, with the dots in the key as underscores. +OWNER = 'labels."k8s-pod/timeslice_io/owner"' +# 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 +detected: dict = {} +discovered_at = -60.0 +credentials = None +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 configuration() -> dict: + """OPEN_RL_GKE_PROJECT, _CLUSTER and _LOCATION, else what the metadata server said; OPEN_RL_GKE_TELEMETRY=0 turns GKE off.""" + fields = {key: os.getenv(f"OPEN_RL_GKE_{key.upper()}") or detected.get(key, "") for key in METADATA} + off = os.getenv("OPEN_RL_GKE_TELEMETRY", "auto").strip().lower() not in ("auto", "1", "true") + return {**fields, "configured": not off and all(fields.values())} + + +async def discover() -> None: + """In a cluster, read the GKE identity from the metadata server; retry a failed read after a minute.""" + global detected, discovered_at + if configuration()["configured"] or not os.getenv("KUBERNETES_SERVICE_HOST") or time.monotonic() - discovered_at < 60: + return + discovered_at = time.monotonic() + async with httpx.AsyncClient(timeout=1, trust_env=False, follow_redirects=False) as metadata: + + async def read(path: str) -> str: + response = await metadata.get(f"http://metadata.google.internal/computeMetadata/v1/{path}", headers={"Metadata-Flavor": "Google"}) + if response.status_code != 200 or response.headers.get("Metadata-Flavor") != "Google" or not 0 < len(response.text.strip()) <= 256: + raise ValueError("Invalid metadata response") + return response.text.strip() + + values = await asyncio.gather(*(read(path) for path in METADATA.values()), return_exceptions=True) + if not any(isinstance(value, Exception) for value in values): # never partial: plain GCE also exposes a project + detected = dict(zip(METADATA, values)) + + +def access_token() -> str: + import google.auth # the cluster extra installs it + from google.auth.transport.requests import Request + + global credentials + with credentials_lock: + if credentials is None: + credentials, _ = 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 this project. + if isinstance(credentials, google.oauth2.credentials.Credentials) and not credentials.quota_project_id: + headers["x-goog-user-project"] = configuration()["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(owner: str, start: str, end: str, severity: str | None, q: str, pod: str | None = None) -> str: + """The pods with the run's owner label in the window, or one of them. Never any other pod in the namespace, so a run id + cannot open another run's logs.""" + config = configuration() + labels = {"project_id": config["project"], "location": config["location"], "cluster_name": config["cluster"], "namespace_name": namespace()} + clauses = ['resource.type="k8s_container"', *(f"resource.labels.{label}={json.dumps(value)}" for label, value in labels.items())] + clauses.append(f"{OWNER}={json.dumps(owner)}") + if pod: + clauses.append(f"resource.labels.pod_name={json.dumps(pod)}") + clauses += [f"timestamp>={json.dumps(start)}", f"timestamp<={json.dumps(end)}"] + if severity in RELABELED: + stderr = json.dumps(f"projects/{config['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") or "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(owner: str, since: str, until: str, *, cursor=None, limit=200, q="", pod=None, severity=None) -> dict: + """A page of logs, newest first. The cursor is Cloud Logging's own page token, valid for the same filters.""" + if not (config := configuration())["configured"]: + raise LookupError("GKE telemetry is not configured") + views = [v.strip() for v in os.getenv("OPEN_RL_LOG_VIEWS", "").split(",") if v.strip()] or [f"projects/{config['project']}"] + query = {"resourceNames": views, "filter": logs_filter(owner, since, until, severity, q, pod), "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=await gcp_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/observability/fleet.py b/src/server/observability/fleet.py index 8c5a37f3f..b5ec39756 100644 --- a/src/server/observability/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/observability/router.py b/src/server/observability/router.py index d2242952c..2ea17c50d 100644 --- a/src/server/observability/router.py +++ b/src/server/observability/router.py @@ -1,13 +1,71 @@ -from fastapi import APIRouter +import asyncio +import os +from typing import Annotated -from server.observability import fleet +import httpx +from fastapi import APIRouter, HTTPException, Query + +from server.model_metadata import decode_model_metadata +from server.observability import client_runs, cloud_logging, fleet +from server.store import get_state_store 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" @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.""" + """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 its client's logs.log")] = 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: the run's worker pods by owner label from Cloud Logging, or with pod=client its client's logs.log. That + file's lines carry no time, so since and until do not apply to it and its cursor is a byte offset.""" + 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) + result = {"run_id": run_id, "source": "gke", "since": since, "until": until, "records": [], "next_cursor": None} + if pod == "client": + run = {"run_id": run_id, "created_at": meta.created_at, "model": meta.base_model} + candidates = await fleet.cached("client directories", 30, lambda: asyncio.to_thread(client_runs.directories)) + if not (directory := await asyncio.to_thread(client_runs.match, run, candidates)): + return {**result, "source": "client", "available": False, "error": f"No client directory under {client_runs.root()} matches this run"} + if cursor and not cursor.isdigit(): + raise HTTPException(400, CURSOR_EXPIRED) + result["source"] = path = os.path.join(directory, "logs.log") + try: + page = await asyncio.to_thread(client_runs.log_page, directory, int(cursor) if cursor else None, limit, q, severity) + return {**result, **page, "available": True} + except OSError as exc: + return {**result, "available": False, "error": f"{path} unreadable: {type(exc).__name__}"} + await cloud_logging.discover() + owner = fleet.owner(meta.runtime(run_id)) + try: + page = await cloud_logging.log_page(owner, since, until, cursor=cursor, limit=limit, q=q, pod=pod, 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"Cloud Logging 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 "Cloud Logging unavailable"} diff --git a/tests/test_observability.py b/tests/test_observability.py index 37000fd62..ed3c0fa0a 100644 --- a/tests/test_observability.py +++ b/tests/test_observability.py @@ -1,48 +1,145 @@ import asyncio +import copy +import json +import os +import tempfile 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 fleet +from server.observability import client_runs, 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() -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)) +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=store)) - return TestClient(api_server.app).get("/api/v1/inspect/inventory").json() + 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.inventory( + 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.inventory(**{"model_meta:good": meta(), "model_meta:bad": "not json"}) + 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"]) + + +GKE = {"OPEN_RL_GKE_PROJECT": "p", "OPEN_RL_GKE_CLUSTER": "c", "OPEN_RL_GKE_LOCATION": "l", "OPEN_RL_WORKER_NAMESPACE": "ns"} + + +class LogsTest(ApiTest): + def test_logs_read_only_the_runs_pods_and_page_with_cloud_loggings_cursor(self): + sent = [] + + async def send(method: str, url: str, json: dict, headers: dict) -> httpx.Response: + sent.append(json) + entry = {"textPayload": "INFO loading", "resource": {"labels": {"pod_name": "orw-r1-t"}}} + 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, GKE)) + self.enterContext(mock.patch.object(cloud_logging, "send", send)) + self.enterContext(mock.patch.object(cloud_logging, "gcp_headers", mock.AsyncMock(return_value={}))) + 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-r1-t&{window}", **{"model_meta:r1": meta(fine_tuning_type="full")}) + second = self.get(f"/api/v1/inspect/logs?run_id=r1&pod=orw-r1-t&{window}&cursor=t2") + + self.assertEqual([(r["pod"], r["message"]) for r in first["records"]], [("orw-r1-t", "INFO loading")]) + self.assertEqual((first["next_cursor"], second["records"], second["next_cursor"]), ("t2", [], None)) + self.assertEqual([s.get("pageToken") for s in sent], [None, "t2"]) + clauses = sent[0]["filter"].split(" AND ") + self.assertIn('labels."k8s-pod/timeslice_io/owner"="r1"', clauses) + self.assertIn('resource.labels.pod_name="orw-r1-t"', clauses) + self.assertIn('resource.labels.namespace_name="ns"', clauses) + + def test_the_client_log_is_found_in_the_e2e_layout_and_paged_newest_first(self): + root = self.enterContext(tempfile.TemporaryDirectory()) + run_dir = os.path.join(root, "open-rl", "runs", "rl_loop", "open-rl-tmp", "fft_gsm8k_rl") + other = os.path.join(root, "open-rl", "runs", "rl_loop", "open-rl-tmp", "lora_gsm8k_rl") + for directory, run_id in ((run_dir, "r1"), (other, "r9")): + os.makedirs(directory) + with open(os.path.join(directory, "checkpoints.jsonl"), "w") as f: + f.write(json.dumps({"state_path": f"tinker://{run_id}/weights/000010"}) + "\n") + with open(os.path.join(run_dir, "logs.log"), "w") as f: + f.write("train:10 [INFO] step 1\ntrain:20 [ERROR] step 2 failed\nTraceback (most recent call last)\n") + self.enterContext(mock.patch.dict(os.environ, {"OPEN_RL_CLIENT_RUNS_ROOT": root})) + client_runs.listings.clear() + + newest = self.get("/api/v1/inspect/logs?run_id=r1&pod=client&limit=1", **{"model_meta:r1": meta()}) + older = self.get(f"/api/v1/inspect/logs?run_id=r1&pod=client&limit=1&cursor={newest['next_cursor']}") + + self.assertEqual(newest["source"], os.path.join(run_dir, "logs.log")) + self.assertEqual([r["message"] for r in newest["records"]], ["train:20 [ERROR] step 2 failed\nTraceback (most recent call last)"]) + self.assertEqual(([r["message"] for r in older["records"]], older["next_cursor"]), (["train:10 [INFO] step 1"], None)) + + if __name__ == "__main__": unittest.main() diff --git a/uv.lock b/uv.lock index a11202399..6d094e7d2 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'" }, From a3f61baff789426e6ac12a1fb298dcae9365eb67 Mon Sep 17 00:00:00 2001 From: ShubyM Date: Thu, 29 Oct 2026 14:41:41 -0700 Subject: [PATCH 3/4] feat(observability): make log reads opt-in and read client logs from a declared path Worker logs are off until OPEN_RL_LOG_STORE names a store. cloud-logging is the only store today. It sits in a small registry, so another store such as Loki can be added later without a route change. The GKE metadata server no longer turns log reads on. OPEN_RL_GKE_TELEMETRY is gone. A run's worker pods are now found by name. The scheduler names a worker pod orw-, and a run's Workloads start with --. The store reads the worker namespace's pods with that prefix instead of the timeslice.io/owner label. The server no longer scans the shared volume for client directories. A client declares its files with the openrl.log_path setting, which is stored on the model. pod=client reads /logs.log only when its resolved path is under OPEN_RL_CLIENT_LOG_ROOTS. OPEN_RL_CLIENT_RUNS_ROOT is gone. --- docs/configuration.md | 35 ++++++---- src/server/api_server.py | 6 ++ src/server/model_metadata.py | 1 + src/server/observability/client_runs.py | 82 +++++------------------ src/server/observability/cloud_logging.py | 27 ++++---- src/server/observability/router.py | 36 ++++++---- tests/test_api_server_paths.py | 5 ++ tests/test_observability.py | 73 ++++++++++++++------ 8 files changed, 133 insertions(+), 132 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 278b23d13..017ef3865 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -138,25 +138,32 @@ that share a runtime show its pods only while they are active. `cluster` lists the nodes with their GPUs and the pods with problems. A source that cannot be read adds a line to `errors`. -`/api/v1/inspect/logs?run_id=` returns a page of the run's worker logs from -Cloud Logging, newest first, with the server's Google application default -credentials. It reads only pods with the run's owner label. -`pod` narrows it to one of them. `since` and `until` take ISO-8601 or Unix -seconds, and `since` also takes a span such as `1h`. The window defaults to the -last 30 minutes and is at most seven days. Pass `next_cursor` back as `cursor` -for the next older page. With `pod=client`, it reads the client's `logs.log` -from the run's client directory instead. That directory is the -`tinker-cookbook` `log_path` on a volume the server mounts, up to five levels -under `OPEN_RL_CLIENT_RUNS_ROOT`. +`/api/v1/inspect/logs?run_id=` returns a page of the run's worker logs, +newest first, from the log store that `OPEN_RL_LOG_STORE` names. Without it, +log reads are off. It reads only the pods in the worker namespace whose names +start with the run's prefix, `orw---`. `pod` narrows it to one of +them. `since` and `until` take ISO-8601 or Unix seconds, and `since` also takes +a span such as `1h`. The window defaults to the last 30 minutes and is at most +seven days. Pass `next_cursor` back as `cursor` for the next older page. +Cloud Logging is the only store today. Other stores, such as Loki, plug in +later as adapters. On a cluster outside GKE, logs need a shipper into a +supported store. + +With `pod=client`, it reads `logs.log` from the directory that the run's +`openrl.log_path` setting names. Whatever launches the client sets it, as a +`TINKER_TAGS` entry or in `user_metadata`, to where the client writes its +files, such as the `tinker-cookbook` `log_path` on a volume the server also +mounts. The server reads the file only when its resolved path is under one of +`OPEN_RL_CLIENT_LOG_ROOTS`. | 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_WORKER_NAMESPACE` | `POD_NAMESPACE`, else `openrl-system` | The namespace whose pods, `Workload`s and logs it reads. | -| `OPEN_RL_GKE_PROJECT`, `OPEN_RL_GKE_CLUSTER`, `OPEN_RL_GKE_LOCATION` | the GKE metadata server | The cluster whose logs it reads. Outside GKE, set all three to read logs. | -| `OPEN_RL_GKE_TELEMETRY` | `auto` | `0` turns off log reads. | -| `OPEN_RL_LOG_VIEWS` | `projects/` | Comma-separated Cloud Logging resources to read, such as a log bucket view. | -| `OPEN_RL_CLIENT_RUNS_ROOT` | `/mnt/shared` | Where it looks for client directories. | +| `OPEN_RL_LOG_STORE` | unset | Where it reads worker logs. Unset turns log reads off. `cloud-logging` reads Cloud Logging with the server's Google application default credentials. | +| `OPEN_RL_GKE_PROJECT`, `OPEN_RL_GKE_CLUSTER`, `OPEN_RL_GKE_LOCATION` | the GKE metadata server | With `cloud-logging`, the cluster whose logs it reads. Outside GKE, set all three. | +| `OPEN_RL_LOG_VIEWS` | `projects/` | With `cloud-logging`, comma-separated Cloud Logging resources to read, such as a log bucket view. | +| `OPEN_RL_CLIENT_LOG_ROOTS` | unset | Comma-separated directories that a run's `openrl.log_path` must be under. Unset, it reads no client logs. | ### Metrics and logs on GKE diff --git a/src/server/api_server.py b/src/server/api_server.py index 0f4d45df2..be82f9332 100644 --- a/src/server/api_server.py +++ b/src/server/api_server.py @@ -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.", ) + log_path: str | None = Field( + default=None, + description="Where the client writes its files, such as the tinker-cookbook log_path, on a volume the server also mounts. " + "The inspection API serves /logs.log when the server allows that path.", + ) # 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, + log_path=settings.log_path, ) 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/model_metadata.py b/src/server/model_metadata.py index 8ec44e568..58e001c3f 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 + log_path: str | None = None status: str = "active" updated_at: float = 0.0 completed_at: float | None = None diff --git a/src/server/observability/client_runs.py b/src/server/observability/client_runs.py index 65e0acc75..29a0b3145 100644 --- a/src/server/observability/client_runs.py +++ b/src/server/observability/client_runs.py @@ -1,84 +1,32 @@ -"""A run's client directory: the tinker-cookbook's log_path on a volume the server also mounts, where the client writes -config.json, checkpoints.jsonl and logs.log. Every read is blocking, so callers run it in a thread.""" +"""A run's client log: logs.log under the openrl.log_path the client declared, such as the tinker-cookbook's log_path on a +volume the server also mounts. Every read is blocking, so callers run it in a thread.""" -import json import os import re -# Run directories as deep as the e2e client's /mnt/shared/open-rl/runs//open-rl-tmp/. -DEPTH = 5 -MATCH_SECONDS = 60 # config.json is written just before the client creates its model SCAN_BYTES = 4 << 20 # of logs.log per page MESSAGE_CHARS = 16384 -TINKER_PATH = re.compile(r'tinker://([^/"]+)/') # The cookbook's file format, "%(name)s:%(lineno)d [%(levelname)s] %(message)s"; other lines continue the record above. HEADER = re.compile(r"\S+:\d+ \[(DEBUG|INFO|WARNING|ERROR|CRITICAL)\] ") -listings: dict[str, tuple[int, list[str], bool]] = {} # directory -> (mtime, subdirectories, is a run directory) -parsed: dict[str, tuple[tuple[int, int], object]] = {} # file -> ((mtime, size), parsed contents) -def root() -> str: - return os.getenv("OPEN_RL_CLIENT_RUNS_ROOT", "/mnt/shared") +def log_file(log_path: str | None) -> str: + """/logs.log, resolved, when it lies under one of OPEN_RL_CLIENT_LOG_ROOTS. The path is the client's word, so + one that leads elsewhere, by .. or a symlink, is refused rather than read.""" + roots = [os.path.realpath(root.strip()) for root in os.getenv("OPEN_RL_CLIENT_LOG_ROOTS", "").split(",") if root.strip()] + if not log_path: + raise LookupError("This run declared no openrl.log_path") + if not roots: + raise LookupError("No client log roots configured. Set OPEN_RL_CLIENT_LOG_ROOTS; see docs/configuration.md.") + path = os.path.realpath(os.path.join(log_path, "logs.log")) + if not any(os.path.commonpath([path, root]) == root for root in roots): + raise LookupError(f"{log_path} is not under OPEN_RL_CLIENT_LOG_ROOTS") + return path -def directories(path: str | None = None, depth: int = DEPTH) -> list[str]: - """Every directory under the root, to DEPTH, that holds config.json or checkpoints.jsonl. A run directory is not - descended into, and a directory is listed again only when its mtime changes.""" - path = path or root() - try: - mtime = os.stat(path).st_mtime_ns - if (listing := listings.get(path)) is None or listing[0] != mtime: - with os.scandir(path) as scan: - entries = list(scan) - run = any(e.name in ("config.json", "checkpoints.jsonl") for e in entries) - below = [] if run else [e.path for e in entries if not e.name.startswith(".") and e.name != "lost+found" and e.is_dir(follow_symlinks=False)] - listing = listings[path] = (mtime, below, run) - except OSError: # no root here, or a directory removed while it was scanned - return [] - if listing[2]: - return [path] - return [found for below in listing[1] for found in directories(below, depth - 1)] if depth > 0 else [] - - -def read(path: str, parse) -> tuple[float, object] | None: - """The file's mtime and parsed contents, read again only when its mtime or size changes. None when it is absent, as - checkpoints.jsonl is until the first save, or is not what the cookbook writes.""" - try: - stat = os.stat(path) - if (entry := parsed.get(path)) is None or entry[0] != (stat.st_mtime_ns, stat.st_size): - with open(path, encoding="utf-8", errors="replace") as f: - entry = parsed[path] = ((stat.st_mtime_ns, stat.st_size), parse(f.read())) - except (OSError, ValueError): - return None - return entry[0][0] / 1e9, entry[1] - - -def started(directory: str, run: dict) -> bool: - """Whether config.json was written within MATCH_SECONDS of the run's creation, for its base model.""" - if not (found := read(os.path.join(directory, "config.json"), json.loads)) or not isinstance(config := found[1], dict): - return False - return abs(found[0] - run["created_at"]) <= MATCH_SECONDS and config.get("model_name") == run["model"] - - -def checkpoint_runs(directory: str) -> set[str]: - found = read(os.path.join(directory, "checkpoints.jsonl"), lambda text: set(TINKER_PATH.findall(text))) - return found[1] if found else set() - - -def match(run: dict, candidates: list[str]) -> str | None: - """The run's directory: the one whose checkpoints.jsonl names tinker:///, else the one started() for it. Of - several, only one that was also started() for it is taken; otherwise none, never a guess.""" - exact = [d for d in candidates if run["run_id"] in checkpoint_runs(d)] - if len(exact) == 1: - return exact[0] - timed = [d for d in exact or candidates if started(d, run)] - return timed[0] if len(timed) == 1 else None - - -def log_page(directory: str, before: int | None, limit: int, q: str = "", severity: str | None = None) -> dict: +def log_page(path: str, before: int | None, limit: int, q: str = "", severity: str | None = None) -> dict: """logs.log, newest first, in the shape of a Cloud Logging page. Its lines carry no time, so a record is a header line and the lines under it, such as a traceback, and next_cursor is the byte offset the next older page ends at.""" - path = os.path.join(directory, "logs.log") with open(path, "rb") as f: size = f.seek(0, os.SEEK_END) end = size if before is None else min(before, size) diff --git a/src/server/observability/cloud_logging.py b/src/server/observability/cloud_logging.py index f4c0bf3c7..c1180e0be 100644 --- a/src/server/observability/cloud_logging.py +++ b/src/server/observability/cloud_logging.py @@ -10,11 +10,7 @@ import httpx -from server.observability.kubernetes import namespace - METADATA = {"project": "project/project-id", "cluster": "instance/attributes/cluster-name", "location": "instance/attributes/cluster-location"} -# GKE copies each pod label onto its log entries as k8s-pod/, with the dots in the key as underscores. -OWNER = 'labels."k8s-pod/timeslice_io/owner"' # 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)") @@ -33,10 +29,9 @@ async def send(method: str, url: str, **kwargs) -> httpx.Response: def configuration() -> dict: - """OPEN_RL_GKE_PROJECT, _CLUSTER and _LOCATION, else what the metadata server said; OPEN_RL_GKE_TELEMETRY=0 turns GKE off.""" + """OPEN_RL_GKE_PROJECT, _CLUSTER and _LOCATION, else what the metadata server said.""" fields = {key: os.getenv(f"OPEN_RL_GKE_{key.upper()}") or detected.get(key, "") for key in METADATA} - off = os.getenv("OPEN_RL_GKE_TELEMETRY", "auto").strip().lower() not in ("auto", "1", "true") - return {**fields, "configured": not off and all(fields.values())} + return {**fields, "configured": all(fields.values())} async def discover() -> None: @@ -99,13 +94,13 @@ def parse(value: str) -> datetime: return start.isoformat(timespec="microseconds"), end.isoformat(timespec="microseconds") -def logs_filter(owner: str, start: str, end: str, severity: str | None, q: str, pod: str | None = None) -> str: - """The pods with the run's owner label in the window, or one of them. Never any other pod in the namespace, so a run id - cannot open another run's logs.""" +def logs_filter(namespace: str, pod_prefix: str, start: str, end: str, severity: str | None, q: str, pod: str | None = None) -> str: + """The namespace's pods whose names start with pod_prefix, in the window, or one of them. Never any other pod, so a run + id cannot open another run's logs.""" config = configuration() - labels = {"project_id": config["project"], "location": config["location"], "cluster_name": config["cluster"], "namespace_name": namespace()} + labels = {"project_id": config["project"], "location": config["location"], "cluster_name": config["cluster"], "namespace_name": namespace} clauses = ['resource.type="k8s_container"', *(f"resource.labels.{label}={json.dumps(value)}" for label, value in labels.items())] - clauses.append(f"{OWNER}={json.dumps(owner)}") + clauses.append(f"resource.labels.pod_name=~{json.dumps('^' + pod_prefix)}") if pod: clauses.append(f"resource.labels.pod_name={json.dumps(pod)}") clauses += [f"timestamp>={json.dumps(start)}", f"timestamp<={json.dumps(end)}"] @@ -142,12 +137,14 @@ def entry_record(entry: dict) -> dict: return record | {"severity": severity, "message": text[:MESSAGE_CHARS], "message_truncated": len(text) > MESSAGE_CHARS} -async def log_page(owner: str, since: str, until: str, *, cursor=None, limit=200, q="", pod=None, severity=None) -> dict: +async def log_page(namespace: str, pod_prefix: str, since: str, until: str, *, cursor=None, limit=200, q="", pod=None, severity=None) -> dict: """A page of logs, newest first. The cursor is Cloud Logging's own page token, valid for the same filters.""" + await discover() if not (config := configuration())["configured"]: - raise LookupError("GKE telemetry is not configured") + raise LookupError("Cloud Logging needs the GKE metadata server or OPEN_RL_GKE_PROJECT, _CLUSTER and _LOCATION") views = [v.strip() for v in os.getenv("OPEN_RL_LOG_VIEWS", "").split(",") if v.strip()] or [f"projects/{config['project']}"] - query = {"resourceNames": views, "filter": logs_filter(owner, since, until, severity, q, pod), "orderBy": "timestamp desc", "pageSize": limit} + filters = logs_filter(namespace, pod_prefix, since, until, severity, q, pod) + query = {"resourceNames": views, "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=await gcp_headers())).json() records = [entry_record(entry) for entry in payload.get("entries", [])] diff --git a/src/server/observability/router.py b/src/server/observability/router.py index 2ea17c50d..f3d2d6efa 100644 --- a/src/server/observability/router.py +++ b/src/server/observability/router.py @@ -7,12 +7,16 @@ from server.model_metadata import decode_model_metadata from server.observability import client_runs, cloud_logging, fleet +from server.observability.kubernetes import namespace 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 picks one. A store reads the namespace's pods whose names start with the prefix, or the one pod named. +LOG_STORES = {"cloud-logging": cloud_logging.log_page} @router.get("/api/v1/inspect/inventory") @@ -35,8 +39,8 @@ async def logs( 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: the run's worker pods by owner label from Cloud Logging, or with pod=client its client's logs.log. That - file's lines carry no time, so since and until do not apply to it and its cursor is a byte offset.""" + """Newest first: the run's worker pods from the OPEN_RL_LOG_STORE log store, or with pod=client the logs.log under its + openrl.log_path. That file's lines carry no time, so since and until do not apply to it and its cursor is a byte offset.""" try: since, until = cloud_logging.window(since, until) except (ValueError, OverflowError, OSError) as exc: @@ -44,28 +48,32 @@ async def logs( 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) - result = {"run_id": run_id, "source": "gke", "since": since, "until": until, "records": [], "next_cursor": None} + store = os.getenv("OPEN_RL_LOG_STORE", "").strip() + result = {"run_id": run_id, "source": store, "since": since, "until": until, "records": [], "next_cursor": None} if pod == "client": - run = {"run_id": run_id, "created_at": meta.created_at, "model": meta.base_model} - candidates = await fleet.cached("client directories", 30, lambda: asyncio.to_thread(client_runs.directories)) - if not (directory := await asyncio.to_thread(client_runs.match, run, candidates)): - return {**result, "source": "client", "available": False, "error": f"No client directory under {client_runs.root()} matches this run"} + try: + result["source"] = path = await asyncio.to_thread(client_runs.log_file, meta.log_path) + except LookupError as exc: + return {**result, "source": "client", "available": False, "error": str(exc)} if cursor and not cursor.isdigit(): raise HTTPException(400, CURSOR_EXPIRED) - result["source"] = path = os.path.join(directory, "logs.log") try: - page = await asyncio.to_thread(client_runs.log_page, directory, int(cursor) if cursor else None, limit, q, severity) + page = await asyncio.to_thread(client_runs.log_page, path, int(cursor) if cursor else None, limit, q, severity) return {**result, **page, "available": True} except OSError as exc: return {**result, "available": False, "error": f"{path} unreadable: {type(exc).__name__}"} - await cloud_logging.discover() - owner = fleet.owner(meta.runtime(run_id)) + 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(store)) is None: + return {**result, "available": False, "error": f"Unknown OPEN_RL_LOG_STORE {store!r}; supported: {', '.join(LOG_STORES)}"} + # 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))}-" try: - page = await cloud_logging.log_page(owner, since, until, cursor=cursor, limit=limit, q=q, pod=pod, severity=severity) + page = await page_of(namespace(), prefix, since, until, cursor=cursor, limit=limit, q=q, pod=pod, 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"Cloud Logging refused the read (HTTP {exc.response.status_code}); check credentials and IAM"} + 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 "Cloud Logging unavailable"} + 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 06283b117..6e54f7645 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_log_path_tag_lands_on_the_model(self) -> None: + session_id = self.post("create_session", {"tags": ["openrl.log_path=/mnt/runs/gsm8k"]}).json()["session_id"] + meta = self.metadata(self.post("create_model", {"base_model": "m", "session_id": session_id}).json()["request_id"]) + self.assertEqual(meta["log_path"], "/mnt/runs/gsm8k") + 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_observability.py b/tests/test_observability.py index ed3c0fa0a..43c366ca4 100644 --- a/tests/test_observability.py +++ b/tests/test_observability.py @@ -1,6 +1,5 @@ import asyncio import copy -import json import os import tempfile import unittest @@ -11,7 +10,7 @@ from server import api_server from server.model_metadata import TrainingModelMetadata -from server.observability import client_runs, cloud_logging, fleet +from server.observability import cloud_logging, fleet from server.store import InMemoryStateStore from server.worker_manager import FAILED, READY, STARTING, STOPPED, WAITING_FOR_PLACEMENT @@ -96,50 +95,80 @@ def test_a_finished_run_on_a_shared_runtime_does_not_claim_its_pods(self): class LogsTest(ApiTest): - def test_logs_read_only_the_runs_pods_and_page_with_cloud_loggings_cursor(self): + def test_logs_read_only_the_runs_pods_and_page_with_the_stores_cursor(self): sent = [] async def send(method: str, url: str, json: dict, headers: dict) -> httpx.Response: sent.append(json) - entry = {"textPayload": "INFO loading", "resource": {"labels": {"pod_name": "orw-r1-t"}}} + 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, GKE)) + self.enterContext(mock.patch.dict(os.environ, GKE | {"OPEN_RL_LOG_STORE": "cloud-logging"})) self.enterContext(mock.patch.object(cloud_logging, "send", send)) self.enterContext(mock.patch.object(cloud_logging, "gcp_headers", mock.AsyncMock(return_value={}))) 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-r1-t&{window}", **{"model_meta:r1": meta(fine_tuning_type="full")}) - second = self.get(f"/api/v1/inspect/logs?run_id=r1&pod=orw-r1-t&{window}&cursor=t2") + 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-r1-t", "INFO loading")]) + 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 sent], [None, "t2"]) clauses = sent[0]["filter"].split(" AND ") - self.assertIn('labels."k8s-pod/timeslice_io/owner"="r1"', clauses) - self.assertIn('resource.labels.pod_name="orw-r1-t"', clauses) self.assertIn('resource.labels.namespace_name="ns"', clauses) - - def test_the_client_log_is_found_in_the_e2e_layout_and_paged_newest_first(self): - root = self.enterContext(tempfile.TemporaryDirectory()) - run_dir = os.path.join(root, "open-rl", "runs", "rl_loop", "open-rl-tmp", "fft_gsm8k_rl") - other = os.path.join(root, "open-rl", "runs", "rl_loop", "open-rl-tmp", "lora_gsm8k_rl") - for directory, run_id in ((run_dir, "r1"), (other, "r9")): - os.makedirs(directory) - with open(os.path.join(directory, "checkpoints.jsonl"), "w") as f: - f.write(json.dumps({"state_path": f"tinker://{run_id}/weights/000010"}) + "\n") + self.assertIn('resource.labels.pod_name=~"^orw-fft-r1-"', clauses) + self.assertIn('resource.labels.pod_name="orw-fft-r1-trainer"', clauses) + self.assertNotIn("owner", sent[0]["filter"]) + + def test_logs_stay_off_without_a_known_log_store(self): + self.enterContext(mock.patch.dict(os.environ, GKE | {"KUBERNETES_SERVICE_HOST": "10.0.0.1"})) + os.environ.pop("OPEN_RL_LOG_STORE", None) + network = self.enterContext(mock.patch.object(httpx.AsyncClient, "send", side_effect=AssertionError("no log read"))) + + 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")) + network.assert_not_called() + + def test_the_client_log_is_read_from_its_declared_path_and_paged_newest_first(self): + root = os.path.realpath(self.enterContext(tempfile.TemporaryDirectory())) + run_dir = os.path.join(root, "runs", "fft_gsm8k_rl") + os.makedirs(run_dir) with open(os.path.join(run_dir, "logs.log"), "w") as f: f.write("train:10 [INFO] step 1\ntrain:20 [ERROR] step 2 failed\nTraceback (most recent call last)\n") - self.enterContext(mock.patch.dict(os.environ, {"OPEN_RL_CLIENT_RUNS_ROOT": root})) - client_runs.listings.clear() + self.enterContext(mock.patch.dict(os.environ, {"OPEN_RL_CLIENT_LOG_ROOTS": f"/nonexistent, {root}"})) - newest = self.get("/api/v1/inspect/logs?run_id=r1&pod=client&limit=1", **{"model_meta:r1": meta()}) + newest = self.get("/api/v1/inspect/logs?run_id=r1&pod=client&limit=1", **{"model_meta:r1": meta(log_path=run_dir)}) older = self.get(f"/api/v1/inspect/logs?run_id=r1&pod=client&limit=1&cursor={newest['next_cursor']}") self.assertEqual(newest["source"], os.path.join(run_dir, "logs.log")) self.assertEqual([r["message"] for r in newest["records"]], ["train:20 [ERROR] step 2 failed\nTraceback (most recent call last)"]) self.assertEqual(([r["message"] for r in older["records"]], older["next_cursor"]), (["train:10 [INFO] step 1"], None)) + def test_a_client_log_outside_the_roots_is_refused(self): + base = os.path.realpath(self.enterContext(tempfile.TemporaryDirectory())) + root, outside = os.path.join(base, "root"), os.path.join(base, "outside") + os.makedirs(root), os.makedirs(outside) + with open(os.path.join(outside, "logs.log"), "w") as f: + f.write("train:10 [INFO] secret\n") + os.symlink(outside, os.path.join(root, "link")) + self.enterContext(mock.patch.dict(os.environ, {"OPEN_RL_CLIENT_LOG_ROOTS": root})) + cases = { + "undeclared": (None, "This run declared no openrl.log_path"), + "dotdot": (os.path.join(root, "..", "outside"), "is not under OPEN_RL_CLIENT_LOG_ROOTS"), + "symlink": (os.path.join(root, "link"), "is not under OPEN_RL_CLIENT_LOG_ROOTS"), + } + for run_id, (log_path, error) in cases.items(): + body = self.get(f"/api/v1/inspect/logs?run_id={run_id}&pod=client", **{f"model_meta:{run_id}": meta(log_path=log_path)}) + self.assertEqual((body["available"], body["records"]), (False, []), run_id) + self.assertIn(error, body["error"], run_id) + os.environ["OPEN_RL_CLIENT_LOG_ROOTS"] = "" + body = self.get("/api/v1/inspect/logs?run_id=r1&pod=client", **{"model_meta:r1": meta(log_path=os.path.join(root, "run"))}) + self.assertIn("Set OPEN_RL_CLIENT_LOG_ROOTS", body["error"]) + if __name__ == "__main__": unittest.main() From 6d10bbd38ccab4272d1fff505b732cbeee268d29 Mon Sep 17 00:00:00 2001 From: ShubyM Date: Thu, 29 Oct 2026 15:24:01 -0700 Subject: [PATCH 4/4] feat(observability): read client logs from the log store by a client id OPEN_RL_LOG_STORE is now the one log setting. cloud-logging reads the project that the Google credentials name. cloud-logging: reads that Cloud Logging resource instead, such as a log bucket view. OPEN_RL_LOG_VIEWS, OPEN_RL_GKE_PROJECT, _CLUSTER and _LOCATION and the metadata server lookup are gone. The worker filter keeps the namespace and the anchored pod name prefix. It no longer pins the project, cluster and location, because owner ids are unique. Client logs now come from the same store. A client sets the session tag openrl.client_log_id=, which is stored on the model. Its log lines carry the label openrl_client=. For a client pod that is a pod label, which GKE copies to k8s-pod/openrl_client. pod=client reads the lines with either label. openrl.log_path, OPEN_RL_CLIENT_LOG_ROOTS and the file reader are gone. A store now takes a selector: a namespace's worker pods by name prefix, or a client id. --- docs/configuration.md | 45 +++------- src/server/api_server.py | 8 +- src/server/model_metadata.py | 2 +- src/server/observability/client_runs.py | 55 ------------ src/server/observability/cloud_logging.py | 90 ++++++++----------- src/server/observability/log_selector.py | 22 +++++ src/server/observability/router.py | 38 ++++---- tests/test_api_server_paths.py | 6 +- tests/test_observability.py | 102 +++++++++++----------- 9 files changed, 139 insertions(+), 229 deletions(-) delete mode 100644 src/server/observability/client_runs.py create mode 100644 src/server/observability/log_selector.py diff --git a/docs/configuration.md b/docs/configuration.md index 017ef3865..4abed401f 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -126,44 +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. It reads pods, nodes and `Workload`s from -Kubernetes. - -Each run has a `lifecycle`. It is `ended`, `failed`, `waiting`, `placed` or -`active`, from the run's `Workload`s. Without `Workload`s it is `active` while a -client session is live. Each run also lists its `pods`, by name with a -`problem` when one is not simply running, and the `nodes` they are on. Runs -that share a runtime show its pods only while they are active. `cluster` lists -the nodes with their GPUs and the pods with problems. A source that cannot be -read adds a line to `errors`. - -`/api/v1/inspect/logs?run_id=` returns a page of the run's worker logs, -newest first, from the log store that `OPEN_RL_LOG_STORE` names. Without it, -log reads are off. It reads only the pods in the worker namespace whose names -start with the run's prefix, `orw---`. `pod` narrows it to one of -them. `since` and `until` take ISO-8601 or Unix seconds, and `since` also takes -a span such as `1h`. The window defaults to the last 30 minutes and is at most -seven days. Pass `next_cursor` back as `cursor` for the next older page. -Cloud Logging is the only store today. Other stores, such as Loki, plug in -later as adapters. On a cluster outside GKE, logs need a shipper into a -supported store. - -With `pod=client`, it reads `logs.log` from the directory that the run's -`openrl.log_path` setting names. Whatever launches the client sets it, as a -`TINKER_TAGS` entry or in `user_metadata`, to where the client writes its -files, such as the `tinker-cookbook` `log_path` on a volume the server also -mounts. The server reads the file only when its resolved path is under one of -`OPEN_RL_CLIENT_LOG_ROOTS`. +`/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_WORKER_NAMESPACE` | `POD_NAMESPACE`, else `openrl-system` | The namespace whose pods, `Workload`s and logs it reads. | -| `OPEN_RL_LOG_STORE` | unset | Where it reads worker logs. Unset turns log reads off. `cloud-logging` reads Cloud Logging with the server's Google application default credentials. | -| `OPEN_RL_GKE_PROJECT`, `OPEN_RL_GKE_CLUSTER`, `OPEN_RL_GKE_LOCATION` | the GKE metadata server | With `cloud-logging`, the cluster whose logs it reads. Outside GKE, set all three. | -| `OPEN_RL_LOG_VIEWS` | `projects/` | With `cloud-logging`, comma-separated Cloud Logging resources to read, such as a log bucket view. | -| `OPEN_RL_CLIENT_LOG_ROOTS` | unset | Comma-separated directories that a run's `openrl.log_path` must be under. Unset, it reads no client logs. | +| `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/src/server/api_server.py b/src/server/api_server.py index be82f9332..a677e856c 100644 --- a/src/server/api_server.py +++ b/src/server/api_server.py @@ -228,10 +228,10 @@ 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.", ) - log_path: str | None = Field( + client_log_id: str | None = Field( default=None, - description="Where the client writes its files, such as the tinker-cookbook log_path, on a volume the server also mounts. " - "The inspection API serves /logs.log when the server allows that path.", + 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( @@ -533,7 +533,7 @@ async def _extract_and_persist_model_metadata( trainer_gpus=settings.trainer_gpus, trainer_cp=settings.trainer_cp, recipe=settings.recipe, - log_path=settings.log_path, + 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/model_metadata.py b/src/server/model_metadata.py index 58e001c3f..7e7961d61 100644 --- a/src/server/model_metadata.py +++ b/src/server/model_metadata.py @@ -49,7 +49,7 @@ class TrainingModelMetadata(BaseModel): trainer_gpus: int = 1 trainer_cp: int = 1 recipe: Literal["rl", "sft"] | None = None - log_path: str | 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/observability/client_runs.py b/src/server/observability/client_runs.py deleted file mode 100644 index 29a0b3145..000000000 --- a/src/server/observability/client_runs.py +++ /dev/null @@ -1,55 +0,0 @@ -"""A run's client log: logs.log under the openrl.log_path the client declared, such as the tinker-cookbook's log_path on a -volume the server also mounts. Every read is blocking, so callers run it in a thread.""" - -import os -import re - -SCAN_BYTES = 4 << 20 # of logs.log per page -MESSAGE_CHARS = 16384 -# The cookbook's file format, "%(name)s:%(lineno)d [%(levelname)s] %(message)s"; other lines continue the record above. -HEADER = re.compile(r"\S+:\d+ \[(DEBUG|INFO|WARNING|ERROR|CRITICAL)\] ") - - -def log_file(log_path: str | None) -> str: - """/logs.log, resolved, when it lies under one of OPEN_RL_CLIENT_LOG_ROOTS. The path is the client's word, so - one that leads elsewhere, by .. or a symlink, is refused rather than read.""" - roots = [os.path.realpath(root.strip()) for root in os.getenv("OPEN_RL_CLIENT_LOG_ROOTS", "").split(",") if root.strip()] - if not log_path: - raise LookupError("This run declared no openrl.log_path") - if not roots: - raise LookupError("No client log roots configured. Set OPEN_RL_CLIENT_LOG_ROOTS; see docs/configuration.md.") - path = os.path.realpath(os.path.join(log_path, "logs.log")) - if not any(os.path.commonpath([path, root]) == root for root in roots): - raise LookupError(f"{log_path} is not under OPEN_RL_CLIENT_LOG_ROOTS") - return path - - -def log_page(path: str, before: int | None, limit: int, q: str = "", severity: str | None = None) -> dict: - """logs.log, newest first, in the shape of a Cloud Logging page. Its lines carry no time, so a record is a header line - and the lines under it, such as a traceback, and next_cursor is the byte offset the next older page ends at.""" - with open(path, "rb") as f: - size = f.seek(0, os.SEEK_END) - end = size if before is None else min(before, size) - start = max(0, end - SCAN_BYTES) - f.seek(start) - block = f.read(end - start) - if start and (cut := block.find(b"\n") + 1): # the first line began before the block - block, start = block[cut:], start + cut - lines, at = [], start - for raw in block.splitlines(keepends=True): - lines.append((at, raw.decode("utf-8", "replace").rstrip("\r\n"))) - at += len(raw) - records, under, cursor = [], [], None - for at, text in reversed(lines): - under.append(text) - # The file's first line starts a record, and so does a block's first line when the block has no header at all. - if not (header := HEADER.match(text)) and at and not (at == start and cursor is None): - continue - message, under, cursor = "\n".join(reversed(under)), [], at - level = header[1] if header else "UNKNOWN" - if (not severity or level == severity) and q.lower() in message.lower(): - record = {"id": f"logs.log:{at}", "timestamp": None, "pod": "client", "role": "client", "node": None, "severity": level} - records.append(record | {"message": message[:MESSAGE_CHARS], "message_truncated": len(message) > MESSAGE_CHARS}) - if len(records) == limit: - break - return {"records": records, "next_cursor": str(cursor) if cursor else None} diff --git a/src/server/observability/cloud_logging.py b/src/server/observability/cloud_logging.py index c1180e0be..f339b2ca8 100644 --- a/src/server/observability/cloud_logging.py +++ b/src/server/observability/cloud_logging.py @@ -1,24 +1,22 @@ -"""A run's worker logs from Cloud Logging, read with the server's Google credentials.""" +"""A run's logs from Cloud Logging, read with the server's Google credentials.""" import asyncio import json -import os import re import threading -import time from datetime import UTC, datetime, timedelta import httpx -METADATA = {"project": "project/project-id", "cluster": "instance/attributes/cluster-name", "location": "instance/attributes/cluster-location"} +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 -detected: dict = {} -discovered_at = -60.0 credentials = None +project: str | None = None # the one the credentials name credentials_lock = threading.Lock() @@ -28,39 +26,14 @@ async def send(method: str, url: str, **kwargs) -> httpx.Response: return (await client.request(method, url, **kwargs)).raise_for_status() -def configuration() -> dict: - """OPEN_RL_GKE_PROJECT, _CLUSTER and _LOCATION, else what the metadata server said.""" - fields = {key: os.getenv(f"OPEN_RL_GKE_{key.upper()}") or detected.get(key, "") for key in METADATA} - return {**fields, "configured": all(fields.values())} - - -async def discover() -> None: - """In a cluster, read the GKE identity from the metadata server; retry a failed read after a minute.""" - global detected, discovered_at - if configuration()["configured"] or not os.getenv("KUBERNETES_SERVICE_HOST") or time.monotonic() - discovered_at < 60: - return - discovered_at = time.monotonic() - async with httpx.AsyncClient(timeout=1, trust_env=False, follow_redirects=False) as metadata: - - async def read(path: str) -> str: - response = await metadata.get(f"http://metadata.google.internal/computeMetadata/v1/{path}", headers={"Metadata-Flavor": "Google"}) - if response.status_code != 200 or response.headers.get("Metadata-Flavor") != "Google" or not 0 < len(response.text.strip()) <= 256: - raise ValueError("Invalid metadata response") - return response.text.strip() - - values = await asyncio.gather(*(read(path) for path in METADATA.values()), return_exceptions=True) - if not any(isinstance(value, Exception) for value in values): # never partial: plain GCE also exposes a project - detected = dict(zip(METADATA, values)) - - def access_token() -> str: import google.auth # the cluster extra installs it from google.auth.transport.requests import Request - global credentials + global credentials, project with credentials_lock: if credentials is None: - credentials, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform.read-only"]) + 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 @@ -71,9 +44,9 @@ async def gcp_headers() -> dict[str, str]: # 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 this project. - if isinstance(credentials, google.oauth2.credentials.Credentials) and not credentials.quota_project_id: - headers["x-goog-user-project"] = configuration()["project"] + # 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 @@ -94,18 +67,21 @@ def parse(value: str) -> datetime: return start.isoformat(timespec="microseconds"), end.isoformat(timespec="microseconds") -def logs_filter(namespace: str, pod_prefix: str, start: str, end: str, severity: str | None, q: str, pod: str | None = None) -> str: - """The namespace's pods whose names start with pod_prefix, in the window, or one of them. Never any other pod, so a run - id cannot open another run's logs.""" - config = configuration() - labels = {"project_id": config["project"], "location": config["location"], "cluster_name": config["cluster"], "namespace_name": namespace} - clauses = ['resource.type="k8s_container"', *(f"resource.labels.{label}={json.dumps(value)}" for label, value in labels.items())] - clauses.append(f"resource.labels.pod_name=~{json.dumps('^' + pod_prefix)}") - if pod: - clauses.append(f"resource.labels.pod_name={json.dumps(pod)}") +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/{config['project']}/logs/stderr") + 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)}") @@ -131,22 +107,24 @@ def entry_record(entry: dict) -> dict: 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") or "unknown" + 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(namespace: str, pod_prefix: str, since: str, until: str, *, cursor=None, limit=200, q="", pod=None, severity=None) -> dict: - """A page of logs, newest first. The cursor is Cloud Logging's own page token, valid for the same filters.""" - await discover() - if not (config := configuration())["configured"]: - raise LookupError("Cloud Logging needs the GKE metadata server or OPEN_RL_GKE_PROJECT, _CLUSTER and _LOCATION") - views = [v.strip() for v in os.getenv("OPEN_RL_LOG_VIEWS", "").split(",") if v.strip()] or [f"projects/{config['project']}"] - filters = logs_filter(namespace, pod_prefix, since, until, severity, q, pod) - query = {"resourceNames": views, "filter": filters, "orderBy": "timestamp desc", "pageSize": limit} +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=await gcp_headers())).json() + 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] diff --git a/src/server/observability/log_selector.py b/src/server/observability/log_selector.py new file mode 100644 index 000000000..5be3dd462 --- /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 index f3d2d6efa..0fa209345 100644 --- a/src/server/observability/router.py +++ b/src/server/observability/router.py @@ -1,4 +1,3 @@ -import asyncio import os from typing import Annotated @@ -6,8 +5,9 @@ from fastapi import APIRouter, HTTPException, Query from server.model_metadata import decode_model_metadata -from server.observability import client_runs, cloud_logging, fleet +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 @@ -15,7 +15,7 @@ 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 picks one. A store reads the namespace's pods whose names start with the prefix, or the one pod named. +# 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} @@ -31,7 +31,7 @@ async def inventory(): @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 its client's logs.log")] = None, + 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, @@ -39,8 +39,7 @@ async def logs( 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: the run's worker pods from the OPEN_RL_LOG_STORE log store, or with pod=client the logs.log under its - openrl.log_path. That file's lines carry no time, so since and until do not apply to it and its cursor is a byte offset.""" + """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: @@ -49,27 +48,22 @@ async def logs( 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 pod == "client": - try: - result["source"] = path = await asyncio.to_thread(client_runs.log_file, meta.log_path) - except LookupError as exc: - return {**result, "source": "client", "available": False, "error": str(exc)} - if cursor and not cursor.isdigit(): - raise HTTPException(400, CURSOR_EXPIRED) - try: - page = await asyncio.to_thread(client_runs.log_page, path, int(cursor) if cursor else None, limit, q, severity) - return {**result, **page, "available": True} - except OSError as exc: - return {**result, "available": False, "error": f"{path} unreadable: {type(exc).__name__}"} 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(store)) is None: + 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)}"} - # 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))}-" + 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(namespace(), prefix, since, until, cursor=cursor, limit=limit, q=q, pod=pod, severity=severity) + 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: diff --git a/tests/test_api_server_paths.py b/tests/test_api_server_paths.py index 6e54f7645..908a9eb93 100644 --- a/tests/test_api_server_paths.py +++ b/tests/test_api_server_paths.py @@ -370,10 +370,10 @@ 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_log_path_tag_lands_on_the_model(self) -> None: - session_id = self.post("create_session", {"tags": ["openrl.log_path=/mnt/runs/gsm8k"]}).json()["session_id"] + 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["log_path"], "/mnt/runs/gsm8k") + 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"]) diff --git a/tests/test_observability.py b/tests/test_observability.py index 43c366ca4..93130e385 100644 --- a/tests/test_observability.py +++ b/tests/test_observability.py @@ -1,7 +1,6 @@ import asyncio import copy import os -import tempfile import unittest from unittest import mock @@ -91,83 +90,80 @@ def test_a_finished_run_on_a_shared_runtime_does_not_claim_its_pods(self): self.assertTrue(finished["shared_runtime"] and active["shared_runtime"]) -GKE = {"OPEN_RL_GKE_PROJECT": "p", "OPEN_RL_GKE_CLUSTER": "c", "OPEN_RL_GKE_LOCATION": "l", "OPEN_RL_WORKER_NAMESPACE": "ns"} - - class LogsTest(ApiTest): - def test_logs_read_only_the_runs_pods_and_page_with_the_stores_cursor(self): - sent = [] + def setUp(self): + super().setUp() + self.sent = [] async def send(method: str, url: str, json: dict, headers: dict) -> httpx.Response: - sent.append(json) + 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, GKE | {"OPEN_RL_LOG_STORE": "cloud-logging"})) + 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, "gcp_headers", mock.AsyncMock(return_value={}))) + 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 sent], [None, "t2"]) - clauses = sent[0]["filter"].split(" AND ") + 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", sent[0]["filter"]) + self.assertNotIn("owner", self.sent[0]["filter"]) - def test_logs_stay_off_without_a_known_log_store(self): - self.enterContext(mock.patch.dict(os.environ, GKE | {"KUBERNETES_SERVICE_HOST": "10.0.0.1"})) - os.environ.pop("OPEN_RL_LOG_STORE", None) - network = self.enterContext(mock.patch.object(httpx.AsyncClient, "send", side_effect=AssertionError("no log read"))) + 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")) - network.assert_not_called() - - def test_the_client_log_is_read_from_its_declared_path_and_paged_newest_first(self): - root = os.path.realpath(self.enterContext(tempfile.TemporaryDirectory())) - run_dir = os.path.join(root, "runs", "fft_gsm8k_rl") - os.makedirs(run_dir) - with open(os.path.join(run_dir, "logs.log"), "w") as f: - f.write("train:10 [INFO] step 1\ntrain:20 [ERROR] step 2 failed\nTraceback (most recent call last)\n") - self.enterContext(mock.patch.dict(os.environ, {"OPEN_RL_CLIENT_LOG_ROOTS": f"/nonexistent, {root}"})) - - newest = self.get("/api/v1/inspect/logs?run_id=r1&pod=client&limit=1", **{"model_meta:r1": meta(log_path=run_dir)}) - older = self.get(f"/api/v1/inspect/logs?run_id=r1&pod=client&limit=1&cursor={newest['next_cursor']}") - - self.assertEqual(newest["source"], os.path.join(run_dir, "logs.log")) - self.assertEqual([r["message"] for r in newest["records"]], ["train:20 [ERROR] step 2 failed\nTraceback (most recent call last)"]) - self.assertEqual(([r["message"] for r in older["records"]], older["next_cursor"]), (["train:10 [INFO] step 1"], None)) - - def test_a_client_log_outside_the_roots_is_refused(self): - base = os.path.realpath(self.enterContext(tempfile.TemporaryDirectory())) - root, outside = os.path.join(base, "root"), os.path.join(base, "outside") - os.makedirs(root), os.makedirs(outside) - with open(os.path.join(outside, "logs.log"), "w") as f: - f.write("train:10 [INFO] secret\n") - os.symlink(outside, os.path.join(root, "link")) - self.enterContext(mock.patch.dict(os.environ, {"OPEN_RL_CLIENT_LOG_ROOTS": root})) - cases = { - "undeclared": (None, "This run declared no openrl.log_path"), - "dotdot": (os.path.join(root, "..", "outside"), "is not under OPEN_RL_CLIENT_LOG_ROOTS"), - "symlink": (os.path.join(root, "link"), "is not under OPEN_RL_CLIENT_LOG_ROOTS"), - } - for run_id, (log_path, error) in cases.items(): - body = self.get(f"/api/v1/inspect/logs?run_id={run_id}&pod=client", **{f"model_meta:{run_id}": meta(log_path=log_path)}) - self.assertEqual((body["available"], body["records"]), (False, []), run_id) - self.assertIn(error, body["error"], run_id) - os.environ["OPEN_RL_CLIENT_LOG_ROOTS"] = "" - body = self.get("/api/v1/inspect/logs?run_id=r1&pod=client", **{"model_meta:r1": meta(log_path=os.path.join(root, "run"))}) - self.assertIn("Set OPEN_RL_CLIENT_LOG_ROOTS", body["error"]) + 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__":