Support S3 and Azure destinations for benchmark results

benchmarking/automation/README.md

@@ -3,13 +3,19 @@
 Scheduled, repeatable benchmark runs of a substrate branch. A CronJob on an
 **orchestration cluster** drives the full build/deploy/run/teardown cycle
 against a separate **test cluster**, once per entry in [tests.yaml](tests.yaml).
-Each run uploads results to GCS via `benchmarking/locust/runner.py`
+Each run uploads results to object storage via `benchmarking/locust/runner.py`
 (`type: locust`) or `benchmarking/nighthawk-ingress/runner.py` (`type: nighthawk-ingress`,
 the router capacity benchmark — see
 [benchmarking/nighthawk-ingress/README.md](../nighthawk-ingress/README.md)).
 To run the cold resume tests from your workstation instead, see
 [ateapi-cold-resume.md](ateapi-cold-resume.md).
 
+Pass `--dest s3://benchmark-results/substrate` (or a `gs://` or
+`az://container/prefix` URL). See the
+[storage configuration](../../docs/guides/benchmarking.md#headless-automation-only)
+for credentials and local-path limitations. The cluster provisioning steps
+below still describe GKE; result storage does not need to be GCS.
+
 ## How it works
 
 1. CronJob fires on the orchestration cluster; pod starts (orchestrator +
@@ -131,7 +137,8 @@ gcloud container clusters create <ORCH_CLUSTER_NAME> \
 
 ## IAM prerequisites
 
-This setup assumes both clusters and the destination GCS bucket already exist.
+This GKE setup assumes both clusters already exist. Create a result bucket or
+container in your chosen provider; the GCS grants below apply only to GCS destinations.
 Two Workload Identity bindings are needed.
 
 Both ServiceAccounts are created by the manifests (`cronjob.yaml` for the

benchmarking/automation/ateapi-cold-resume.md

@@ -117,14 +117,18 @@ PY
 
 $S/benchmarking/automation/venv/bin/python -u $D/run.py \
   --repo file://$S --branch $(git branch --show-current) \
-  --dest gs://<results-bucket>/<prefix> \
+  --dest s3://<results-bucket>/<prefix> \
   --tests $S/benchmarking/automation/tests.yaml \
   --test-name ateapi_cold_resume_pause_fake_smoke \
   --target-cluster-dir $D/target-clusters \
   --manifests-dir $S/benchmarking/automation/manifests \
   --junit-output $D/junit.xml 2>&1 | tee $D/orchestrator.log
 ```
 
+Configure storage credentials on the runner Job and on your workstation for
+retrieval; see [result storage](../../docs/guides/benchmarking.md#headless-automation-only).
+The example uses S3; GCS and Azure destinations work with the same commands.
+
 Keep your workstation awake for the whole run. If it suspends, the
 orchestrator pauses too, and the teardown and the next test wait for it. For
 long ladders, run the command under `setsid nohup` or in a VM.
@@ -137,8 +141,11 @@ explains the per-step rows and where the run path is in the log.
 lists the files. To copy them all down:
 
 ```bash
-st=$(grep -o 'gs://[^ ]*/status.json' $D/orchestrator.log | head -1)
-mkdir -p $D/results && gcloud storage cp "${st%/status.json}/*" $D/results/
+python3 -m venv "$D/results-venv"
+"$D/results-venv/bin/pip" install -r "$S/benchmarking/requirements.txt"
+st=$(grep -Eo '(gs|s3|az)://[^ ]*/status.json' "$D/orchestrator.log" | head -1)
+"$D/results-venv/bin/python" "$S/benchmarking/result_store.py" download \
+  "${st%/status.json}" "$D/results"
 ```
 
 The capacity is the highest step whose `resume_rps` keeps up with its

benchmarking/automation/manifests/cronjob.yaml.tmpl

@@ -68,7 +68,7 @@ spec:
             - "--branch"
             - "<FILL IN: main>"
             - "--dest"
-            - "<FILL IN: gs://your-bucket/benchmark-results>"
+            - "<FILL IN: s3://your-bucket/benchmark-results>"
             env:
             - name: DOCKER_HOST
               value: tcp://localhost:2375

benchmarking/locust/Dockerfile

@@ -18,8 +18,9 @@
 # Stage 1: Python dependency builder
 FROM python:3.11-slim AS pydeps
 WORKDIR /app
-COPY benchmarking/locust/requirements.txt .
-RUN pip install --no-cache-dir --target=/app/deps -r requirements.txt
+COPY benchmarking/requirements.txt /requirements/requirements.txt
+COPY benchmarking/locust/requirements.txt /requirements/locust/requirements.txt
+RUN pip install --no-cache-dir --target=/app/deps -r /requirements/locust/requirements.txt
 
 # Stage 2: Python gRPC clients, generated from the protos rather than checked
 # in. See codegen/generate.sh.
@@ -64,5 +65,7 @@ COPY benchmarking/locust/ladder_steps.py /app/ladder_steps.py
 COPY benchmarking/locust/turnover_window.py /app/turnover_window.py
 COPY benchmarking/analysis/phase_report.py /app/phase_report.py
 
+COPY benchmarking/result_store.py /app/result_store.py
+
 ENV PYTHONPATH=/app:/app/deps
 ENV PYTHONUNBUFFERED=1

benchmarking/locust/codegen/requirements.txt

@@ -15,7 +15,7 @@
 # Build-time dependencies for generate.sh and for the nighthawk-ingress
 # image's descriptor build. Kept apart from the load test's runtime
 # requirements.txt so that generating the protos does not pull in locust, the
-# opentelemetry exporters and google-cloud-storage.
+# opentelemetry exporters.
 #
 # grpcio-tools bundles protoc along with the Python and gRPC codegen plugins.
 # It is pinned because the generated files stamp the version that produced

benchmarking/locust/requirements.txt

@@ -26,5 +26,5 @@ grpcio>=1.84.1
 opentelemetry-exporter-otlp
 opentelemetry-instrumentation-grpc
 opentelemetry-instrumentation-requests
-google-cloud-storage
+-r ../requirements.txt
 kubernetes

benchmarking/locust/runner.py

@@ -15,7 +15,7 @@
 """Headless locust runner that publishes results as JSONL + CSVs.
 
 Runs locust with the given flags, converts the resulting stats CSV
-to JSONL, and uploads everything to either GCS or local disk under
+to JSONL, and uploads everything to object storage or local disk under
 <dest>/runs/<tag>/<timestamp>/.
 
 When the test target is glutton.py, also spawns the boomer-worker Go
@@ -33,7 +33,6 @@
 import json
 import os
 import re
-import shutil
 import signal
 import subprocess
 import sys
@@ -43,6 +42,10 @@
 from pathlib import Path
 from typing import IO, Any, TextIO
 
+# Also find the shared module when running directly from the source tree.
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+from result_store import upload
+
 from cluster_facts import (
     EMPTY_FACTS,
     append_trial_summary,
@@ -110,7 +113,7 @@ def parse_args() -> argparse.Namespace:
     p.add_argument(
         "--dest",
         required=True,
-        help="Root destination (gs://bucket/path or local path)",
+        help="Root destination (gs://bucket/path, s3://bucket/path, az://container/path, or local path)",
     )
     p.add_argument(
         "--allow-empty-stats",
@@ -494,25 +497,6 @@ def stats_to_jsonl(stats_csv: Path, jsonl_path: Path, timestamp: str, tag: str,
     return rows_written
 
 
-def upload_to_gcs(local_path: Path, gcs_uri: str) -> None:
-    # Imported here so non-GCS use doesn't require google-cloud-storage.
-    from google.cloud import storage
-
-    bucket_name, _, blob_path = gcs_uri[len("gs://"):].partition("/")
-    storage.Client().bucket(bucket_name).blob(blob_path).upload_from_filename(
-        str(local_path)
-    )
-
-
-def upload(src: Path, dest: str) -> None:
-    if dest.startswith("gs://"):
-        upload_to_gcs(src, dest)
-    else:
-        dest_path = Path(dest)
-        dest_path.parent.mkdir(parents=True, exist_ok=True)
-        shutil.copy(src, dest_path)
-
-
 def collect_cluster_facts(
     args: argparse.Namespace, logs: TextIO
 ) -> dict[str, Any]:

benchmarking/locust/unit_tests/test_server_telemetry.py

@@ -785,6 +785,16 @@ def test_telemetry_survives_a_missing_stats_csv(self, _run, _up, telemetry):
         self.assertEqual(telemetry.call_args.kwargs["lag_s"], 70)
         self.assertNotIn("worker_pod_count", telemetry.call_args.kwargs)
 
+    @mock.patch.object(runner, "extract_and_record_server_telemetry")
+    @mock.patch.object(runner, "upload", side_effect=OSError("storage unavailable"))
+    @mock.patch.object(runner, "run_test", return_value=0)
+    def test_upload_failure_is_not_success(self, _run, _upload, _telemetry):
+        with mock.patch.object(sys, "argv", ARGV + ["--no-cluster-facts", "--allow-empty-stats"]), \
+             contextlib.redirect_stdout(io.StringIO()):
+            # An uncaught upload error makes the runner process exit non-zero.
+            with self.assertRaisesRegex(OSError, "storage unavailable"):
+                runner.main()
+
     # Ladder rows follow the telemetry harvest, whose wait covers their
     # last samples, and a failure there still uploads the run.
     @mock.patch.object(runner, "upload")

benchmarking/nighthawk-ingress/Dockerfile

@@ -74,14 +74,15 @@ FROM envoyproxy/nighthawk-dev@${NIGHTHAWK_IMAGE_DIGEST}
 
 # Deps must be installed with this image's own interpreter: grpcio/protobuf
 # native extensions are python-version-specific.
-COPY benchmarking/nighthawk-ingress/requirements.txt /tmp/requirements.txt
+COPY benchmarking/requirements.txt /requirements/requirements.txt
+COPY benchmarking/nighthawk-ingress/requirements.txt /requirements/nighthawk-ingress/requirements.txt
 RUN apt-get update \
     && apt-get install -y --no-install-recommends python3 python3-pip ca-certificates \
     && python3 -m pip install --no-cache-dir --break-system-packages \
-        --target=/app/deps -r /tmp/requirements.txt \
+        --target=/app/deps -r /requirements/nighthawk-ingress/requirements.txt \
     && apt-get purge -y python3-pip \
     && apt-get autoremove -y \
-    && rm -rf /var/lib/apt/lists/* /tmp/requirements.txt
+    && rm -rf /var/lib/apt/lists/* /requirements
 
 WORKDIR /app
 COPY --from=protogen /out/nighthawk.desc /app/nighthawk.desc
@@ -99,6 +100,8 @@ COPY benchmarking/nighthawk-ingress/runner.py \
      /app/
 COPY benchmarking/nighthawk-ingress/tests /app/tests
 
+COPY benchmarking/result_store.py /app/result_store.py
+
 ENV PYTHONPATH=/app:/app/deps
 ENV PYTHONUNBUFFERED=1
 ENV NIGHTHAWK_DESC=/app/nighthawk.desc

benchmarking/nighthawk-ingress/requirements.txt

@@ -18,4 +18,4 @@
 protobuf>=7.35.1
 grpcio>=1.84.1
 requests
-google-cloud-storage
+-r ../requirements.txt

benchmarking/nighthawk-ingress/run-dev.sh

@@ -39,7 +39,7 @@ Usage: $0 [options]
   --actors N                actor fleet size; needs at least one worker Running (default: ${ACTORS})
   --tail-latency-slo-ms N   SLO bound; 0 disables (default: ${TAIL_LATENCY_SLO_MS})
   --atespace NAME           actor namespace (default: ${ATESPACE})
-  --dest gs://...           results root (default: gs://\$BUCKET_NAME/nighthawk-ingress-results)
+  --dest URL                results root (default: gs://\$BUCKET_NAME/nighthawk-ingress-results)
 EOF
   exit "${1:-1}"
 }
@@ -68,10 +68,13 @@ source hack/util/venv.sh
 }
 # shellcheck disable=SC1091
 source .ate-dev-env.sh
-for var in PROJECT_ID CLUSTER_NAME CLUSTER_LOCATION KO_DOCKER_REPO BUCKET_NAME; do
+for var in PROJECT_ID CLUSTER_NAME CLUSTER_LOCATION KO_DOCKER_REPO; do
   [[ -n "${!var:-}" ]] || { echo "ERROR: ${var} not set by .ate-dev-env.sh" >&2; exit 1; }
 done
-DEST="${DEST:-gs://${BUCKET_NAME}/nighthawk-ingress-results}"
+if [[ -z "${DEST}" ]]; then
+  : "${BUCKET_NAME:?Set BUCKET_NAME or pass --dest}"
+  DEST="gs://${BUCKET_NAME}/nighthawk-ingress-results"
+fi
 
 docker info >/dev/null 2>&1 || {
   echo "ERROR: docker daemon not running (the runner image is built locally)" >&2
@@ -110,6 +113,7 @@ if ! "${VENV}/bin/python" -c 'import yaml' 2>/dev/null; then
   "${VENV}/bin/pip" install --quiet pyyaml ||
     "${VENV}/bin/pip" install --quiet --index-url https://pypi.org/simple/ pyyaml
 fi
+venv_sync_requirements "${VENV}" "benchmarking/requirements.txt"
 PY="${VENV}/bin/python"
 
 # --- pin the router if its current cpu differs --------------------------------
@@ -194,10 +198,11 @@ kill "${LOGS_PID}" 2>/dev/null || true
 
 # --- verdict --------------------------------------------------------------------
 if [[ "${STATUS}" == "complete" ]]; then
-  CAPACITY="$(gcloud storage ls "${DEST}/runs/${NAME}/**/run_tag=${TAG}/capacity.json" 2>/dev/null | tail -1)"
+  CAPACITY="$("${PY}" benchmarking/result_store.py list "${DEST}/runs/${NAME}" \
+    --match "*/run_tag=${TAG}/capacity.json" | tail -1)"
   if [[ -n "${CAPACITY}" ]]; then
     echo ">>> ${CAPACITY}"
-    gcloud storage cat "${CAPACITY}"
+    "${PY}" benchmarking/result_store.py cat "${CAPACITY}"
   fi
   kubectl delete job "${JOB}" -n "${NAMESPACE}" >/dev/null
   echo ">>> ${NAME}: complete (tag ${TAG})"

benchmarking/nighthawk-ingress/runner.py

@@ -30,7 +30,6 @@
 import json
 import os
 import re
-import shutil
 import socket
 import subprocess
 import sys
@@ -40,6 +39,10 @@
 from pathlib import Path
 from typing import IO, TextIO
 
+# Also find the shared module when running directly from the source tree.
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+from result_store import upload
+
 import actors
 import output as output_mod
 import spec as spec_mod
@@ -62,12 +65,12 @@ def parse_duration_seconds(s: str) -> int:
 
 def parse_args() -> argparse.Namespace:
     p = argparse.ArgumentParser(description=__doc__)
-    p.add_argument("--name", required=True, help="Test name (GCS path component)")
+    p.add_argument("--name", required=True, help="Test name (result path component)")
     p.add_argument("--tag", required=True, help="Tag for this run (commit sha)")
     p.add_argument(
         "--dest",
         required=True,
-        help="Root destination (gs://bucket/path or local path)",
+        help="Root destination (gs://bucket/path, s3://bucket/path, az://container/path, or local path)",
     )
     p.add_argument("--envoy-cpu", required=True, type=int, dest="envoy_cpu")
     p.add_argument("--actors", type=int, default=100)
@@ -179,25 +182,6 @@ def run_adaptive_client(
     return proc.wait()
 
 
-def upload_to_gcs(local_path: Path, gcs_uri: str) -> None:
-    # Imported here so non-GCS use doesn't require google-cloud-storage.
-    from google.cloud import storage
-
-    bucket_name, _, blob_path = gcs_uri[len("gs://"):].partition("/")
-    storage.Client().bucket(bucket_name).blob(blob_path).upload_from_filename(
-        str(local_path)
-    )
-
-
-def upload(src: Path, dest: str) -> None:
-    if dest.startswith("gs://"):
-        upload_to_gcs(src, dest)
-    else:
-        dest_path = Path(dest)
-        dest_path.parent.mkdir(parents=True, exist_ok=True)
-        shutil.copy(src, dest_path)
-
-
 def log_run_config(args: argparse.Namespace, prefix: str, logs: TextIO) -> None:
     lines = [
         "==== Run config ====",

benchmarking/nighthawk-ingress/tests/test_runner.py

@@ -0,0 +1,55 @@
+# Copyright 2026 The Agent Substrate Authors
+#
+# 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.
+
+import importlib.util
+import sys
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+
+@pytest.mark.parametrize("upload_fails", [False, True])
+def test_converged_run_requires_successful_uploads(tmp_path, monkeypatch, upload_fails):
+    # Actor RPCs are outside this test; their generated clients only exist in
+    # the runner image. Keep all workload and network operations mocked.
+    actors = mock.Mock()
+    spec = importlib.util.spec_from_file_location(
+        "nighthawk_runner", Path(__file__).resolve().parents[1] / "runner.py"
+    )
+    runner = importlib.util.module_from_spec(spec)
+    with mock.patch.dict(sys.modules, {"actors": actors}):
+        spec.loader.exec_module(runner)
+    monkeypatch.setattr(sys, "argv", [
+        "runner.py", "--name", "test", "--tag", "test", "--dest", str(tmp_path / "results"),
+        "--envoy-cpu", "1",
+    ])
+    # Place the otherwise timestamp-named work directory under this test.
+    monkeypatch.setattr(runner, "Path", lambda _: tmp_path / "work")
+    monkeypatch.setattr(runner.spec_mod, "build_spec_dict", lambda **_: {})
+    monkeypatch.setattr(runner.spec_mod, "spec_dict_to_textproto", lambda *_: "spec")
+    monkeypatch.setattr(runner, "start_service", lambda *_: mock.Mock())
+    def adaptive(_spec, output, _logs):
+        output.write_text("output")
+        return 0
+    monkeypatch.setattr(runner, "run_adaptive_client", adaptive)
+    monkeypatch.setattr(runner.output_mod, "parse_output_textproto", lambda *_: {})
+    monkeypatch.setattr(runner.output_mod, "stage_rows", lambda *_: [{"stage": "testing"}])
+    monkeypatch.setattr(runner.output_mod, "stats_records", lambda *_: [])
+    monkeypatch.setattr(runner.output_mod, "capacity_summary", lambda *_, **__: {})
+    if upload_fails:
+        monkeypatch.setattr(runner, "upload", mock.Mock(side_effect=OSError("storage unavailable")))
+    with pytest.raises(SystemExit) as exc:
+        runner.main()
+    assert exc.value.code == (1 if upload_fails else 0)