feat(benchmarking): add pluggable config paths and support optional token auth

benchmarking/automation/manifests/runner-job.yaml.tmpl

@@ -67,10 +67,15 @@ spec:
         env:
         - name: OTEL_EXPORTER_OTLP_ENDPOINT
           value: http://opentelemetry-collector.gke-managed-otel.svc.cluster.local:4317
+        - name: ATE_ATEAPI_CLIENT_AUTH
+          value: "${ATE_ATEAPI_CLIENT_AUTH}"
         volumeMounts:
         - name: servicedns-ca
           mountPath: /run/servicedns-ca
           readOnly: true
+        - name: ateapi-token
+          mountPath: /run/ateapi-token
+          readOnly: true
         - name: podidentity
           mountPath: /run/podidentity.podcert.ate.dev
           readOnly: true
@@ -88,6 +93,13 @@ spec:
                 matchLabels:
                   podcert.ate.dev/canarying: live
               path: ca.crt
+      - name: ateapi-token
+        projected:
+          sources:
+          - serviceAccountToken:
+              audience: api.ate-system.svc
+              expirationSeconds: 3600
+              path: token
       - name: podidentity
         projected:
           sources:

benchmarking/automation/orchestrator.py

@@ -63,6 +63,16 @@ def parse_args() -> argparse.Namespace:
         default="/etc/orchestrator/tests.yaml",
         help="Path to the tests YAML file (mounted from a ConfigMap)",
     )
+    p.add_argument(
+        "--target-cluster-dir",
+        default=TARGET_CLUSTER_DIR,
+        help="Directory containing target cluster configuration scripts (<cluster>.sh)",
+    )
+    p.add_argument(
+        "--runner-job-tmpl",
+        default=RUNNER_JOB_TMPL,
+        help="Path to the runner job YAML template file",
+    )
     return p.parse_args()
 
 
@@ -90,14 +100,16 @@ def source_env(path: str) -> None:
         os.environ[k] = v
 
 
-def apply_target_cluster(target_cluster: str) -> None:
-    """Copy /etc/orchestrator/target-clusters/<name>.sh into the cloned
+def apply_target_cluster(
+    target_cluster: str, target_cluster_dir: str = TARGET_CLUSTER_DIR
+) -> None:
+    """Copy target_cluster_dir/<name>.sh into the cloned
     substrate repo as .ate-dev-env.sh (so install-ate.sh / deploy.sh source
     it) and merge it into this process's env (so the orchestrator's own
     gcloud / docker / kubectl calls see the same values). Resets os.environ
     to the startup baseline first so vars defined by a previous target
     cluster don't bleed into the next one."""
