benchmarking: add OOM kills to server telemetry

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

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

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}

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

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,