diff --git a/crates/ecstore/src/rpc/internode_data_transport.rs b/crates/ecstore/src/rpc/internode_data_transport.rs index fd83e81ef..7892cbb90 100644 --- a/crates/ecstore/src/rpc/internode_data_transport.rs +++ b/crates/ecstore/src/rpc/internode_data_transport.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::disk::error::Result; +use crate::disk::error::{Error, Result}; use crate::disk::{FileReader, FileWriter}; use crate::rpc::build_auth_headers; use async_trait::async_trait; @@ -21,9 +21,43 @@ use rustfs_config::{DEFAULT_INTERNODE_DATA_TRANSPORT, ENV_RUSTFS_INTERNODE_DATA_ use rustfs_rio::{HttpReader, HttpWriter}; use std::sync::{Arc, OnceLock}; use std::time::Duration; -use tracing::warn; -static INTERNODE_DATA_TRANSPORT: OnceLock> = OnceLock::new(); +pub const INTERNODE_DATA_TRANSPORT_TCP: &str = "tcp"; +static INTERNODE_DATA_TRANSPORT: OnceLock, String>> = OnceLock::new(); + +const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream"; +const PUT_FILE_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream"; +const WALK_DIR_PATH: &str = "/rustfs/rpc/walk_dir"; +const CONTENT_TYPE_JSON: &str = "application/json"; + +fn unsupported_transport_message(transport: &str) -> String { + format!( + "invalid {ENV_RUSTFS_INTERNODE_DATA_TRANSPORT}={transport:?}; supported values: {DEFAULT_INTERNODE_DATA_TRANSPORT}, {INTERNODE_DATA_TRANSPORT_TCP}" + ) +} + +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +pub struct InternodeDataTransportCapabilities { + pub stream_read: bool, + pub stream_write: bool, + pub walk_dir: bool, + pub registered_memory: bool, + pub scatter_gather: bool, + pub zero_copy_receive: bool, +} + +impl InternodeDataTransportCapabilities { + pub const fn tcp_http() -> Self { + Self { + stream_read: true, + stream_write: true, + walk_dir: true, + registered_memory: false, + scatter_gather: false, + zero_copy_receive: false, + } + } +} #[derive(Debug, Clone)] pub struct ReadStreamRequest { @@ -59,6 +93,7 @@ pub trait InternodeDataTransport: Send + Sync + std::fmt::Debug { async fn open_write(&self, request: WriteStreamRequest) -> Result; async fn open_walk_dir(&self, request: WalkDirStreamRequest) -> Result; fn name(&self) -> &'static str; + fn capabilities(&self) -> InternodeDataTransportCapabilities; } #[derive(Debug, Default)] @@ -67,41 +102,22 @@ 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")); + let url = build_read_file_stream_url(&request); + let mut headers = json_headers(); 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")); + let url = build_put_file_stream_url(&request); + let mut headers = json_headers(); 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")); + let url = build_walk_dir_url(&request); + let mut headers = json_headers(); 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?, @@ -111,55 +127,195 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport { 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), + fn capabilities(&self) -> InternodeDataTransportCapabilities { + InternodeDataTransportCapabilities::tcp_http() } } -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())), +fn build_read_file_stream_url(request: &ReadStreamRequest) -> String { + format!( + "{}{}?disk={}&volume={}&path={}&offset={}&length={}", + request.endpoint, + READ_FILE_STREAM_PATH, + urlencoding::encode(&request.disk), + urlencoding::encode(&request.volume), + urlencoding::encode(&request.path), + request.offset, + request.length ) } +fn build_put_file_stream_url(request: &WriteStreamRequest) -> String { + format!( + "{}{}?disk={}&volume={}&path={}&append={}&size={}", + request.endpoint, + PUT_FILE_STREAM_PATH, + urlencoding::encode(&request.disk), + urlencoding::encode(&request.volume), + urlencoding::encode(&request.path), + request.append, + request.size + ) +} + +fn build_walk_dir_url(request: &WalkDirStreamRequest) -> String { + format!("{}{}?disk={}", request.endpoint, WALK_DIR_PATH, urlencoding::encode(&request.disk)) +} + +fn json_headers() -> HeaderMap { + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static(CONTENT_TYPE_JSON)); + headers +} + +fn build_internode_data_transport_result( + configured_transport: Option<&str>, +) -> std::result::Result, String> { + match configured_transport.map(str::trim).filter(|transport| !transport.is_empty()) { + None => Ok(Arc::new(TcpHttpInternodeDataTransport)), + Some(transport) + if transport.eq_ignore_ascii_case(DEFAULT_INTERNODE_DATA_TRANSPORT) + || transport.eq_ignore_ascii_case(INTERNODE_DATA_TRANSPORT_TCP) => + { + Ok(Arc::new(TcpHttpInternodeDataTransport)) + } + Some(transport) => Err(unsupported_transport_message(transport)), + } +} + +pub fn build_internode_data_transport(configured_transport: Option<&str>) -> Result> { + build_internode_data_transport_result(configured_transport).map_err(Error::other) +} + +pub fn build_internode_data_transport_from_env() -> Result> { + let configured_transport = std::env::var(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT).ok(); + #[cfg(test)] + { + build_internode_data_transport(configured_transport.as_deref()) + } + + #[cfg(not(test))] + INTERNODE_DATA_TRANSPORT + .get_or_init(|| build_internode_data_transport_result(configured_transport.as_deref())) + .as_ref() + .map(Arc::clone) + .map_err(|err| Error::other(err.clone())) +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn tcp_http_capabilities_are_behavior_preserving() { + let transport = TcpHttpInternodeDataTransport; + + assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT); + assert_eq!( + transport.capabilities(), + InternodeDataTransportCapabilities { + stream_read: true, + stream_write: true, + walk_dir: true, + registered_memory: false, + scatter_gather: false, + zero_copy_receive: false, + } + ); + } + + #[test] + fn read_file_stream_url_encodes_query_values() { + let url = build_read_file_stream_url(&ReadStreamRequest { + endpoint: "http://node1:9000".to_string(), + disk: "http://node1:9000/data/rustfs0".to_string(), + volume: ".rustfs.sys".to_string(), + path: "pool.bin/../part.1".to_string(), + offset: 7, + length: 11, + }); + + assert_eq!( + url, + "http://node1:9000/rustfs/rpc/read_file_stream?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0&volume=.rustfs.sys&path=pool.bin%2F..%2Fpart.1&offset=7&length=11" + ); + } + + #[test] + fn put_file_stream_url_encodes_query_values() { + let url = build_put_file_stream_url(&WriteStreamRequest { + endpoint: "http://node1:9000".to_string(), + disk: "http://node1:9000/data/rustfs0".to_string(), + volume: "bucket".to_string(), + path: "object/part.1".to_string(), + append: false, + size: 4096, + }); + + assert_eq!( + url, + "http://node1:9000/rustfs/rpc/put_file_stream?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0&volume=bucket&path=object%2Fpart.1&append=false&size=4096" + ); + } + + #[test] + fn walk_dir_url_encodes_disk_ref() { + let url = build_walk_dir_url(&WalkDirStreamRequest { + endpoint: "http://node1:9000".to_string(), + disk: "http://node1:9000/data/rustfs0".to_string(), + body: Vec::new(), + stall_timeout: None, + }); + + assert_eq!( + url, + "http://node1:9000/rustfs/rpc/walk_dir?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0" + ); + } + #[test] fn transport_config_defaults_to_tcp_http() { - let transport = build_internode_data_transport(None); + let transport = build_internode_data_transport(None).unwrap(); assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT); } #[test] - fn transport_config_accepts_tcp_http() { - let transport = build_internode_data_transport(Some("TCP-HTTP")); + fn transport_config_blank_value_falls_back_to_default() { + let transport = build_internode_data_transport(Some(" ")).unwrap(); 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")); + fn transport_config_accepts_tcp_aliases() { + for configured in [ + DEFAULT_INTERNODE_DATA_TRANSPORT, + INTERNODE_DATA_TRANSPORT_TCP, + "TCP-HTTP", + "TCP", + ] { + let transport = build_internode_data_transport(Some(configured)).unwrap(); - assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT); + assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT); + } + } + + #[test] + fn transport_config_rejects_unknown_backend() { + let err = build_internode_data_transport(Some("rdma")).expect_err("unknown backend should fail closed"); + + assert!(err.to_string().contains(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT)); + assert!(err.to_string().contains("rdma")); + } + + #[test] + fn cached_transport_config_error_uses_raw_message() { + let err = build_internode_data_transport_result(Some("rdma")).expect_err("unknown backend should fail closed"); + + assert!(!err.starts_with("io error ")); + assert!(err.contains(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT)); + assert!(err.contains("rdma")); } } diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index 7db819148..8e180527b 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -129,7 +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(), + data_transport: build_internode_data_transport_from_env()?, }; record_drive_runtime_state(ep, RuntimeDriveHealthState::Online); @@ -1674,6 +1674,7 @@ impl DiskAPI for RemoteDisk { mod tests { use super::*; use rustfs_common::GLOBAL_CONN_MAP; + use rustfs_config::ENV_RUSTFS_INTERNODE_DATA_TRANSPORT; use std::sync::Once; use tokio::io::duplex; use tokio::net::TcpListener; @@ -1694,6 +1695,13 @@ mod tests { }); } + async fn new_remote_disk_with_default_transport(ep: &Endpoint, opt: &DiskOption) -> Result { + temp_env::async_with_vars([(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT, None::<&str>)], async { + RemoteDisk::new(ep, opt).await + }) + .await + } + #[tokio::test] async fn test_remote_disk_creation() { let url = url::Url::parse("http://example.com:9000/path").unwrap(); @@ -1710,7 +1718,7 @@ mod tests { health_check: false, }; - let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap(); + let remote_disk = new_remote_disk_with_default_transport(&endpoint, &disk_option).await.unwrap(); assert!(!remote_disk.is_local()); assert_eq!(remote_disk.endpoint.url, url); @@ -1720,6 +1728,55 @@ mod tests { assert_eq!(remote_disk.host_name(), "example.com:9000"); } + #[tokio::test] + async fn test_remote_disk_creation_rejects_unknown_data_transport_backend() { + let url = url::Url::parse("http://example.com:9000/path").unwrap(); + let endpoint = Endpoint { + url, + is_local: false, + pool_idx: 0, + set_idx: 1, + disk_idx: 2, + }; + let disk_option = DiskOption { + cleanup: false, + health_check: false, + }; + + temp_env::async_with_vars([(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT, Some("rdma"))], async { + let err = RemoteDisk::new(&endpoint, &disk_option) + .await + .expect_err("unknown backend should fail closed"); + assert!(err.to_string().contains(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT)); + assert!(err.to_string().contains("rdma")); + }) + .await; + } + + #[tokio::test] + async fn test_remote_disk_creation_accepts_tcp_alias_backend() { + let url = url::Url::parse("http://example.com:9000/path").unwrap(); + let endpoint = Endpoint { + url, + is_local: false, + pool_idx: 0, + set_idx: 1, + disk_idx: 2, + }; + let disk_option = DiskOption { + cleanup: false, + health_check: false, + }; + + temp_env::async_with_vars([(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT, Some("tcp"))], async { + let remote_disk = RemoteDisk::new(&endpoint, &disk_option) + .await + .expect("tcp alias should be accepted"); + assert_eq!(remote_disk.host_name(), "example.com:9000"); + }) + .await; + } + #[tokio::test] async fn test_remote_disk_basic_properties() { let url = url::Url::parse("http://remote-server:9000").unwrap(); @@ -1736,7 +1793,7 @@ mod tests { health_check: false, }; - let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap(); + let remote_disk = new_remote_disk_with_default_transport(&endpoint, &disk_option).await.unwrap(); // Test basic properties assert!(!remote_disk.is_local()); @@ -1768,7 +1825,7 @@ mod tests { health_check: false, }; - let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap(); + let remote_disk = new_remote_disk_with_default_transport(&endpoint, &disk_option).await.unwrap(); let path = remote_disk.path(); // Remote disk path should be based on the URL path @@ -1794,7 +1851,7 @@ mod tests { health_check: false, }; - let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap(); + let remote_disk = new_remote_disk_with_default_transport(&endpoint, &disk_option).await.unwrap(); assert!(remote_disk.is_online().await); drop(listener); @@ -1825,7 +1882,7 @@ mod tests { health_check: true, }; - let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap(); + let remote_disk = new_remote_disk_with_default_transport(&endpoint, &disk_option).await.unwrap(); remote_disk.enable_health_check(); // wait for health check connect timeout @@ -1868,7 +1925,7 @@ mod tests { health_check: false, }; - let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap(); + let remote_disk = new_remote_disk_with_default_transport(&endpoint, &disk_option).await.unwrap(); // Initially, disk ID should be None let initial_id = remote_disk.get_disk_id().await.unwrap(); @@ -1903,7 +1960,7 @@ mod tests { health_check: false, }; - let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap(); + let remote_disk = new_remote_disk_with_default_transport(&endpoint, &disk_option).await.unwrap(); assert_eq!(remote_disk.disk_ref().await, endpoint.to_string()); let disk_id = Uuid::new_v4(); @@ -1936,7 +1993,7 @@ mod tests { health_check: false, }; - let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap(); + let remote_disk = new_remote_disk_with_default_transport(&endpoint, &disk_option).await.unwrap(); assert!(!remote_disk.is_local()); assert_eq!(remote_disk.host_name(), expected_hostname); @@ -1962,7 +2019,9 @@ mod tests { health_check: false, }; - let remote_disk = RemoteDisk::new(&valid_endpoint, &disk_option).await.unwrap(); + let remote_disk = new_remote_disk_with_default_transport(&valid_endpoint, &disk_option) + .await + .unwrap(); let location = remote_disk.get_disk_location(); assert!(location.valid()); assert_eq!(location.pool_idx, Some(0)); @@ -1978,7 +2037,9 @@ mod tests { disk_idx: -1, }; - let remote_disk_invalid = RemoteDisk::new(&invalid_endpoint, &disk_option).await.unwrap(); + let remote_disk_invalid = new_remote_disk_with_default_transport(&invalid_endpoint, &disk_option) + .await + .unwrap(); let invalid_location = remote_disk_invalid.get_disk_location(); assert!(!invalid_location.valid()); assert_eq!(invalid_location.pool_idx, None); @@ -2002,7 +2063,7 @@ mod tests { health_check: false, }; - let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap(); + let remote_disk = new_remote_disk_with_default_transport(&endpoint, &disk_option).await.unwrap(); // Test close operation (should succeed) let result = remote_disk.close().await; @@ -2020,7 +2081,7 @@ mod tests { disk_idx: 0, }; - let remote_disk = RemoteDisk::new( + let remote_disk = new_remote_disk_with_default_transport( &endpoint, &DiskOption { cleanup: false, @@ -2056,7 +2117,7 @@ mod tests { disk_idx: 0, }; - let remote_disk = RemoteDisk::new( + let remote_disk = new_remote_disk_with_default_transport( &endpoint, &DiskOption { cleanup: false, @@ -2094,7 +2155,7 @@ mod tests { disk_idx: 0, }; - let remote_disk = RemoteDisk::new( + let remote_disk = new_remote_disk_with_default_transport( &endpoint, &DiskOption { cleanup: false, @@ -2133,7 +2194,7 @@ mod tests { disk_idx: 0, }; - let remote_disk = RemoteDisk::new( + let remote_disk = new_remote_disk_with_default_transport( &endpoint, &DiskOption { cleanup: false, @@ -2176,7 +2237,7 @@ mod tests { disk_idx: 0, }; - let remote_disk = RemoteDisk::new( + let remote_disk = new_remote_disk_with_default_transport( &endpoint, &DiskOption { cleanup: false, @@ -2223,7 +2284,7 @@ mod tests { disk_idx: 0, }; - let remote_disk = RemoteDisk::new( + let remote_disk = new_remote_disk_with_default_transport( &endpoint, &DiskOption { cleanup: false, @@ -2275,7 +2336,7 @@ mod tests { disk_idx: 0, }; - let remote_disk = RemoteDisk::new( + let remote_disk = new_remote_disk_with_default_transport( &endpoint, &DiskOption { cleanup: false, @@ -2327,7 +2388,7 @@ mod tests { disk_idx: 0, }; - let remote_disk = RemoteDisk::new( + let remote_disk = new_remote_disk_with_default_transport( &endpoint, &DiskOption { cleanup: false, diff --git a/scripts/run_internode_transport_baseline.sh b/scripts/run_internode_transport_baseline.sh index 700d3cfd8..056b6bedf 100755 --- a/scripts/run_internode_transport_baseline.sh +++ b/scripts/run_internode_transport_baseline.sh @@ -3,6 +3,7 @@ set -euo pipefail # Internode transport baseline runner. # Reuses scripts/run_object_batch_bench.sh and exports reproducible artifacts: +# - run manifest with scenario/tool metadata and git revision # - object benchmark summaries per scenario/workload/concurrency # - optional internode operation metric deltas from a Prometheus text endpoint @@ -61,6 +62,7 @@ Optional: Notes: - This baseline covers S3 PUT/GET workloads and records internode metric deltas when --metrics-url is set. + - The run manifest intentionally omits access keys, secret keys, and extra args to avoid writing credentials to artifacts. - Healing/replication-specific workloads should be run separately and appended to the same artifact directory. USAGE } @@ -129,6 +131,41 @@ setup_output() { fi } +write_run_manifest() { + local manifest="${OUT_DIR}/run_manifest.txt" + local git_commit git_dirty rustc_version + + git_commit="$(git -C "${PROJECT_ROOT}" rev-parse HEAD 2>/dev/null || echo "unknown")" + if [[ -n "$(git -C "${PROJECT_ROOT}" status --porcelain 2>/dev/null || true)" ]]; then + git_dirty="true" + else + git_dirty="false" + fi + rustc_version="$(rustc --version 2>/dev/null || echo "unknown")" + + { + echo "created_utc=$(date -u +%Y-%m-%dT%H:%M:%SZ)" + echo "git_commit=${git_commit}" + echo "git_dirty=${git_dirty}" + echo "rustc_version=${rustc_version}" + echo "kernel=$(uname -srvmo 2>/dev/null || echo "unknown")" + echo "tool=${TOOL}" + echo "region=${REGION}" + echo "bucket_prefix=${BUCKET_PREFIX}" + echo "scenarios=${SCENARIOS}" + echo "sizes=${SIZES}" + echo "concurrencies=${CONCURRENCIES}" + echo "duration=${DURATION}" + echo "samples=${SAMPLES}" + echo "insecure=${INSECURE}" + echo "metrics_url=${INTERNODE_METRICS_URL:-N/A}" + echo "out_dir=${OUT_DIR}" + echo "extra_args_present=$([[ -n "${EXTRA_ARGS}" ]] && echo true || echo false)" + echo "access_key=REDACTED" + echo "secret_key=REDACTED" + } > "${manifest}" +} + collect_internode_snapshot() { local snapshot_file="$1" if [[ -z "${INTERNODE_METRICS_URL}" ]]; then @@ -315,9 +352,11 @@ main() { fi setup_output + write_run_manifest run_all echo "Artifacts:" + echo " ${OUT_DIR}/run_manifest.txt" echo " ${OUT_DIR}/summary.csv" if [[ -n "${INTERNODE_METRICS_URL}" ]]; then echo " ${OUT_DIR}/internode_metric_deltas.csv" diff --git a/scripts/run_object_batch_bench.sh b/scripts/run_object_batch_bench.sh index d266fb86f..0f35ceef2 100755 --- a/scripts/run_object_batch_bench.sh +++ b/scripts/run_object_batch_bench.sh @@ -78,6 +78,34 @@ require_cmd() { fi } +print_dry_run_command() { + local redact_next=false + local arg + + printf '[DRY-RUN]' + for arg in "$@"; do + if [[ "${redact_next}" == "true" ]]; then + printf ' %q' "REDACTED" + redact_next=false + continue + fi + + case "${arg}" in + --access-key|--secret-key) + printf ' %q' "${arg}" + redact_next=true + ;; + -accessKey=*|-secretKey=*) + printf ' %q' "${arg%%=*}=REDACTED" + ;; + *) + printf ' %q' "${arg}" + ;; + esac + done + printf '\n' +} + normalize_warp_host() { local raw="$1" raw="${raw#http://}" @@ -215,8 +243,7 @@ run_one() { fi if [[ "$DRY_RUN" == "true" ]]; then - printf '[DRY-RUN] %q ' "${cmd[@]}" - printf '\n' + print_dry_run_command "${cmd[@]}" echo "size=$size tool=$TOOL dry_run" > "$log_file" else if ! "${cmd[@]}" 2>&1 | tee "$log_file"; then @@ -243,8 +270,7 @@ run_one() { fi if [[ "$DRY_RUN" == "true" ]]; then - printf '[DRY-RUN] %q ' "${cmd[@]}" - printf '\n' + print_dry_run_command "${cmd[@]}" echo "size=$size tool=$TOOL dry_run" > "$log_file" else if ! "${cmd[@]}" 2>&1 | tee "$log_file"; then @@ -275,10 +301,12 @@ main() { validate_args resolve_bucket require_cmd rg - if [[ "$TOOL" == "warp" ]]; then - require_cmd "$WARP_BIN" - else - require_cmd "$S3BENCH_BIN" + if [[ "$DRY_RUN" != "true" ]]; then + if [[ "$TOOL" == "warp" ]]; then + require_cmd "$WARP_BIN" + else + require_cmd "$S3BENCH_BIN" + fi fi setup_output