-    src = Path(TARGET_CLUSTER_DIR) / f"{target_cluster}.sh"
+    src = Path(target_cluster_dir) / f"{target_cluster}.sh"
     if not src.exists():
         raise FileNotFoundError(
             f"target cluster {target_cluster!r} not found at {src}"
@@ -308,7 +320,13 @@ def teardown_workloads() -> None:
     run_no_check(["benchmarking/workloads/deploy.sh", "--delete"])
 
 
-def run_test(test: dict[str, Any], image: str, dest: str, commit: str) -> str:
+def run_test(
+    test: dict[str, Any],
+    image: str,
+    dest: str,
+    commit: str,
+    runner_job_tmpl: str = RUNNER_JOB_TMPL,
+) -> str:
     name = test["name"]
     job_name = f"runner-{sanitize(name)}-{commit[:7]}-{uuid.uuid4().hex[:6]}"
     subs = {
@@ -320,8 +338,9 @@ def run_test(test: dict[str, Any], image: str, dest: str, commit: str) -> str:
         "TAG": commit,
         "NAME": name,
         "DEST": dest,
+        "ATE_ATEAPI_CLIENT_AUTH": os.environ.get("ATE_ATEAPI_CLIENT_AUTH", "cert"),
     }
-    manifest = render_template(RUNNER_JOB_TMPL, subs, test.get("flags", []))
+    manifest = render_template(runner_job_tmpl, subs, test.get("flags", []))
     wait_for_no_active_runners()
     print(f"Submitting Job {job_name}", flush=True)
     subprocess.run(
@@ -382,7 +401,7 @@ def main() -> None:
         )
 
         try:
-            apply_target_cluster(target_cluster)
+            apply_target_cluster(target_cluster, args.target_cluster_dir)
         except Exception as e:
             print(
                 f"Failed to apply target cluster {target_cluster!r}: {e}",
@@ -410,7 +429,13 @@ def main() -> None:
                 deploy_substrate()
                 deploy_workloads(test.get("workerCount", 1))
                 try:
-                    status = run_test(test, locust_image, args.dest, commit)
+                    status = run_test(
+                        test,
+                        locust_image,
+                        args.dest,
+                        commit,
+                        args.runner_job_tmpl,
+                    )
                 except Exception as e:
                     print(f"Test {test['name']} crashed: {e}", flush=True)
             except Exception as e:

benchmarking/locust/runner.py

@@ -212,6 +212,8 @@ def run_test(args: argparse.Namespace, csv_prefix: Path, logs: TextIO, traces: T
         cfg_json = build_config_json(args.locust_extra)
         if cfg_json:
             boomer_cmd += ["--config-json", cfg_json]
+        if os.environ.get("ATE_ATEAPI_CLIENT_AUTH") == "token":
+            boomer_cmd += ["--use-token-auth"]
         tee(logs, f"Running: {' '.join(boomer_cmd)}")
         boomer_proc = subprocess.Popen(
             boomer_cmd,

cmd/benchmarking/boomer-glutton/main.go

@@ -42,6 +42,7 @@ func main() {
 		promAddr      = flag.String("prometheus-addr", ":8001", "Address for the Prometheus /metrics endpoint.")
 		configJSON    = flag.String("config-json", "", "Initial dynconfig as a JSON object (keys: trace_probability, min_wait_time, max_wait_time in seconds). Unset fields keep their built-in defaults.")
 		masterWebPort = flag.Int("master-web-port", 0, "If non-zero, fetch dynconfig from http://{master-host}:{master-web-port}/boomer-config on each spawn message and fail fatally on error. {master-host} comes from boomer's existing --master-host flag.")
+		useTokenAuth  = flag.Bool("use-token-auth", false, "Use Kubernetes ServiceAccount token for ateapi auth instead of client certificate.")
 	)
 	// boomer.Run will call flag.Parse() if we haven't yet; calling here so
 	// our flag-derived values are usable before that.
@@ -68,7 +69,7 @@ func main() {
 		_ = tp.Shutdown(shutdownCtx)
 	}()
 
-	conn, apiStub, err := glutton.DialControl(*apiEndpoint)
+	conn, apiStub, err := glutton.DialControl(*apiEndpoint, *useTokenAuth)
 	if err != nil {
 		slog.Error("failed to dial ateapi", slog.String("err", err.Error()))
 		os.Exit(1)

internal/benchmarking/boomer/glutton/grpcclient.go

@@ -41,9 +41,11 @@ const (
 // certificate — the same client auth the base install gives ate-controller
 // and atenet-router. ClientCredBundle re-reads the bundle on every handshake,
 // so rotations are picked up without a restart.
-func DialControl(endpoint string) (*grpc.ClientConn, ateapipb.ControlClient, error) {
+func DialControl(endpoint string, useTokenAuth bool) (*grpc.ClientConn, ateapipb.ControlClient, error) {
 	dialOpts, err := ateapiauth.DialOptions(ateapiauth.ClientConfig{
+		UseTokenAuth:     useTokenAuth,
 		CAFile:           ateapiCAFile,
+		TokenFile:        "/run/ateapi-token/token",
 		ClientCredBundle: ateapiCredBundle,
 		ServerName:       ateapiServerName,
 	})