test(benchmark): add cloud benchmark suites and CloudMode fixes for running them

golem-test-framework/src/benchmark/mod.rs

@@ -16,7 +16,9 @@ mod config;
 mod results;
 
 pub use config::{BenchmarkConfig, BenchmarkSuite, BenchmarkSuiteItem, RunConfig};
-pub use results::{BenchmarkResult, BenchmarkRunResult, BenchmarkSuiteResult, ResultKey};
+pub use results::{
+    BenchmarkResult, BenchmarkRunResult, BenchmarkSuiteResult, ResultKey, RunMetadata,
+};
 
 use crate::config::benchmark::TestMode;
 use async_trait::async_trait;

golem-test-framework/src/benchmark/results.rs

@@ -484,13 +484,106 @@ impl Display for BenchmarkResultView {
     }
 }
 
+/// Cloud-mode run metadata collected by the buildspec and passed via environment variables.
+/// All fields are optional — missing env vars produce `None` rather than failing the run.
+#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct RunMetadata {
+    /// The `golem-oss` commit SHA that was built and deployed.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub golem_oss_commit_sha: Option<String>,
+    /// The `golem-cloud` (kubernetes manifests) commit SHA that was deployed.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub kubernetes_manifest_commit_sha: Option<String>,
+    /// Number of Ready `worker-executor` pods observed at run start.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub observed_cluster_size: Option<u32>,
+    /// Container image tag of the deployed `worker-executor`.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub worker_executor_image_tag: Option<String>,
+    /// Container image tag of the deployed `registry-service`.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub registry_service_image_tag: Option<String>,
+    /// Container image tag of the deployed `worker-service`.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub worker_service_image_tag: Option<String>,
+    /// Aurora ACU capacity for the main (`golem_dev`) cluster at run start.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub aurora_acu_main: Option<f64>,
+    /// Aurora ACU capacity for the indexed-storage cluster at run start.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub aurora_acu_indexed: Option<f64>,
+    /// Aurora ACU capacity for the keyvalue-storage cluster at run start.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub aurora_acu_keyvalue: Option<f64>,
+    /// Ready replica count for `worker-executor` at run start.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub worker_executor_replicas: Option<u32>,
+    /// Ready replica count for `worker-service` at run start.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub worker_service_replicas: Option<u32>,
+    /// Ready replica count for `registry-service` at run start.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub registry_service_replicas: Option<u32>,
+    /// Ready replica count for `compilation-service` at run start.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub compilation_service_replicas: Option<u32>,
+    /// Ready replica count for `debugging-service` at run start.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub debugging_service_replicas: Option<u32>,
+    /// Free-form note from the `workflow_dispatch` trigger.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub note: Option<String>,
+}
+
+impl RunMetadata {
+    /// Reads all `GOLEM_BENCH_*` environment variables and returns a populated
+    /// `RunMetadata`.  Missing variables produce `None` for that field.
+    pub fn from_env() -> Self {
+        fn env_str(key: &str) -> Option<String> {
+            std::env::var(key).ok().filter(|v| !v.is_empty())
+        }
+        fn env_u32(key: &str) -> Option<u32> {
+            env_str(key).and_then(|v| v.parse().ok())
+        }
+        fn env_f64(key: &str) -> Option<f64> {
+            env_str(key).and_then(|v| v.parse().ok())
+        }
+
+        Self {
+            golem_oss_commit_sha: env_str("GOLEM_BENCH_OSS_COMMIT_SHA"),
+            kubernetes_manifest_commit_sha: env_str("GOLEM_BENCH_K8S_MANIFEST_COMMIT_SHA"),
+            observed_cluster_size: env_u32("GOLEM_BENCH_OBSERVED_CLUSTER_SIZE"),
+            worker_executor_image_tag: env_str("GOLEM_BENCH_WORKER_EXECUTOR_IMAGE_TAG"),
+            registry_service_image_tag: env_str("GOLEM_BENCH_REGISTRY_SERVICE_IMAGE_TAG"),
+            worker_service_image_tag: env_str("GOLEM_BENCH_WORKER_SERVICE_IMAGE_TAG"),
+            aurora_acu_main: env_f64("GOLEM_BENCH_AURORA_ACU_MAIN"),
+            aurora_acu_indexed: env_f64("GOLEM_BENCH_AURORA_ACU_INDEXED"),
+            aurora_acu_keyvalue: env_f64("GOLEM_BENCH_AURORA_ACU_KEYVALUE"),
+            worker_executor_replicas: env_u32("GOLEM_BENCH_WORKER_EXECUTOR_REPLICAS"),
+            worker_service_replicas: env_u32("GOLEM_BENCH_WORKER_SERVICE_REPLICAS"),
+            registry_service_replicas: env_u32("GOLEM_BENCH_REGISTRY_SERVICE_REPLICAS"),
+            compilation_service_replicas: env_u32("GOLEM_BENCH_COMPILATION_SERVICE_REPLICAS"),
+            debugging_service_replicas: env_u32("GOLEM_BENCH_DEBUGGING_SERVICE_REPLICAS"),
+            note: env_str("GOLEM_BENCH_RUN_NOTE"),
+        }
+    }
+
+    /// Returns `true` if every field is `None` (nothing was read from env).
+    pub fn is_empty(&self) -> bool {
+        self == &Self::default()
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct BenchmarkSuiteResultCollection {
     pub runs: Vec<BenchmarkSuiteResult>,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct BenchmarkSuiteResult {
+    /// Result format version. Always `1` for results produced by this binary.
+    pub schema_version: u32,
     pub suite: String,
     pub environment: String,
     pub version: String,
@@ -499,6 +592,10 @@ pub struct BenchmarkSuiteResult {
     /// cross-run correlation and garbage collection of orphaned state.
     #[serde(skip_serializing_if = "Option::is_none", default)]
     pub run_id: Option<String>,
+    /// Cloud-mode run metadata populated from `GOLEM_BENCH_*` environment variables.
+    /// `None` in Spawned or Provided modes where cluster metadata is not available.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub run_metadata: Option<RunMetadata>,
     pub results: Vec<BenchmarkResult>,
 }
 
@@ -530,11 +627,13 @@ impl BenchmarkSuiteResult {
         );
 
         Self {
+            schema_version: 1,
             suite: suite.to_string(),
             environment,
             version: golem_common::golem_version().to_string(),
             timestamp: Utc::now(),
             run_id: None,
+            run_metadata: None,
             results: vec![],
         }
     }

golem-test-framework/src/components/shard_manager/mod.rs

@@ -31,7 +31,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use tonic::codec::CompressionEncoding;
 use tonic::transport::Channel;
-use tracing::Level;
+use tracing::{Level, warn};
 
 #[async_trait]
 pub trait ShardManager: Send + Sync {
@@ -47,25 +47,30 @@ pub trait ShardManager: Send + Sync {
     async fn restart(&self, number_of_shards_override: Option<usize>);
 
     async fn get_routing_table(&self) -> crate::Result<RoutingTable> {
-        let routing_table = self
-            .client()
-            .await
-            .get_routing_table(GetRoutingTableRequest {})
-            .await
-            .expect("Unable to fetch the routing table from shard-manager-service");
-
-        match routing_table.into_inner() {
-            shardmanager::v1::GetRoutingTableResponse {
-                result:
-                    Some(shardmanager::v1::get_routing_table_response::Result::Success(routing_table)),
-            } => Ok(routing_table
-                .try_into()
-                .map_err(|e| anyhow!("Failed converting routing table: {e}"))?),
-            shardmanager::v1::GetRoutingTableResponse {
-                result: Some(shardmanager::v1::get_routing_table_response::Result::Failure(err)),
-            } => Err(anyhow!("Failed to get routing table: {err:?}")),
-            _ => Err(anyhow!("Failed to get routing table")),
+        // Retry with backoff to tolerate transient port-forward reconnects.
+        // The port-forward watchdog restarts in ~500ms, so 10 attempts with
+        // 1s delay gives ~10s of tolerance before giving up.
+        let max_attempts = 10;
+        let retry_delay = Duration::from_secs(1);
+        let mut last_err = anyhow!("get_routing_table: no attempts made");
+
+        for attempt in 1..=max_attempts {
+            match try_get_routing_table(&self.grpc_host(), self.grpc_port()).await {
+                Ok(rt) => return Ok(rt),
+                Err(err) => {
+                    warn!(
+                        attempt,
+                        max_attempts,
+                        error = %err,
+                        "Failed to fetch routing table, retrying..."
+                    );
+                    last_err = err;
+                    tokio::time::sleep(retry_delay).await;
+                }
+            }
         }
+
+        Err(last_err)
     }
 }
 
@@ -77,6 +82,34 @@ async fn new_client(host: &str, grpc_port: u16) -> ShardManagerServiceClient<Cha
         .accept_compressed(CompressionEncoding::Gzip)
 }
 
+async fn try_get_routing_table(host: &str, grpc_port: u16) -> crate::Result<RoutingTable> {
+    let mut client = ShardManagerServiceClient::connect(format!("http://{host}:{grpc_port}"))
+        .await
+        .map_err(|e| anyhow!("Failed to connect to shard-manager: {e}"))?
+        .send_compressed(CompressionEncoding::Gzip)
+        .accept_compressed(CompressionEncoding::Gzip);
+
+    let routing_table = client
+        .get_routing_table(GetRoutingTableRequest {})
+        .await
+        .map_err(|e| {
+            anyhow!("Unable to fetch the routing table from shard-manager-service: {e}")
+        })?;
+
+    match routing_table.into_inner() {
+        shardmanager::v1::GetRoutingTableResponse {
+            result:
+                Some(shardmanager::v1::get_routing_table_response::Result::Success(routing_table)),
+        } => Ok(routing_table
+            .try_into()
+            .map_err(|e| anyhow!("Failed converting routing table: {e}"))?),
+        shardmanager::v1::GetRoutingTableResponse {
+            result: Some(shardmanager::v1::get_routing_table_response::Result::Failure(err)),
+        } => Err(anyhow!("Failed to get routing table: {err:?}")),
+        _ => Err(anyhow!("Failed to get routing table")),
+    }
+}
+
 async fn wait_for_startup(
     host: &str,
     grpc_port: u16,

golem-test-framework/src/components/shard_manager/provided.rs

@@ -40,10 +40,10 @@ impl ShardManager for ProvidedShardManager {
     }
 
     async fn kill(&self) {
-        panic!("Cannot kill provided shard manager");
+        // Nothing to do — we do not own this shard manager process.
     }
 
     async fn restart(&self, _number_of_shards_override: Option<usize>) {
-        panic!("Cannot restart provided shard manager");
+        // Nothing to do — we do not own this shard manager process.
     }
 }

golem-test-framework/src/config/benchmark.rs

@@ -280,10 +280,14 @@ pub enum TestMode {
         #[arg(long)]
         admin_account_token: String,
         /// UUID of the builtin-plugin-owner account.
-        #[arg(long)]
+        /// Only needed for environment-plugin-grant tests; benchmarks do not
+        /// use it so the default (nil UUID) is fine for benchmark runs.
+        #[arg(long, default_value_t = Uuid::nil())]
         builtin_plugin_owner_account_id: Uuid,
         /// UUID of the default plan on the target cluster.
-        #[arg(long)]
+        /// Only needed for environment-plugin-grant tests; benchmarks do not
+        /// use it so the default (nil UUID) is fine for benchmark runs.
+        #[arg(long, default_value_t = Uuid::nil())]
         default_plan_id: Uuid,
         /// Optional shard-manager gRPC hostname for a kubectl port-forward
         /// (e.g. `localhost`). When set together with

integration-tests/benchmark_suites/cloud-density-saturation.yaml

@@ -0,0 +1,68 @@
+# Cloud throughput-saturation benchmark suite.
+#
+# Unlike cloud-perf's throughput benchmarks (which keep `size` small enough that
+# all workers fit in memory), this suite deliberately ramps the number of
+# active, memory-holding agents up to and past the executor's memory ceiling to
+# find the per-pod active-agent capacity and the throughput sustained once
+# memory is exhausted.
+#
+# Each agent retains a deterministic, per-agent-distinct amount of resident
+# memory, so the fleet presents a mix of footprints near the limit (exercising
+# the admission/eviction path). The measured phase drives one in-flight
+# `busy_for` call per agent and records aggregate throughput.
+#
+# Run with the benchmarks binary's `cloud` subcommand (same flags as cloud-perf):
+#
+#   benchmarks suite integration-tests/benchmark_suites/cloud-density-saturation.yaml \
+#     --save-to-json result.json \
+#     cloud --api-url https://<host> --apps-base-domain <domain> \
+#       --admin-account-token <token> --builtin-plugin-owner-account-id <uuid> \
+#       --default-plan-id <uuid> --component-directory <path-to-wasm-components>
+#
+# Reading the result: plot `saturation-throughput-ops-per-sec` and
+# invocation-retries/timeouts against `size`. Throughput climbs with `size`
+# until the pod's memory is exhausted, then plateaus or drops while retries and
+# eviction churn rise — that knee is the active-agent ceiling.
+#
+# `clusterSize` is ignored in cloud mode (single observed cluster).
+
+name: cloud-density-saturation
+benchmarks:
+  # # Rust echo agents — lean per-instance linear memory (the ~900 KB module is
+  # # charged once per component, shared across all agents; what scales per agent
+  # # is the small instance heap). The previous run reached the top of the sweep
+  # # (12000) without saturating pod memory, so the knee here is throughput /
+  # # eviction-churn rather than memory. Dropped the low points that told us
+  # # nothing and pushed the range up with coarser steps.
+  # - name: throughput-saturation-echo-rust
+  #   iterations: 3
+  #   clusterSize: [2]
+  #   size: [2000, 3000, 4000, 5000, 10000, 15000, 20000]
+  #   length: [0]
+
+  # # TypeScript echo agents — each instance instantiates its own QuickJS runtime
+  # # and JS heap in its own linear memory (the 17.4 MB module is shared once per
+  # # component; the per-instance runtime state is the heavy per-agent cost).
+  # # Heavier per agent than the Rust variant, so a lower knee — but the previous
+  # # run reached 2000 without saturating, so push higher and drop the low points.
+  # - name: throughput-saturation-echo-ts
+  #   iterations: 3
+  #   clusterSize: [2]
+  #   size: [1000, 2000, 3000]
+  #   length: [0]
+
+  # Synthetic footprint — each agent retains a deterministic per-agent-distinct
+  # amount of resident memory, exercising the admission/eviction path with a
+  # controllable footprint near the limit. Run first: this is the variant that
+  # actually fills memory and drives the gate to its reject/evict path.
+  # size   = number of active, memory-holding agents (the ramp axis)
+  # length = base per-agent memory footprint in bytes; each agent retains a
+  #          deterministic multiple (1x..8x), averaging ~4.5x. 16 MiB base =>
+  #          ~72 MiB average per agent, filling a ~10 GiB usable pool around
+  #          ~145 agents. The sweep brackets that ceiling and pushes well past it
+  #          so the admission gate's reject/evict behaviour near OOM is exercised.
+  - name: throughput-saturation-counters
+    iterations: 1
+    clusterSize: [2]
+    size: [50, 100, 150, 200, 300]
+    length: [16777216]