From d2b8e73cac33cd0eec13c0fcc312affa30720f7d Mon Sep 17 00:00:00 2001 From: Nishanth Kotla Date: Thu, 8 Oct 2026 21:45:30 +0000 Subject: [PATCH] benchmarking: add OOM kills to server telemetry --- benchmarking/README.md | 26 +++- .../automation/manifests/runner-job.yaml.tmpl | 4 + benchmarking/locust/cluster_facts.py | 106 ++++++++++++++- benchmarking/locust/runner.py | 38 +++++- benchmarking/locust/server_telemetry.py | 69 +++++++++- .../locust/unit_tests/test_cluster_facts.py | 123 +++++++++++++++++- .../unit_tests/test_server_telemetry.py | 65 ++++++++- benchmarking/monitoring.yaml | 4 +- 8 files changed, 419 insertions(+), 16 deletions(-) diff --git a/benchmarking/README.md b/benchmarking/README.md index e8cfcd135a..9b6ef90939 100644 --- a/benchmarking/README.md +++ b/benchmarking/README.md @@ -118,7 +118,8 @@ Four flags control the optional post-run measurements described in [Benchmark output files](#benchmark-output-files): * `--cluster-facts` / `--no-cluster-facts`: read node capacity and worker pod - count from the Kubernetes API once the run ends, to derive density frontiers. + count from the Kubernetes API once the run ends, to derive density frontiers, + and worker pod restarts and evictions around the run, for `oom_events`. On by default. Pass `--no-cluster-facts` to skip Kubernetes API discovery. * `--prometheus-url`: the Prometheus to harvest server-side telemetry from. Defaults to the in-cluster service installed by @@ -720,6 +721,29 @@ actually did, independent of what the load generator reported. `--actor-stats-poll-interval` (default 1m), so these move in steps of a minute or more. `actor` matches one worker pod's actors only when the node runs one worker pod. +* `oom_events.node`, `pod`: OOM kills over the run, from cAdvisor's + `container_oom_events_total`. `node` counts only global OOMs. `pod` counts + kills inside a worker pod (actor sandboxes and the worker container) and is + best effort: a non-zero value is real, but a kill whose cgroup is gone before + the scrape is missed. A node-allocatable OOM (`oom_memcg=/kubepods.slice`) + isn't global, so it isn't in `node`, and `pod` usually misses it; the kubelet + evictions that often come with it show in `evicted`. OOMs inside a microVM + guest are invisible and show up as task failures. Memory PSI (`node_psi` / + `pod_psi`) is the early warning. +* `oom_events.worker_restarts`, `worker_oomkilled_pods`, `evicted`, + `worker_pods_lost`: from the Kubernetes API, read before and after the run. + Restarts on a pod deleted during the run are lost. `worker_oomkilled_pods` + counts pods whose last restart was `OOMKilled`, not kills; a group kill of + the whole worker (`memory.oom.group=1`) shows here. `evicted` is kubelet + eviction under node pressure (usually memory), not a kernel OOM kill. It also + counts replacement pods the kubelet refused to start while the node was under + pressure, as they get the same `Evicted` reason. It comes from `Evicted` + events, which last about 1h, so a run longer than that can undercount. + `worker_pods_lost` is pods gone by the end for any reason (eviction, + scale-down, recycling), not an OOM count; a pod created and deleted within + the run isn't counted. A node OOM can also show as a worker `OOMKilled` + restart, so never add the fields. A field is `null` when its Kubernetes read + fails, and all are with `--no-cluster-facts`. Every distribution reports p50, p90, p95 and p99 over the steady-state window. The steady-state window runs from the first to the last Locust sample at 90% or diff --git a/benchmarking/automation/manifests/runner-job.yaml.tmpl b/benchmarking/automation/manifests/runner-job.yaml.tmpl index eb5e4acc5d..5303dcb2fb 100644 --- a/benchmarking/automation/manifests/runner-job.yaml.tmpl +++ b/benchmarking/automation/manifests/runner-job.yaml.tmpl @@ -81,6 +81,10 @@ rules: - apiGroups: [""] resources: ["pods/log"] verbs: ["get"] +# Worker pod events, for the evictions in oom_events (cluster_facts.py). +- apiGroups: [""] + resources: ["events"] + verbs: ["list"] --- # The atelet pod logs hold the other half of the phase breakdown. apiVersion: rbac.authorization.k8s.io/v1 diff --git a/benchmarking/locust/cluster_facts.py b/benchmarking/locust/cluster_facts.py index 8bb03d9a82..1f983565c5 100644 --- a/benchmarking/locust/cluster_facts.py +++ b/benchmarking/locust/cluster_facts.py @@ -19,7 +19,8 @@ API, then derives the actor-density frontiers (actors per node / vCPU / GB RAM and the actors-per-pod percentiles) for a completed trial. Also records the nodes' Kubernetes version, the worker pods' declared capacity and sandbox -class, the run window, and any caller-supplied --metadata. +class, the run window, and any caller-supplied --metadata, and reads worker +pod restarts and evictions around a run, for server telemetry's `oom_events`. """ import argparse @@ -389,3 +390,106 @@ def append_trial_summary( with open(jsonl_path, "a", encoding="utf-8") as f: f.write(json.dumps(summary_entry) + "\n") _log(logs, f"Appended trial_summary to {jsonl_path}") + + +def snapshot_worker_pods( + logs: TextIO | None = None, +) -> dict[str, dict[str, Any]] | None: + """Per worker pod uid: worker restarts, last exit OOMKilled, evicted. + + An evicted pod is usually gone by now; read_evicted_pods covers it. None if + the read failed. + """ + if not _load_kube_config(logs): + return None + try: + # Consistent read, so the snapshot right after the run is exact. + pods = client.CoreV1Api().list_namespaced_pod( + namespace=WORKER_POOL_NAMESPACE, + label_selector=WORKER_POOL_LABEL, + _request_timeout=API_TIMEOUT_SECONDS, + ).items + except Exception as e: + reason = getattr(e, "reason", e) + _log(logs, + f"Notice: could not list pods in {WORKER_POOL_NAMESPACE}: {reason}") + return None + out: dict[str, dict[str, Any]] = {} + for pod in pods: + # By name, so a sidecar can't skew it; a Pending pod may have none yet. + cs = next((c for c in pod.status.container_statuses or [] + if c.name == WORKER_CONTAINER), None) + last = cs.last_state.terminated if cs and cs.last_state else None + out[pod.metadata.uid] = { + "restarts": (cs.restart_count or 0) if cs else 0, + "last_oomkilled": bool(last and last.reason == "OOMKilled"), + "evicted": pod.status.reason == "Evicted", + } + return out + + +def read_evicted_pods( + since_ts: int, + logs: TextIO | None = None, +) -> set[str] | None: + """Uids of pods evicted at or after since_ts, from Evicted events. + + atecontroller deletes an evicted worker pod at once, so the pod is gone by + the next snapshot; its event stays for about 1h. Events can't be filtered by + label, so this covers every pod in the namespace, where only workers run. + None if the read failed. + """ + if not _load_kube_config(logs): + return None + try: + events = client.CoreV1Api().list_namespaced_event( + namespace=WORKER_POOL_NAMESPACE, + field_selector="reason=Evicted,involvedObject.kind=Pod", + _request_timeout=API_TIMEOUT_SECONDS, + ).items + except Exception as e: + reason = getattr(e, "reason", e) + _log(logs, + f"Notice: could not list events in {WORKER_POOL_NAMESPACE}: {reason}") + return None + uids: set[str] = set() + for ev in events: + # Kubelet sets last_timestamp; events.k8s.io writers set event_time. + t = ev.last_timestamp or ev.event_time or ev.metadata.creation_timestamp + if t and t.timestamp() >= since_ts and ev.involved_object.uid: + uids.add(ev.involved_object.uid) + return uids + + +def diff_worker_pods( + before: dict[str, dict[str, Any]] | None, + after: dict[str, dict[str, Any]] | None, + evicted_uids: set[str] | None, +) -> dict[str, int | None]: + """Worker restarts, OOMKilled pods, evictions and lost pods over the run. + + `worker_oomkilled_pods` counts pods, not kills: lastState keeps only the + last exit. A node OOM can also restart a worker. `evicted` is kubelet + eviction or admission rejection, not a kernel OOM kill: pods evicted in the + run, from events and from the snapshots, once per uid. `worker_pods_lost` + is pods gone by the second snapshot, for any reason. A field is None when + its source failed. + """ + restarts: int | None = None + oomkilled: int | None = None + lost: int | None = None + status_evicted: set[str] = set() + if before is not None and after is not None: + grown = oom = 0 + for uid, now in after.items(): + was = before.get(uid, {"restarts": 0, "evicted": False}) + grew = max(now["restarts"] - was["restarts"], 0) + grown += grew + oom += bool(grew and now["last_oomkilled"]) + if now["evicted"] and not was["evicted"]: + status_evicted.add(uid) + restarts, oomkilled = grown, oom + lost = len(before.keys() - after.keys()) + evicted = None if evicted_uids is None else len(evicted_uids | status_evicted) + return {"worker_restarts": restarts, "worker_oomkilled_pods": oomkilled, + "evicted": evicted, "worker_pods_lost": lost} diff --git a/benchmarking/locust/runner.py b/benchmarking/locust/runner.py index 5f94e78dc2..be7503321e 100644 --- a/benchmarking/locust/runner.py +++ b/benchmarking/locust/runner.py @@ -46,7 +46,10 @@ from cluster_facts import ( EMPTY_FACTS, append_trial_summary, + diff_worker_pods, get_cluster_hardware_facts, + read_evicted_pods, + snapshot_worker_pods, ) from common.boomer_config import build_config_json from server_telemetry import extract_and_record_server_telemetry @@ -154,7 +157,8 @@ def parse_args() -> argparse.Namespace: default=True, help=( "Read node capacity and worker pod count from the Kubernetes API " - "after the run to derive density frontiers. Pass " + "after the run to derive density frontiers, and worker pod restarts " + "and evictions around it for oom_events. Pass " "--no-cluster-facts to skip Kubernetes API discovery" ), ) @@ -517,6 +521,32 @@ def collect_cluster_facts( return get_cluster_hardware_facts(logs) +def read_worker_pods( + args: argparse.Namespace, logs: TextIO +) -> dict[str, dict[str, Any]] | None: + """Worker pod state, or None when discovery is off or the read fails.""" + if not args.cluster_facts: + return None + try: + return snapshot_worker_pods(logs) + except Exception as e: + tee(logs, f"Warning: Failed to read worker pods: {e}") + return None + + +def read_evictions( + args: argparse.Namespace, since_ts: int, logs: TextIO +) -> set[str] | None: + """Pods evicted since since_ts, or None when discovery is off or it fails.""" + if not args.cluster_facts: + return None + try: + return read_evicted_pods(since_ts, logs) + except Exception as e: + tee(logs, f"Warning: Failed to read evictions: {e}") + return None + + def main() -> None: args = parse_args() now = datetime.now(timezone.utc) @@ -547,8 +577,13 @@ def main() -> None: traces.write("\t".join(TRACE_COLUMNS) + "\n") traces.flush() log_run_config(args, prefix, work_dir, logs) + # Worker pods around the run, for OOMKilled restarts and evictions. + pods_before = read_worker_pods(args, logs) exit_code = run_test(args, csv_prefix, logs, traces) run_end_ts = int(datetime.now(timezone.utc).timestamp()) + pods_after = read_worker_pods(args, logs) + # Right after the run, as Evicted events expire after about 1h. + evicted_uids = read_evictions(args, run_ts, logs) stats_generated = False if stats_csv.exists(): @@ -618,6 +653,7 @@ def main() -> None: test_name=args.name, logs=logs, lag_s=args.atelet_lag_s, + worker_pods=diff_worker_pods(pods_before, pods_after, evicted_uids), ) except Exception as e: tee(logs, f"Warning: Failed to harvest server telemetry: {e}") diff --git a/benchmarking/locust/server_telemetry.py b/benchmarking/locust/server_telemetry.py index 087edde0a7..ba9d6a90ac 100644 --- a/benchmarking/locust/server_telemetry.py +++ b/benchmarking/locust/server_telemetry.py @@ -16,7 +16,7 @@ Queries Prometheus over [T_start, T_end] and the steady-state window inside it to capture cluster packing, node and pod PSI stalls, snapshot sizes, latencies -and throughput, running actors and memory working set. +and throughput, running actors, memory working set and OOM kills. """ import csv @@ -697,6 +697,60 @@ def summarize(series: dict[Any, dict[int, float]]) -> dict[str, Any]: return out +# A global OOM logs no oom_memcg, so cAdvisor counts it on `/` (container="node"). +OOM_NODE = 'container="node"' +# Kills in a worker pod: nested actor sandbox cgroups, then the worker container's. +OOM_POD = ( + r'pod="", id=~".*/cri-containerd-[^/]+\\.scope/.+"', + 'namespace="benchmark-workloads", container="worker"', +) + + +def _oom_growth_query(selector: str, start_ts: int, window_s: int) -> str: + """OOMs in the window, clamped per series; `or vector(0)` so [] = failed read.""" + s = f"container_oom_events_total{{{selector}}}" + peak = f"max_over_time({s}[{window_s}s])" + return f"sum(clamp_min({peak} - ({s} @ {start_ts} or {peak} * 0), 0)) or vector(0)" + + +# From the Kubernetes API, read in runner.py. +WORKER_POD_FIELDS = ( + "worker_restarts", "worker_oomkilled_pods", "evicted", "worker_pods_lost", +) + + +def _count_oom_events(prom_url: str, start_ts: int, end_ts: int) -> dict[str, Any]: + """OOM kills over the run; a failed read is null, not 0. + + `node` counts global OOMs only. `pod` is best effort: a kill counts only if + its cgroup lasts until the scrape, so it usually misses a node-allocatable + OOM and a group kill of the whole worker, but a non-zero value is real. + """ + window_s = max(end_ts - start_ts, 1) + out: dict[str, Any] = { + "node": None, + "pod": None, + "window": {"start_ts": start_ts, "end_ts": end_ts}, + } + if not query_prometheus_instant( + prom_url, "count(container_oom_events_total)", time_ts=end_ts + ): + return out # Metric not scraped. + + def grown(selector: str) -> int | None: + v = _parse_instant_float(query_prometheus_instant( + prom_url, _oom_growth_query(selector, start_ts, window_s), + time_ts=end_ts, + )) + return None if v is None else int(round(v)) + + out["node"] = grown(OOM_NODE) + # Two reads, as `@` takes a selector, not (A or B); half a read is no read. + pod = [grown(sel) for sel in OOM_POD] + out["pod"] = None if None in pod else sum(pod) + return out + + def harvest_server_telemetry( prom_url: str, start_ts: int, @@ -708,8 +762,8 @@ def harvest_server_telemetry( """Harvests the ground truth metric streams from Prometheus. Only the atelet-exported snapshot, active actor and actor working set - blocks are read late, by `lag_s // 2`; packing, PSI and the cAdvisor - working set aren't shifted and are read over the run window as is. + blocks are read late, by `lag_s // 2`; packing, PSI, the cAdvisor working + set and OOM counts aren't shifted and are read over the run window as is. """ steady_end = end_ts if steady_end_ts is None else steady_end_ts return { @@ -731,6 +785,7 @@ def harvest_server_telemetry( "working_set": _measure_working_set( prom_url, start_ts, end_ts, steady_start_ts, steady_end, lag_s ), + "oom_events": _count_oom_events(prom_url, start_ts, end_ts), } @@ -746,11 +801,13 @@ def extract_and_record_server_telemetry( test_name: str, logs: TextIO | None = None, lag_s: int = 70, + worker_pods: dict[str, int | None] | None = None, ) -> None: """Entry point called by runner.py to query Prometheus and persist artifacts. Waits until `end_ts + lag_s` first, so the atelet's last export of the run has been scraped; lag_s must cover its export interval plus one scrape. + `worker_pods` (WORKER_POD_FIELDS) joins `oom_events`; null when not given. """ def log(msg: str) -> None: if logs: @@ -776,6 +833,9 @@ def log(msg: str) -> None: prom_url, start_ts, end_ts, steady_start, lag_s, steady_end_ts=steady_end, ) + telemetry.setdefault("oom_events", {}).update( + {k: (worker_pods or {}).get(k) for k in WORKER_POD_FIELDS} + ) full_artifact = { "metadata": { @@ -832,6 +892,9 @@ def log(msg: str) -> None: ws = working_set.get(block, {}).get("summary", {}) for p in (*ps, "max"): measurements[f"working_set_{block}_{p}_gb"] = ws.get(p) + oom = telemetry.get("oom_events", {}) + for key in ("node", "pod", *WORKER_POD_FIELDS): + measurements[f"oom_events_{key}"] = oom.get(key) jsonl_row = { "timestamp": data_ts, diff --git a/benchmarking/locust/unit_tests/test_cluster_facts.py b/benchmarking/locust/unit_tests/test_cluster_facts.py index 713748b009..13ad41b08b 100644 --- a/benchmarking/locust/unit_tests/test_cluster_facts.py +++ b/benchmarking/locust/unit_tests/test_cluster_facts.py @@ -27,6 +27,7 @@ import sys import tempfile import unittest +from datetime import datetime, timezone from pathlib import Path from types import SimpleNamespace from unittest import mock @@ -77,13 +78,29 @@ def pod(phase="Running", node_name="node-a", containers=()): status=SimpleNamespace(phase=phase)) -def fake_api(nodes=None, ns_pods=None): +def worker_pod(uid, restarts=0, last=None, evicted=False, statuses=True): + """A worker pod for the snapshot: the worker container's restarts and last exit reason.""" + worker = SimpleNamespace( + name="worker", restart_count=restarts, + last_state=SimpleNamespace( + terminated=SimpleNamespace(reason=last) if last else None)) + sidecar = SimpleNamespace(name="sidecar", restart_count=9, last_state=None) + return SimpleNamespace( + metadata=SimpleNamespace(uid=uid), + status=SimpleNamespace( + container_statuses=[sidecar, worker] if statuses else None, + reason="Evicted" if evicted else None)) + + +def fake_api(nodes=None, ns_pods=None, events=None): """Stand-in CoreV1Api; None raises 403 Forbidden.""" def forbidden(*_args, **_kwargs): raise ApiException(status=403, reason="Forbidden") api = mock.Mock() - for attr, items in (("list_node", nodes), ("list_namespaced_pod", ns_pods)): + lists = (("list_node", nodes), ("list_namespaced_pod", ns_pods), + ("list_namespaced_event", events)) + for attr, items in lists: if items is None: getattr(api, attr).side_effect = forbidden else: @@ -98,6 +115,13 @@ def discover(api): return cluster_facts.get_cluster_hardware_facts() +def snapshot(api): + with mock.patch.object(cluster_facts, "_load_kube_config", return_value=True), \ + mock.patch.object(cluster_facts.client, "CoreV1Api", return_value=api), \ + contextlib.redirect_stdout(io.StringIO()): + return cluster_facts.snapshot_worker_pods() + + def summarize(facts, directory, stats=STATS_HEADER + ",Aggregated,100,25\n", users=10, user_counts=None, actors_per_user=None, run_start=None, run_end=None, metadata=None): @@ -433,6 +457,101 @@ def test_per_rpc_failure_ratios(self): self.assertNotIn("resume_actor_failure_ratio", row["measurements"]) self.assertNotIn("suspend_actor_failure_ratio", row["measurements"]) + def test_worker_pod_snapshot(self): + api = fake_api(ns_pods=[ + worker_pod("a", restarts=2, last="OOMKilled"), + worker_pod("b", restarts=1, last="Error"), + worker_pod("p", statuses=False), # Pending, no statuses yet + worker_pod("e", evicted=True, statuses=False), + ]) + # worker container only: the sidecar's 9 restarts never show. + self.assertEqual(snapshot(api), { + "a": {"restarts": 2, "last_oomkilled": True, "evicted": False}, + "b": {"restarts": 1, "last_oomkilled": False, "evicted": False}, + "p": {"restarts": 0, "last_oomkilled": False, "evicted": False}, + "e": {"restarts": 0, "last_oomkilled": False, "evicted": True}, + }) + api.list_namespaced_pod.assert_called_once_with( + namespace="benchmark-workloads", + label_selector="ate.dev/worker-pool", + _request_timeout=5, + ) + # A denied read or no credentials is unknown, not an empty pool. + self.assertIsNone(snapshot(fake_api(ns_pods=None))) + with mock.patch.object(cluster_facts, "_load_kube_config", return_value=False): + self.assertIsNone(cluster_facts.snapshot_worker_pods()) + + def test_evicted_pods_from_events(self): + def event(uid, last=None, event_time=None, created=None): + def ts(s): + return datetime.fromtimestamp(s, timezone.utc) if s else None + return SimpleNamespace( + involved_object=SimpleNamespace(uid=uid), + last_timestamp=ts(last), event_time=ts(event_time), + metadata=SimpleNamespace(creation_timestamp=ts(created))) + + api = fake_api(events=[ + event("a", last=120), + event("a", last=150), # same pod again: counted once + event("old", last=90), # evicted before the run + event("b", event_time=110), # events.k8s.io: no last_timestamp + event("c", created=130), # only creation_timestamp + ]) + with mock.patch.object(cluster_facts, "_load_kube_config", return_value=True), \ + mock.patch.object(cluster_facts.client, "CoreV1Api", return_value=api), \ + contextlib.redirect_stdout(io.StringIO()): + self.assertEqual(cluster_facts.read_evicted_pods(100), {"a", "b", "c"}) + api.list_namespaced_event.assert_called_once_with( + namespace="benchmark-workloads", + field_selector="reason=Evicted,involvedObject.kind=Pod", + _request_timeout=5, + ) + # A denied read is unknown, not zero evictions. + api.list_namespaced_event.side_effect = ApiException(status=403) + self.assertIsNone(cluster_facts.read_evicted_pods(100)) + + def test_worker_pod_diff(self): + def state(restarts=0, oom=False, evicted=False): + return {"restarts": restarts, "last_oomkilled": oom, "evicted": evicted} + + before = {"a": state(), "b": state(1, oom=True), "e": state(evicted=True), + "gone": state()} # deleted in the run: lost + after = { + "a": state(2, oom=True), # 2 restarts, last OOMKilled: 1 pod + "b": state(1, oom=True), # an OOMKilled from before the run + "e": state(evicted=True), # evicted before the run + "c": state(1, oom=True), # new pod, restarted once + "f": state(evicted=True), # evicted in the run, still listed + } + # "f" is in both sources, "g" was deleted before the second snapshot. + self.assertEqual(cluster_facts.diff_worker_pods(before, after, {"f", "g"}), { + "worker_restarts": 3, "worker_oomkilled_pods": 2, "evicted": 2, + "worker_pods_lost": 1}) + # Each field is null only when its own source failed. + self.assertEqual(cluster_facts.diff_worker_pods(None, after, {"g"}), { + "worker_restarts": None, "worker_oomkilled_pods": None, "evicted": 1, + "worker_pods_lost": None}) + self.assertEqual(cluster_facts.diff_worker_pods(before, after, None), { + "worker_restarts": 3, "worker_oomkilled_pods": 2, "evicted": None, + "worker_pods_lost": 1}) + + def test_worker_pod_reads_follow_cluster_facts_flag(self): + with mock.patch.object(runner, "snapshot_worker_pods") as read, \ + mock.patch.object(runner, "read_evicted_pods") as events: + off = parse("--no-cluster-facts") + self.assertIsNone(runner.read_worker_pods(off, io.StringIO())) + self.assertIsNone(runner.read_evictions(off, 100, io.StringIO())) + read.assert_not_called() + events.assert_not_called() + # A failed read never stops the run. + with mock.patch.object(runner, "snapshot_worker_pods", + side_effect=RuntimeError("boom")), \ + mock.patch.object(runner, "read_evicted_pods", + side_effect=RuntimeError("boom")), \ + contextlib.redirect_stdout(io.StringIO()): + self.assertIsNone(runner.read_worker_pods(parse(), io.StringIO())) + self.assertIsNone(runner.read_evictions(parse(), 100, io.StringIO())) + if __name__ == "__main__": unittest.main() diff --git a/benchmarking/locust/unit_tests/test_server_telemetry.py b/benchmarking/locust/unit_tests/test_server_telemetry.py index 968e6ea1bc..47ed4539d4 100644 --- a/benchmarking/locust/unit_tests/test_server_telemetry.py +++ b/benchmarking/locust/unit_tests/test_server_telemetry.py @@ -108,18 +108,24 @@ def parse(*extra): } -def snapshot_prom(quantile=None, start=None, end=None, at=(100, 105)): +def snapshot_prom(quantile=None, start=None, end=None, at=(100, 105), oom=None): """A query_prometheus_instant fake answering by query and timestamp. Every quantile read at the window end returns `quantile`; `start` and `end` map a SUMS name to its value at the two instants in `at`. A window delta query subtracts its `@` value, or zero when that series is absent. Anything - else, including a read at the wrong time, returns no series. + else, including a read at the wrong time, returns no series. Every OOM + query returns `oom`, or with a dict, the value of the selector it holds. """ by_time = {at[0]: start or {}, at[1]: end or {}} def fake(_url, query, time_ts=None): - if query.startswith("histogram_quantile"): + if "container_oom_events_total" in query: + val = oom + if isinstance(oom, dict): + # The existence check has no selector, so it finds the metric. + val = next((v for k, v in oom.items() if k in query), "1") + elif query.startswith("histogram_quantile"): val = quantile if time_ts == at[1] else None else: name = next((n for n, s in SUMS.items() if s in query), None) @@ -261,7 +267,7 @@ def test_packing_and_checkpoint_math(self, mock_instant, mock_range): end={"count": "150", "size": str(100 * MB), "bytes": str(100 * MB), "seconds": "50", "restore_sum": "4", "restore_count": "8", "checkpoint_count": "25"}, - at=(135, 140)) + at=(135, 140), oom="0") with tempfile.TemporaryDirectory() as td: out_json = Path(td) / "server_summary.json" @@ -279,6 +285,8 @@ def test_packing_and_checkpoint_math(self, mock_instant, mock_range): data_ts="2026-01-01", tag="unit", test_name="unit-run", + worker_pods={"worker_restarts": 1, "worker_oomkilled_pods": 1, + "evicted": 0, "worker_pods_lost": 0}, ) summary = json.loads(out_json.read_text()) row = json.loads(out_jsonl.read_text().splitlines()[0]) @@ -292,7 +300,7 @@ def test_packing_and_checkpoint_math(self, mock_instant, mock_range): self.assertEqual( set(summary), {"metadata", "cluster_packing", "node_psi", "pod_psi", "snapshots", - "active_actors", "working_set"}, + "active_actors", "working_set", "oom_events"}, ) packing = summary["cluster_packing"] self.assertEqual(packing["summary"]["p50"], 0.8) # 3 + 1 busy / 5 workers @@ -312,7 +320,7 @@ def test_packing_and_checkpoint_math(self, mock_instant, mock_range): self.assertEqual(row["metric"], "server_summary") m = row["measurements"] # Every source answered, so a key wired to a wrong name would read None. - self.assertEqual(len(m), 74) + self.assertEqual(len(m), 80) self.assertEqual([k for k, v in m.items() if v is None], []) # Every value a string, so one row's types match every other row's. self.assertTrue(all(isinstance(v, str) for v in m.values())) @@ -478,6 +486,20 @@ def test_steady_window_bounds_packing_and_psi(self, mock_instant, mock_range): self.assertEqual(out["node_psi"]["cpu_stall_pct"]["max"], 0.5) self.assertEqual(out["pod_psi"]["cpu_stall_pct"]["max"], 0.5) + @mock.patch("server_telemetry.query_prometheus_instant") + def test_oom_pod_needs_both_reads(self, mock_instant): + leaf, worker = 'pod=""', 'namespace="benchmark-workloads"' + mock_instant.side_effect = snapshot_prom( + oom={'container="node"': "2", leaf: "3", worker: "1"}) + out = server_telemetry._count_oom_events("http://p", 100, 105) + self.assertEqual(out, {"node": 2, "pod": 4, + "window": {"start_ts": 100, "end_ts": 105}}) + # Half a best-effort read is no read. + mock_instant.side_effect = snapshot_prom( + oom={'container="node"': "2", leaf: None, worker: "1"}) + out = server_telemetry._count_oom_events("http://p", 100, 105) + self.assertEqual((out["node"], out["pod"]), (2, None)) + @mock.patch("server_telemetry.query_prometheus_range") def test_working_set_gib_and_steady_window(self, mock_range): gib = 2**30 @@ -532,6 +554,37 @@ def test_working_set_queries_keep_live_workers(self): self.assertIn('container="node"', q["node"]) self.assertIn("and on (pod)", q["worker"]) + def test_oom_query_shape(self): + q = server_telemetry._oom_growth_query('container="node"', 100, 600) + s = 'container_oom_events_total{container="node"}' + # Clamped per series inside the sum, so one reset can't hide a kill. + self.assertTrue(q.startswith(f"sum(clamp_min(max_over_time({s}[600s]) - ")) + self.assertIn(f"({s} @ 100 or max_over_time({s}[600s]) * 0), 0))", q) + self.assertTrue(q.endswith(" or vector(0)")) + # One backslash pair in PromQL, so the regex reads a literal dot. + self.assertIn(r"\\.scope/", server_telemetry.OOM_POD[0]) + + @mock.patch("server_telemetry.query_prometheus_instant") + def test_oom_metric_missing_is_null(self, mock_instant): + mock_instant.side_effect = snapshot_prom(oom=None) + self.assertEqual(server_telemetry._count_oom_events("http://p", 100, 105), { + "node": None, "pod": None, "window": {"start_ts": 100, "end_ts": 105}}) + + @mock.patch("server_telemetry.harvest_server_telemetry", + return_value={"oom_events": {"node": 0, "pod": 0}}) + def test_worker_pod_fields_null_when_not_read(self, _harvest): + with tempfile.TemporaryDirectory() as td: + out_json = Path(td) / "server_summary.json" + with contextlib.redirect_stdout(io.StringIO()): + server_telemetry.extract_and_record_server_telemetry( + "http://p", 100, 105, Path(td) / "missing.csv", out_json, + Path(td) / "stats.jsonl", "2026-01-01", "unit", "unit-run", + lag_s=0) + oom = json.loads(out_json.read_text())["oom_events"] + self.assertEqual(oom, {"node": 0, "pod": 0, "worker_restarts": None, + "worker_oomkilled_pods": None, "evicted": None, + "worker_pods_lost": None}) + @mock.patch("server_telemetry.query_prometheus_instant") def test_failed_delta_read_is_unknown(self, mock_instant): # A failed delta query is not taken as a series born in the window. diff --git a/benchmarking/monitoring.yaml b/benchmarking/monitoring.yaml index 66c13fdf22..28702ac342 100644 --- a/benchmarking/monitoring.yaml +++ b/benchmarking/monitoring.yaml @@ -188,10 +188,10 @@ data: # cAdvisor sends a very large number of metrics. Keep only the # metrics that the benchmarks read: kernel pressure stall (PSI), # container network bandwidth, disk I/O, memory RSS/working-set, - # CPU utilization, and CFS throttling. + # CPU utilization, CFS throttling, and OOM kill events. - source_labels: [__name__] action: keep - regex: container_((.*pressure.*)|network_(receive|transmit)_bytes_total|fs_(reads|writes)_bytes_total|memory_rss|memory_working_set_bytes|cpu_usage_seconds_total|cpu_cfs_.*|spec_memory_limit_bytes) + regex: container_((.*pressure.*)|network_(receive|transmit)_bytes_total|fs_(reads|writes)_bytes_total|memory_rss|memory_working_set_bytes|cpu_usage_seconds_total|cpu_cfs_.*|spec_memory_limit_bytes|oom_events_total) # Drop the series that have an empty container label only for cumulative # memory working set and CPU usage. cAdvisor gives a rollup for the whole # pod cgroup next to the series for each container; keeping both counts