From 5b8d6b8aa48a8e8100f97931ec3db7dd6ee9b9d4 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Tue, 19 May 2026 10:35:49 +0000 Subject: [PATCH] feat(internode): harden data transport baseline --- .../src/rpc/internode_data_transport.rs | 238 ++++++++++++++---- crates/ecstore/src/rpc/remote_disk.rs | 2 +- scripts/run_internode_transport_baseline.sh | 44 +++- scripts/run_object_batch_bench.sh | 44 +++- 4 files changed, 273 insertions(+), 55 deletions(-) diff --git a/crates/ecstore/src/rpc/internode_data_transport.rs b/crates/ecstore/src/rpc/internode_data_transport.rs index 708f7fe15..0e1209db3 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; @@ -20,10 +20,38 @@ use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE}; use rustfs_rio::{HttpReader, HttpWriter}; use std::sync::Arc; use std::time::Duration; -use tracing::warn; pub const ENV_RUSTFS_INTERNODE_DATA_TRANSPORT: &str = "RUSTFS_INTERNODE_DATA_TRANSPORT"; pub const INTERNODE_DATA_TRANSPORT_TCP_HTTP: &str = "tcp-http"; +pub const INTERNODE_DATA_TRANSPORT_TCP: &str = "tcp"; + +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"; + +#[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 +87,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 +96,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,23 +121,163 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport { fn name(&self) -> &'static str { INTERNODE_DATA_TRANSPORT_TCP_HTTP } -} -pub fn build_internode_data_transport_from_env() -> Arc { - match std::env::var(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT) { - Ok(transport) => { - if transport.eq_ignore_ascii_case(INTERNODE_DATA_TRANSPORT_TCP_HTTP) { - Arc::new(TcpHttpInternodeDataTransport) - } else { - warn!( - env = ENV_RUSTFS_INTERNODE_DATA_TRANSPORT, - requested = %transport, - fallback = INTERNODE_DATA_TRANSPORT_TCP_HTTP, - "unknown internode data transport, using default backend" - ); - Arc::new(TcpHttpInternodeDataTransport) - } - } - Err(_) => Arc::new(TcpHttpInternodeDataTransport), + fn capabilities(&self) -> InternodeDataTransportCapabilities { + InternodeDataTransportCapabilities::tcp_http() + } +} + +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 +} + +pub fn build_internode_data_transport(configured_transport: Option<&str>) -> Result> { + match configured_transport.map(str::trim).filter(|transport| !transport.is_empty()) { + None => Ok(Arc::new(TcpHttpInternodeDataTransport)), + Some(transport) + if transport.eq_ignore_ascii_case(INTERNODE_DATA_TRANSPORT_TCP_HTTP) + || transport.eq_ignore_ascii_case(INTERNODE_DATA_TRANSPORT_TCP) => + { + Ok(Arc::new(TcpHttpInternodeDataTransport)) + } + Some(transport) => Err(Error::other(format!( + "unsupported internode data transport {transport:?}; supported values: {INTERNODE_DATA_TRANSPORT_TCP_HTTP}, {INTERNODE_DATA_TRANSPORT_TCP}" + ))), + } +} + +pub fn build_internode_data_transport_from_env() -> Result> { + build_internode_data_transport(std::env::var(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT).ok().as_deref()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn tcp_http_capabilities_are_behavior_preserving() { + let transport = TcpHttpInternodeDataTransport; + + assert_eq!(transport.name(), INTERNODE_DATA_TRANSPORT_TCP_HTTP); + 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).unwrap(); + + assert_eq!(transport.name(), INTERNODE_DATA_TRANSPORT_TCP_HTTP); + } + + #[test] + fn transport_config_accepts_tcp_aliases() { + for configured in [ + INTERNODE_DATA_TRANSPORT_TCP_HTTP, + INTERNODE_DATA_TRANSPORT_TCP, + "TCP-HTTP", + "TCP", + ] { + let transport = build_internode_data_transport(Some(configured)).unwrap(); + + assert_eq!(transport.name(), INTERNODE_DATA_TRANSPORT_TCP_HTTP); + } + } + + #[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("unsupported internode data transport")); } } diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index ffca09bd0..05a02cd58 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); diff --git a/scripts/run_internode_transport_baseline.sh b/scripts/run_internode_transport_baseline.sh index 4e71f4b6d..c843b11e9 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 } @@ -127,6 +129,41 @@ setup_output() { echo "scenario,workload,concurrency,size,metric,operation,before,after,delta" > "${OUT_DIR}/internode_metric_deltas.csv" } +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 @@ -300,17 +337,20 @@ main() { parse_args "$@" validate_args require_cmd awk - require_cmd curl - require_cmd rg + 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 + write_run_manifest run_all echo "Artifacts:" + echo " ${OUT_DIR}/run_manifest.txt" echo " ${OUT_DIR}/summary.csv" echo " ${OUT_DIR}/internode_metric_deltas.csv" } diff --git a/scripts/run_object_batch_bench.sh b/scripts/run_object_batch_bench.sh index 120231a72..7196bd2a5 100755 --- a/scripts/run_object_batch_bench.sh +++ b/scripts/run_object_batch_bench.sh @@ -74,6 +74,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://}" @@ -200,8 +228,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 @@ -228,8 +255,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 @@ -251,10 +277,12 @@ main() { parse_args "$@" validate_args 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