From 58b76a3d45b43e909923be0884429f8218912b4b Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 27 Jun 2026 13:27:16 +0800 Subject: [PATCH] feat(get): add codec engine ab matrix (#3940) * upgrade version * feat(get): add codec engine ab matrix * chore(get): drop unrelated dependency drift * upgrade version * upgrade version * fix cargo deny --- Cargo.lock | 34 +- Cargo.toml | 4 +- crates/ecstore/src/diagnostics/get.rs | 4 + .../src/erasure/coding/decode_reader.rs | 83 ++-- crates/ecstore/src/set_disk/mod.rs | 13 +- crates/ecstore/src/set_disk/read.rs | 37 +- deny.toml | 1 + scripts/run_get_codec_streaming_smoke.sh | 405 +++++++++++++++--- 8 files changed, 440 insertions(+), 141 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1c1c57f87..59a5cfb2b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -260,9 +260,9 @@ dependencies = [ [[package]] name = "anyhow" -version = "1.0.102" +version = "1.0.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] name = "apache-avro" @@ -3607,7 +3607,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3945,7 +3945,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -5350,7 +5350,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -6584,7 +6584,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -6757,7 +6757,7 @@ version = "5.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d" dependencies = [ - "base64 0.21.7", + "base64 0.22.1", "chrono", "getrandom 0.2.17", "http 1.4.2", @@ -7019,9 +7019,9 @@ dependencies = [ [[package]] name = "opentelemetry-semantic-conventions" -version = "0.32.0" +version = "0.32.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ca2f98a0437b427b4b08f19f1caa3c44db885a202bc12cfea13d6c702243d68" +checksum = "c913ac17a6c451661ee255f4625d143e51647ae78ebd969b75e41c4442f4fe47" [[package]] name = "opentelemetry-stdout" @@ -7927,7 +7927,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ "heck", - "itertools 0.10.5", + "itertools 0.14.0", "log", "multimap", "once_cell", @@ -7947,7 +7947,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck", - "itertools 0.10.5", + "itertools 0.14.0", "log", "multimap", "petgraph 0.8.3", @@ -7968,7 +7968,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.10.5", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.118", @@ -7981,7 +7981,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.10.5", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.118", @@ -10232,7 +10232,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -10306,7 +10306,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -11484,7 +11484,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -12610,7 +12610,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 0ac645085..06f403769 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -206,7 +206,7 @@ tokio-postgres = { version = "0.7.18", default-features = false, features = ["ru tokio-postgres-rustls = "0.14.0" # Utilities and Tools -anyhow = "1.0.102" +anyhow = "1.0.103" arc-swap = "1.9.1" astral-tokio-tar = "0.6.3" atoi = "3.1.0" @@ -312,7 +312,7 @@ opentelemetry = { version = "0.32.0" } opentelemetry-appender-tracing = { version = "0.32.0", features = ["experimental_span_attributes", "experimental_metadata_attributes"] } opentelemetry-otlp = { version = "0.32.0", features = ["gzip-http", "reqwest-rustls"] } opentelemetry_sdk = { version = "0.32.1", features = ["rt-tokio"] } -opentelemetry-semantic-conventions = { version = "0.32.0", features = ["semconv_experimental"] } +opentelemetry-semantic-conventions = { version = "0.32.1", features = ["semconv_experimental"] } opentelemetry-stdout = { version = "0.32.0" } pyroscope = { version = "=2.0.5", features = ["backend-pprof-rs", "backend-jemalloc"] } diff --git a/crates/ecstore/src/diagnostics/get.rs b/crates/ecstore/src/diagnostics/get.rs index 7df1edf95..501722c88 100644 --- a/crates/ecstore/src/diagnostics/get.rs +++ b/crates/ecstore/src/diagnostics/get.rs @@ -17,6 +17,8 @@ use crate::error::StorageError; use std::io; pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING: &str = "codec_streaming"; +pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE: &str = "codec_streaming_legacy_engine"; +pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE: &str = "codec_streaming_rustfs_engine"; pub(crate) const GET_OBJECT_PATH_EMPTY: &str = "empty"; pub(crate) const GET_OBJECT_PATH_LEGACY_DUPLEX: &str = "legacy_duplex"; pub(crate) const GET_OBJECT_PATH_REMOTE_TRANSITION: &str = "remote_transition"; @@ -216,6 +218,8 @@ mod tests { assert_eq!(GetObjectFailureReason::DecodeError.as_str(), "decode_error"); assert_eq!(GET_READER_BUFFER_OUTPUT, "output"); assert_eq!(GET_READER_BUFFER_PREFETCH, "prefetch"); + assert_eq!(GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, "codec_streaming_legacy_engine"); + assert_eq!(GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, "codec_streaming_rustfs_engine"); assert_eq!(GET_READER_PREFETCH_DIRECT, "direct"); assert_eq!(GET_READER_PREFETCH_STORED, "stored"); assert_eq!(GET_READER_PREFETCH_EOF, "eof"); diff --git a/crates/ecstore/src/erasure/coding/decode_reader.rs b/crates/ecstore/src/erasure/coding/decode_reader.rs index be7b4568d..576e9eef1 100644 --- a/crates/ecstore/src/erasure/coding/decode_reader.rs +++ b/crates/ecstore/src/erasure/coding/decode_reader.rs @@ -43,6 +43,7 @@ pub(crate) struct ErasureDecodeReader where E: ErasureDecodeEngine, { + metrics_path: &'static str, source: Option, engine: E, workspace: Option, @@ -62,6 +63,15 @@ where E: ErasureDecodeEngine + Clone + Send + Sync + 'static, { pub(crate) fn new(source: S, engine: E, total_length: usize) -> io::Result { + Self::new_with_metrics_path(source, engine, total_length, GET_OBJECT_PATH_CODEC_STREAMING) + } + + pub(crate) fn new_with_metrics_path( + source: S, + engine: E, + total_length: usize, + metrics_path: &'static str, + ) -> io::Result { if engine.data_shards() == 0 { return Err(io::Error::new(ErrorKind::InvalidInput, "erasure reader requires data shards")); } @@ -73,6 +83,7 @@ where let workspace = engine.prepare_workspace(shard_len)?; Ok(Self { + metrics_path, source: Some(source), engine, workspace: Some(workspace), @@ -97,25 +108,26 @@ where }; let engine = self.engine.clone(); + let metrics_path = self.metrics_path; let remaining = self.remaining; self.fill = Some(tokio::spawn(async move { let fill_stage_start = Instant::now(); let stripe_read_stage_start = Instant::now(); let state = source.read_next_stripe().await; rustfs_io_metrics::record_get_object_stage_duration( - GET_OBJECT_PATH_CODEC_STREAMING, + metrics_path, GET_STAGE_STRIPE_READ, stripe_read_stage_start.elapsed().as_secs_f64(), ); let decode_stage_start = Instant::now(); - let result = decode_stripe(&engine, &mut workspace, state, remaining); + let result = decode_stripe(metrics_path, &engine, &mut workspace, state, remaining); rustfs_io_metrics::record_get_object_stage_duration( - GET_OBJECT_PATH_CODEC_STREAMING, + metrics_path, GET_STAGE_DECODE, decode_stage_start.elapsed().as_secs_f64(), ); rustfs_io_metrics::record_get_object_stage_duration( - GET_OBJECT_PATH_CODEC_STREAMING, + metrics_path, GET_STAGE_FILL, fill_stage_start.elapsed().as_secs_f64(), ); @@ -153,8 +165,8 @@ where if buf.is_empty() && self.remaining > 0 { return Poll::Ready(Err(DiskError::LessData.into())); } - rustfs_io_metrics::record_get_object_reader_stripe(GET_OBJECT_PATH_CODEC_STREAMING); - rustfs_io_metrics::record_get_object_reader_bytes(GET_OBJECT_PATH_CODEC_STREAMING, buf.len()); + rustfs_io_metrics::record_get_object_reader_stripe(self.metrics_path); + rustfs_io_metrics::record_get_object_reader_bytes(self.metrics_path, buf.len()); self.remaining -= buf.len(); Poll::Ready(Ok(Some(buf))) } @@ -182,7 +194,7 @@ where Poll::Ready(result) => { if let Some(started_at) = self.prefetch_wait_started_at.take() { rustfs_io_metrics::record_get_object_reader_prefetch_wait( - GET_OBJECT_PATH_CODEC_STREAMING, + self.metrics_path, started_at.elapsed().as_secs_f64(), ); } @@ -194,48 +206,28 @@ where match fill { Ok(Some(buf)) => { if self.output_pos < self.output_buf.len() { - rustfs_io_metrics::record_get_object_reader_prefetch( - GET_OBJECT_PATH_CODEC_STREAMING, - GET_READER_PREFETCH_STORED, - ); - rustfs_io_metrics::record_get_object_reader_buffer( - GET_OBJECT_PATH_CODEC_STREAMING, - GET_READER_BUFFER_PREFETCH, - buf.len(), - ); + rustfs_io_metrics::record_get_object_reader_prefetch(self.metrics_path, GET_READER_PREFETCH_STORED); + rustfs_io_metrics::record_get_object_reader_buffer(self.metrics_path, GET_READER_BUFFER_PREFETCH, buf.len()); self.prefetched_buf = Some(buf); } else { - rustfs_io_metrics::record_get_object_reader_prefetch( - GET_OBJECT_PATH_CODEC_STREAMING, - GET_READER_PREFETCH_DIRECT, - ); - rustfs_io_metrics::record_get_object_reader_buffer( - GET_OBJECT_PATH_CODEC_STREAMING, - GET_READER_BUFFER_OUTPUT, - buf.len(), - ); + rustfs_io_metrics::record_get_object_reader_prefetch(self.metrics_path, GET_READER_PREFETCH_DIRECT); + rustfs_io_metrics::record_get_object_reader_buffer(self.metrics_path, GET_READER_BUFFER_OUTPUT, buf.len()); self.output_buf = buf; self.output_pos = 0; } Poll::Ready(Ok(())) } Ok(None) => { - rustfs_io_metrics::record_get_object_reader_prefetch(GET_OBJECT_PATH_CODEC_STREAMING, GET_READER_PREFETCH_EOF); + rustfs_io_metrics::record_get_object_reader_prefetch(self.metrics_path, GET_READER_PREFETCH_EOF); Poll::Ready(Ok(())) } Err(err) => { if self.output_pos < self.output_buf.len() { - rustfs_io_metrics::record_get_object_reader_prefetch( - GET_OBJECT_PATH_CODEC_STREAMING, - GET_READER_PREFETCH_ERROR_DEFERRED, - ); + rustfs_io_metrics::record_get_object_reader_prefetch(self.metrics_path, GET_READER_PREFETCH_ERROR_DEFERRED); self.prefetch_error = Some(err); Poll::Ready(Ok(())) } else { - rustfs_io_metrics::record_get_object_reader_prefetch( - GET_OBJECT_PATH_CODEC_STREAMING, - GET_READER_PREFETCH_ERROR_IMMEDIATE, - ); + rustfs_io_metrics::record_get_object_reader_prefetch(self.metrics_path, GET_READER_PREFETCH_ERROR_IMMEDIATE); Poll::Ready(Err(err)) } } @@ -281,7 +273,7 @@ where self.output_pos += copy_len; if copy_len > 0 { rustfs_io_metrics::record_get_object_reader_copy( - GET_OBJECT_PATH_CODEC_STREAMING, + self.metrics_path, copy_len, read_buf_remaining_before, output_remaining_before, @@ -292,11 +284,7 @@ where } if let Some(next_buf) = self.prefetched_buf.take() { - rustfs_io_metrics::record_get_object_reader_buffer( - GET_OBJECT_PATH_CODEC_STREAMING, - GET_READER_BUFFER_OUTPUT, - next_buf.len(), - ); + rustfs_io_metrics::record_get_object_reader_buffer(self.metrics_path, GET_READER_BUFFER_OUTPUT, next_buf.len()); self.output_buf = next_buf; self.output_pos = 0; continue; @@ -380,6 +368,7 @@ where } fn decode_stripe( + metrics_path: &'static str, engine: &E, workspace: &mut E::Workspace, state: StripeReadState, @@ -398,14 +387,14 @@ where let reconstruct_stage_start = Instant::now(); if state.data_shards_complete(engine.data_shards()) { rustfs_io_metrics::record_get_object_stage_duration( - GET_OBJECT_PATH_CODEC_STREAMING, + metrics_path, GET_STAGE_RECONSTRUCT, reconstruct_stage_start.elapsed().as_secs_f64(), ); let emit_stage_start = Instant::now(); let output = emit_data_shards(&state, engine.data_shards(), engine.block_size(), remaining)?; rustfs_io_metrics::record_get_object_stage_duration( - GET_OBJECT_PATH_CODEC_STREAMING, + metrics_path, GET_STAGE_EMIT, emit_stage_start.elapsed().as_secs_f64(), ); @@ -415,14 +404,14 @@ where let (mut shards, _errs) = state.into_parts(); if let Err(err) = engine.reconstruct_into(&mut shards, workspace) { rustfs_io_metrics::record_get_object_stage_duration( - GET_OBJECT_PATH_CODEC_STREAMING, + metrics_path, GET_STAGE_RECONSTRUCT, reconstruct_stage_start.elapsed().as_secs_f64(), ); return Err(err); } rustfs_io_metrics::record_get_object_stage_duration( - GET_OBJECT_PATH_CODEC_STREAMING, + metrics_path, GET_STAGE_RECONSTRUCT, reconstruct_stage_start.elapsed().as_secs_f64(), ); @@ -446,11 +435,7 @@ where let copy_len = shard.len().min(remaining - output.len()); output.extend_from_slice(&shard[..copy_len]); } - rustfs_io_metrics::record_get_object_stage_duration( - GET_OBJECT_PATH_CODEC_STREAMING, - GET_STAGE_EMIT, - emit_stage_start.elapsed().as_secs_f64(), - ); + rustfs_io_metrics::record_get_object_stage_duration(metrics_path, GET_STAGE_EMIT, emit_stage_start.elapsed().as_secs_f64()); Ok(Some(output)) } diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 8c9d898ce..4f4e525ab 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -24,8 +24,10 @@ use crate::bucket::versioning_sys::BucketVersioningSys; use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl}; use crate::cluster::rpc::heal_bucket_local_on_disks; use crate::diagnostics::get::{ - GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_REMOTE_TRANSITION, - GET_STAGE_EMIT, GET_STAGE_METADATA, classify_storage_error, record_get_object_pipeline_failure, + GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, + GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_LEGACY_DUPLEX, + GET_OBJECT_PATH_REMOTE_TRANSITION, GET_STAGE_EMIT, GET_STAGE_METADATA, classify_storage_error, + record_get_object_pipeline_failure, }; use crate::disk::error_reduce::{ BUCKET_OP_IGNORED_ERRS, OBJECT_OP_IGNORED_ERRS, build_write_quorum_failure_summary, count_errs, reduce_read_quorum_errs, @@ -410,6 +412,13 @@ fn build_get_codec_streaming_decode_engine(erasure: coding::Erasure) -> std::io: } } +fn get_codec_streaming_metrics_path() -> &'static str { + match get_codec_streaming_engine() { + GetCodecStreamingEngine::Legacy => GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, + GetCodecStreamingEngine::Rustfs => GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, + } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum GetCodecStreamingDecision { Use, diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 19c9bcecc..dddc8c898 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -2244,8 +2244,9 @@ impl SetDisks { BitrotReaderSetupMode::VerifyReconstruction, ) .await; + let metrics_path = get_codec_streaming_metrics_path(); rustfs_io_metrics::record_get_object_stage_duration( - GET_OBJECT_PATH_CODEC_STREAMING, + metrics_path, GET_STAGE_READER_SETUP, reader_setup_stage_start.elapsed().as_secs_f64(), ); @@ -2254,14 +2255,10 @@ impl SetDisks { if available_shards < erasure.data_shards { if let Some(read_err) = reduce_read_quorum_errs(&reader_setup.errors, OBJECT_OP_IGNORED_ERRS, erasure.data_shards) { let reason = classify_disk_error(&read_err); - record_get_object_pipeline_failure_for_path(GET_OBJECT_PATH_CODEC_STREAMING, GET_STAGE_READER_SETUP, reason); + record_get_object_pipeline_failure_for_path(metrics_path, GET_STAGE_READER_SETUP, reason); return Err(to_object_err(read_err.into(), vec![bucket, object])); } - record_get_object_pipeline_failure_for_path( - GET_OBJECT_PATH_CODEC_STREAMING, - GET_STAGE_READER_SETUP, - GetObjectFailureReason::ReadQuorum, - ); + record_get_object_pipeline_failure_for_path(metrics_path, GET_STAGE_READER_SETUP, GetObjectFailureReason::ReadQuorum); return Err(Error::other(format!("not enough disks to read: {:?}", reader_setup.errors))); } @@ -2287,11 +2284,16 @@ impl SetDisks { erasure.clone(), 0, part_size, - Some(GET_OBJECT_PATH_CODEC_STREAMING), + Some(metrics_path), read_costs, ); let engine = build_get_codec_streaming_decode_engine(erasure)?; - let reader = crate::erasure::coding::decode_reader::ErasureDecodeReader::new(source, engine, part_length)?; + let reader = crate::erasure::coding::decode_reader::ErasureDecodeReader::new_with_metrics_path( + source, + engine, + part_length, + metrics_path, + )?; Ok(Box::new(crate::erasure::coding::decode_reader::SyncErasureDecodeReader::new(reader))) } } @@ -3362,6 +3364,23 @@ mod tests { }); } + #[test] + fn codec_streaming_metrics_path_matches_selected_engine() { + temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, None::<&str>, || { + assert_eq!( + get_codec_streaming_metrics_path(), + crate::diagnostics::get::GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE + ); + }); + + temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS), || { + assert_eq!( + get_codec_streaming_metrics_path(), + crate::diagnostics::get::GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE + ); + }); + } + #[test] fn codec_streaming_fallback_metric_labels_are_stable() { assert_eq!(GetCodecStreamingFallbackReason::Disabled.as_str(), "disabled"); diff --git a/deny.toml b/deny.toml index 554d76272..53f5e5660 100644 --- a/deny.toml +++ b/deny.toml @@ -39,6 +39,7 @@ allow-git = [ # Custom S3 server library with minio compatibility patches not yet upstreamed. # Pinned to a specific commit in workspace Cargo.toml. "https://github.com/rustfs/s3s", + "https://github.com/apache/datafusion.git", ] [bans] diff --git a/scripts/run_get_codec_streaming_smoke.sh b/scripts/run_get_codec_streaming_smoke.sh index 64fa81f66..8cc134640 100755 --- a/scripts/run_get_codec_streaming_smoke.sh +++ b/scripts/run_get_codec_streaming_smoke.sh @@ -24,6 +24,7 @@ ROUNDS=3 RETRY_PER_ROUND=1 ROUND_COOLDOWN_SECS=0 MODE="both" +CODEC_ENGINES="legacy" OUT_DIR="" RUSTFS_BIN="${PROJECT_ROOT}/target/release/rustfs" WARP_BIN="warp" @@ -49,6 +50,7 @@ Purpose: Core options: --mode Which profile(s) to run (default: both) + --codec-engine Codec engine(s) for codec profiles (default: legacy) --address RustFS listen address (default: 127.0.0.1:19030) --bucket Benchmark bucket (default: rustfs-get-codec-smoke) --sizes Object sizes (default: 1MiB,4MiB,10MiB) @@ -76,13 +78,17 @@ Credentials: Output: /legacy/warp/median_summary.csv - /codec/warp/median_summary.csv - /codec/warp/baseline_compare.csv when --mode both - /compat_summary.csv when --mode both + /codec-legacy/warp/median_summary.csv + /codec-rustfs/warp/median_summary.csv + /engine_compare.csv when legacy and codec profiles both run + /compat_summary.csv when legacy and codec profiles both run + /metrics_summary.csv /body_sha256_legacy.txt when legacy profile runs - /body_sha256_new.txt when codec profile runs + /body_sha256_codec_legacy.txt when codec-legacy profile runs + /body_sha256_codec_rustfs.txt when codec-rustfs profile runs /response_headers_legacy.json when legacy profile runs - /response_headers_new.json when codec profile runs + /response_headers_codec_legacy.json when codec-legacy profile runs + /response_headers_codec_rustfs.json when codec-rustfs profile runs //manifest.env //metrics_summary.csv //compat/compat_summary.csv @@ -92,7 +98,7 @@ Output: Example: scripts/run_get_codec_streaming_smoke.sh \ - --mode both --sizes 1MiB,4MiB,10MiB --concurrency 64 --duration 30s + --mode both --codec-engine legacy,rustfs --sizes 1MiB,4MiB,10MiB --concurrency 64 --duration 30s USAGE } @@ -131,6 +137,7 @@ parse_args() { while [[ $# -gt 0 ]]; do case "$1" in --mode) MODE="$2"; shift 2 ;; + --codec-engine) CODEC_ENGINES="$2"; shift 2 ;; --address) ADDRESS="$2"; shift 2 ;; --bucket) BUCKET="$2"; shift 2 ;; --sizes) SIZES="$2"; shift 2 ;; @@ -166,6 +173,17 @@ validate_args() { *) die "--mode must be legacy, codec, or both" ;; esac + local raw engine + IFS=',' read -r -a engines <<< "$CODEC_ENGINES" + [[ "${#engines[@]}" -gt 0 ]] || die "--codec-engine must not be empty" + for raw in "${engines[@]}"; do + engine="${raw//[[:space:]]/}" + case "$engine" in + legacy|rustfs) ;; + *) die "--codec-engine supports only legacy,rustfs, got: $engine" ;; + esac + done + [[ -n "$ADDRESS" ]] || die "--address must not be empty" [[ -n "$ACCESS_KEY" ]] || die "--access-key must not be empty" [[ -n "$SECRET_KEY" ]] || die "--secret-key must not be empty" @@ -210,7 +228,26 @@ profile_codec_enabled() { local profile="$1" case "$profile" in legacy) echo "false" ;; - codec) echo "true" ;; + codec-legacy|codec-rustfs) echo "true" ;; + *) die "unknown profile: $profile" ;; + esac +} + +profile_codec_engine() { + local profile="$1" + case "$profile" in + legacy|codec-legacy) echo "legacy" ;; + codec-rustfs) echo "rustfs" ;; + *) die "unknown profile: $profile" ;; + esac +} + +profile_metrics_path() { + local profile="$1" + case "$profile" in + legacy) echo "legacy_duplex" ;; + codec-legacy) echo "codec_streaming_legacy_engine" ;; + codec-rustfs) echo "codec_streaming_rustfs_engine" ;; *) die "unknown profile: $profile" ;; esac } @@ -234,6 +271,8 @@ write_manifest() { local profile_dir="$2" local codec_enabled="$3" local volumes="$4" + local codec_engine="$5" + local metrics_path="$6" local git_head git_dirty_count git_head="$(git -C "$PROJECT_ROOT" rev-parse HEAD)" @@ -261,7 +300,9 @@ compat_object_key=${COMPAT_OBJECT_KEY} compat_object_size=${COMPAT_OBJECT_SIZE} rustfs_volumes=${volumes} RUSTFS_GET_CODEC_STREAMING_ENABLE=${codec_enabled} +RUSTFS_GET_CODEC_STREAMING_ENGINE=${codec_engine} RUSTFS_GET_CODEC_STREAMING_MIN_SIZE=${CODEC_MIN_SIZE} +metrics_path=${metrics_path} RUSTFS_SCANNER_ENABLED=false RUSTFS_SCANNER_START_DELAY_SECS=3600 RUSTFS_SCANNER_CYCLE=3600 @@ -307,15 +348,19 @@ start_server() { local data_root local volumes local codec_enabled + local codec_engine + local metrics_path local rustfs_log data_root="$(profile_data_root "$profile")" volumes="$(profile_volumes "$data_root")" codec_enabled="$(profile_codec_enabled "$profile")" + codec_engine="$(profile_codec_engine "$profile")" + metrics_path="$(profile_metrics_path "$profile")" rustfs_log="${profile_dir}/rustfs.log" mkdir -p "${data_root}/disk1" "${data_root}/disk2" "${data_root}/disk3" "${data_root}/disk4" - write_manifest "$profile" "$profile_dir" "$codec_enabled" "$volumes" + write_manifest "$profile" "$profile_dir" "$codec_enabled" "$volumes" "$codec_engine" "$metrics_path" if [[ "$DRY_RUN" == "true" ]]; then log "[DRY-RUN] start RustFS profile=${profile} endpoint=$(endpoint_url)" @@ -332,6 +377,7 @@ start_server() { export RUSTFS_VOLUMES="$volumes" export RUSTFS_CONSOLE_ENABLE=false export RUSTFS_GET_CODEC_STREAMING_ENABLE="$codec_enabled" + export RUSTFS_GET_CODEC_STREAMING_ENGINE="$codec_engine" export RUSTFS_GET_CODEC_STREAMING_MIN_SIZE="$CODEC_MIN_SIZE" export RUSTFS_REGION="$REGION" export RUSTFS_RPC_SECRET="rustfs-get-codec-smoke-rpc-secret" @@ -440,6 +486,7 @@ import http.client import json import pathlib import sys +from typing import Dict, List, Optional, Tuple import urllib.parse endpoint, access_key, secret_key, region, bucket, object_key, object_size_raw, out_dir_raw, profile = sys.argv[1:] @@ -481,7 +528,7 @@ def canonical_uri(path: str) -> str: return "/".join(urllib.parse.quote(part, safe="-_.~/") for part in path.split("/")) or "/" -def signed_headers(method: str, path: str, body: bytes, extra_headers: list[tuple[str, str]]) -> dict[str, str]: +def signed_headers(method: str, path: str, body: bytes, extra_headers: List[Tuple[str, str]]) -> Dict[str, str]: parsed = urllib.parse.urlsplit(endpoint) amz_date = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ") date_stamp = amz_date[:8] @@ -509,7 +556,7 @@ def signed_headers(method: str, path: str, body: bytes, extra_headers: list[tupl return result -def request(method: str, path: str, body: bytes = b"", extra_headers: list[tuple[str, str]] | None = None): +def request(method: str, path: str, body: bytes = b"", extra_headers: Optional[List[Tuple[str, str]]] = None): parsed = urllib.parse.urlsplit(endpoint) headers = signed_headers(method, path, body, extra_headers or []) if body and "content-length" not in headers: @@ -529,8 +576,8 @@ def request(method: str, path: str, body: bytes = b"", extra_headers: list[tuple return snapshot, response_body -def header_map(headers: list[list[str] | tuple[str, str]]) -> dict[str, list[str]]: - result: dict[str, list[str]] = {} +def header_map(headers) -> Dict[str, List[str]]: + result: Dict[str, List[str]] = {} for name, value in headers: result.setdefault(name.lower(), []).append(value) for value in result.values(): @@ -609,22 +656,33 @@ copy_profile_compat_artifacts() { fi } -write_ab_compat_summary() { +write_root_compat_summary() { local legacy_snapshot="${OUT_DIR}/legacy/compat/snapshot.json" - local codec_snapshot="${OUT_DIR}/codec/compat/snapshot.json" + local codec_snapshots=() + local profile snapshot - if [[ ! -f "$legacy_snapshot" || ! -f "$codec_snapshot" ]]; then - return - fi + [[ -f "$legacy_snapshot" ]] || return - "$PYTHON_BIN" - "$legacy_snapshot" "$codec_snapshot" "${OUT_DIR}/compat_summary.csv" <<'PY' + for profile in "$@"; do + [[ "$profile" == legacy ]] && continue + snapshot="${OUT_DIR}/${profile}/compat/snapshot.json" + if [[ -f "$snapshot" ]]; then + codec_snapshots+=("$snapshot") + fi + done + + [[ "${#codec_snapshots[@]}" -gt 0 ]] || return 0 + + "$PYTHON_BIN" - "${OUT_DIR}/compat_summary.csv" "$legacy_snapshot" "${codec_snapshots[@]}" <<'PY' import csv import json import sys -legacy_path, codec_path, out_csv = sys.argv[1:] +out_csv = sys.argv[1] +legacy_path = sys.argv[2] +codec_paths = sys.argv[3:] legacy = json.load(open(legacy_path, encoding="utf-8")) -codec = json.load(open(codec_path, encoding="utf-8")) +codecs = [json.load(open(path, encoding="utf-8")) for path in codec_paths] def values(snapshot, name): @@ -636,23 +694,6 @@ def prefixed(snapshot, prefix): return {key: headers[key] for key in sorted(headers) if key.startswith(prefix)} -body_match = legacy.get("body_sha256") == codec.get("body_sha256") -content_length_match = values(legacy, "content-length") == values(codec, "content-length") -etag_match = values(legacy, "etag") == values(codec, "etag") -content_range_match = values(legacy, "content-range") == values(codec, "content-range") -checksum_headers_match = prefixed(legacy, "x-amz-checksum") == prefixed(codec, "x-amz-checksum") -sse_headers_match = prefixed(legacy, "x-amz-server-side-encryption") == prefixed(codec, "x-amz-server-side-encryption") -status_code_match = legacy.get("get_status") == codec.get("get_status") and legacy.get("head_status") == codec.get("head_status") -checks = [ - body_match, - content_length_match, - etag_match, - content_range_match, - checksum_headers_match, - sse_headers_match, - status_code_match, -] - with open(out_csv, "w", encoding="utf-8", newline="") as handle: writer = csv.writer(handle) writer.writerow( @@ -673,24 +714,239 @@ with open(out_csv, "w", encoding="utf-8", newline="") as handle: "new_status_code", ] ) - writer.writerow( - [ - codec.get("size", legacy.get("size", "")), - "codec_streaming", - str(body_match).lower(), - str(content_length_match).lower(), - str(etag_match).lower(), - str(content_range_match).lower(), - str(checksum_headers_match).lower(), - str(sse_headers_match).lower(), - str(status_code_match).lower(), - sum(1 for check in checks if not check), - legacy.get("body_sha256", ""), - codec.get("body_sha256", ""), - legacy.get("get_status", ""), - codec.get("get_status", ""), + for codec in codecs: + body_match = legacy.get("body_sha256") == codec.get("body_sha256") + content_length_match = values(legacy, "content-length") == values(codec, "content-length") + etag_match = values(legacy, "etag") == values(codec, "etag") + content_range_match = values(legacy, "content-range") == values(codec, "content-range") + checksum_headers_match = prefixed(legacy, "x-amz-checksum") == prefixed(codec, "x-amz-checksum") + sse_headers_match = prefixed(legacy, "x-amz-server-side-encryption") == prefixed(codec, "x-amz-server-side-encryption") + status_code_match = legacy.get("get_status") == codec.get("get_status") and legacy.get("head_status") == codec.get("head_status") + checks = [ + body_match, + content_length_match, + etag_match, + content_range_match, + checksum_headers_match, + sse_headers_match, + status_code_match, ] + writer.writerow( + [ + codec.get("size", legacy.get("size", "")), + codec.get("path", ""), + str(body_match).lower(), + str(content_length_match).lower(), + str(etag_match).lower(), + str(content_range_match).lower(), + str(checksum_headers_match).lower(), + str(sse_headers_match).lower(), + str(status_code_match).lower(), + sum(1 for check in checks if not check), + legacy.get("body_sha256", ""), + codec.get("body_sha256", ""), + legacy.get("get_status", ""), + codec.get("get_status", ""), + ] + ) +PY +} + +write_root_metrics_summary() { + local profile_dirs=() + local profile + + for profile in "$@"; do + profile_dirs+=("${OUT_DIR}/${profile}") + done + + [[ "${#profile_dirs[@]}" -gt 0 ]] || return + + "$PYTHON_BIN" - "${OUT_DIR}/metrics_summary.csv" "${OUT_DIR}/engine_compare.csv" "${profile_dirs[@]}" <<'PY' +import csv +import pathlib +import sys + +metrics_csv = pathlib.Path(sys.argv[1]) +engine_compare_csv = pathlib.Path(sys.argv[2]) +profile_dirs = [pathlib.Path(value) for value in sys.argv[3:]] + + +def load_manifest(path): + data = {} + for line in path.read_text(encoding="utf-8").splitlines(): + if "=" not in line: + continue + key, value = line.split("=", 1) + data[key] = value + return data + + +def parse_float(value): + if value in ("", "N/A", None): + return None + return float(value) + + +def delta_pct(new_value, baseline_value): + if new_value is None or baseline_value in (None, 0.0): + return "" + return f"{((new_value - baseline_value) / baseline_value) * 100:.2f}" + + +def performance_note(throughput_delta, latency_delta): + if throughput_delta == "" or latency_delta == "": + return "" + throughput_delta = float(throughput_delta) + latency_delta = float(latency_delta) + if throughput_delta < 0 and latency_delta > 0: + return "slower_than_legacy_and_higher_latency" + if throughput_delta < 0: + return "slower_than_legacy" + if latency_delta > 0: + return "higher_latency_than_legacy" + return "at_or_better_than_legacy" + + +rows = [] +by_profile = {} +for profile_dir in profile_dirs: + manifest = load_manifest(profile_dir / "manifest.env") + profile = manifest["profile"] + with open(profile_dir / "warp" / "median_summary.csv", encoding="utf-8", newline="") as handle: + profile_rows = list(csv.DictReader(handle)) + by_profile[profile] = profile_rows + for row in profile_rows: + rows.append( + { + "profile": profile, + "read_path": manifest.get("metrics_path", profile), + "codec_streaming_enabled": manifest.get("RUSTFS_GET_CODEC_STREAMING_ENABLE", ""), + "codec_engine": manifest.get("RUSTFS_GET_CODEC_STREAMING_ENGINE", ""), + "size": row["size"], + "tool": row["tool"], + "concurrency": row["concurrency"], + "successful_rounds": row["successful_rounds"], + "failed_rounds": row["failed_rounds"], + "median_throughput_bps": row["median_throughput_bps"], + "median_reqps": row["median_reqps"], + "median_latency_ms": row["median_latency_ms"], + } + ) + +legacy_by_size = {row["size"]: row for row in by_profile.get("legacy", [])} +for row in rows: + baseline = legacy_by_size.get(row["size"]) + if baseline is None or row["profile"] == "legacy": + row["baseline_profile"] = "" + row["delta_throughput_pct_vs_legacy"] = "" + row["delta_reqps_pct_vs_legacy"] = "" + row["delta_latency_pct_vs_legacy"] = "" + row["performance_note"] = "" + continue + throughput_delta = delta_pct(parse_float(row["median_throughput_bps"]), parse_float(baseline["median_throughput_bps"])) + reqps_delta = delta_pct(parse_float(row["median_reqps"]), parse_float(baseline["median_reqps"])) + latency_delta = delta_pct(parse_float(row["median_latency_ms"]), parse_float(baseline["median_latency_ms"])) + row["baseline_profile"] = "legacy" + row["delta_throughput_pct_vs_legacy"] = throughput_delta + row["delta_reqps_pct_vs_legacy"] = reqps_delta + row["delta_latency_pct_vs_legacy"] = latency_delta + row["performance_note"] = performance_note(throughput_delta, latency_delta) + +fieldnames = [ + "profile", + "read_path", + "codec_streaming_enabled", + "codec_engine", + "size", + "tool", + "concurrency", + "successful_rounds", + "failed_rounds", + "median_throughput_bps", + "median_reqps", + "median_latency_ms", + "baseline_profile", + "delta_throughput_pct_vs_legacy", + "delta_reqps_pct_vs_legacy", + "delta_latency_pct_vs_legacy", + "performance_note", +] +with open(metrics_csv, "w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(rows) + +codec_legacy_by_size = {row["size"]: row for row in by_profile.get("codec-legacy", [])} +codec_rustfs_by_size = {row["size"]: row for row in by_profile.get("codec-rustfs", [])} +if not legacy_by_size: + raise SystemExit(0) + +compare_fields = [ + "size", + "legacy_median_throughput_bps", + "codec_legacy_median_throughput_bps", + "codec_rustfs_median_throughput_bps", + "legacy_median_reqps", + "codec_legacy_median_reqps", + "codec_rustfs_median_reqps", + "legacy_median_latency_ms", + "codec_legacy_median_latency_ms", + "codec_rustfs_median_latency_ms", + "codec_legacy_delta_throughput_pct_vs_legacy", + "codec_rustfs_delta_throughput_pct_vs_legacy", + "codec_rustfs_delta_throughput_pct_vs_codec_legacy", + "codec_legacy_delta_latency_pct_vs_legacy", + "codec_rustfs_delta_latency_pct_vs_legacy", + "codec_rustfs_delta_latency_pct_vs_codec_legacy", +] +compare_rows = [] +for size, legacy in legacy_by_size.items(): + codec_legacy = codec_legacy_by_size.get(size, {}) + codec_rustfs = codec_rustfs_by_size.get(size, {}) + compare_rows.append( + { + "size": size, + "legacy_median_throughput_bps": legacy.get("median_throughput_bps", ""), + "codec_legacy_median_throughput_bps": codec_legacy.get("median_throughput_bps", ""), + "codec_rustfs_median_throughput_bps": codec_rustfs.get("median_throughput_bps", ""), + "legacy_median_reqps": legacy.get("median_reqps", ""), + "codec_legacy_median_reqps": codec_legacy.get("median_reqps", ""), + "codec_rustfs_median_reqps": codec_rustfs.get("median_reqps", ""), + "legacy_median_latency_ms": legacy.get("median_latency_ms", ""), + "codec_legacy_median_latency_ms": codec_legacy.get("median_latency_ms", ""), + "codec_rustfs_median_latency_ms": codec_rustfs.get("median_latency_ms", ""), + "codec_legacy_delta_throughput_pct_vs_legacy": delta_pct( + parse_float(codec_legacy.get("median_throughput_bps", "")), + parse_float(legacy.get("median_throughput_bps", "")), + ), + "codec_rustfs_delta_throughput_pct_vs_legacy": delta_pct( + parse_float(codec_rustfs.get("median_throughput_bps", "")), + parse_float(legacy.get("median_throughput_bps", "")), + ), + "codec_rustfs_delta_throughput_pct_vs_codec_legacy": delta_pct( + parse_float(codec_rustfs.get("median_throughput_bps", "")), + parse_float(codec_legacy.get("median_throughput_bps", "")), + ), + "codec_legacy_delta_latency_pct_vs_legacy": delta_pct( + parse_float(codec_legacy.get("median_latency_ms", "")), + parse_float(legacy.get("median_latency_ms", "")), + ), + "codec_rustfs_delta_latency_pct_vs_legacy": delta_pct( + parse_float(codec_rustfs.get("median_latency_ms", "")), + parse_float(legacy.get("median_latency_ms", "")), + ), + "codec_rustfs_delta_latency_pct_vs_codec_legacy": delta_pct( + parse_float(codec_rustfs.get("median_latency_ms", "")), + parse_float(codec_legacy.get("median_latency_ms", "")), + ), + } ) + +with open(engine_compare_csv, "w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=compare_fields) + writer.writeheader() + writer.writerows(compare_rows) PY } @@ -722,27 +978,52 @@ main() { trap stop_server EXIT INT TERM log "Output dir: $OUT_DIR" + local codec_profiles=() + local profiles=() + local raw engine profile + IFS=',' read -r -a engines <<< "$CODEC_ENGINES" + for raw in "${engines[@]}"; do + engine="${raw//[[:space:]]/}" + [[ -n "$engine" ]] || continue + profile="codec-${engine}" + codec_profiles+=("$profile") + done + case "$MODE" in legacy) - run_profile legacy - copy_profile_compat_artifacts legacy legacy + profiles=("legacy") ;; codec) - run_profile codec - copy_profile_compat_artifacts codec new + profiles=("${codec_profiles[@]}") ;; both) - run_profile legacy - copy_profile_compat_artifacts legacy legacy - run_profile codec "${OUT_DIR}/legacy/warp/median_summary.csv" - copy_profile_compat_artifacts codec new - write_ab_compat_summary + profiles=("legacy" "${codec_profiles[@]}") ;; esac + local legacy_baseline_csv="" + for profile in "${profiles[@]}"; do + if [[ "$profile" == "legacy" ]]; then + run_profile "$profile" + legacy_baseline_csv="${OUT_DIR}/legacy/warp/median_summary.csv" + else + run_profile "$profile" "$legacy_baseline_csv" + fi + copy_profile_compat_artifacts "$profile" "${profile//-/_}" + done + + write_root_metrics_summary "${profiles[@]}" + write_root_compat_summary "${profiles[@]}" + if [[ -f "${OUT_DIR}/compat_summary.csv" ]]; then log "Compatibility compare: ${OUT_DIR}/compat_summary.csv" fi + if [[ -f "${OUT_DIR}/engine_compare.csv" ]]; then + log "Engine compare: ${OUT_DIR}/engine_compare.csv" + fi + if [[ -f "${OUT_DIR}/metrics_summary.csv" ]]; then + log "Metrics summary: ${OUT_DIR}/metrics_summary.csv" + fi log "GET codec streaming smoke finished." }