From e929814edcc0dcf0f79e0e83a3f31ff624ec2793 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Wed, 20 May 2026 06:41:34 +0800 Subject: [PATCH] feat(ecstore): add internode transport boundary and TCP baseline runner (#3010) * docs: add internode data transport RFC * feat: add internode operation metrics * fix feedback * fix(ci): fallback protoc token to github.token * feat(ecstore): add internode transport boundary and baseline runner * feat(internode): harden data transport baseline * Revert "feat(internode): harden data transport baseline" This reverts commit 5b8d6b8aa48a8e8100f97931ec3db7dd6ee9b9d4. * fix(internode): address baseline review comments * fix(ci): pin setup-protoc to stable release * fix(ci): install protoc via apt on linux * fix(ci): restore protoc install for macos and windows --------- Co-authored-by: Henry Guo Co-authored-by: houseme --- .github/actions/setup/action.yml | 7 +- INTERNODE_DATA_TRANSPORT_RFC.md | 10 + crates/config/src/constants/internode.rs | 6 + .../src/rpc/internode_data_transport.rs | 165 +++++++++ crates/ecstore/src/rpc/mod.rs | 1 + crates/ecstore/src/rpc/remote_disk.rs | 112 +++--- scripts/run_internode_transport_baseline.sh | 327 ++++++++++++++++++ 7 files changed, 558 insertions(+), 70 deletions(-) create mode 100644 crates/ecstore/src/rpc/internode_data_transport.rs create mode 100755 scripts/run_internode_transport_baseline.sh diff --git a/.github/actions/setup/action.yml b/.github/actions/setup/action.yml index ac7b9c363..d30454ea1 100644 --- a/.github/actions/setup/action.yml +++ b/.github/actions/setup/action.yml @@ -53,13 +53,14 @@ runs: musl-tools \ build-essential \ pkg-config \ - libssl-dev + libssl-dev \ + protobuf-compiler - name: Install protoc uses: rustfs/setup-protoc@v3.0.1 with: - version: "34.1" - repo-token: ${{ inputs.github-token || github.token }} + version: "29.3" + repo-token: ${{ github.token }} - name: Install flatc uses: Nugine/setup-flatc@v1 diff --git a/INTERNODE_DATA_TRANSPORT_RFC.md b/INTERNODE_DATA_TRANSPORT_RFC.md index 117ff23be..6d76672d5 100644 --- a/INTERNODE_DATA_TRANSPORT_RFC.md +++ b/INTERNODE_DATA_TRANSPORT_RFC.md @@ -169,6 +169,7 @@ Existing benchmark assets: - `scripts/run_object_batch_bench_enhanced.sh` - `scripts/run_object_batch_bench_abc.sh` - `scripts/run_four_node_cluster_failover_bench.sh` +- `scripts/run_internode_transport_baseline.sh` (scenario matrix wrapper for local vs distributed TCP baseline artifacts) - Criterion benches under `crates/ecstore/benches/` These mostly cover S3/object workload or erasure coding performance. They do @@ -222,6 +223,15 @@ The baseline should produce a machine-readable artifact, for example `target/bench/internode-transport//summary.csv`, plus the exact commands and configuration used. +### Baseline runner entry point + +Use `scripts/run_internode_transport_baseline.sh` to execute a reproducible +S3 PUT/GET matrix against `local` and `distributed` scenarios and export: + +- `summary.csv` (throughput/latency summary per workload and object size) +- `internode_metric_deltas.csv` (operation-level internode metric deltas when + `--metrics-url` is provided) + ## Transport Abstraction Proposal ### Design principle diff --git a/crates/config/src/constants/internode.rs b/crates/config/src/constants/internode.rs index aa9b861e6..c45febf74 100644 --- a/crates/config/src/constants/internode.rs +++ b/crates/config/src/constants/internode.rs @@ -32,6 +32,10 @@ pub const DEFAULT_INTERNODE_HTTP2_KEEPALIVE_TIMEOUT_SECS: u64 = 3; pub const ENV_INTERNODE_RPC_TIMEOUT_SECS: &str = "RUSTFS_INTERNODE_RPC_TIMEOUT_SECS"; pub const DEFAULT_INTERNODE_RPC_TIMEOUT_SECS: u64 = 10; +/// Environment variable for selecting the internode data-plane transport backend. +pub const ENV_RUSTFS_INTERNODE_DATA_TRANSPORT: &str = "RUSTFS_INTERNODE_DATA_TRANSPORT"; +pub const DEFAULT_INTERNODE_DATA_TRANSPORT: &str = "tcp-http"; + #[cfg(test)] mod tests { use super::*; @@ -58,5 +62,7 @@ mod tests { "RUSTFS_INTERNODE_HTTP2_KEEPALIVE_TIMEOUT_SECS" ); assert_eq!(ENV_INTERNODE_RPC_TIMEOUT_SECS, "RUSTFS_INTERNODE_RPC_TIMEOUT_SECS"); + assert_eq!(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT, "RUSTFS_INTERNODE_DATA_TRANSPORT"); + assert_eq!(DEFAULT_INTERNODE_DATA_TRANSPORT, "tcp-http"); } } diff --git a/crates/ecstore/src/rpc/internode_data_transport.rs b/crates/ecstore/src/rpc/internode_data_transport.rs new file mode 100644 index 000000000..fd83e81ef --- /dev/null +++ b/crates/ecstore/src/rpc/internode_data_transport.rs @@ -0,0 +1,165 @@ +// Copyright 2024 RustFS Team +// +// 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. + +use crate::disk::error::Result; +use crate::disk::{FileReader, FileWriter}; +use crate::rpc::build_auth_headers; +use async_trait::async_trait; +use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE}; +use rustfs_config::{DEFAULT_INTERNODE_DATA_TRANSPORT, ENV_RUSTFS_INTERNODE_DATA_TRANSPORT}; +use rustfs_rio::{HttpReader, HttpWriter}; +use std::sync::{Arc, OnceLock}; +use std::time::Duration; +use tracing::warn; + +static INTERNODE_DATA_TRANSPORT: OnceLock> = OnceLock::new(); + +#[derive(Debug, Clone)] +pub struct ReadStreamRequest { + pub endpoint: String, + pub disk: String, + pub volume: String, + pub path: String, + pub offset: usize, + pub length: usize, +} + +#[derive(Debug, Clone)] +pub struct WriteStreamRequest { + pub endpoint: String, + pub disk: String, + pub volume: String, + pub path: String, + pub append: bool, + pub size: i64, +} + +#[derive(Debug, Clone)] +pub struct WalkDirStreamRequest { + pub endpoint: String, + pub disk: String, + pub body: Vec, + pub stall_timeout: Option, +} + +#[async_trait] +pub trait InternodeDataTransport: Send + Sync + std::fmt::Debug { + async fn open_read(&self, request: ReadStreamRequest) -> Result; + async fn open_write(&self, request: WriteStreamRequest) -> Result; + async fn open_walk_dir(&self, request: WalkDirStreamRequest) -> Result; + fn name(&self) -> &'static str; +} + +#[derive(Debug, Default)] +pub struct TcpHttpInternodeDataTransport; + +#[async_trait] +impl InternodeDataTransport for TcpHttpInternodeDataTransport { + async fn open_read(&self, request: ReadStreamRequest) -> Result { + let url = format!( + "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}", + request.endpoint, + urlencoding::encode(&request.disk), + urlencoding::encode(&request.volume), + urlencoding::encode(&request.path), + request.offset, + request.length + ); + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + build_auth_headers(&url, &Method::GET, &mut headers)?; + Ok(Box::new(HttpReader::new(url, Method::GET, headers, None).await?)) + } + + async fn open_write(&self, request: WriteStreamRequest) -> Result { + let url = format!( + "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}", + request.endpoint, + urlencoding::encode(&request.disk), + urlencoding::encode(&request.volume), + urlencoding::encode(&request.path), + request.append, + request.size + ); + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + build_auth_headers(&url, &Method::PUT, &mut headers)?; + Ok(Box::new(HttpWriter::new(url, Method::PUT, headers).await?)) + } + + async fn open_walk_dir(&self, request: WalkDirStreamRequest) -> Result { + let url = format!("{}/rustfs/rpc/walk_dir?disk={}", request.endpoint, urlencoding::encode(&request.disk)); + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + build_auth_headers(&url, &Method::GET, &mut headers)?; + Ok(Box::new( + HttpReader::new_with_stall_timeout(url, Method::GET, headers, Some(request.body), request.stall_timeout).await?, + )) + } + + fn name(&self) -> &'static str { + DEFAULT_INTERNODE_DATA_TRANSPORT + } +} + +fn build_internode_data_transport(configured_transport: Option<&str>) -> Arc { + match configured_transport.map(str::trim).filter(|transport| !transport.is_empty()) { + Some(transport) if transport.eq_ignore_ascii_case(DEFAULT_INTERNODE_DATA_TRANSPORT) => { + Arc::new(TcpHttpInternodeDataTransport) + } + Some(transport) => { + warn!( + env = ENV_RUSTFS_INTERNODE_DATA_TRANSPORT, + requested = %transport, + fallback = DEFAULT_INTERNODE_DATA_TRANSPORT, + "unknown internode data transport, using default backend" + ); + Arc::new(TcpHttpInternodeDataTransport) + } + None => Arc::new(TcpHttpInternodeDataTransport), + } +} + +pub fn build_internode_data_transport_from_env() -> Arc { + Arc::clone( + INTERNODE_DATA_TRANSPORT + .get_or_init(|| build_internode_data_transport(std::env::var(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT).ok().as_deref())), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn transport_config_defaults_to_tcp_http() { + let transport = build_internode_data_transport(None); + + assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT); + } + + #[test] + fn transport_config_accepts_tcp_http() { + let transport = build_internode_data_transport(Some("TCP-HTTP")); + + assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT); + } + + #[test] + fn transport_config_falls_back_to_tcp_http_for_unknown_backend() { + let transport = build_internode_data_transport(Some("rdma")); + + assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT); + } +} diff --git a/crates/ecstore/src/rpc/mod.rs b/crates/ecstore/src/rpc/mod.rs index f2c0c77c4..643ee4bdc 100644 --- a/crates/ecstore/src/rpc/mod.rs +++ b/crates/ecstore/src/rpc/mod.rs @@ -15,6 +15,7 @@ mod client; mod context_propagation; mod http_auth; +mod internode_data_transport; mod peer_rest_client; mod peer_s3_client; mod remote_disk; diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index 4a575d9bd..7db819148 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::disk::error::{Error, Result}; use crate::disk::{ CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, FileInfoVersions, FileReader, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, @@ -27,14 +28,12 @@ use crate::disk::{disk_store::DiskHealthTracker, error::DiskError, local::ScanGu use crate::rpc::client::{ TonicInterceptor, gen_tonic_signature_interceptor, is_network_like_disk_error, node_service_time_out_client, }; -use crate::set_disk::DEFAULT_READ_BUFFER_SIZE; -use crate::{ - disk::error::{Error, Result}, - rpc::build_auth_headers, +use crate::rpc::internode_data_transport::{ + InternodeDataTransport, ReadStreamRequest, WalkDirStreamRequest, WriteStreamRequest, build_internode_data_transport_from_env, }; +use crate::set_disk::DEFAULT_READ_BUFFER_SIZE; use bytes::Bytes; use futures::lock::Mutex; -use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE}; use metrics::counter; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; use rustfs_io_metrics::internode_metrics::{ @@ -49,7 +48,6 @@ use rustfs_protos::proto_gen::node_service::{ RenameFileRequest, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, WriteMetadataRequest, node_service_client::NodeServiceClient, }; -use rustfs_rio::{HttpReader, HttpWriter}; use serde::{Serialize, de::DeserializeOwned}; use std::{ io::Cursor, @@ -109,6 +107,7 @@ pub struct RemoteDisk { health: Arc, /// Cancellation token for monitoring tasks cancel_token: CancellationToken, + data_transport: Arc, } impl RemoteDisk { @@ -130,6 +129,7 @@ impl RemoteDisk { health_check: opt.health_check && env_health_check, health: Arc::new(DiskHealthTracker::new()), cancel_token: CancellationToken::new(), + data_transport: build_internode_data_transport_from_env(), }; record_drive_runtime_state(ep, RuntimeDriveHealthState::Online); @@ -1200,23 +1200,16 @@ impl DiskAPI for RemoteDisk { "walk_dir", || async { let disk = self.disk_ref().await; - - let url = format!("{}/rustfs/rpc/walk_dir?disk={}", self.endpoint.grid_host(), urlencoding::encode(&disk),); - let opts = serde_json::to_vec(&opts)?; - - let mut headers = HeaderMap::new(); - headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); - build_auth_headers(&url, &Method::GET, &mut headers)?; - - let mut reader = HttpReader::new_with_stall_timeout( - url, - Method::GET, - headers, - Some(opts), - Some(get_drive_walkdir_stall_timeout()), - ) - .await?; + let mut reader = self + .data_transport + .open_walk_dir(WalkDirStreamRequest { + endpoint: self.endpoint.grid_host(), + disk, + body: opts, + stall_timeout: Some(get_drive_walkdir_stall_timeout()), + }) + .await?; copy_stream_with_buffer(&mut reader, wr, DEFAULT_READ_BUFFER_SIZE).await?; @@ -1248,21 +1241,16 @@ impl DiskAPI for RemoteDisk { return Err(DiskError::FaultyDisk); } let disk = self.disk_ref().await; - - let url = format!( - "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}", - self.endpoint.grid_host(), - urlencoding::encode(&disk), - urlencoding::encode(volume), - urlencoding::encode(path), - offset, - length - ); - - let mut headers = HeaderMap::new(); - headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); - build_auth_headers(&url, &Method::GET, &mut headers)?; - Ok(Box::new(HttpReader::new(url, Method::GET, headers, None).await?)) + self.data_transport + .open_read(ReadStreamRequest { + endpoint: self.endpoint.grid_host(), + disk, + volume: volume.to_string(), + path: path.to_string(), + offset, + length, + }) + .await } /// Zero-copy read for remote disks falls back to efficient network read. @@ -1291,21 +1279,16 @@ impl DiskAPI for RemoteDisk { return Err(DiskError::FaultyDisk); } let disk = self.disk_ref().await; - - let url = format!( - "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}", - self.endpoint.grid_host(), - urlencoding::encode(&disk), - urlencoding::encode(volume), - urlencoding::encode(path), - true, - 0 - ); - - let mut headers = HeaderMap::new(); - headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); - build_auth_headers(&url, &Method::PUT, &mut headers)?; - Ok(Box::new(HttpWriter::new(url, Method::PUT, headers).await?)) + self.data_transport + .open_write(WriteStreamRequest { + endpoint: self.endpoint.grid_host(), + disk, + volume: volume.to_string(), + path: path.to_string(), + append: true, + size: 0, + }) + .await } #[tracing::instrument(level = "debug", skip(self))] @@ -1322,21 +1305,16 @@ impl DiskAPI for RemoteDisk { return Err(DiskError::FaultyDisk); } let disk = self.disk_ref().await; - - let url = format!( - "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}", - self.endpoint.grid_host(), - urlencoding::encode(&disk), - urlencoding::encode(volume), - urlencoding::encode(path), - false, - file_size - ); - - let mut headers = HeaderMap::new(); - headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); - build_auth_headers(&url, &Method::PUT, &mut headers)?; - Ok(Box::new(HttpWriter::new(url, Method::PUT, headers).await?)) + self.data_transport + .open_write(WriteStreamRequest { + endpoint: self.endpoint.grid_host(), + disk, + volume: volume.to_string(), + path: path.to_string(), + append: false, + size: file_size, + }) + .await } #[tracing::instrument(level = "debug", skip(self))] diff --git a/scripts/run_internode_transport_baseline.sh b/scripts/run_internode_transport_baseline.sh new file mode 100755 index 000000000..700d3cfd8 --- /dev/null +++ b/scripts/run_internode_transport_baseline.sh @@ -0,0 +1,327 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Internode transport baseline runner. +# Reuses scripts/run_object_batch_bench.sh and exports reproducible artifacts: +# - object benchmark summaries per scenario/workload/concurrency +# - optional internode operation metric deltas from a Prometheus text endpoint + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PROJECT_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)" +OBJECT_BENCH_SCRIPT="${PROJECT_ROOT}/scripts/run_object_batch_bench.sh" + +TOOL="warp" +ACCESS_KEY="" +SECRET_KEY="" +REGION="us-east-1" +BUCKET_PREFIX="rustfs-internode-bench" +OUT_DIR="" +SIZES="4KiB,1MiB,16MiB,128MiB,1GiB" +CONCURRENCIES="1,16,64,128" +DURATION="90s" +SAMPLES=20000 +INSECURE=false +EXTRA_ARGS="" +WARP_BIN="warp" +S3BENCH_BIN="s3bench" + +# Scenario format: name=endpoint +SCENARIOS="local=http://127.0.0.1:9000,distributed=http://127.0.0.1:9001" + +# Optional Prometheus text exposition URL. +# If empty, metrics delta collection is skipped. +INTERNODE_METRICS_URL="" + +usage() { + cat <<'USAGE' +Usage: + scripts/run_internode_transport_baseline.sh --access-key --secret-key [options] + +Required: + --access-key + --secret-key + +Optional: + --tool Default: warp + --region Default: us-east-1 + --bucket-prefix Default: rustfs-internode-bench + --scenarios Default: local=http://127.0.0.1:9000,distributed=http://127.0.0.1:9001 + --sizes Default: 4KiB,1MiB,16MiB,128MiB,1GiB + --concurrencies Default: 1,16,64,128 + --duration Default: 90s + --samples Default: 20000 (for s3bench) + --warp-bin Default: warp + --s3bench-bin Default: s3bench + --metrics-url Prometheus text endpoint for internode metrics delta + --out-dir Default: target/bench/internode-transport- + --extra-args "" Passed to run_object_batch_bench.sh --extra-args + --insecure TLS insecure (self-signed) + --dry-run Print commands only + -h, --help + +Notes: + - This baseline covers S3 PUT/GET workloads and records internode metric deltas when --metrics-url is set. + - Healing/replication-specific workloads should be run separately and appended to the same artifact directory. +USAGE +} + +require_cmd() { + if ! command -v "$1" >/dev/null 2>&1; then + echo "ERROR: command not found: $1" >&2 + exit 1 + fi +} + +DRY_RUN=false +parse_args() { + while [[ $# -gt 0 ]]; do + case "$1" in + --access-key) ACCESS_KEY="$2"; shift 2 ;; + --secret-key) SECRET_KEY="$2"; shift 2 ;; + --tool) TOOL="$2"; shift 2 ;; + --region) REGION="$2"; shift 2 ;; + --bucket-prefix) BUCKET_PREFIX="$2"; shift 2 ;; + --scenarios) SCENARIOS="$2"; shift 2 ;; + --sizes) SIZES="$2"; shift 2 ;; + --concurrencies) CONCURRENCIES="$2"; shift 2 ;; + --duration) DURATION="$2"; shift 2 ;; + --samples) SAMPLES="$2"; shift 2 ;; + --warp-bin) WARP_BIN="$2"; shift 2 ;; + --s3bench-bin) S3BENCH_BIN="$2"; shift 2 ;; + --metrics-url) INTERNODE_METRICS_URL="$2"; shift 2 ;; + --out-dir) OUT_DIR="$2"; shift 2 ;; + --extra-args) EXTRA_ARGS="$2"; shift 2 ;; + --insecure) INSECURE=true; shift ;; + --dry-run) DRY_RUN=true; shift ;; + -h|--help) usage; exit 0 ;; + *) + echo "ERROR: unknown arg: $1" >&2 + usage + exit 1 + ;; + esac + done +} + +validate_args() { + if [[ -z "${ACCESS_KEY}" || -z "${SECRET_KEY}" ]]; then + echo "ERROR: --access-key and --secret-key are required" >&2 + exit 1 + fi + if [[ "${TOOL}" != "warp" && "${TOOL}" != "s3bench" ]]; then + echo "ERROR: --tool must be warp or s3bench" >&2 + exit 1 + fi + if [[ -z "${SCENARIOS}" ]]; then + echo "ERROR: --scenarios cannot be empty" >&2 + exit 1 + fi +} + +setup_output() { + if [[ -z "${OUT_DIR}" ]]; then + OUT_DIR="target/bench/internode-transport-$(date +%Y%m%d-%H%M%S)" + fi + mkdir -p "${OUT_DIR}" + echo "scenario,endpoint,workload,concurrency,size,status,throughput,requests_per_sec,avg_latency,log_file,run_dir" > "${OUT_DIR}/summary.csv" + if [[ -n "${INTERNODE_METRICS_URL}" ]]; then + echo "scenario,workload,concurrency,size,metric,operation,before,after,delta" > "${OUT_DIR}/internode_metric_deltas.csv" + fi +} + +collect_internode_snapshot() { + local snapshot_file="$1" + if [[ -z "${INTERNODE_METRICS_URL}" ]]; then + : > "${snapshot_file}" + return 0 + fi + if [[ "${DRY_RUN}" == "true" ]]; then + : > "${snapshot_file}" + return 0 + fi + if ! curl -fsSL "${INTERNODE_METRICS_URL}" > "${snapshot_file}"; then + echo "WARN: failed to fetch metrics from ${INTERNODE_METRICS_URL}, skipping metrics delta for this run" >&2 + : > "${snapshot_file}" + fi +} + +extract_internode_rows() { + local src="$1" + if [[ ! -s "${src}" ]]; then + return 0 + fi + awk ' + $1 ~ /^rustfs_system_network_internode_operation_/ { + metric = $1 + op = "all" + if (match($0, /operation="[^"]+"/)) { + op = substr($0, RSTART + 11, RLENGTH - 12) + } + n = split($0, parts, " ") + value = parts[n] + gsub(/[[:space:]]+/, "", value) + if (value ~ /^[0-9]+([.][0-9]+)?$/) { + print metric "," op "," value + } + }' "${src}" +} + +append_metric_deltas() { + local scenario="$1" + local workload="$2" + local conc="$3" + local size="$4" + local before_file="$5" + local after_file="$6" + + local before_rows after_rows + before_rows="$(mktemp)" + after_rows="$(mktemp)" + extract_internode_rows "${before_file}" > "${before_rows}" + extract_internode_rows "${after_file}" > "${after_rows}" + + awk -F',' -v scenario="${scenario}" -v workload="${workload}" -v conc="${conc}" -v size="${size}" ' + FNR==NR { + key = $1 SUBSEP $2 + before[key] = $3 + next + } + { + key = $1 SUBSEP $2 + metric = $1 + operation = $2 + afterv = $3 + 0 + beforev = (key in before ? before[key] + 0 : 0) + delta = afterv - beforev + printf "%s,%s,%s,%s,%s,%s,%.0f,%.0f,%.0f\n", scenario, workload, conc, size, metric, operation, beforev, afterv, delta + } + ' "${before_rows}" "${after_rows}" >> "${OUT_DIR}/internode_metric_deltas.csv" + + rm -f "${before_rows}" "${after_rows}" +} + +append_object_summary() { + local scenario="$1" + local endpoint="$2" + local workload="$3" + local conc="$4" + local run_dir="$5" + + local src="${run_dir}/summary.csv" + if [[ ! -f "${src}" ]]; then + echo "WARN: missing summary file: ${src}" >&2 + return 0 + fi + + awk -F',' -v scenario="${scenario}" -v endpoint="${endpoint}" -v workload="${workload}" -v run_dir="${run_dir}" ' + NR == 1 { next } + { + printf "%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s\n", scenario, endpoint, workload, $3, $1, $4, $5, $6, $7, $8, run_dir + } + ' "${src}" >> "${OUT_DIR}/summary.csv" +} + +run_workload() { + local scenario="$1" + local endpoint="$2" + local workload="$3" + local conc="$4" + + local run_dir="${OUT_DIR}/${scenario}/${workload}/concurrency-${conc}" + local bucket="${BUCKET_PREFIX}-${scenario}-${workload}-c${conc}" + local before_metrics="${run_dir}/metrics_before.prom" + local after_metrics="${run_dir}/metrics_after.prom" + + mkdir -p "${run_dir}" + if [[ -n "${INTERNODE_METRICS_URL}" ]]; then + collect_internode_snapshot "${before_metrics}" + fi + + local cmd=( + "${OBJECT_BENCH_SCRIPT}" + --tool "${TOOL}" + --endpoint "${endpoint}" + --access-key "${ACCESS_KEY}" + --secret-key "${SECRET_KEY}" + --bucket "${bucket}" + --region "${REGION}" + --concurrency "${conc}" + --sizes "${SIZES}" + --out-dir "${run_dir}" + ) + if [[ "${TOOL}" == "warp" ]]; then + cmd+=(--duration "${DURATION}" --warp-mode "${workload}" --warp-bin "${WARP_BIN}") + else + cmd+=(--samples "${SAMPLES}" --s3bench-bin "${S3BENCH_BIN}") + fi + if [[ "${INSECURE}" == "true" ]]; then + cmd+=(--insecure) + fi + if [[ -n "${EXTRA_ARGS}" ]]; then + cmd+=(--extra-args "${EXTRA_ARGS}") + fi + if [[ "${DRY_RUN}" == "true" ]]; then + cmd+=(--dry-run) + fi + + printf '==> scenario=%s workload=%s concurrency=%s endpoint=%s\n' "${scenario}" "${workload}" "${conc}" "${endpoint}" + "${cmd[@]}" + + append_object_summary "${scenario}" "${endpoint}" "${workload}" "${conc}" "${run_dir}" + if [[ -n "${INTERNODE_METRICS_URL}" ]]; then + collect_internode_snapshot "${after_metrics}" + append_metric_deltas "${scenario}" "${workload}" "${conc}" "all_sizes" "${before_metrics}" "${after_metrics}" + fi +} + +run_all() { + local scenario_pair scenario endpoint + local workload conc + + IFS=',' read -r -a scenario_list <<< "${SCENARIOS}" + IFS=',' read -r -a conc_list <<< "${CONCURRENCIES}" + + for scenario_pair in "${scenario_list[@]}"; do + scenario="${scenario_pair%%=*}" + endpoint="${scenario_pair#*=}" + if [[ -z "${scenario}" || -z "${endpoint}" || "${scenario}" == "${endpoint}" ]]; then + echo "ERROR: invalid scenario entry: ${scenario_pair} (expected name=url)" >&2 + exit 1 + fi + + for workload in put get; do + for conc in "${conc_list[@]}"; do + conc="$(echo "${conc}" | xargs)" + if ! [[ "${conc}" =~ ^[0-9]+$ ]] || [[ "${conc}" -le 0 ]]; then + echo "ERROR: invalid concurrency: ${conc}" >&2 + exit 1 + fi + run_workload "${scenario}" "${endpoint}" "${workload}" "${conc}" + done + done + done +} + +main() { + parse_args "$@" + validate_args + require_cmd awk + if [[ -n "${INTERNODE_METRICS_URL}" && "${DRY_RUN}" != "true" ]]; then + require_cmd curl + fi + if [[ ! -x "${OBJECT_BENCH_SCRIPT}" ]]; then + echo "ERROR: benchmark script not executable: ${OBJECT_BENCH_SCRIPT}" >&2 + exit 1 + fi + + setup_output + run_all + + echo "Artifacts:" + echo " ${OUT_DIR}/summary.csv" + if [[ -n "${INTERNODE_METRICS_URL}" ]]; then + echo " ${OUT_DIR}/internode_metric_deltas.csv" + fi +} + +main "$@"