nighthawk-ingress: record router CPU and memory per stage

benchmarking/nighthawk-ingress/Dockerfile

@@ -96,6 +96,11 @@ COPY benchmarking/nighthawk-ingress/runner.py \
      benchmarking/nighthawk-ingress/spec.py \
      benchmarking/nighthawk-ingress/output.py \
      benchmarking/nighthawk-ingress/actors.py \
+     benchmarking/nighthawk-ingress/kubeapi.py \
+     benchmarking/nighthawk-ingress/resources_common.py \
+     benchmarking/nighthawk-ingress/resource_sampler.py \
+     benchmarking/nighthawk-ingress/envoy_stats.py \
+     benchmarking/nighthawk-ingress/fetch_run_resources.py \
      /app/
 COPY benchmarking/nighthawk-ingress/tests /app/tests
 

benchmarking/nighthawk-ingress/README.md

@@ -141,7 +141,7 @@ Then each benchmark run is one command, ~8–10 min:
    `capacity.json`.
 
 Flags: `--envoy-cpu`, `--actors`, `--tail-latency-slo-ms`, `--atespace`,
-`--dest` (see
+`--dest`, `--sample-resources` (see
 `--help`). It needs a running Docker daemon and gcloud/kubectl credentials.
 
 Notes:
@@ -167,6 +167,147 @@ Start with `capacity.json`, drill into `stats.jsonl`:
 | `results.json` | full session, 1:1 with Nighthawk's output |
 | `spec.textproto`, `output.textproto` | exact Nighthawk input/output, for reproducibility |
 | `status.json`, `logs.txt` | run metadata + process logs |
+| `resources.jsonl`, `resources-by-stage.json`, `envoy-stats.jsonl` | only with `--sample-resources`; see "Resource usage" below |
+
+## Resource usage
+
+Three tools put CPU and memory next to `stats.jsonl`. All three write the
+`stats.jsonl` envelope (`timestamp`, `tag`, `test_name`, `metric`,
+`measurements`) and need only the Python standard library. Their tests are
+`unittest` cases: `python3 -m unittest discover -s tests` runs them, but
+silently skips the directory's older pytest-style `test_output.py` and
+`test_spec.py`. `python3 -m pytest tests` runs all of them.
+
+A sample covers the window `(timestamp - window_s, timestamp]` and counts
+toward every stage that window overlaps. `resources-by-stage.json` gives
+mean, max, p95 and `n` per stage and container. For router containers
+it adds `cpu_fraction_of_pin`, which is `cpu_cores / --envoy-cpu`.
+
+`null` means no sample reached that stage and container. A missing source
+reads as `null`, never as zero. `missing` lists containers that produced
+no sample at all.
+
+### During a run: `resource_sampler.py` and `--sample-resources`
+
+This is the only source of router CPU and memory on the benchmark
+cluster today. `runner.py --sample-resources [--sample-interval 2]` polls
+each relevant node's kubelet cAdvisor endpoint through the API server,
+`/api/v1/nodes/<node>/proxy/metrics/cadvisor`, and writes
+`resources.jsonl` and `resources-by-stage.json` to the work dir, which
+the upload step copies next to `stats.jsonl`. It samples:
+
+- router: `ate-system`, `app=atenet-router`, containers `envoy` and
+  `atenet-router`;
+- runner: `benchmarking`, `app=substrate-benchmark-runner`, container
+  `runner`, narrowed to the runner's own pod via `HOSTNAME`;
+- workers: `benchmark-workloads`, `ate.dev/worker-pool`, container `ateom`.
+
+Per container it records `cpu_cores`, the rate of
+`cpu_usage_seconds_total` between two cAdvisor readings,
+`cfs_throttled_ratio`, `working_set_bytes` and `rss_bytes`. Each target
+pod also gets a `POD` row with pod network and PSI. The flag is off by
+default. Standalone, with the caller's kubeconfig:
+
+```
+python3 resource_sampler.py --duration 60 --interval 2 --out /tmp/rs \
+  [--stats stats.jsonl] [--envoy-cpu 2] [--context <ctx>] \
+  [--target component=namespace:selector[:c1,c2]]
+```
+
+The runner's ServiceAccount needs `nodes/proxy` get cluster-wide and
+`pods` list in `ate-system`, `benchmarking` and `benchmark-workloads`.
+`resource-sampler-rbac.yaml` grants both; apply it after those namespaces
+exist. The orchestrator does not apply it. `nodes/proxy` reaches every
+kubelet endpoint, including exec, so grant it only on
+benchmark clusters.
+
+Limits:
+
+- The sampler keeps the series `benchmarking/monitoring.yaml` keeps. Like
+  that config, it drops `cpu_usage_seconds_total` and
+  `memory_working_set_bytes` on the pod cgroup row, `container=""`. That
+  row is the sum of the containers, and keeping both double counts.
+- cAdvisor refreshes each container every 10 to 20 s, whatever the poll
+  interval. A poll that finds no new reading writes no value for it, and
+  a rate's `window_s` is the real gap between cAdvisor readings. Expect
+  roughly one CPU point per 10 s stage.
+- CFS throttling appears only for containers with a CPU limit. A
+  BestEffort router has no `cfs_throttled_ratio`.
+- Actors run inside the worker's `ateom` container, so a worker row
+  covers every actor on that worker.
+
+### After a run: `fetch_run_resources.py`
+
+```
+SSL_CERT_FILE=/etc/ssl/cert.pem python3 fetch_run_resources.py \
+  gs://<bucket>/runs/<name>/run_date=.../run_ts=.../run_tag=<sha> \
+  --project <project> --cluster <cluster> --envoy-cpu 2 --out /tmp/res
+```
+
+`--project` and `--cluster` default to `PROJECT_ID` and `CLUSTER_NAME`
+from the environment, so `source .ate-dev-env.sh` covers both. The
+tool reads `stats.jsonl` and queries Managed Prometheus for the run's
+time range. By default it fetches what the benchmark cluster exports today:
+
+- `ate.actor.stats.*` from the run's atelet pods, matched by the commit
+  sha in the tag or named with `--atelet`: actor CPU,
+  working set and sampled actor count, summed per atelet pod;
+- router request rate from `atenet.router.route.duration`, for the
+  `atenet-router-*` pod on the same nodes as those atelet pods.
+
+Two Substrate installs can report under one cluster label, so the tool
+picks the run's pods by name, never by cluster alone. `--system-metrics`
+adds `kubernetes.io/container/*` queries for router, runner and worker
+containers. They return nothing until the cluster exports GKE system
+metrics.
+`queries` in `resources-by-stage.json` records every query with its
+series count or error.
+
+The caller needs `roles/monitoring.viewer` on the project and
+`storage.objects.get` on the bucket. The token comes from `--token-file`,
+`GOOGLE_OAUTH_ACCESS_TOKEN`, or
+`gcloud auth application-default print-access-token`. The python.org
+build of Python ships no CA bundle, hence `SSL_CERT_FILE`.
+
+Limits:
+
+- Points are 60 s rates, and stages last about 10 s. One point spans
+  several stages, so per-stage values are smeared and `n` is often 1
+  or 2.
+- Monitoring keeps data for six weeks.
+- Actor stats are per atelet pod, the sum of the actors it hosts, not
+  per actor.
+
+### Envoy admin stats: `envoy_stats.py`
+
+With `--sample-resources`, the runner also snapshots the router's Envoy
+`/stats` before and after the session into `envoy-stats.jsonl`. It keeps
+`server.memory_*`, `server.worker_*`, `cluster.*.upstream_cx_total`,
+`http.*.downstream_rq_*` and `listener.*.downstream_cx_*`. Subtract
+`before` from `after` for session counters.
+
+```
+python3 envoy_stats.py --via proxy --label now --out envoy-stats.jsonl [--context <ctx>]
+```
+
+The admin port is 9901 on the pod IP. The `atenet-router` Service does
+not expose it, so a Service address cannot reach it.
+
+- `--via direct` is the runner default. It reads pod IPs from the
+  `atenet-router` EndpointSlices, which Role `atelet-endpointslices`
+  already grants the runner. It then GETs `http://<pod IP>:9901/stats`.
+  The runner Job runs in the benchmark cluster, so this needs no new
+  RBAC. It works only while no NetworkPolicy blocks `ate-system` ingress.
+- `--via proxy` goes through the API server:
+  `/api/v1/namespaces/ate-system/pods/<pod>:9901/proxy/stats`. It needs
+  `pods` list and `pods/proxy` get in `ate-system` and works off-cluster.
+
+A Prow job runs outside the benchmark cluster and cannot reach pod IPs.
+It would need `--via proxy` with those two permissions for its
+credentials, or a `9901` port on a Service it can reach.
+
+`stats` is `null`, with `error` set, when a pod's admin port did not
+answer.
 
 ## Tuning
 

benchmarking/nighthawk-ingress/envoy_stats.py

@@ -0,0 +1,198 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Snapshot the atenet-router Envoy admin /stats into envoy-stats.jsonl.
+
+Keeps server.memory_*, server.worker_*, cluster.*.upstream_cx_total,
+http.*.downstream_rq_* and listener.*.downstream_cx_*. Each call appends
+one record per router pod, labeled (for example before or after).
+
+Reaching the admin port (9901, bound on the pod IP; the atenet-router
+Service does not expose it):
+  --via direct  GET http://<pod IP>:9901/stats. Pod IPs come from the
+                atenet-router EndpointSlices, which the runner's
+                ServiceAccount can already read. Works from inside the
+                cluster only.
+  --via proxy   GET /api/v1/namespaces/ate-system/pods/<pod>:9901/proxy/stats
+                through the API server. Needs pods list and pods/proxy get
+                in ate-system; works off-cluster with kubectl.
+
+  python3 envoy_stats.py --via proxy --label before --out envoy-stats.jsonl
+"""
+
+import argparse
+import json
+import re
+import sys
+import urllib.parse
+import urllib.request
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Callable
+
+import kubeapi
+import resources_common as rc
+
+ADMIN_PORT = 9901
+NAMESPACE = "ate-system"
+SERVICE = "atenet-router"
+POD_SELECTOR = "app=atenet-router"
+
+KEEP = re.compile(
+    r"server\.memory_.*"
+    r"|server\.worker_.*"
+    r"|cluster\.[^:]+\.upstream_cx_total"
+    r"|http\.[^:]+\.downstream_rq_.*"
+    r"|listener\.[^:]+\.downstream_cx_.*"
+)
+
+
+def parse_stats(text: str) -> dict[str, float]:
+    """Kept counters and gauges from Envoy's text /stats. Histograms, whose
+    values are not numbers, are skipped."""
+    out = {}
+    for line in text.splitlines():
+        name, sep, value = line.partition(": ")
+        if not sep or not KEEP.fullmatch(name):
+            continue
+        try:
+            out[name] = float(value)
+        except ValueError:
+            continue
+    return out
+
+
+def router_endpoints(client: kubeapi.KubeClient) -> list[tuple[str, str]]:
+    """(pod name, address) for each atenet-router endpoint."""
+    q = urllib.parse.urlencode({"labelSelector": f"kubernetes.io/service-name={SERVICE}"})
+    body = client.get_json(f"/apis/discovery.k8s.io/v1/namespaces/{NAMESPACE}/endpointslices?{q}")
+    seen = {}
+    for sl in body.get("items", []):
+        for ep in sl.get("endpoints") or []:
+            addrs = ep.get("addresses") or []
+            pod = (ep.get("targetRef") or {}).get("name", "")
+            if addrs and pod not in seen:
+                seen[pod] = addrs[0]
+    return sorted(seen.items())
+
+
+def http_get(url: str, timeout_s: float = 10) -> str:
+    with urllib.request.urlopen(url, timeout=timeout_s) as r:
+        return r.read().decode()
+
+
+def admin_url(addr: str, port: int = ADMIN_PORT) -> str:
+    host = f"[{addr}]" if ":" in addr else addr
+    return f"http://{host}:{port}/stats"
+
+
+def snapshot(
+    client: kubeapi.KubeClient,
+    via: str,
+    get: Callable[[str], str] = http_get,
+    port: int = ADMIN_PORT,
+) -> list[tuple[str, dict[str, float] | None, str]]:
+    """(pod, stats or None, error) per router pod."""
+    out = []
+    if via == "direct":
+        targets = [(pod, admin_url(addr, port)) for pod, addr in router_endpoints(client)]
+        fetch = get
+    else:
+        pods = client.list_pods(NAMESPACE, POD_SELECTOR)
+        targets = [
+            (p["metadata"]["name"], f"/api/v1/namespaces/{NAMESPACE}/pods/{p['metadata']['name']}:{port}/proxy/stats")
+            for p in pods
+            if p.get("status", {}).get("phase") == "Running"
+        ]
+        fetch = client.get_raw
+    for pod, url in targets:
+        try:
+            out.append((pod, parse_stats(fetch(url)), ""))
+        except Exception as e:
+            out.append((pod, None, str(e)))
+    return out
+
+
+def records(
+    snaps: list[tuple[str, dict[str, float] | None, str]],
+    label: str,
+    via: str,
+    tag: str,
+    test_name: str,
+    now: datetime,
+) -> list[dict]:
+    """stats.jsonl envelope; stats null when the admin port did not answer."""
+    return [
+        {
+            "timestamp": rc.format_time(now),
+            "tag": tag,
+            "test_name": test_name,
+            "metric": f"envoy-stats/{label}",
+            "measurements": {
+                "pod": pod,
+                "label": label,
+                "via": via,
+                "error": err or None,
+                "stats": stats,
+            },
+        }
+        for pod, stats, err in snaps
+    ]
+
+
+def append(
+    client: kubeapi.KubeClient,
+    out: Path,
+    label: str,
+    via: str = "direct",
+    tag: str = "",
+    test_name: str = "",
+    get: Callable[[str], str] = http_get,
+    now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
+) -> list[dict]:
+    recs = records(snapshot(client, via, get), label, via, tag, test_name, now())
+    out.parent.mkdir(parents=True, exist_ok=True)
+    with open(out, "a") as f:
+        for r in recs:
+            f.write(json.dumps(r) + "\n")
+    return recs
+
+
+def main(argv: list[str] | None = None) -> int:
+    p = argparse.ArgumentParser(
+        prog="envoy_stats",
+        description=__doc__,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    p.add_argument("--out", required=True, help="envoy-stats.jsonl to append to")
+    p.add_argument("--label", required=True, help="Snapshot label, e.g. before or after")
+    p.add_argument("--via", choices=("direct", "proxy"), default="direct")
+    p.add_argument("--context", help="kubectl context (default: in-cluster, else current)")
+    p.add_argument("--tag", default="")
+    p.add_argument("--test-name", default="")
+    args = p.parse_args(argv)
+    recs = append(
+        kubeapi.default_client(args.context), Path(args.out), args.label, args.via, args.tag, args.test_name
+    )
+    for r in recs:
+        m = r["measurements"]
+        n = len(m["stats"]) if m["stats"] is not None else 0
+        print(f"{m['pod']}: {n} stats" + (f", error: {m['error']}" if m["error"] else ""))
+    if not recs:
+        print("no atenet-router pods found", file=sys.stderr)
+    return 0 if recs and all(r["measurements"]["stats"] is not None for r in recs) else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())