From 5a78a9c416c8b858ed92bbf3f8ab76cd89382168 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 26 Jun 2026 20:09:09 +0800 Subject: [PATCH] feat(get): add v2 metrics compatibility harness (#3913) * feat(get): add v2 metrics compatibility harness * feat(get): observe metadata fanout quorum (#3915) * feat(get): observe metadata fanout quorum * fix(app): use storage ECStore type in app boundary ---- Co-Authored-By: heihutu --- crates/ecstore/src/get_diagnostics.rs | 37 +++ crates/ecstore/src/set_disk/read.rs | 389 ++++++++++++++++++++++- crates/io-metrics/src/lib.rs | 154 +++++++++ rustfs/src/app/storage_api.rs | 2 +- scripts/run_get_codec_streaming_smoke.sh | 348 ++++++++++++++++++++ 5 files changed, 916 insertions(+), 14 deletions(-) diff --git a/crates/ecstore/src/get_diagnostics.rs b/crates/ecstore/src/get_diagnostics.rs index 0c375d869..64485f21a 100644 --- a/crates/ecstore/src/get_diagnostics.rs +++ b/crates/ecstore/src/get_diagnostics.rs @@ -24,15 +24,25 @@ pub(crate) const GET_OBJECT_PATH_REMOTE_TRANSITION: &str = "remote_transition"; pub(crate) const GET_STAGE_DECODE: &str = "decode"; pub(crate) const GET_STAGE_EMIT: &str = "emit"; pub(crate) const GET_STAGE_FILL: &str = "fill"; +pub(crate) const GET_STAGE_FIRST_BYTE: &str = "first_byte"; +pub(crate) const GET_STAGE_FIRST_METADATA_RESPONSE: &str = "first_metadata_response"; +pub(crate) const GET_STAGE_FIRST_VALID_METADATA_RESPONSE: &str = "first_valid_metadata_response"; +pub(crate) const GET_STAGE_FIRST_SHARD_READ: &str = "first_shard_read"; +pub(crate) const GET_STAGE_FULL_BODY: &str = "full_body"; pub(crate) const GET_STAGE_METADATA: &str = "metadata"; +pub(crate) const GET_STAGE_METADATA_FANOUT: &str = "metadata_fanout"; pub(crate) const GET_STAGE_OUTPUT_LOCK_WAIT: &str = "output_lock_wait"; pub(crate) const GET_STAGE_OUTPUT_POLL: &str = "output_poll"; +pub(crate) const GET_STAGE_QUORUM_REACHED: &str = "quorum_reached"; pub(crate) const GET_STAGE_RANGE: &str = "range"; pub(crate) const GET_STAGE_READER_SETUP: &str = "reader_setup"; pub(crate) const GET_STAGE_RECONSTRUCT: &str = "reconstruct"; +pub(crate) const GET_STAGE_RESPONSE_HANDOFF: &str = "response_handoff"; +pub(crate) const GET_STAGE_SLOWEST_METADATA_RESPONSE: &str = "slowest_metadata_response"; pub(crate) const GET_STAGE_STRIPE_READ: &str = "stripe_read"; pub(crate) const GET_STAGE_STRIPE_READ_FIRST_SHARD: &str = "stripe_read_first_shard"; pub(crate) const GET_STAGE_STRIPE_READ_QUORUM: &str = "stripe_read_quorum"; +pub(crate) const GET_STAGE_BITROT_VERIFY: &str = "bitrot_verify"; pub(crate) const GET_READER_BUFFER_OUTPUT: &str = "output"; pub(crate) const GET_READER_BUFFER_PREFETCH: &str = "prefetch"; @@ -51,6 +61,15 @@ pub(crate) const GET_SHARD_READ_OUTCOME_SUCCESS: &str = "success"; pub(crate) const GET_SHARD_ROLE_DATA: &str = "data"; pub(crate) const GET_SHARD_ROLE_PARITY: &str = "parity"; +pub(crate) const GET_METADATA_RESPONSE_CORRUPT: &str = "corrupt"; +pub(crate) const GET_METADATA_RESPONSE_DISK_NOT_FOUND: &str = "disk_not_found"; +pub(crate) const GET_METADATA_RESPONSE_ERROR: &str = "error"; +pub(crate) const GET_METADATA_RESPONSE_IGNORED: &str = "ignored"; +pub(crate) const GET_METADATA_RESPONSE_NOT_FOUND: &str = "not_found"; +pub(crate) const GET_METADATA_RESPONSE_TIMEOUT: &str = "timeout"; +pub(crate) const GET_METADATA_RESPONSE_VALID: &str = "valid"; +pub(crate) const GET_METADATA_RESPONSE_VERSION_NOT_FOUND: &str = "version_not_found"; + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum GetObjectFailureReason { BitrotMismatch, @@ -195,19 +214,37 @@ mod tests { assert_eq!(GET_STAGE_DECODE, "decode"); assert_eq!(GET_STAGE_EMIT, "emit"); assert_eq!(GET_STAGE_FILL, "fill"); + assert_eq!(GET_STAGE_FIRST_BYTE, "first_byte"); + assert_eq!(GET_STAGE_FIRST_METADATA_RESPONSE, "first_metadata_response"); + assert_eq!(GET_STAGE_FIRST_VALID_METADATA_RESPONSE, "first_valid_metadata_response"); + assert_eq!(GET_STAGE_FIRST_SHARD_READ, "first_shard_read"); + assert_eq!(GET_STAGE_FULL_BODY, "full_body"); assert_eq!(GET_STAGE_METADATA, "metadata"); + assert_eq!(GET_STAGE_METADATA_FANOUT, "metadata_fanout"); assert_eq!(GET_STAGE_OUTPUT_LOCK_WAIT, "output_lock_wait"); assert_eq!(GET_STAGE_OUTPUT_POLL, "output_poll"); + assert_eq!(GET_STAGE_QUORUM_REACHED, "quorum_reached"); assert_eq!(GET_STAGE_RANGE, "range"); assert_eq!(GET_STAGE_READER_SETUP, "reader_setup"); assert_eq!(GET_STAGE_RECONSTRUCT, "reconstruct"); + assert_eq!(GET_STAGE_RESPONSE_HANDOFF, "response_handoff"); + assert_eq!(GET_STAGE_SLOWEST_METADATA_RESPONSE, "slowest_metadata_response"); assert_eq!(GET_STAGE_STRIPE_READ, "stripe_read"); assert_eq!(GET_STAGE_STRIPE_READ_FIRST_SHARD, "stripe_read_first_shard"); assert_eq!(GET_STAGE_STRIPE_READ_QUORUM, "stripe_read_quorum"); + assert_eq!(GET_STAGE_BITROT_VERIFY, "bitrot_verify"); assert_eq!(GET_SHARD_READ_OUTCOME_ERROR, "error"); assert_eq!(GET_SHARD_READ_OUTCOME_MISSING, "missing"); assert_eq!(GET_SHARD_READ_OUTCOME_SUCCESS, "success"); assert_eq!(GET_SHARD_ROLE_DATA, "data"); assert_eq!(GET_SHARD_ROLE_PARITY, "parity"); + assert_eq!(GET_METADATA_RESPONSE_CORRUPT, "corrupt"); + assert_eq!(GET_METADATA_RESPONSE_DISK_NOT_FOUND, "disk_not_found"); + assert_eq!(GET_METADATA_RESPONSE_ERROR, "error"); + assert_eq!(GET_METADATA_RESPONSE_IGNORED, "ignored"); + assert_eq!(GET_METADATA_RESPONSE_NOT_FOUND, "not_found"); + assert_eq!(GET_METADATA_RESPONSE_TIMEOUT, "timeout"); + assert_eq!(GET_METADATA_RESPONSE_VALID, "valid"); + assert_eq!(GET_METADATA_RESPONSE_VERSION_NOT_FOUND, "version_not_found"); } } diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 5fc4da587..85fd79b0f 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -14,8 +14,11 @@ use super::*; use crate::get_diagnostics::{ - GET_OBJECT_PATH_CODEC_STREAMING, GET_STAGE_DECODE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP, GetObjectFailureReason, - classify_disk_error, record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path, + GET_METADATA_RESPONSE_CORRUPT, GET_METADATA_RESPONSE_DISK_NOT_FOUND, GET_METADATA_RESPONSE_ERROR, + GET_METADATA_RESPONSE_IGNORED, GET_METADATA_RESPONSE_NOT_FOUND, GET_METADATA_RESPONSE_TIMEOUT, GET_METADATA_RESPONSE_VALID, + GET_METADATA_RESPONSE_VERSION_NOT_FOUND, GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_DECODE, + GET_STAGE_RANGE, GET_STAGE_READER_SETUP, GetObjectFailureReason, classify_disk_error, record_get_object_pipeline_failure, + record_get_object_pipeline_failure_for_path, }; use metrics::counter; use rustfs_config::{DEFAULT_OBJECT_ZERO_COPY_ENABLE, ENV_OBJECT_ZERO_COPY_ENABLE}; @@ -35,6 +38,152 @@ const READ_REPAIR_HEAL_DEDUP_MAX_ENTRIES: usize = 4096; static READ_REPAIR_HEAL_CACHE: OnceLock>> = OnceLock::new(); +#[derive(Clone, Copy, Debug)] +struct MetadataFanoutObservation { + outcome: &'static str, + elapsed: Duration, + valid: bool, + ignored: bool, +} + +impl MetadataFanoutObservation { + fn from_file_info(file_info: &FileInfo, elapsed: Duration) -> Self { + if file_info.is_valid() { + Self { + outcome: GET_METADATA_RESPONSE_VALID, + elapsed, + valid: true, + ignored: false, + } + } else { + Self { + outcome: GET_METADATA_RESPONSE_ERROR, + elapsed, + valid: false, + ignored: false, + } + } + } + + fn from_error(err: &DiskError, elapsed: Duration) -> Self { + Self { + outcome: classify_metadata_response_error(err), + elapsed, + valid: false, + ignored: is_metadata_fanout_ignored_error(err), + } + } +} + +#[derive(Clone, Debug, Default)] +struct MetadataFanoutDiagnostics { + fanout_duration: Duration, + observations: Vec, +} + +impl MetadataFanoutDiagnostics { + fn new(fanout_duration: Duration, observations: Vec) -> Self { + Self { + fanout_duration, + observations, + } + } + + fn total_responses(&self) -> usize { + self.observations.len() + } + + fn valid_responses(&self) -> usize { + self.observations.iter().filter(|observation| observation.valid).count() + } + + fn ignored_responses(&self) -> usize { + self.observations.iter().filter(|observation| observation.ignored).count() + } + + fn error_responses(&self) -> usize { + self.total_responses().saturating_sub(self.valid_responses()) + } + + fn first_response_latency(&self) -> Option { + self.observations.iter().map(|observation| observation.elapsed).min() + } + + fn first_valid_response_latency(&self) -> Option { + self.observations + .iter() + .filter(|observation| observation.valid) + .map(|observation| observation.elapsed) + .min() + } + + fn slowest_response_latency(&self) -> Option { + self.observations.iter().map(|observation| observation.elapsed).max() + } + + fn quorum_candidate_latency(&self, read_quorum: usize) -> Option { + if read_quorum == 0 { + return Some(Duration::ZERO); + } + + let mut valid_latencies = self + .observations + .iter() + .filter(|observation| observation.valid) + .map(|observation| observation.elapsed) + .collect::>(); + valid_latencies.sort_unstable(); + valid_latencies.get(read_quorum.saturating_sub(1)).copied() + } + + fn record(&self, path: &'static str) { + rustfs_io_metrics::record_get_object_metadata_fanout_duration(path, self.fanout_duration.as_secs_f64()); + if let Some(latency) = self.first_response_latency() { + rustfs_io_metrics::record_get_object_first_metadata_response_latency(path, latency.as_secs_f64()); + } + if let Some(latency) = self.first_valid_response_latency() { + rustfs_io_metrics::record_get_object_first_valid_metadata_response_latency(path, latency.as_secs_f64()); + } + if let Some(latency) = self.slowest_response_latency() { + rustfs_io_metrics::record_get_object_slowest_metadata_response_latency(path, latency.as_secs_f64()); + } + rustfs_io_metrics::record_get_object_metadata_fanout_shape( + path, + self.total_responses(), + self.valid_responses(), + self.ignored_responses(), + self.error_responses(), + ); + for observation in &self.observations { + rustfs_io_metrics::record_get_object_metadata_response(path, observation.outcome); + } + } + + fn record_quorum_candidate_latency(&self, path: &'static str, read_quorum: usize) { + if let Some(latency) = self.quorum_candidate_latency(read_quorum) { + rustfs_io_metrics::record_get_object_quorum_reached_latency(path, latency.as_secs_f64()); + } + } +} + +fn classify_metadata_response_error(err: &DiskError) -> &'static str { + match err { + DiskError::FileNotFound | DiskError::VolumeNotFound => GET_METADATA_RESPONSE_NOT_FOUND, + DiskError::FileVersionNotFound => GET_METADATA_RESPONSE_VERSION_NOT_FOUND, + DiskError::DiskNotFound => GET_METADATA_RESPONSE_DISK_NOT_FOUND, + DiskError::FileCorrupt | DiskError::CorruptedFormat | DiskError::CorruptedBackend | DiskError::OutdatedXLMeta => { + GET_METADATA_RESPONSE_CORRUPT + } + DiskError::Timeout => GET_METADATA_RESPONSE_TIMEOUT, + DiskError::FaultyDisk | DiskError::FaultyRemoteDisk => GET_METADATA_RESPONSE_IGNORED, + _ => GET_METADATA_RESPONSE_ERROR, + } +} + +fn is_metadata_fanout_ignored_error(err: &DiskError) -> bool { + OBJECT_OP_IGNORED_ERRS.iter().any(|ignored| ignored == err) +} + #[derive(Clone, Debug, Eq, Hash, PartialEq)] struct ReadRepairHealCacheKey { bucket: String, @@ -522,8 +671,62 @@ impl SetDisks { healing: bool, incl_free_versions: bool, ) -> disk::error::Result<(Vec, Vec>)> { + let (ress, errors, _) = Self::read_all_fileinfo_inner( + disks, + org_bucket, + bucket, + object, + version_id, + read_data, + healing, + incl_free_versions, + false, + ) + .await?; + Ok((ress, errors)) + } + + #[allow(clippy::too_many_arguments)] + async fn read_all_fileinfo_observed( + disks: &[Option], + org_bucket: &str, + bucket: &str, + object: &str, + version_id: &str, + read_data: bool, + healing: bool, + incl_free_versions: bool, + ) -> disk::error::Result<(Vec, Vec>, MetadataFanoutDiagnostics)> { + Self::read_all_fileinfo_inner( + disks, + org_bucket, + bucket, + object, + version_id, + read_data, + healing, + incl_free_versions, + true, + ) + .await + } + + #[allow(clippy::too_many_arguments)] + async fn read_all_fileinfo_inner( + disks: &[Option], + org_bucket: &str, + bucket: &str, + object: &str, + version_id: &str, + read_data: bool, + healing: bool, + incl_free_versions: bool, + observe: bool, + ) -> disk::error::Result<(Vec, Vec>, MetadataFanoutDiagnostics)> { + let fanout_start = observe.then(Instant::now); let mut ress = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len()); + let mut observations = observe.then(|| Vec::with_capacity(disks.len())); let opts = Arc::new(ReadOptions { incl_free_versions, read_data, @@ -541,11 +744,14 @@ impl SetDisks { let object = object.clone(); let version_id = version_id.clone(); tokio::spawn(async move { - if let Some(disk) = disk { + let response_start = observe.then(Instant::now); + let result = if let Some(disk) = disk { disk.read_version(&org_bucket, &bucket, &object, &version_id, &opts).await } else { Err(DiskError::DiskNotFound) - } + }; + let elapsed = response_start.map(|start| start.elapsed()); + (result, elapsed) }) }); @@ -554,23 +760,37 @@ impl SetDisks { for result in results { match result { - Ok(res) => match res { + Ok((res, elapsed)) => match res { Ok(file_info) => { + if let (Some(observations), Some(elapsed)) = (&mut observations, elapsed) { + observations.push(MetadataFanoutObservation::from_file_info(&file_info, elapsed)); + } ress.push(file_info); errors.push(None); } Err(e) => { + if let (Some(observations), Some(elapsed)) = (&mut observations, elapsed) { + observations.push(MetadataFanoutObservation::from_error(&e, elapsed)); + } ress.push(FileInfo::default()); errors.push(Some(e)); } }, Err(_) => { + let err = DiskError::Unexpected; + if let (Some(observations), Some(fanout_start)) = (&mut observations, fanout_start) { + observations.push(MetadataFanoutObservation::from_error(&err, fanout_start.elapsed())); + } ress.push(FileInfo::default()); - errors.push(Some(DiskError::Unexpected)); + errors.push(Some(err)); } } } - Ok((ress, errors)) + let diagnostics = match (fanout_start, observations) { + (Some(fanout_start), Some(observations)) => MetadataFanoutDiagnostics::new(fanout_start.elapsed(), observations), + _ => MetadataFanoutDiagnostics::default(), + }; + Ok((ress, errors, diagnostics)) } pub async fn read_version_optimized( @@ -912,8 +1132,10 @@ impl SetDisks { let vid = opts.version_id.clone().unwrap_or_default(); // TODO: optimize concurrency and break once enough slots are available - let (parts_metadata, errs) = - Self::read_all_fileinfo(&disks, "", bucket, object, vid.as_str(), read_data, false, opts.incl_free_versions).await?; + let (parts_metadata, errs, metadata_fanout_diagnostics) = + Self::read_all_fileinfo_observed(&disks, "", bucket, object, vid.as_str(), read_data, false, opts.incl_free_versions) + .await?; + metadata_fanout_diagnostics.record(GET_OBJECT_PATH_LEGACY_DUPLEX); // warn!("get_object_fileinfo parts_metadata {:?}", &parts_metadata); // warn!("get_object_fileinfo {}/{} errs {:?}", bucket, object, &errs); @@ -928,15 +1150,18 @@ impl SetDisks { return Err(e); } }; + let read_quorum = + usize::try_from(read_quorum).map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?; + metadata_fanout_diagnostics.record_quorum_candidate_latency(GET_OBJECT_PATH_LEGACY_DUPLEX, read_quorum); - if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum as usize) { + if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) { error!("reduce_read_quorum_errs: {:?}, bucket: {}, object: {}", &err, bucket, object); return Err(to_object_err(err.into(), vec![bucket, object])); } - let (op_online_disks, mot_time, etag) = Self::list_online_disks(&disks, &parts_metadata, &errs, read_quorum as usize); + let (op_online_disks, mot_time, etag) = Self::list_online_disks(&disks, &parts_metadata, &errs, read_quorum); - let fi = Self::pick_valid_fileinfo(&parts_metadata, mot_time, etag, read_quorum as usize)?; + let fi = Self::pick_valid_fileinfo(&parts_metadata, mot_time, etag, read_quorum)?; if errs.iter().any(|err| err.is_some()) { let version_id = resolved_read_repair_version_id(&fi, opts.version_id.as_deref()); submit_read_repair_heal( @@ -950,7 +1175,7 @@ impl SetDisks { ) .await; } else if use_metadata_cache { - self.cache_get_object_fileinfo(bucket, object, &fi, &parts_metadata, &op_online_disks, read_quorum as usize) + self.cache_get_object_fileinfo(bucket, object, &fi, &parts_metadata, &op_online_disks, read_quorum) .await; } // debug!("get_object_fileinfo pick fi {:?}", &fi); @@ -1861,6 +2086,16 @@ mod metadata_cache_tests { mod tests { use super::*; + fn metadata_fanout_test_fileinfo(object: &str) -> FileInfo { + let mut fi = FileInfo::new(object, 2, 2); + fi.volume = "bucket".to_string(); + fi.name = object.to_string(); + fi.size = 1; + fi.erasure.index = 1; + fi.metadata.insert("etag".to_string(), "etag-1".to_string()); + fi + } + fn codec_streaming_test_fileinfo(size: i64, part_count: usize) -> FileInfo { let mut fi = FileInfo::new("object", 4, 2); fi.volume = "bucket".to_string(); @@ -1888,6 +2123,134 @@ mod tests { fi } + #[test] + fn metadata_fanout_diagnostics_classifies_response_outcomes() { + let valid = metadata_fanout_test_fileinfo("object"); + let invalid = FileInfo::default(); + + let observations = vec![ + MetadataFanoutObservation::from_file_info(&valid, Duration::from_millis(3)), + MetadataFanoutObservation::from_file_info(&invalid, Duration::from_millis(4)), + MetadataFanoutObservation::from_error(&DiskError::FileNotFound, Duration::from_millis(5)), + MetadataFanoutObservation::from_error(&DiskError::FileVersionNotFound, Duration::from_millis(6)), + MetadataFanoutObservation::from_error(&DiskError::DiskNotFound, Duration::from_millis(7)), + MetadataFanoutObservation::from_error(&DiskError::FileCorrupt, Duration::from_millis(8)), + MetadataFanoutObservation::from_error(&DiskError::Timeout, Duration::from_millis(9)), + MetadataFanoutObservation::from_error(&DiskError::FaultyDisk, Duration::from_millis(10)), + MetadataFanoutObservation::from_error(&DiskError::Unexpected, Duration::from_millis(11)), + ]; + let diagnostics = MetadataFanoutDiagnostics::new(Duration::from_millis(12), observations); + let outcomes = diagnostics + .observations + .iter() + .map(|observation| observation.outcome) + .collect::>(); + + assert_eq!( + outcomes, + vec![ + GET_METADATA_RESPONSE_VALID, + GET_METADATA_RESPONSE_ERROR, + GET_METADATA_RESPONSE_NOT_FOUND, + GET_METADATA_RESPONSE_VERSION_NOT_FOUND, + GET_METADATA_RESPONSE_DISK_NOT_FOUND, + GET_METADATA_RESPONSE_CORRUPT, + GET_METADATA_RESPONSE_TIMEOUT, + GET_METADATA_RESPONSE_IGNORED, + GET_METADATA_RESPONSE_ERROR, + ] + ); + assert_eq!(diagnostics.total_responses(), 9); + assert_eq!(diagnostics.valid_responses(), 1); + assert_eq!(diagnostics.error_responses(), 8); + } + + #[test] + fn metadata_fanout_diagnostics_tracks_ignored_errors_without_hiding_outcomes() { + let diagnostics = MetadataFanoutDiagnostics::new( + Duration::from_millis(15), + vec![ + MetadataFanoutObservation::from_error(&DiskError::DiskNotFound, Duration::from_millis(1)), + MetadataFanoutObservation::from_error(&DiskError::FaultyRemoteDisk, Duration::from_millis(2)), + MetadataFanoutObservation::from_error(&DiskError::FileNotFound, Duration::from_millis(3)), + ], + ); + + assert_eq!(diagnostics.ignored_responses(), 2); + assert_eq!(diagnostics.error_responses(), 3); + assert_eq!(diagnostics.observations[0].outcome, GET_METADATA_RESPONSE_DISK_NOT_FOUND); + assert_eq!(diagnostics.observations[1].outcome, GET_METADATA_RESPONSE_IGNORED); + assert_eq!(diagnostics.observations[2].outcome, GET_METADATA_RESPONSE_NOT_FOUND); + } + + #[test] + fn metadata_fanout_quorum_candidate_latency_ignores_slow_trailing_valid_response() { + let valid = metadata_fanout_test_fileinfo("object"); + let diagnostics = MetadataFanoutDiagnostics::new( + Duration::from_millis(250), + vec![ + MetadataFanoutObservation::from_error(&DiskError::FileNotFound, Duration::from_millis(2)), + MetadataFanoutObservation::from_file_info(&valid, Duration::from_millis(5)), + MetadataFanoutObservation::from_file_info(&valid, Duration::from_millis(7)), + MetadataFanoutObservation::from_file_info(&valid, Duration::from_millis(250)), + ], + ); + + assert_eq!(diagnostics.first_response_latency(), Some(Duration::from_millis(2))); + assert_eq!(diagnostics.first_valid_response_latency(), Some(Duration::from_millis(5))); + assert_eq!(diagnostics.slowest_response_latency(), Some(Duration::from_millis(250))); + assert_eq!(diagnostics.quorum_candidate_latency(2), Some(Duration::from_millis(7))); + } + + #[test] + fn metadata_fanout_quorum_candidate_latency_requires_enough_valid_responses() { + let valid = metadata_fanout_test_fileinfo("object"); + let diagnostics = MetadataFanoutDiagnostics::new( + Duration::from_millis(8), + vec![ + MetadataFanoutObservation::from_file_info(&valid, Duration::from_millis(5)), + MetadataFanoutObservation::from_error(&DiskError::FileNotFound, Duration::from_millis(6)), + ], + ); + + assert_eq!(diagnostics.quorum_candidate_latency(0), Some(Duration::ZERO)); + assert_eq!(diagnostics.quorum_candidate_latency(1), Some(Duration::from_millis(5))); + assert_eq!(diagnostics.quorum_candidate_latency(2), None); + } + + #[test] + fn metadata_fanout_counts_delete_marker_and_conflicting_versions_as_valid_observations_only() { + let mut latest = metadata_fanout_test_fileinfo("object"); + latest.version_id = Some(Uuid::parse_str("00000000-0000-0000-0000-000000000001").expect("static uuid should parse")); + + let mut historical = metadata_fanout_test_fileinfo("object"); + historical.version_id = Some(Uuid::parse_str("00000000-0000-0000-0000-000000000002").expect("static uuid should parse")); + + let mut delete_marker = metadata_fanout_test_fileinfo("object"); + delete_marker.deleted = true; + delete_marker.version_id = + Some(Uuid::parse_str("00000000-0000-0000-0000-000000000003").expect("static uuid should parse")); + + let diagnostics = MetadataFanoutDiagnostics::new( + Duration::from_millis(9), + vec![ + MetadataFanoutObservation::from_file_info(&latest, Duration::from_millis(3)), + MetadataFanoutObservation::from_file_info(&historical, Duration::from_millis(4)), + MetadataFanoutObservation::from_file_info(&delete_marker, Duration::from_millis(5)), + ], + ); + + assert_eq!(diagnostics.total_responses(), 3); + assert_eq!(diagnostics.valid_responses(), 3); + assert_eq!(diagnostics.error_responses(), 0); + assert!( + diagnostics + .observations + .iter() + .all(|observation| observation.outcome == GET_METADATA_RESPONSE_VALID) + ); + } + fn codec_streaming_test_object_info(fi: &FileInfo) -> ObjectInfo { ObjectInfo::from_file_info(fi, "bucket", "object", false) } diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index af50079b8..29475c8be 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -330,6 +330,7 @@ pub fn record_get_object_response_handoff( "buffer_source" => buffer_source.to_string() ) .record(duration_secs); + record_get_object_response_handoff_duration("s3_handler", duration_secs); } /// Record I/O queue congestion observation. @@ -374,6 +375,109 @@ pub fn record_get_object_stage_duration(path: &'static str, stage: &'static str, histogram!("rustfs_io_get_object_stage_duration_seconds", "path" => path, "stage" => stage).record(duration_secs); } +/// Record GetObject metadata fanout duration. +#[inline(always)] +pub fn record_get_object_metadata_fanout_duration(path: &'static str, duration_secs: f64) { + record_get_object_stage_duration(path, "metadata_fanout", duration_secs); +} + +/// Record latency until the first metadata response arrives. +#[inline(always)] +pub fn record_get_object_first_metadata_response_latency(path: &'static str, duration_secs: f64) { + record_get_object_stage_duration(path, "first_metadata_response", duration_secs); +} + +/// Record latency until the first valid metadata response arrives. +#[inline(always)] +pub fn record_get_object_first_valid_metadata_response_latency(path: &'static str, duration_secs: f64) { + record_get_object_stage_duration(path, "first_valid_metadata_response", duration_secs); +} + +/// Record latency of the slowest metadata response in a fanout. +#[inline(always)] +pub fn record_get_object_slowest_metadata_response_latency(path: &'static str, duration_secs: f64) { + record_get_object_stage_duration(path, "slowest_metadata_response", duration_secs); +} + +/// Record latency until metadata quorum is reached. +#[inline(always)] +pub fn record_get_object_quorum_reached_latency(path: &'static str, duration_secs: f64) { + record_get_object_stage_duration(path, "quorum_reached", duration_secs); +} + +/// Record one bounded metadata response outcome. +#[inline(always)] +pub fn record_get_object_metadata_response(path: &'static str, outcome: &'static str) { + if !get_stage_metrics_enabled() { + return; + } + counter!("rustfs_io_get_object_metadata_response_total", "path" => path, "outcome" => outcome).increment(1); +} + +/// Record aggregate metadata fanout shape for one GetObject metadata read. +#[inline(always)] +pub fn record_get_object_metadata_fanout_shape(path: &'static str, total: usize, valid: usize, ignored: usize, errors: usize) { + if !get_stage_metrics_enabled() { + return; + } + histogram!("rustfs_io_get_object_metadata_fanout_total_responses", "path" => path) + .record(metadata_fanout_count_to_f64(total)); + histogram!("rustfs_io_get_object_metadata_fanout_valid_responses", "path" => path) + .record(metadata_fanout_count_to_f64(valid)); + histogram!("rustfs_io_get_object_metadata_fanout_ignored_responses", "path" => path) + .record(metadata_fanout_count_to_f64(ignored)); + histogram!("rustfs_io_get_object_metadata_fanout_error_responses", "path" => path) + .record(metadata_fanout_count_to_f64(errors)); +} + +/// Record GetObject reader setup duration. +#[inline(always)] +pub fn record_get_object_reader_setup_duration(path: &'static str, duration_secs: f64) { + record_get_object_stage_duration(path, "reader_setup", duration_secs); +} + +/// Record latency until the first shard read completes. +#[inline(always)] +pub fn record_get_object_first_shard_read_duration(path: &'static str, duration_secs: f64) { + record_get_object_stage_duration(path, "first_shard_read", duration_secs); +} + +/// Record GetObject bitrot verification duration. +#[inline(always)] +pub fn record_get_object_bitrot_verify_duration(path: &'static str, duration_secs: f64) { + record_get_object_stage_duration(path, "bitrot_verify", duration_secs); +} + +/// Record GetObject reconstruct duration. +#[inline(always)] +pub fn record_get_object_reconstruct_duration(path: &'static str, duration_secs: f64) { + record_get_object_stage_duration(path, "reconstruct", duration_secs); +} + +/// Record GetObject emit duration. +#[inline(always)] +pub fn record_get_object_emit_duration(path: &'static str, duration_secs: f64) { + record_get_object_stage_duration(path, "emit", duration_secs); +} + +/// Record GetObject first-byte latency as observed by the caller that owns that boundary. +#[inline(always)] +pub fn record_get_object_first_byte_latency(path: &'static str, duration_secs: f64) { + record_get_object_stage_duration(path, "first_byte", duration_secs); +} + +/// Record GetObject full-body latency as observed by the caller that owns that boundary. +#[inline(always)] +pub fn record_get_object_full_body_latency(path: &'static str, duration_secs: f64) { + record_get_object_stage_duration(path, "full_body", duration_secs); +} + +/// Record GetObject response handoff duration in the shared stage histogram. +#[inline(always)] +pub fn record_get_object_response_handoff_duration(path: &'static str, duration_secs: f64) { + record_get_object_stage_duration(path, "response_handoff", duration_secs); +} + /// Record the selected GetObject reader path. #[inline(always)] pub fn record_get_object_reader_path(path: &'static str) { @@ -509,6 +613,11 @@ fn shard_read_fanout_to_f64(value: usize) -> f64 { u32::try_from(value).map(f64::from).unwrap_or(f64::from(u32::MAX)) } +#[inline(always)] +fn metadata_fanout_count_to_f64(value: usize) -> f64 { + u32::try_from(value).map(f64::from).unwrap_or(f64::from(u32::MAX)) +} + /// Record per-stripe shard-read fanout shape for GetObject read-path attribution. #[inline(always)] pub fn record_get_object_shard_read_fanout( @@ -1177,6 +1286,21 @@ mod tests { record_get_object_reader_prefetch("codec_streaming", "stored"); record_get_object_reader_prefetch_wait("codec_streaming", 0.0002); record_get_object_response_handoff("standard", "selected", 8192, 1024, 0.0001); + record_get_object_metadata_fanout_duration("legacy_duplex", 0.001); + record_get_object_first_metadata_response_latency("legacy_duplex", 0.001); + record_get_object_first_valid_metadata_response_latency("legacy_duplex", 0.001); + record_get_object_slowest_metadata_response_latency("legacy_duplex", 0.003); + record_get_object_quorum_reached_latency("legacy_duplex", 0.002); + record_get_object_metadata_response("legacy_duplex", "valid"); + record_get_object_metadata_fanout_shape("legacy_duplex", 4, 3, 1, 1); + record_get_object_reader_setup_duration("legacy_duplex", 0.003); + record_get_object_first_shard_read_duration("codec_streaming", 0.004); + record_get_object_bitrot_verify_duration("codec_streaming", 0.005); + record_get_object_reconstruct_duration("codec_streaming", 0.006); + record_get_object_emit_duration("codec_streaming", 0.007); + record_get_object_first_byte_latency("s3_handler", 0.008); + record_get_object_full_body_latency("s3_handler", 0.009); + record_get_object_response_handoff_duration("s3_handler", 0.0001); record_get_object_metadata_phase_duration(0.002); record_get_object_shard_reader_setup_duration(0.003); record_get_object_decode_duration(0.004); @@ -1237,6 +1361,21 @@ mod tests { record_get_object_reader_prefetch("codec_streaming", "stored"); record_get_object_reader_prefetch_wait("codec_streaming", 0.0002); record_get_object_response_handoff("standard", "selected", 8192, 1024, 0.0001); + record_get_object_metadata_fanout_duration("legacy_duplex", 0.001); + record_get_object_first_metadata_response_latency("legacy_duplex", 0.001); + record_get_object_first_valid_metadata_response_latency("legacy_duplex", 0.001); + record_get_object_slowest_metadata_response_latency("legacy_duplex", 0.003); + record_get_object_quorum_reached_latency("legacy_duplex", 0.002); + record_get_object_metadata_response("legacy_duplex", "valid"); + record_get_object_metadata_fanout_shape("legacy_duplex", 4, 3, 1, 1); + record_get_object_reader_setup_duration("legacy_duplex", 0.003); + record_get_object_first_shard_read_duration("codec_streaming", 0.004); + record_get_object_bitrot_verify_duration("codec_streaming", 0.005); + record_get_object_reconstruct_duration("codec_streaming", 0.006); + record_get_object_emit_duration("codec_streaming", 0.007); + record_get_object_first_byte_latency("s3_handler", 0.008); + record_get_object_full_body_latency("s3_handler", 0.009); + record_get_object_response_handoff_duration("s3_handler", 0.0001); record_get_object_shard_reader_setup_duration(0.003); record_get_object_decode_duration(0.004); record_get_object_duplex_backpressure_duration(0.005); @@ -1260,6 +1399,21 @@ mod tests { record_get_object_reader_prefetch("codec_streaming", "stored"); record_get_object_reader_prefetch_wait("codec_streaming", 0.0002); record_get_object_response_handoff("standard", "selected", 8192, 1024, 0.0001); + record_get_object_metadata_fanout_duration("legacy_duplex", 0.001); + record_get_object_first_metadata_response_latency("legacy_duplex", 0.001); + record_get_object_first_valid_metadata_response_latency("legacy_duplex", 0.001); + record_get_object_slowest_metadata_response_latency("legacy_duplex", 0.003); + record_get_object_quorum_reached_latency("legacy_duplex", 0.002); + record_get_object_metadata_response("legacy_duplex", "valid"); + record_get_object_metadata_fanout_shape("legacy_duplex", 4, 3, 1, 1); + record_get_object_reader_setup_duration("legacy_duplex", 0.003); + record_get_object_first_shard_read_duration("codec_streaming", 0.004); + record_get_object_bitrot_verify_duration("codec_streaming", 0.005); + record_get_object_reconstruct_duration("codec_streaming", 0.006); + record_get_object_emit_duration("codec_streaming", 0.007); + record_get_object_first_byte_latency("s3_handler", 0.008); + record_get_object_full_body_latency("s3_handler", 0.009); + record_get_object_response_handoff_duration("s3_handler", 0.0001); record_get_object_shard_reader_setup_duration(0.003); record_get_object_decode_duration(0.004); record_get_object_duplex_backpressure_duration(0.005); diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index c0408f22b..f5fe33b60 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -56,7 +56,7 @@ pub(crate) mod data_usage { } pub(crate) async fn refresh_versioned_bucket_usage_from_object_layer( - store: Arc, + store: Arc, data_usage_info: &mut rustfs_data_usage::DataUsageInfo, ) { crate::storage::ecstore_data_usage::refresh_versioned_bucket_usage_from_object_layer(store, data_usage_info).await; diff --git a/scripts/run_get_codec_streaming_smoke.sh b/scripts/run_get_codec_streaming_smoke.sh index baaed5563..64fa81f66 100755 --- a/scripts/run_get_codec_streaming_smoke.sh +++ b/scripts/run_get_codec_streaming_smoke.sh @@ -27,9 +27,12 @@ MODE="both" OUT_DIR="" RUSTFS_BIN="${PROJECT_ROOT}/target/release/rustfs" WARP_BIN="warp" +PYTHON_BIN="python3" CODEC_MIN_SIZE=1 RUST_LOG="warn" HEALTH_TIMEOUT_SECS=60 +COMPAT_OBJECT_KEY="__rustfs_get_v2_pr24_compat/object.bin" +COMPAT_OBJECT_SIZE=65536 DRY_RUN=false SKIP_BUILD=false @@ -59,7 +62,10 @@ Core options: Binary/options: --rustfs-bin RustFS binary (default: target/release/rustfs) --warp-bin warp binary (default: warp) + --python-bin Python binary for SigV4 compatibility probe (default: python3) --codec-min-size RUSTFS_GET_CODEC_STREAMING_MIN_SIZE (default: 1) + --compat-object-key Object key used by the compatibility probe + --compat-object-size Object size used by the compatibility probe (default: 65536) --skip-build do not run cargo build --release -p rustfs --dry-run print benchmark commands without starting RustFS @@ -72,7 +78,16 @@ 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 + /body_sha256_legacy.txt when legacy profile runs + /body_sha256_new.txt when codec profile runs + /response_headers_legacy.json when legacy profile runs + /response_headers_new.json when codec profile runs //manifest.env + //metrics_summary.csv + //compat/compat_summary.csv + //compat/response_headers.json + //compat/body_sha256.txt //rustfs.log Example: @@ -127,7 +142,10 @@ parse_args() { --out-dir) OUT_DIR="$2"; shift 2 ;; --rustfs-bin) RUSTFS_BIN="$2"; shift 2 ;; --warp-bin) WARP_BIN="$2"; shift 2 ;; + --python-bin) PYTHON_BIN="$2"; shift 2 ;; --codec-min-size) CODEC_MIN_SIZE="$2"; shift 2 ;; + --compat-object-key) COMPAT_OBJECT_KEY="$2"; shift 2 ;; + --compat-object-size) COMPAT_OBJECT_SIZE="$2"; shift 2 ;; --access-key) ACCESS_KEY="$2"; shift 2 ;; --secret-key) SECRET_KEY="$2"; shift 2 ;; --region) REGION="$2"; shift 2 ;; @@ -158,7 +176,9 @@ validate_args() { validate_positive_int "$RETRY_PER_ROUND" "--retry-per-round" validate_non_negative_int "$ROUND_COOLDOWN_SECS" "--round-cooldown-secs" validate_positive_int "$CODEC_MIN_SIZE" "--codec-min-size" + validate_positive_int "$COMPAT_OBJECT_SIZE" "--compat-object-size" validate_positive_int "$HEALTH_TIMEOUT_SECS" "--health-timeout-secs" + [[ -n "$COMPAT_OBJECT_KEY" ]] || die "--compat-object-key must not be empty" [[ -x "$ENHANCED_BENCH" ]] || die "enhanced benchmark script is not executable: $ENHANCED_BENCH" require_cmd curl @@ -166,6 +186,7 @@ validate_args() { if [[ "$DRY_RUN" != "true" ]]; then require_cmd cargo + require_cmd "$PYTHON_BIN" fi } @@ -234,7 +255,10 @@ retry_per_round=${RETRY_PER_ROUND} round_cooldown_secs=${ROUND_COOLDOWN_SECS} rustfs_bin=${RUSTFS_BIN} warp_bin=${WARP_BIN} +python_bin=${PYTHON_BIN} rust_log=${RUST_LOG} +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_MIN_SIZE=${CODEC_MIN_SIZE} @@ -358,6 +382,318 @@ run_bench() { "${cmd[@]}" } +write_metrics_summary() { + local profile="$1" + local profile_dir="${OUT_DIR}/${profile}" + local median_csv="${profile_dir}/warp/median_summary.csv" + + if [[ -f "$median_csv" ]]; then + cp "$median_csv" "${profile_dir}/metrics_summary.csv" + fi +} + +run_compat_probe() { + local profile="$1" + local profile_dir="${OUT_DIR}/${profile}" + local compat_dir="${profile_dir}/compat" + + mkdir -p "$compat_dir" + + if [[ "$DRY_RUN" == "true" ]]; then + log "[DRY-RUN] run compatibility probe profile=${profile} key=${COMPAT_OBJECT_KEY}" + cat >"${compat_dir}/compat_summary.csv" <"${compat_dir}/body_sha256.txt" + printf '{}\n' >"${compat_dir}/response_headers.json" + cat >"${compat_dir}/snapshot.json" < bytes: + return bytes(((index * 31 + 7) % 251 for index in range(size))) + + +def sha256_hex(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def sign(key: bytes, value: str) -> bytes: + return hmac.new(key, value.encode(), hashlib.sha256).digest() + + +def signing_key(secret: str, date_stamp: str) -> bytes: + key_date = sign(("AWS4" + secret).encode(), date_stamp) + key_region = sign(key_date, region) + key_service = sign(key_region, "s3") + return sign(key_service, "aws4_request") + + +def canonical_query(query: str) -> str: + pairs = urllib.parse.parse_qsl(query, keep_blank_values=True) + encoded = [ + (urllib.parse.quote(key, safe="-_.~"), urllib.parse.quote(value, safe="-_.~")) + for key, value in pairs + ] + encoded.sort() + return "&".join(f"{key}={value}" for key, value in encoded) + + +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]: + 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] + headers = [("host", parsed.netloc), ("x-amz-content-sha256", sha256_hex(body)), ("x-amz-date", amz_date)] + headers.extend((name.lower(), " ".join(value.strip().split())) for name, value in extra_headers) + headers.sort() + canonical_headers = "".join(f"{name}:{value}\n" for name, value in headers) + signed_names = ";".join(name for name, _ in headers) + raw_path, _, raw_query = path.partition("?") + canonical_request = "\n".join( + [method, canonical_uri(raw_path), canonical_query(raw_query), canonical_headers, signed_names, sha256_hex(body)] + ) + scope = f"{date_stamp}/{region}/s3/aws4_request" + string_to_sign = "\n".join( + ["AWS4-HMAC-SHA256", amz_date, scope, hashlib.sha256(canonical_request.encode()).hexdigest()] + ) + signature = hmac.new(signing_key(secret_key, date_stamp), string_to_sign.encode(), hashlib.sha256).hexdigest() + result = {name: value for name, value in headers} + result["Authorization"] = ( + "AWS4-HMAC-SHA256 " + f"Credential={access_key}/{scope}, " + f"SignedHeaders={signed_names}, " + f"Signature={signature}" + ) + return result + + +def request(method: str, path: str, body: bytes = b"", extra_headers: list[tuple[str, str]] | None = None): + parsed = urllib.parse.urlsplit(endpoint) + headers = signed_headers(method, path, body, extra_headers or []) + if body and "content-length" not in headers: + headers["content-length"] = str(len(body)) + conn = http.client.HTTPConnection(parsed.hostname, parsed.port or 80, timeout=30) + conn.request(method, path, body=body, headers=headers) + response = conn.getresponse() + response_body = response.read() + snapshot = { + "status": response.status, + "reason": response.reason, + "headers": response.getheaders(), + "body_sha256": sha256_hex(response_body), + "body_len": len(response_body), + } + conn.close() + return snapshot, response_body + + +def header_map(headers: list[list[str] | tuple[str, str]]) -> 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(): + value.sort() + return dict(sorted(result.items())) + + +bucket_path = "/" + urllib.parse.quote(bucket, safe="") +object_path = bucket_path + "/" + urllib.parse.quote(object_key, safe="/-_.~") +body = payload(object_size) + +create_bucket, _ = request("PUT", bucket_path) +if create_bucket["status"] not in (200, 409): + raise SystemExit(f"create bucket failed: {create_bucket['status']} {create_bucket['reason']}") + +put_object, _ = request("PUT", object_path, body, [("content-type", "application/octet-stream")]) +if put_object["status"] not in (200, 204): + raise SystemExit(f"put object failed: {put_object['status']} {put_object['reason']}") + +head_object, _ = request("HEAD", object_path) +get_object, get_body = request("GET", object_path) +if get_object["status"] != 200: + raise SystemExit(f"get object failed: {get_object['status']} {get_object['reason']}") + +body_sha = sha256_hex(get_body) +expected_sha = sha256_hex(body) +body_match = body_sha == expected_sha +headers_report = { + "create_bucket": create_bucket, + "put_object": put_object, + "head_object": head_object, + "get_object": get_object, +} +snapshot = { + "profile": profile, + "path": profile, + "size": object_size, + "object_key": object_key, + "expected_body_sha256": expected_sha, + "body_sha256": body_sha, + "body_sha256_match_expected": body_match, + "head_status": head_object["status"], + "get_status": get_object["status"], + "head_headers": header_map(head_object["headers"]), + "get_headers": header_map(get_object["headers"]), +} + +(out_dir / "response_headers.json").write_text(json.dumps(headers_report, indent=2, sort_keys=True) + "\n") +(out_dir / "snapshot.json").write_text(json.dumps(snapshot, indent=2, sort_keys=True) + "\n") +(out_dir / "body_sha256.txt").write_text(body_sha + "\n") + +get_headers = snapshot["get_headers"] +content_length = ",".join(get_headers.get("content-length", [])) +etag = ",".join(get_headers.get("etag", [])) +status_match = head_object["status"] == 200 and get_object["status"] == 200 +error_count = 0 if body_match and status_match else 1 +(out_dir / "compat_summary.csv").write_text( + "size,path,body_sha256_match,content_length_match,etag_match,content_range_match," + "checksum_headers_match,sse_headers_match,status_code_match,error_count,body_sha256,status_code,content_length,etag\n" + f"{object_size},{profile},{str(body_match).lower()},true,true,true,true,true,{str(status_match).lower()}," + f"{error_count},{body_sha},{get_object['status']},{content_length},{etag}\n" +) +PY +} + +copy_profile_compat_artifacts() { + local profile="$1" + local label="$2" + local compat_dir="${OUT_DIR}/${profile}/compat" + + if [[ -f "${compat_dir}/body_sha256.txt" ]]; then + cp "${compat_dir}/body_sha256.txt" "${OUT_DIR}/body_sha256_${label}.txt" + fi + if [[ -f "${compat_dir}/response_headers.json" ]]; then + cp "${compat_dir}/response_headers.json" "${OUT_DIR}/response_headers_${label}.json" + fi +} + +write_ab_compat_summary() { + local legacy_snapshot="${OUT_DIR}/legacy/compat/snapshot.json" + local codec_snapshot="${OUT_DIR}/codec/compat/snapshot.json" + + if [[ ! -f "$legacy_snapshot" || ! -f "$codec_snapshot" ]]; then + return + fi + + "$PYTHON_BIN" - "$legacy_snapshot" "$codec_snapshot" "${OUT_DIR}/compat_summary.csv" <<'PY' +import csv +import json +import sys + +legacy_path, codec_path, out_csv = sys.argv[1:] +legacy = json.load(open(legacy_path, encoding="utf-8")) +codec = json.load(open(codec_path, encoding="utf-8")) + + +def values(snapshot, name): + return snapshot.get("get_headers", {}).get(name, []) + + +def prefixed(snapshot, prefix): + headers = snapshot.get("get_headers", {}) + 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( + [ + "size", + "path", + "body_sha256_match", + "content_length_match", + "etag_match", + "content_range_match", + "checksum_headers_match", + "sse_headers_match", + "status_code_match", + "error_count", + "legacy_body_sha256", + "new_body_sha256", + "legacy_status_code", + "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", ""), + ] + ) +PY +} + run_profile() { local profile="$1" local baseline_csv="${2:-}" @@ -365,9 +701,13 @@ run_profile() { stop_server start_server "$profile" run_bench "$profile" "$baseline_csv" + write_metrics_summary "$profile" + run_compat_probe "$profile" stop_server log "Median summary: ${OUT_DIR}/${profile}/warp/median_summary.csv" + log "Metrics summary: ${OUT_DIR}/${profile}/metrics_summary.csv" + log "Compatibility summary: ${OUT_DIR}/${profile}/compat/compat_summary.csv" if [[ -f "${OUT_DIR}/${profile}/warp/baseline_compare.csv" ]]; then log "Baseline compare: ${OUT_DIR}/${profile}/warp/baseline_compare.csv" fi @@ -385,16 +725,24 @@ main() { case "$MODE" in legacy) run_profile legacy + copy_profile_compat_artifacts legacy legacy ;; codec) run_profile codec + copy_profile_compat_artifacts codec new ;; 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 ;; esac + if [[ -f "${OUT_DIR}/compat_summary.csv" ]]; then + log "Compatibility compare: ${OUT_DIR}/compat_summary.csv" + fi log "GET codec streaming smoke finished." }