From 9c7326a71f6b75856ca40564123c30e58efe59ef Mon Sep 17 00:00:00 2001 From: kmatasfp <33095685+kmatas@users.noreply.github.com> Date: Wed, 17 Jun 2026 01:31:06 -0700 Subject: [PATCH] test(benchmark): add cloud benchmark suites and CloudMode fixes for running them --- golem-test-framework/src/benchmark/mod.rs | 4 +- golem-test-framework/src/benchmark/results.rs | 99 ++++ .../src/components/shard_manager/mod.rs | 71 ++- .../src/components/shard_manager/provided.rs | 4 +- golem-test-framework/src/config/benchmark.rs | 8 +- .../cloud-density-saturation.yaml | 68 +++ .../benchmark_suites/cloud-perf.yaml | 130 ++++++ integration-tests/src/benchmarks/all.rs | 35 +- integration-tests/src/benchmarks/mod.rs | 1 + .../src/benchmarks/throughput.rs | 47 +- .../src/benchmarks/throughput_saturation.rs | 425 ++++++++++++++++++ test-components/agent-counters/Cargo.toml | 6 + test-components/agent-counters/src/lib.rs | 91 +++- 13 files changed, 960 insertions(+), 29 deletions(-) create mode 100644 integration-tests/benchmark_suites/cloud-density-saturation.yaml create mode 100644 integration-tests/benchmark_suites/cloud-perf.yaml create mode 100644 integration-tests/src/benchmarks/throughput_saturation.rs diff --git a/golem-test-framework/src/benchmark/mod.rs b/golem-test-framework/src/benchmark/mod.rs index fd246b2be7..5e82adde15 100644 --- a/golem-test-framework/src/benchmark/mod.rs +++ b/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; diff --git a/golem-test-framework/src/benchmark/results.rs b/golem-test-framework/src/benchmark/results.rs index a309d1d5e5..afb7319a7d 100644 --- a/golem-test-framework/src/benchmark/results.rs +++ b/golem-test-framework/src/benchmark/results.rs @@ -484,6 +484,97 @@ 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, + /// The `golem-cloud` (kubernetes manifests) commit SHA that was deployed. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub kubernetes_manifest_commit_sha: Option, + /// Number of Ready `worker-executor` pods observed at run start. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub observed_cluster_size: Option, + /// Container image tag of the deployed `worker-executor`. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub worker_executor_image_tag: Option, + /// Container image tag of the deployed `registry-service`. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub registry_service_image_tag: Option, + /// Container image tag of the deployed `worker-service`. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub worker_service_image_tag: Option, + /// 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, + /// Aurora ACU capacity for the indexed-storage cluster at run start. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub aurora_acu_indexed: Option, + /// Aurora ACU capacity for the keyvalue-storage cluster at run start. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub aurora_acu_keyvalue: Option, + /// Ready replica count for `worker-executor` at run start. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub worker_executor_replicas: Option, + /// Ready replica count for `worker-service` at run start. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub worker_service_replicas: Option, + /// Ready replica count for `registry-service` at run start. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub registry_service_replicas: Option, + /// Ready replica count for `compilation-service` at run start. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub compilation_service_replicas: Option, + /// Ready replica count for `debugging-service` at run start. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub debugging_service_replicas: Option, + /// Free-form note from the `workflow_dispatch` trigger. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub note: Option, +} + +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 { + std::env::var(key).ok().filter(|v| !v.is_empty()) + } + fn env_u32(key: &str) -> Option { + env_str(key).and_then(|v| v.parse().ok()) + } + fn env_f64(key: &str) -> Option { + 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, @@ -491,6 +582,8 @@ pub struct BenchmarkSuiteResultCollection { #[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, + /// 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, pub results: Vec, } @@ -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![], } } diff --git a/golem-test-framework/src/components/shard_manager/mod.rs b/golem-test-framework/src/components/shard_manager/mod.rs index 67b10051bf..91ed2ed2da 100644 --- a/golem-test-framework/src/components/shard_manager/mod.rs +++ b/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); async fn get_routing_table(&self) -> crate::Result { - 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 crate::Result { + 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, diff --git a/golem-test-framework/src/components/shard_manager/provided.rs b/golem-test-framework/src/components/shard_manager/provided.rs index d7e4ff1305..84d213a5fb 100644 --- a/golem-test-framework/src/components/shard_manager/provided.rs +++ b/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) { - panic!("Cannot restart provided shard manager"); + // Nothing to do — we do not own this shard manager process. } } diff --git a/golem-test-framework/src/config/benchmark.rs b/golem-test-framework/src/config/benchmark.rs index e97a58c47d..34ac140d23 100644 --- a/golem-test-framework/src/config/benchmark.rs +++ b/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 diff --git a/integration-tests/benchmark_suites/cloud-density-saturation.yaml b/integration-tests/benchmark_suites/cloud-density-saturation.yaml new file mode 100644 index 0000000000..1d7a477661 --- /dev/null +++ b/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:// --apps-base-domain \ +# --admin-account-token --builtin-plugin-owner-account-id \ +# --default-plan-id --component-directory +# +# 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] diff --git a/integration-tests/benchmark_suites/cloud-perf.yaml b/integration-tests/benchmark_suites/cloud-perf.yaml new file mode 100644 index 0000000000..ef8dd7d61f --- /dev/null +++ b/integration-tests/benchmark_suites/cloud-perf.yaml @@ -0,0 +1,130 @@ +# Cloud-perf benchmark suite — runs the full benchmark suite against a +# deployed Golem environment via Gateway-API hostnames (TestMode::Cloud). +# +# Run with the benchmarks binary's `cloud` subcommand: +# +# benchmarks suite integration-tests/benchmark_suites/cloud-perf.yaml \ +# --save-to-json result.json \ +# cloud \ +# --api-url https:// \ +# --apps-base-domain \ +# --admin-account-token \ +# --builtin-plugin-owner-account-id \ +# --default-plan-id \ +# --component-directory +# +# Note: clusterSize is ignored in Cloud mode (the observed cluster size is +# read from shard-manager at run start and recorded in result metadata). +# +# Suite order rationale: throughput benchmarks run first because they involve +# RPC worker pairs and HTTP deployments — the most complex setup. Running them +# early surfaces infrastructure issues (stuck workers, port-forward drops) +# before spending time on the simpler benchmarks. + +name: cloud-perf +benchmarks: + # Throughput — measures invocation throughput across six implementations: + # rust agent (gRPC), TS agent (gRPC), rust agent (HTTP), TS agent (HTTP), + # TS RPC pair, rust RPC pair. + # size = number of workers per implementation (×6 implementations total) + # length = unused for echo + - name: throughput-echo + iterations: 3 + clusterSize: [2] + size: [1, 50, 100, 250] + length: [1000] + + # size = number of workers per implementation + # length = payload size in bytes sent to large_input + # NOTE: large payloads grow worker linear memory, so this is the throughput + # benchmark most relevant to the memory-admission investigation — sized to + # match throughput-echo so it exercises real density. + - name: throughput-large-input + iterations: 3 + clusterSize: [2] + size: [1, 50, 100, 250] + length: [100, 10000] + + # size = number of workers per implementation + # length = CPU work length passed to cpu_intensive + - name: throughput-cpu-intensive + iterations: 3 + clusterSize: [2] + size: [1, 50, 100, 250] + length: [100] + + # Cold-start: compilation cache disabled — measures true cold-start latency + # with no warm compiled artefact available. + # size = number of unique components created (each in its own env) + # length = seconds to wait per component for pre-compilation warm-up + - name: cold-start-unknown-small + iterations: 3 + clusterSize: [2] + size: [1, 5, 25, 50] + length: [2] + disableCompilationCache: true + + - name: cold-start-unknown-medium + iterations: 3 + clusterSize: [2] + size: [1, 5, 25, 50] + length: [5] + disableCompilationCache: true + + # Cold-start: compilation cache enabled — measures latency once the compiled + # artefact is available in the cache. + # size = number of unique components created (each in its own env) + # length = seconds to wait per component for pre-compilation warm-up + # NOTE: if results here are close to the cache-disabled entries above, the + # warm-up wait is too short and compilation hasn't finished — bump length. + - name: cold-start-unknown-small + iterations: 3 + clusterSize: [2] + size: [1, 5, 25, 50] + length: [2] + + - name: cold-start-unknown-medium + iterations: 3 + clusterSize: [2] + size: [1, 5, 25, 50] + length: [5] + + # Invocation latency — hot and cold paths through the Gateway NLB. + # Large worker counts to stress the load balancer and connection pool. + # size = number of workers created + # length = number of hot invocations per worker after the first cold one + - name: latency-small + iterations: 3 + clusterSize: [2] + size: [100, 500, 1000, 2000, 5000] + length: [2] + + - name: latency-medium + iterations: 3 + clusterSize: [2] + size: [100, 500, 1000, 2000] + length: [5] + + # Sleep — measures worker suspension and resumption under real network + # conditions. High residency: all `size` workers held in memory sleeping at + # once, so this also probes how many resident workers fit (memory-admission + # relevant) — pushed past the ~2000 echo proved out. + # size = number of workers launched in parallel + # length = sleep duration in milliseconds + - name: sleep + iterations: 3 + clusterSize: [2] + size: [10, 100, 500, 1000, 2000] + length: [10000] + + # Durability overhead — measures the cost of durable vs ephemeral execution + # across four variants (durable-persistent, durable-non-persistent, + # ephemeral, durable-persistent-commit). size workers concurrent per phase; + # sized up to put real load on the oplog/persistence/storage path. + # size = number of workers per variant + # length = loop iteration count passed to oplog_heavy + - name: durability-overhead + iterations: 3 + clusterSize: [2] + size: [10, 50, 100, 200] + length: [5000] diff --git a/integration-tests/src/benchmarks/all.rs b/integration-tests/src/benchmarks/all.rs index 6efe3eadd7..f08c71793e 100644 --- a/integration-tests/src/benchmarks/all.rs +++ b/integration-tests/src/benchmarks/all.rs @@ -21,7 +21,7 @@ use golem_common::model::environment::{EnvironmentCreation, EnvironmentName}; use golem_common::{agent_id, data_value}; use golem_test_framework::benchmark::{ Benchmark, BenchmarkApi, BenchmarkConfig, BenchmarkResult, BenchmarkSuite, BenchmarkSuiteItem, - BenchmarkSuiteResult, + BenchmarkSuiteResult, RunMetadata, }; use golem_test_framework::config::benchmark::{TestMode, cloud_bench_run_id}; use golem_test_framework::config::{ @@ -133,6 +133,30 @@ async fn main() { >(mode, verbosity, item, primary_only, otlp)) }), ); + benchmarks_by_name.insert( + "throughput-saturation-counters", + Box::new(|mode, verbosity, item, primary_only, otlp| { + Box::pin(run_benchmark::< + integration_tests::benchmarks::throughput_saturation::ThroughputSaturationCounters, + >(mode, verbosity, item, primary_only, otlp)) + }), + ); + benchmarks_by_name.insert( + "throughput-saturation-echo-rust", + Box::new(|mode, verbosity, item, primary_only, otlp| { + Box::pin(run_benchmark::< + integration_tests::benchmarks::throughput_saturation::ThroughputSaturationEchoRust, + >(mode, verbosity, item, primary_only, otlp)) + }), + ); + benchmarks_by_name.insert( + "throughput-saturation-echo-ts", + Box::new(|mode, verbosity, item, primary_only, otlp| { + Box::pin(run_benchmark::< + integration_tests::benchmarks::throughput_saturation::ThroughputSaturationEchoTs, + >(mode, verbosity, item, primary_only, otlp)) + }), + ); let params = BenchmarkCliParameters::parse_from(std::env::args_os()); let tracer_provider = BenchmarkTestDependencies::init_logging(¶ms); @@ -233,9 +257,16 @@ async fn main() { // no else: we already validated all names above } - // Attach the run_id to result metadata (cloud mode only). + // Attach the run_id and run_metadata to result metadata (cloud mode only). if let Some(run_id) = cloud_bench_run_id() { suite_result.run_id = Some(format!("bench-{run_id}")); + + // Read GOLEM_BENCH_* env vars set by the buildspec before invoking + // the binary. Missing vars produce None rather than failing the run. + let metadata = RunMetadata::from_env(); + if !metadata.is_empty() { + suite_result.run_metadata = Some(metadata); + } } if let Some(path) = save_to_json { diff --git a/integration-tests/src/benchmarks/mod.rs b/integration-tests/src/benchmarks/mod.rs index 7a94e72570..5b1896c87f 100644 --- a/integration-tests/src/benchmarks/mod.rs +++ b/integration-tests/src/benchmarks/mod.rs @@ -35,6 +35,7 @@ pub mod durability_overhead; pub mod latency; pub mod sleep; pub mod throughput; +pub mod throughput_saturation; // Re-export cleanup helpers so callers can use the flat `benchmarks::*` path. pub use cleanup::{cleanup_account, cleanup_env_and_app, cleanup_user_state}; diff --git a/integration-tests/src/benchmarks/throughput.rs b/integration-tests/src/benchmarks/throughput.rs index 97207db1a3..e6bbff878b 100644 --- a/integration-tests/src/benchmarks/throughput.rs +++ b/integration-tests/src/benchmarks/throughput.rs @@ -29,7 +29,7 @@ use golem_common::model::http_api_deployment::{ }; use golem_common::model::{AgentId, RoutingTable}; use golem_common::{agent_id, data_value}; -use golem_test_framework::benchmark::{Benchmark, BenchmarkRecorder, RunConfig}; +use golem_test_framework::benchmark::{Benchmark, BenchmarkRecorder, ResultKey, RunConfig}; use golem_test_framework::config::benchmark::TestMode; use golem_test_framework::config::dsl_impl::TestUserContext; use golem_test_framework::config::{BenchmarkTestDependencies, TestDependencies}; @@ -38,6 +38,7 @@ use indoc::indoc; use reqwest::{Body, Method, Request, Url}; use serde_json::json; use std::collections::BTreeMap; +use std::time::Instant; use tracing::{Instrument, Level, info}; pub struct ThroughputEcho { @@ -460,6 +461,31 @@ fn agent_ids_to_agent_ids(component_id: ComponentId, ids: &[LegacyParsedAgentId] .collect() } +/// Records aggregate throughput (invocations per second) for a measurement +/// block as a `count` result under the key `{prefix}throughput-ops-per-sec`. +/// +/// `total_calls` is the total number of invocations issued across all targets +/// in the block; `elapsed` is the wall-clock duration of the concurrently +/// executed block. Throughput is therefore the realised aggregate rate the +/// cluster sustained for this implementation, not a per-call latency. +fn record_throughput( + recorder: &BenchmarkRecorder, + prefix: &str, + total_calls: usize, + elapsed: std::time::Duration, +) { + let secs = elapsed.as_secs_f64(); + if secs <= 0.0 || total_calls == 0 { + return; + } + let ops_per_sec = (total_calls as f64 / secs).round() as u64; + info!("{prefix}throughput: {total_calls} calls in {secs:.3}s = {ops_per_sec} ops/sec"); + recorder.count( + &ResultKey::primary(format!("{prefix}throughput-ops-per-sec")), + ops_per_sec, + ); +} + impl ThroughputBenchmark { pub async fn new( rust_method_name: &str, @@ -796,7 +822,10 @@ impl ThroughputBenchmark { }) .collect::>(); + let started = Instant::now(); let results = result_futures.join().await; + let elapsed = started.elapsed(); + record_throughput(recorder, prefix, targets.len() * call_count, elapsed); for (idx, (results, target)) in results.iter().zip(targets).enumerate() { let prefix = target.prefix(prefix, routing_table); for result in results { @@ -903,7 +932,15 @@ impl ThroughputBenchmark { }) .collect::>(); + let started = Instant::now(); let results = result_futures.join().await; + let elapsed = started.elapsed(); + record_throughput( + &recorder, + "rust-agent-http-", + iteration.rust_agent_ids_for_http.len() * self.call_count, + elapsed, + ); for (idx, results) in results.iter().enumerate() { for result in results { result.record(&recorder, "rust-agent-http-", idx.to_string().as_str()); @@ -936,7 +973,15 @@ impl ThroughputBenchmark { }) .collect::>(); + let started = Instant::now(); let results = result_futures.join().await; + let elapsed = started.elapsed(); + record_throughput( + &recorder, + "ts-agent-http-", + iteration.ts_agent_ids_for_http.len() * self.call_count, + elapsed, + ); for (idx, results) in results.iter().enumerate() { for result in results { result.record(&recorder, "ts-agent-http-", idx.to_string().as_str()); diff --git a/integration-tests/src/benchmarks/throughput_saturation.rs b/integration-tests/src/benchmarks/throughput_saturation.rs new file mode 100644 index 0000000000..2e31934fdc --- /dev/null +++ b/integration-tests/src/benchmarks/throughput_saturation.rs @@ -0,0 +1,425 @@ +// Copyright 2024-2026 Golem Cloud +// +// Licensed under the Golem Source License v1.1 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://license.golem.cloud/LICENSE +// +// 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. + +//! Throughput-under-memory-saturation benchmarks. +//! +//! Unlike the regular throughput benchmark — which keeps `size` small enough +//! that all workers fit comfortably in memory — these benchmarks deliberately +//! ramp the number of *active* agents up to and past the executor's memory +//! ceiling, to find the knee: the agent count where the pod can still keep +//! everything resident (latency flat, throughput scaling linearly) just before +//! it starts evicting and replaying (latency spikes, throughput craters). +//! +//! The measured `run` phase drives sustained load over a fixed window: each +//! agent repeatedly does a short unit of work then goes idle for [`IDLE_GAP`]. +//! During that gap the agent has no in-flight work and becomes a `LoadedIdle` +//! eviction candidate, so under memory pressure it can be evicted and then must +//! reload (oplog replay + re-admission) on its next call — the churn that makes +//! throughput crater past the knee. Starts are staggered so the fleet is not +//! synchronised. +//! +//! Three variants: +//! - `throughput-saturation-counters`: agent-counters with a synthetic, +//! per-agent-distinct retained footprint (`allocate_memory`) plus CPU work +//! (`busy_for`). The footprint is controllable via `length`. +//! - `throughput-saturation-echo-rust` / `throughput-saturation-echo-ts`: the +//! benchmark `echo` agent (Rust / TS) called repeatedly. No synthetic +//! footprint — the per-agent memory is the agent's natural footprint, which +//! for the TS agent includes the QuickJS runtime. Answers "how many actively +//! invoked echo agents fit per pod". +//! +//! Parameters: +//! - `size` = number of active agents in this step (the ramp axis). +//! - `length` = for the counters variant, the base per-agent memory footprint in +//! bytes (agent `i` retains a deterministic multiple); ignored by the echo +//! variants. + +use crate::benchmarks::{cleanup_user_state, delete_workers, invoke_and_await_agent}; +use async_trait::async_trait; +use futures_concurrency::future::Join; +use golem_common::base_model::agent::{DataValue, LegacyParsedAgentId}; +use golem_common::model::AgentId; +use golem_common::model::component::ComponentDto; +use golem_common::model::environment::EnvironmentId; +use golem_common::{agent_id, data_value}; +use golem_test_framework::benchmark::{Benchmark, BenchmarkRecorder, ResultKey, RunConfig}; +use golem_test_framework::config::benchmark::TestMode; +use golem_test_framework::config::dsl_impl::TestUserContext; +use golem_test_framework::config::{BenchmarkTestDependencies, TestDependencies}; +use golem_test_framework::dsl::{TestDsl, TestDslExtended}; +use indoc::indoc; +use std::time::{Duration, Instant}; +use tracing::{Instrument, Level, info}; + +/// Number of distinct footprint buckets the synthetic per-agent memory spread +/// cycles through, so the fleet holds a mix of sizes rather than a uniform +/// amount. +const SPREAD_BUCKETS: usize = 8; + +/// CPU busy time (ms) per `busy_for` invocation (counters variant only). +const BUSY_MILLIS: u32 = 50; + +/// Idle gap each agent sleeps between calls. During this gap the agent has no +/// in-flight work and becomes a `LoadedIdle` eviction candidate. Under memory +/// pressure it may be evicted and then must reload on its next call — the churn +/// this benchmark exists to measure. +const IDLE_GAP: Duration = Duration::from_millis(200); + +/// Total measured wall-clock duration of the sustained-load phase. Throughput +/// and churn are measured over this fixed window so steps with different `size` +/// are comparable. Held long enough that the high-residency plateau persists for +/// at least a minute, so steady-state behaviour at the memory ceiling (not just +/// the initial burst) is observed. +const RUN_DURATION: Duration = Duration::from_secs(90); + +/// Maximum per-agent start stagger, so the fleet is not synchronised: at any +/// instant some agents are mid-call (demanding memory) while others sit idle +/// (evictable). +const MAX_STAGGER: Duration = Duration::from_millis(250); + +/// Resident memory (bytes) the synthetic-footprint agent `index` retains for a +/// given `base`. Spreads deterministically across [`SPREAD_BUCKETS`] buckets +/// (`base * 1` .. `base * SPREAD_BUCKETS`) so different agents hold different +/// amounts and some sit much closer to the limit than others. +fn agent_memory_bytes(index: usize, base: usize) -> u32 { + let bucket = (index % SPREAD_BUCKETS) + 1; + (base.saturating_mul(bucket)).min(u32::MAX as usize) as u32 +} + +/// Per-agent start offset derived deterministically from the index, spread +/// across `[0, MAX_STAGGER)`. +fn agent_stagger(index: usize) -> Duration { + let frac = (index as u32).wrapping_mul(2_654_435_761) % 1000; + MAX_STAGGER.checked_mul(frac).unwrap_or_default() / 1000 +} + +/// Describes one saturation variant: which component to load, which agent type +/// and method to actively invoke, and whether to pre-load a synthetic footprint. +struct SaturationVariant { + /// WASM file name (without `.wasm`) in the component directory. + wasm_name: &'static str, + /// Registry display name for the component. + component_name: &'static str, + /// Agent type to instantiate. + agent_type: &'static str, + /// Method invoked repeatedly during the measured phase. + active_method: &'static str, + /// Builds the parameter for one `active_method` call. + active_params: fn() -> DataValue, + /// When set, each agent calls this method once in warmup with its + /// deterministic footprint (`allocate_memory`-style). `None` for the echo + /// variants, whose footprint is the agent's natural memory. + allocate_method: Option<&'static str>, +} + +const COUNTERS_VARIANT: SaturationVariant = SaturationVariant { + wasm_name: "it_agent_counters_release", + component_name: "it:agent-counters", + agent_type: "Counter", + active_method: "busy_for", + active_params: || data_value!(BUSY_MILLIS), + allocate_method: Some("allocate_memory"), +}; + +const ECHO_RUST_VARIANT: SaturationVariant = SaturationVariant { + wasm_name: "benchmark_agent_rust_release", + component_name: "benchmark:agent-rust", + agent_type: "RustBenchmarkAgent", + active_method: "echo", + active_params: || data_value!("saturation"), + allocate_method: None, +}; + +const ECHO_TS_VARIANT: SaturationVariant = SaturationVariant { + wasm_name: "benchmark_agent_ts", + component_name: "benchmark:agent-ts", + agent_type: "BenchmarkAgent", + active_method: "echo", + active_params: || data_value!("saturation"), + allocate_method: None, +}; + +pub struct SaturationBenchmarkContext { + deps: BenchmarkTestDependencies, +} + +pub struct SaturationIterationContext { + user: TestUserContext, + component: ComponentDto, + agent_ids: Vec, + base_memory_bytes: usize, + env_id: EnvironmentId, +} + +/// Shared implementation for all saturation variants. The variant-specific +/// config is supplied by the wrapper types' `variant()`. +async fn create_context( + mode: &TestMode, + verbosity: Level, + cluster_size: usize, + disable_compilation_cache: bool, + otlp: bool, +) -> SaturationBenchmarkContext { + SaturationBenchmarkContext { + deps: BenchmarkTestDependencies::new( + mode, + verbosity, + cluster_size, + disable_compilation_cache, + otlp, + ) + .await, + } +} + +async fn setup_iteration( + variant: &SaturationVariant, + config: &RunConfig, + benchmark_context: &SaturationBenchmarkContext, +) -> SaturationIterationContext { + let user = benchmark_context.deps.user().await.unwrap(); + let (_, env) = user.app_and_env().await.unwrap(); + + info!("Registering component {}", variant.component_name); + let component = user + .component(&env.id, variant.wasm_name) + .name(variant.component_name) + .store() + .await + .unwrap(); + + let mut agent_ids = vec![]; + for n in 0..config.size { + agent_ids.push(agent_id!(variant.agent_type, format!("saturation-{n}"))); + } + + SaturationIterationContext { + user, + component, + agent_ids, + base_memory_bytes: config.length, + env_id: env.id, + } +} + +async fn warmup(variant: &SaturationVariant, context: &SaturationIterationContext) { + let Some(allocate_method) = variant.allocate_method else { + // Echo variants: nothing to pre-load; the agent's natural footprint is + // established on first invocation. + return; + }; + + async { + let base = context.base_memory_bytes; + let result_futures = context + .agent_ids + .iter() + .enumerate() + .map(move |(idx, agent_id)| async move { + let user_clone = context.user.clone(); + let bytes = agent_memory_bytes(idx, base); + invoke_and_await_agent( + &user_clone, + &context.component, + agent_id, + allocate_method, + data_value!(bytes), + ) + .await + }) + .collect::>(); + let _ = result_futures.join().await; + } + .instrument(tracing::info_span!( + "warmup_allocate_memory", + agent_count = context.agent_ids.len() + )) + .await; +} + +async fn run( + variant: &SaturationVariant, + context: &SaturationIterationContext, + recorder: BenchmarkRecorder, +) { + let agent_count = context.agent_ids.len(); + let deadline = Instant::now() + RUN_DURATION; + + let result_futures = context + .agent_ids + .iter() + .enumerate() + .map(|(idx, agent_id)| { + let recorder = recorder.clone(); + async move { + let user_clone = context.user.clone(); + + tokio::time::sleep(agent_stagger(idx)).await; + + let mut calls = 0u64; + while Instant::now() < deadline { + let result = invoke_and_await_agent( + &user_clone, + &context.component, + agent_id, + variant.active_method, + (variant.active_params)(), + ) + .await; + result.record(&recorder, "", idx.to_string().as_str()); + calls += 1; + tokio::time::sleep(IDLE_GAP).await; + } + calls + } + }) + .collect::>(); + + let started = Instant::now(); + let per_agent_calls = result_futures.join().await; + let elapsed = started.elapsed(); + + // Aggregate sustained throughput over the fixed run window. Across `size` + // steps, this reveals where added active agents stop adding throughput + // (memory saturation / eviction churn dominates) — the knee we are after. + let total_calls: u64 = per_agent_calls.iter().sum(); + let secs = elapsed.as_secs_f64(); + if secs > 0.0 { + let ops_per_sec = (total_calls as f64 / secs).round() as u64; + info!( + "saturation: {agent_count} agents, {total_calls} calls in {secs:.1}s = {ops_per_sec} ops/sec" + ); + recorder.count( + &ResultKey::primary("saturation-throughput-ops-per-sec"), + ops_per_sec, + ); + } +} + +async fn cleanup_iteration(context: SaturationIterationContext) { + let agent_ids: Vec = context + .agent_ids + .iter() + .filter_map(|agent_id| AgentId::from_agent_id(context.component.id, agent_id).ok()) + .collect(); + delete_workers(&context.user, &agent_ids).await; + cleanup_user_state(&context.user, &context.env_id).await; +} + +/// Generates a `Benchmark` impl wrapper for a saturation variant. +macro_rules! saturation_benchmark { + ($ty:ident, $bench_name:literal, $variant:expr, $description:literal) => { + pub struct $ty { + config: RunConfig, + } + + #[async_trait] + impl Benchmark for $ty { + type BenchmarkContext = SaturationBenchmarkContext; + type IterationContext = SaturationIterationContext; + + fn name() -> &'static str { + $bench_name + } + + fn description() -> &'static str { + indoc! { $description } + } + + async fn create_benchmark_context( + mode: &TestMode, + verbosity: Level, + cluster_size: usize, + disable_compilation_cache: bool, + otlp: bool, + ) -> Self::BenchmarkContext { + create_context( + mode, + verbosity, + cluster_size, + disable_compilation_cache, + otlp, + ) + .await + } + + async fn cleanup(benchmark_context: Self::BenchmarkContext) { + benchmark_context.deps.kill_all().await; + } + + async fn create(_mode: &TestMode, config: RunConfig) -> Self { + Self { config } + } + + async fn setup_iteration( + &self, + benchmark_context: &Self::BenchmarkContext, + ) -> Self::IterationContext { + setup_iteration(&$variant, &self.config, benchmark_context).await + } + + async fn warmup( + &self, + _benchmark_context: &Self::BenchmarkContext, + context: &Self::IterationContext, + ) { + warmup(&$variant, context).await + } + + async fn run( + &self, + _benchmark_context: &Self::BenchmarkContext, + context: &Self::IterationContext, + recorder: BenchmarkRecorder, + ) { + run(&$variant, context, recorder).await + } + + async fn cleanup_iteration( + &self, + _benchmark_context: &Self::BenchmarkContext, + context: Self::IterationContext, + ) { + cleanup_iteration(context).await + } + } + }; +} + +saturation_benchmark!( + ThroughputSaturationCounters, + "throughput-saturation-counters", + COUNTERS_VARIANT, + "Ramps `size` active agents that each retain a deterministic, per-agent-distinct + synthetic memory footprint (controlled by `length`) and do CPU work, measuring + sustained throughput to locate the memory-saturation knee." +); + +saturation_benchmark!( + ThroughputSaturationEchoRust, + "throughput-saturation-echo-rust", + ECHO_RUST_VARIANT, + "Ramps `size` actively-invoked Rust `echo` agents to find how many fit resident + per pod before eviction churn craters throughput. The per-agent footprint is the + agent's natural memory (no synthetic allocation)." +); + +saturation_benchmark!( + ThroughputSaturationEchoTs, + "throughput-saturation-echo-ts", + ECHO_TS_VARIANT, + "Ramps `size` actively-invoked TypeScript `echo` agents to find how many fit + resident per pod before eviction churn craters throughput. The per-agent + footprint is the agent's natural memory, including the QuickJS runtime." +); diff --git a/test-components/agent-counters/Cargo.toml b/test-components/agent-counters/Cargo.toml index c7567da5a5..069f9180f3 100644 --- a/test-components/agent-counters/Cargo.toml +++ b/test-components/agent-counters/Cargo.toml @@ -3,6 +3,12 @@ name = "it_agent_counters" version = "0.0.1" edition = "2024" +# Standalone workspace root: this component is excluded from the golem-oss +# workspace, and when built nested inside another repo's workspace (e.g. the +# cloud-perf CI checkout under golem-cloud) cargo would otherwise walk up and +# attach it to that unrelated workspace. An empty table stops that search. +[workspace] + [profile.release] opt-level = "s" lto = true diff --git a/test-components/agent-counters/src/lib.rs b/test-components/agent-counters/src/lib.rs index b2ac7d4d44..b14840512d 100644 --- a/test-components/agent-counters/src/lib.rs +++ b/test-components/agent-counters/src/lib.rs @@ -3,6 +3,43 @@ pub mod repository; use golem_rust::{agent_definition, agent_implementation, generate_idempotency_key}; +/// Page size used when touching retained memory so the OS backs it with real +/// resident pages rather than leaving it as untouched (non-resident) reservation. +const PAGE_SIZE: usize = 4096; + +/// Spins doing cheap arithmetic for approximately `millis` milliseconds, polling +/// the monotonic clock between batches of work rather than on every iteration so +/// the workload is CPU-bound, not clock-syscall-bound. Returns an accumulated +/// value so the work cannot be optimised away. +fn busy_loop(millis: u32) -> u32 { + let deadline = std::time::Duration::from_millis(millis as u64); + let start = std::time::Instant::now(); + let mut acc: u32 = 0; + loop { + for i in 0..10_000u32 { + acc = acc.wrapping_add(i).wrapping_mul(31).wrapping_add(7); + } + if start.elapsed() >= deadline { + break; + } + } + acc +} + +/// Grows `buffer` to hold `bytes` and touches one byte per page so the memory +/// becomes resident (real RSS), not just reserved address space. +fn retain_memory(buffer: &mut Vec, bytes: u32) { + let bytes = bytes as usize; + buffer.clear(); + buffer.shrink_to_fit(); + buffer.resize(bytes, 0); + let mut page = 0; + while page < bytes { + buffer[page] = buffer[page].wrapping_add(1); + page += PAGE_SIZE; + } +} + #[agent_definition] trait Counter { fn new(id: String) -> Self; @@ -10,17 +47,32 @@ trait Counter { async fn increment_through_rpc(&mut self) -> u32; async fn increment_through_rpc_to_ephemeral(&mut self) -> u32; async fn increment_through_rpc_to_ephemeral_phantom(&mut self) -> u32; + + /// Spins for `millis` milliseconds of cheap CPU work, then increments and + /// returns the counter. Used to define an "active" agent without making the + /// workload oplog-bound on a tight loop. + fn busy_for(&mut self, millis: u32) -> u32; + + /// Retains `bytes` of resident linear memory in the agent's state and + /// increments the counter. The memory stays resident across invocations so + /// the agent contributes a controllable footprint to the executor's pool. + fn allocate_memory(&mut self, bytes: u32) -> u32; } struct CounterImpl { count: u32, id: String, + retained: Vec, } #[agent_implementation] impl Counter for CounterImpl { fn new(id: String) -> Self { - Self { id, count: 0 } + Self { + id, + count: 0, + retained: Vec::new(), + } } fn increment(&mut self) -> u32 { @@ -42,29 +94,64 @@ impl Counter for CounterImpl { let mut client = EphemeralSingletonCounterClient::new_phantom(); client.increment().await } + + fn busy_for(&mut self, millis: u32) -> u32 { + let _ = busy_loop(millis); + self.count += 1; + self.count + } + + fn allocate_memory(&mut self, bytes: u32) -> u32 { + retain_memory(&mut self.retained, bytes); + self.count += 1; + self.count + } } #[agent_definition(ephemeral)] trait EphemeralCounter { fn new(id: String) -> Self; fn increment(&mut self) -> u32; + + /// See [`Counter::busy_for`]. + fn busy_for(&mut self, millis: u32) -> u32; + + /// See [`Counter::allocate_memory`]. + fn allocate_memory(&mut self, bytes: u32) -> u32; } struct EphemeralCounterImpl { count: u32, _id: String, + retained: Vec, } #[agent_implementation] impl EphemeralCounter for EphemeralCounterImpl { fn new(id: String) -> Self { - Self { _id: id, count: 0 } + Self { + _id: id, + count: 0, + retained: Vec::new(), + } } fn increment(&mut self) -> u32 { self.count += 1; self.count } + + fn busy_for(&mut self, millis: u32) -> u32 { + let _ = busy_loop(millis); + self.count += 1; + self.count + } + + fn allocate_memory(&mut self, bytes: u32) -> u32 { + retain_memory(&mut self.retained, bytes); + self.count += 1; + self.count + } }