diff --git a/crates/ecstore/src/diagnostics/get.rs b/crates/ecstore/src/diagnostics/get.rs index 501722c88..c71a6e7b7 100644 --- a/crates/ecstore/src/diagnostics/get.rs +++ b/crates/ecstore/src/diagnostics/get.rs @@ -22,6 +22,15 @@ pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE: &str = "codec_st 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"; +pub(crate) const GET_CODEC_STREAMING_DECISION_USE: &str = "use"; +pub(crate) const GET_CODEC_STREAMING_DECISION_FALLBACK: &str = "fallback"; +pub(crate) const GET_CODEC_STREAMING_REASON_NONE: &str = "none"; +pub(crate) const GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART: &str = "plain_single_part"; +pub(crate) const GET_CODEC_STREAMING_OBJECT_CLASS_RANGE: &str = "range"; +pub(crate) const GET_CODEC_STREAMING_OBJECT_CLASS_ENCRYPTED: &str = "encrypted"; +pub(crate) const GET_CODEC_STREAMING_OBJECT_CLASS_COMPRESSED: &str = "compressed"; +pub(crate) const GET_CODEC_STREAMING_OBJECT_CLASS_REMOTE: &str = "remote"; +pub(crate) const GET_CODEC_STREAMING_OBJECT_CLASS_MULTIPART: &str = "multipart"; pub(crate) const GET_STAGE_DECODE: &str = "decode"; pub(crate) const GET_STAGE_EMIT: &str = "emit"; @@ -220,6 +229,15 @@ mod tests { 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_CODEC_STREAMING_DECISION_USE, "use"); + assert_eq!(GET_CODEC_STREAMING_DECISION_FALLBACK, "fallback"); + assert_eq!(GET_CODEC_STREAMING_REASON_NONE, "none"); + assert_eq!(GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, "plain_single_part"); + assert_eq!(GET_CODEC_STREAMING_OBJECT_CLASS_RANGE, "range"); + assert_eq!(GET_CODEC_STREAMING_OBJECT_CLASS_ENCRYPTED, "encrypted"); + assert_eq!(GET_CODEC_STREAMING_OBJECT_CLASS_COMPRESSED, "compressed"); + assert_eq!(GET_CODEC_STREAMING_OBJECT_CLASS_REMOTE, "remote"); + assert_eq!(GET_CODEC_STREAMING_OBJECT_CLASS_MULTIPART, "multipart"); 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/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 4f4e525ab..b8efb0c26 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -307,11 +307,19 @@ const GET_OBJECT_METADATA_CACHE_MAX_ENTRIES: usize = 1024; const ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE: &str = "RUSTFS_GET_CODEC_STREAMING_ENABLE"; const ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE: &str = "RUSTFS_GET_CODEC_STREAMING_MIN_SIZE"; const ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE: &str = "RUSTFS_GET_CODEC_STREAMING_ENGINE"; +const ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT: &str = "RUSTFS_GET_CODEC_STREAMING_ROLLOUT"; +const ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED"; +const ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED"; const DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE: bool = false; const DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE: usize = MI_B; const DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENGINE: &str = GET_CODEC_STREAMING_ENGINE_LEGACY; +const DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT: &str = "off"; const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ENABLE"; const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: bool = false; +const ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT: &str = "RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT"; +const DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT: u32 = 100; +const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT"; +const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: u32 = 100; static OBJECT_LOCK_DIAG_ENABLED: OnceLock = OnceLock::new(); mod heal; @@ -362,28 +370,107 @@ pub fn get_object_lock_diag_slow_hold_threshold() -> Duration { /// Check if lock optimization is enabled. /// When enabled, read locks are released after metadata read instead of /// being held for the entire data transfer duration. +/// +/// **Note**: Cached via `OnceLock` — env var changes require process restart. pub fn is_lock_optimization_enabled() -> bool { - rustfs_utils::get_env_bool( - rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, - rustfs_config::DEFAULT_OBJECT_LOCK_OPTIMIZATION_ENABLE, - ) + static CACHED: OnceLock = OnceLock::new(); + *CACHED.get_or_init(|| { + rustfs_utils::get_env_bool( + rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, + rustfs_config::DEFAULT_OBJECT_LOCK_OPTIMIZATION_ENABLE, + ) + }) } /// Check if deadlock detection is enabled. /// When enabled, lock operations are recorded for deadlock analysis. +/// +/// **Note**: Cached via `OnceLock` — env var changes require process restart. pub fn is_deadlock_detection_enabled() -> bool { - rustfs_utils::get_env_bool( - rustfs_config::ENV_OBJECT_DEADLOCK_DETECTION_ENABLE, - rustfs_config::DEFAULT_OBJECT_DEADLOCK_DETECTION_ENABLE, - ) + static CACHED: OnceLock = OnceLock::new(); + *CACHED.get_or_init(|| { + rustfs_utils::get_env_bool( + rustfs_config::ENV_OBJECT_DEADLOCK_DETECTION_ENABLE, + rustfs_config::DEFAULT_OBJECT_DEADLOCK_DETECTION_ENABLE, + ) + }) } +/// **Note**: Cached via `OnceLock` — env var changes require process restart. fn is_get_codec_streaming_enabled() -> bool { - rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE) + #[cfg(test)] + let enabled = rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE); + #[cfg(not(test))] + static CACHED: OnceLock = OnceLock::new(); + #[cfg(not(test))] + let enabled = *CACHED.get_or_init(|| { + rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE) + }); + enabled } +/// **Note**: Cached via `OnceLock` — env var changes require process restart. fn is_get_metadata_early_stop_enabled() -> bool { - rustfs_utils::get_env_bool(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE) + static CACHED: OnceLock = OnceLock::new(); + *CACHED.get_or_init(|| { + rustfs_utils::get_env_bool(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE) + }) +} + +/// Determine if an optimization should be enabled for a specific request. +/// +/// Uses a stable hash of `(bucket, object)` to ensure the same object +/// always gets consistent behavior. This enables percentage-based gradual rollout. +/// +/// **Note**: `base_enabled` must be pre-cached (via `OnceLock`) to avoid +/// per-request `std::env::var()` calls. +fn is_optimization_enabled_for_request(base_enabled: bool, rollout_pct: u32, bucket: &str, object: &str) -> bool { + if !base_enabled || rollout_pct == 0 { + return false; + } + if rollout_pct >= 100 { + return true; + } + + // Stable hash: same (bucket, object) always produces the same result + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + bucket.hash(&mut hasher); + object.hash(&mut hasher); + let hash = hasher.finish() % 100; + + (hash as u32) < rollout_pct +} + +fn get_codec_streaming_rollout_pct() -> u32 { + static CACHED: OnceLock = OnceLock::new(); + *CACHED.get_or_init(|| { + rustfs_utils::get_env_u32(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT) + }) +} + +fn get_metadata_early_stop_rollout_pct() -> u32 { + static CACHED: OnceLock = OnceLock::new(); + *CACHED.get_or_init(|| { + rustfs_utils::get_env_u32( + ENV_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT, + DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT, + ) + }) +} + +/// Should this specific request use codec streaming? +pub fn should_use_codec_streaming(bucket: &str, object: &str) -> bool { + let base = is_get_codec_streaming_enabled(); + let pct = get_codec_streaming_rollout_pct(); + is_optimization_enabled_for_request(base, pct, bucket, object) +} + +/// Should this specific request use metadata early-stop? +pub fn should_use_metadata_early_stop(bucket: &str, object: &str) -> bool { + let base = is_get_metadata_early_stop_enabled(); + let pct = get_metadata_early_stop_rollout_pct(); + is_optimization_enabled_for_request(base, pct, bucket, object) } fn get_codec_streaming_min_size() -> usize { @@ -405,6 +492,23 @@ fn get_codec_streaming_engine() -> GetCodecStreamingEngine { } } +fn get_codec_streaming_rollout() -> GetCodecStreamingRollout { + let rollout = rustfs_utils::get_env_str(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT); + match rollout.trim() { + value if value.eq_ignore_ascii_case("internal") => GetCodecStreamingRollout::Internal, + value if value.eq_ignore_ascii_case("benchmark") => GetCodecStreamingRollout::Benchmark, + _ => GetCodecStreamingRollout::Off, + } +} + +fn is_get_codec_streaming_body_compat_confirmed() -> bool { + rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, false) +} + +fn is_get_codec_streaming_header_compat_confirmed() -> bool { + rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, false) +} + fn build_get_codec_streaming_decode_engine(erasure: coding::Erasure) -> std::io::Result { match get_codec_streaming_engine() { GetCodecStreamingEngine::Legacy => Ok(CodecStreamingDecodeEngine::legacy(erasure)), @@ -425,9 +529,25 @@ enum GetCodecStreamingDecision { Fallback(GetCodecStreamingFallbackReason), } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum GetCodecStreamingRollout { + Off, + Internal, + Benchmark, +} + +impl GetCodecStreamingRollout { + const fn is_opted_in(self) -> bool { + !matches!(self, Self::Off) + } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum GetCodecStreamingFallbackReason { Disabled, + RolloutNotOptedIn, + BodyCompatibilityUnconfirmed, + HeaderCompatibilityUnconfirmed, LockOptimizationDisabled, Range, BelowMinSize, @@ -436,12 +556,16 @@ enum GetCodecStreamingFallbackReason { Remote, Multipart, InvalidMinSize, + ReadQuorumNotSafe, } impl GetCodecStreamingFallbackReason { const fn as_str(self) -> &'static str { match self { Self::Disabled => "disabled", + Self::RolloutNotOptedIn => "rollout_not_opted_in", + Self::BodyCompatibilityUnconfirmed => "body_compatibility_unconfirmed", + Self::HeaderCompatibilityUnconfirmed => "header_compatibility_unconfirmed", Self::LockOptimizationDisabled => "lock_optimization_disabled", Self::Range => "range", Self::BelowMinSize => "below_min_size", @@ -450,46 +574,162 @@ impl GetCodecStreamingFallbackReason { Self::Remote => "remote", Self::Multipart => "multipart", Self::InvalidMinSize => "invalid_min_size", + Self::ReadQuorumNotSafe => "read_quorum_not_safe", } } } -fn get_codec_streaming_reader_decision( +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum GetCodecStreamingObjectClass { + PlainSinglePart, + Range, + Encrypted, + Compressed, + Remote, + Multipart, +} + +impl GetCodecStreamingObjectClass { + const fn as_str(self) -> &'static str { + match self { + Self::PlainSinglePart => crate::diagnostics::get::GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, + Self::Range => crate::diagnostics::get::GET_CODEC_STREAMING_OBJECT_CLASS_RANGE, + Self::Encrypted => crate::diagnostics::get::GET_CODEC_STREAMING_OBJECT_CLASS_ENCRYPTED, + Self::Compressed => crate::diagnostics::get::GET_CODEC_STREAMING_OBJECT_CLASS_COMPRESSED, + Self::Remote => crate::diagnostics::get::GET_CODEC_STREAMING_OBJECT_CLASS_REMOTE, + Self::Multipart => crate::diagnostics::get::GET_CODEC_STREAMING_OBJECT_CLASS_MULTIPART, + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct GetCodecStreamingGate { + object_class: GetCodecStreamingObjectClass, + decision: GetCodecStreamingDecision, +} + +fn record_get_codec_streaming_gate_decision(object_class: GetCodecStreamingObjectClass, decision: GetCodecStreamingDecision) { + let (outcome, reason) = match decision { + GetCodecStreamingDecision::Use => ( + crate::diagnostics::get::GET_CODEC_STREAMING_DECISION_USE, + crate::diagnostics::get::GET_CODEC_STREAMING_REASON_NONE, + ), + GetCodecStreamingDecision::Fallback(reason) => { + (crate::diagnostics::get::GET_CODEC_STREAMING_DECISION_FALLBACK, reason.as_str()) + } + }; + rustfs_io_metrics::record_get_object_codec_streaming_decision(outcome, object_class.as_str(), reason); +} + +fn classify_get_codec_streaming_object_class( + range: &Option, + object_info: &ObjectInfo, + fi: &FileInfo, +) -> GetCodecStreamingObjectClass { + if range.is_some() { + return GetCodecStreamingObjectClass::Range; + } + if object_info.is_encrypted() { + return GetCodecStreamingObjectClass::Encrypted; + } + if object_info.is_compressed() { + return GetCodecStreamingObjectClass::Compressed; + } + if object_info.is_remote() { + return GetCodecStreamingObjectClass::Remote; + } + if fi.parts.len() != 1 { + return GetCodecStreamingObjectClass::Multipart; + } + GetCodecStreamingObjectClass::PlainSinglePart +} + +fn get_codec_streaming_reader_gate( range: &Option, object_info: &ObjectInfo, fi: &FileInfo, lock_optimization_enabled: bool, -) -> GetCodecStreamingDecision { +) -> GetCodecStreamingGate { + let object_class = classify_get_codec_streaming_object_class(range, object_info, fi); + if !is_get_codec_streaming_enabled() { - return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Disabled); + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Disabled), + }; + } + if !get_codec_streaming_rollout().is_opted_in() { + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::RolloutNotOptedIn), + }; + } + if !is_get_codec_streaming_body_compat_confirmed() { + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BodyCompatibilityUnconfirmed), + }; + } + if !is_get_codec_streaming_header_compat_confirmed() { + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::HeaderCompatibilityUnconfirmed), + }; + } + if object_class == GetCodecStreamingObjectClass::Range { + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Range), + }; } if !lock_optimization_enabled { - return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::LockOptimizationDisabled); - } - if range.is_some() { - return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Range); + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::LockOptimizationDisabled), + }; } let Ok(min_size) = i64::try_from(get_codec_streaming_min_size()) else { - return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::InvalidMinSize); + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::InvalidMinSize), + }; }; if object_info.size < min_size { - return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize); + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize), + }; } - if object_info.is_encrypted() { - return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Encrypted); + if object_class == GetCodecStreamingObjectClass::Encrypted { + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Encrypted), + }; } - if object_info.is_compressed() { - return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Compressed); + if object_class == GetCodecStreamingObjectClass::Compressed { + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Compressed), + }; } - if object_info.is_remote() { - return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Remote); + if object_class == GetCodecStreamingObjectClass::Remote { + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Remote), + }; } - if fi.parts.len() != 1 { - return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Multipart); + if object_class == GetCodecStreamingObjectClass::Multipart { + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Multipart), + }; } - GetCodecStreamingDecision::Use + GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Use, + } } fn is_confirmed_complete_part_missing(err: &str) -> bool { @@ -1233,10 +1473,11 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { return Ok(reader); } - let codec_streaming_decision = get_codec_streaming_reader_decision(&range, &object_info, &fi, lock_optimization_enabled); + let codec_streaming_gate = get_codec_streaming_reader_gate(&range, &object_info, &fi, lock_optimization_enabled); if object_info.is_remote() { - if let GetCodecStreamingDecision::Fallback(reason) = codec_streaming_decision { + if let GetCodecStreamingDecision::Fallback(reason) = codec_streaming_gate.decision { + record_get_codec_streaming_gate_decision(codec_streaming_gate.object_class, codec_streaming_gate.decision); rustfs_io_metrics::record_get_object_codec_streaming_fallback(reason.as_str()); } rustfs_io_metrics::record_get_object_reader_path(GET_OBJECT_PATH_REMOTE_TRANSITION); @@ -1267,24 +1508,40 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { read_lock_guard }; - match codec_streaming_decision { + match codec_streaming_gate.decision { GetCodecStreamingDecision::Use => { - rustfs_io_metrics::record_get_object_reader_path(GET_OBJECT_PATH_CODEC_STREAMING); - let stream = Self::get_object_decode_reader_with_fileinfo( + match Self::get_object_decode_reader_with_fileinfo( bucket, object, - fi, - files, + &fi, + &files, &disks, self.set_index, self.pool_index, opts.skip_verify_bitrot, ) - .await?; - let (reader, _offset, _length) = GetObjectReader::new(stream, range, &object_info, opts, &h).await?; - return Ok(reader); + .await? + { + read::GetCodecStreamingReaderBuildOutcome::Reader(stream) => { + record_get_codec_streaming_gate_decision( + codec_streaming_gate.object_class, + GetCodecStreamingDecision::Use, + ); + rustfs_io_metrics::record_get_object_reader_path(GET_OBJECT_PATH_CODEC_STREAMING); + let (reader, _offset, _length) = GetObjectReader::new(stream, range, &object_info, opts, &h).await?; + return Ok(reader); + } + read::GetCodecStreamingReaderBuildOutcome::Fallback(reason) => { + record_get_codec_streaming_gate_decision( + codec_streaming_gate.object_class, + GetCodecStreamingDecision::Fallback(reason), + ); + rustfs_io_metrics::record_get_object_codec_streaming_fallback(reason.as_str()); + } + } } GetCodecStreamingDecision::Fallback(reason) => { + record_get_codec_streaming_gate_decision(codec_streaming_gate.object_class, codec_streaming_gate.decision); rustfs_io_metrics::record_get_object_codec_streaming_fallback(reason.as_str()); } } diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index d2286489c..a585f4c24 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -48,6 +48,15 @@ const READ_REPAIR_HEAL_DEDUP_MAX_ENTRIES: usize = 4096; static READ_REPAIR_HEAL_CACHE: OnceLock>> = OnceLock::new(); +pub(super) enum GetCodecStreamingReaderBuildOutcome { + Reader(Box), + Fallback(GetCodecStreamingFallbackReason), +} + +pub(super) fn codec_streaming_reader_setup_fallback_reason(missing_shards: usize) -> Option { + (missing_shards > 0).then_some(GetCodecStreamingFallbackReason::ReadQuorumNotSafe) +} + #[derive(Clone, Copy, Debug)] struct MetadataFanoutObservation { outcome: &'static str, @@ -1646,8 +1655,13 @@ impl SetDisks { opts: &ObjectOptions, read_data: bool, ) -> Result<(FileInfo, Vec, Vec>)> { + let vid = opts.version_id.clone().unwrap_or_default(); + let use_metadata_cache = self.is_get_object_metadata_cache_enabled(bucket, opts, read_data).await; - if use_metadata_cache && let Some(cached) = self.cached_get_object_fileinfo(bucket, object).await { + if use_metadata_cache + && vid.is_empty() + && let Some(cached) = self.cached_get_object_fileinfo(bucket, object).await + { return Ok((cached.fi, cached.parts_metadata, cached.online_disks)); } @@ -1655,8 +1669,6 @@ impl SetDisks { let disks = disks.clone(); - let vid = opts.version_id.clone().unwrap_or_default(); - // TODO: optimize concurrency and break once enough slots are available let (parts_metadata, errs, metadata_fanout_diagnostics) = Self::read_all_fileinfo_observed( &disks, @@ -2186,14 +2198,14 @@ impl SetDisks { pub(super) async fn get_object_decode_reader_with_fileinfo( bucket: &str, object: &str, - fi: FileInfo, - files: Vec, + fi: &FileInfo, + files: &[FileInfo], disks: &[Option], - set_index: usize, - pool_index: usize, + _set_index: usize, + _pool_index: usize, skip_verify_bitrot: bool, - ) -> Result> { - let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, &files, &fi); + ) -> Result { + let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi); if fi.parts.len() != 1 { return Err(Error::other("codec streaming reader only supports single-part plain objects")); } @@ -2263,18 +2275,8 @@ impl SetDisks { } let missing_shards = reader_setup.completed_failed_shards(); - if missing_shards > 0 { - let version_id = fi.version_id.as_ref().map(ToString::to_string); - submit_read_repair_heal( - bucket, - object, - version_id.as_deref(), - pool_index, - set_index, - Some(part_number), - "missing_shards_streaming", - ) - .await; + if let Some(reason) = codec_streaming_reader_setup_fallback_reason(missing_shards) { + return Ok(GetCodecStreamingReaderBuildOutcome::Fallback(reason)); } let readers = reader_setup.readers; @@ -2294,9 +2296,9 @@ impl SetDisks { part_length, metrics_path, )?; - Ok(Box::new( + Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new( crate::erasure::coding::decode_reader::SyncErasureDecodeReader::new_with_metrics_path(reader, metrics_path), - )) + ))) } } @@ -3253,18 +3255,21 @@ mod tests { temp_env::with_vars( [ (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")), + (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")), (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), ], || { let fi = codec_streaming_test_fileinfo(1024, 1); let object_info = codec_streaming_test_object_info(&fi); assert_eq!( - get_codec_streaming_reader_decision(&None, &object_info, &fi, true), + get_codec_streaming_reader_gate(&None, &object_info, &fi, true).decision, GetCodecStreamingDecision::Use ); assert_eq!( - get_codec_streaming_reader_decision(&None, &object_info, &fi, false), + get_codec_streaming_reader_gate(&None, &object_info, &fi, false).decision, GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::LockOptimizationDisabled) ); @@ -3274,14 +3279,14 @@ mod tests { end: 1, }); assert_eq!( - get_codec_streaming_reader_decision(&range, &object_info, &fi, true), + get_codec_streaming_reader_gate(&range, &object_info, &fi, true).decision, GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Range) ); let multipart_fi = codec_streaming_test_fileinfo(1024, 2); let multipart_object_info = codec_streaming_test_object_info(&multipart_fi); assert_eq!( - get_codec_streaming_reader_decision(&None, &multipart_object_info, &multipart_fi, true), + get_codec_streaming_reader_gate(&None, &multipart_object_info, &multipart_fi, true).decision, GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Multipart) ); @@ -3291,7 +3296,7 @@ mod tests { .insert("x-amz-server-side-encryption".to_string(), "AES256".to_string()); let encrypted = codec_streaming_test_object_info(&encrypted_fi); assert_eq!( - get_codec_streaming_reader_decision(&None, &encrypted, &fi, true), + get_codec_streaming_reader_gate(&None, &encrypted, &encrypted_fi, true).decision, GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Encrypted) ); @@ -3299,14 +3304,14 @@ mod tests { insert_str(&mut compressed_fi.metadata, SUFFIX_COMPRESSION, "lz4".to_string()); let compressed = codec_streaming_test_object_info(&compressed_fi); assert_eq!( - get_codec_streaming_reader_decision(&None, &compressed, &fi, true), + get_codec_streaming_reader_gate(&None, &compressed, &compressed_fi, true).decision, GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Compressed) ); let small_fi = codec_streaming_test_fileinfo(0, 1); let small_object_info = codec_streaming_test_object_info(&small_fi); assert_eq!( - get_codec_streaming_reader_decision(&None, &small_object_info, &small_fi, true), + get_codec_streaming_reader_gate(&None, &small_object_info, &small_fi, true).decision, GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize) ); @@ -3314,7 +3319,7 @@ mod tests { remote_fi.transition_status = crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string(); let remote = codec_streaming_test_object_info(&remote_fi); assert_eq!( - get_codec_streaming_reader_decision(&None, &remote, &remote_fi, true), + get_codec_streaming_reader_gate(&None, &remote, &remote_fi, true).decision, GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Remote) ); }, @@ -3341,6 +3346,9 @@ mod tests { temp_env::with_vars( [ (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("false")), + (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, None::<&str>), + (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, None::<&str>), + (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, None::<&str>), (ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)), (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), ], @@ -3349,7 +3357,7 @@ mod tests { let object_info = codec_streaming_test_object_info(&fi); assert_eq!( - get_codec_streaming_reader_decision(&None, &object_info, &fi, true), + get_codec_streaming_reader_gate(&None, &object_info, &fi, true).decision, GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Disabled) ); }, @@ -3386,6 +3394,15 @@ mod tests { #[test] fn codec_streaming_fallback_metric_labels_are_stable() { assert_eq!(GetCodecStreamingFallbackReason::Disabled.as_str(), "disabled"); + assert_eq!(GetCodecStreamingFallbackReason::RolloutNotOptedIn.as_str(), "rollout_not_opted_in"); + assert_eq!( + GetCodecStreamingFallbackReason::BodyCompatibilityUnconfirmed.as_str(), + "body_compatibility_unconfirmed" + ); + assert_eq!( + GetCodecStreamingFallbackReason::HeaderCompatibilityUnconfirmed.as_str(), + "header_compatibility_unconfirmed" + ); assert_eq!( GetCodecStreamingFallbackReason::LockOptimizationDisabled.as_str(), "lock_optimization_disabled" @@ -3397,6 +3414,13 @@ mod tests { assert_eq!(GetCodecStreamingFallbackReason::Remote.as_str(), "remote"); assert_eq!(GetCodecStreamingFallbackReason::Multipart.as_str(), "multipart"); assert_eq!(GetCodecStreamingFallbackReason::InvalidMinSize.as_str(), "invalid_min_size"); + assert_eq!(GetCodecStreamingFallbackReason::ReadQuorumNotSafe.as_str(), "read_quorum_not_safe"); + assert_eq!(GetCodecStreamingObjectClass::PlainSinglePart.as_str(), "plain_single_part"); + assert_eq!(GetCodecStreamingObjectClass::Range.as_str(), "range"); + assert_eq!(GetCodecStreamingObjectClass::Encrypted.as_str(), "encrypted"); + assert_eq!(GetCodecStreamingObjectClass::Compressed.as_str(), "compressed"); + assert_eq!(GetCodecStreamingObjectClass::Remote.as_str(), "remote"); + assert_eq!(GetCodecStreamingObjectClass::Multipart.as_str(), "multipart"); } #[test] @@ -3404,6 +3428,9 @@ mod tests { temp_env::with_vars( [ (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, None::<&str>), + (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, None::<&str>), + (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, None::<&str>), + (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, None::<&str>), (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), ], || { @@ -3411,13 +3438,121 @@ mod tests { let object_info = codec_streaming_test_object_info(&fi); assert_eq!( - get_codec_streaming_reader_decision(&None, &object_info, &fi, true), + get_codec_streaming_reader_gate(&None, &object_info, &fi, true).decision, GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Disabled) ); }, ); } + #[test] + fn codec_streaming_reader_gate_requires_explicit_rollout_and_compat_confirmation() { + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("off")), + (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, None::<&str>), + (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, None::<&str>), + ], + || { + let fi = codec_streaming_test_fileinfo(1024, 1); + let object_info = codec_streaming_test_object_info(&fi); + + assert_eq!( + get_codec_streaming_reader_gate(&None, &object_info, &fi, true).decision, + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::RolloutNotOptedIn) + ); + }, + ); + + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")), + (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("false")), + (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")), + ], + || { + let fi = codec_streaming_test_fileinfo(1024, 1); + let object_info = codec_streaming_test_object_info(&fi); + + assert_eq!( + get_codec_streaming_reader_gate(&None, &object_info, &fi, true).decision, + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BodyCompatibilityUnconfirmed) + ); + }, + ); + + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")), + (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("false")), + ], + || { + let fi = codec_streaming_test_fileinfo(1024, 1); + let object_info = codec_streaming_test_object_info(&fi); + + assert_eq!( + get_codec_streaming_reader_gate(&None, &object_info, &fi, true).decision, + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::HeaderCompatibilityUnconfirmed) + ); + }, + ); + } + + #[test] + fn codec_streaming_reader_gate_records_object_classes() { + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")), + (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + ], + || { + let fi = codec_streaming_test_fileinfo(1024, 1); + let object_info = codec_streaming_test_object_info(&fi); + assert_eq!( + get_codec_streaming_reader_gate(&None, &object_info, &fi, true).object_class, + GetCodecStreamingObjectClass::PlainSinglePart + ); + + let range = Some(HTTPRangeSpec { + is_suffix_length: false, + start: 0, + end: 1, + }); + assert_eq!( + get_codec_streaming_reader_gate(&range, &object_info, &fi, true).object_class, + GetCodecStreamingObjectClass::Range + ); + }, + ); + } + + #[tokio::test] + async fn codec_streaming_reader_build_falls_back_when_read_quorum_is_not_safe() { + let setup = setup_inline_bitrot_readers( + vec![None, Some(b"bbbb"), Some(b"cccc"), Some(b"dddd")], + 2, + 2, + BitrotReaderSetupMode::VerifyReconstruction, + ) + .await; + + assert_eq!(setup.completed_failed_shards(), 1); + assert_eq!( + codec_streaming_reader_setup_fallback_reason(setup.completed_failed_shards()), + Some(GetCodecStreamingFallbackReason::ReadQuorumNotSafe) + ); + } + #[tokio::test] async fn collect_read_multiple_results_fails_early_when_quorum_is_impossible() { let started = std::time::Instant::now(); diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 74f94d077..e9a7ee1e6 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -597,6 +597,21 @@ pub fn record_get_object_codec_streaming_fallback(reason: &'static str) { counter!("rustfs_io_get_object_codec_streaming_fallback_total", "reason" => reason).increment(1); } +/// Record the final codec-streaming rollout decision for a GET request. +#[inline(always)] +pub fn record_get_object_codec_streaming_decision(outcome: &'static str, object_class: &'static str, reason: &'static str) { + if !get_stage_metrics_enabled() { + return; + } + counter!( + "rustfs_io_get_object_codec_streaming_decision_total", + "outcome" => outcome, + "object_class" => object_class, + "reason" => reason + ) + .increment(1); +} + /// Record one decoded reader stripe processed by a GetObject read path. #[inline(always)] pub fn record_get_object_reader_stripe(path: &'static str) { @@ -1568,6 +1583,8 @@ mod tests { record_get_object_stage_duration("s3_handler", "request_context", 0.001); record_get_object_reader_path("codec_streaming"); record_get_object_codec_streaming_fallback("range"); + record_get_object_codec_streaming_decision("fallback", "range", "range"); + record_get_object_codec_streaming_decision("use", "plain_single_part", "none"); record_get_object_reader_stripe("codec_streaming"); record_get_object_reader_bytes("codec_streaming", 1024); record_get_object_reader_buffer("codec_streaming", "output", 1024); diff --git a/scripts/run_get_codec_streaming_smoke.sh b/scripts/run_get_codec_streaming_smoke.sh index 15d65476f..43088a24c 100755 --- a/scripts/run_get_codec_streaming_smoke.sh +++ b/scripts/run_get_codec_streaming_smoke.sh @@ -328,6 +328,9 @@ codec_engines=${CODEC_ENGINES} metadata_early_stop=${METADATA_EARLY_STOP} shard_locality_preference=${SHARD_LOCALITY_PREFERENCE} codec_max_inflight=${CODEC_MAX_INFLIGHT} +codec_rollout_codec_profile=benchmark +codec_body_compat_confirmed_codec_profile=true +codec_header_compat_confirmed_codec_profile=true output_handoff_attribution=${OUTPUT_HANDOFF_ATTRIBUTION} address=${ADDRESS} bucket=${BUCKET} @@ -377,6 +380,33 @@ profile_codec_engine() { esac } +profile_rollout_target() { + local profile="$1" + case "$profile" in + legacy) echo "off" ;; + codec-legacy|codec-rustfs) echo "benchmark" ;; + *) die "unknown profile: $profile" ;; + esac +} + +profile_body_compat_confirmed() { + local profile="$1" + case "$profile" in + legacy) echo "false" ;; + codec-legacy|codec-rustfs) echo "true" ;; + *) die "unknown profile: $profile" ;; + esac +} + +profile_header_compat_confirmed() { + local profile="$1" + case "$profile" in + legacy) echo "false" ;; + codec-legacy|codec-rustfs) echo "true" ;; + *) die "unknown profile: $profile" ;; + esac +} + profile_metrics_path() { local profile="$1" case "$profile" in @@ -408,7 +438,11 @@ write_manifest() { local volumes="$4" local codec_engine="$5" local metrics_path="$6" + local rollout_target body_compat_confirmed header_compat_confirmed local git_head git_dirty_count + rollout_target="$(profile_rollout_target "$profile")" + body_compat_confirmed="$(profile_body_compat_confirmed "$profile")" + header_compat_confirmed="$(profile_header_compat_confirmed "$profile")" git_head="$(git -C "$PROJECT_ROOT" rev-parse HEAD)" git_dirty_count="$(git -C "$PROJECT_ROOT" status --porcelain | awk 'END { print NR + 0 }')" @@ -436,6 +470,9 @@ 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_ROLLOUT=${rollout_target} +RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED=${body_compat_confirmed} +RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED=${header_compat_confirmed} RUSTFS_GET_METADATA_EARLY_STOP_ENABLE=$(bool_from_on_off "$METADATA_EARLY_STOP") RUSTFS_GET_SHARD_LOCALITY_PREFERENCE_ENABLE=$(bool_from_on_off "$SHARD_LOCALITY_PREFERENCE") RUSTFS_GET_CODEC_STREAMING_MAX_INFLIGHT=${CODEC_MAX_INFLIGHT} @@ -489,6 +526,9 @@ start_server() { local codec_enabled local codec_engine local metrics_path + local rollout_target + local body_compat_confirmed + local header_compat_confirmed local rustfs_log data_root="$(profile_data_root "$profile")" @@ -496,6 +536,9 @@ start_server() { codec_enabled="$(profile_codec_enabled "$profile")" codec_engine="$(profile_codec_engine "$profile")" metrics_path="$(profile_metrics_path "$profile")" + rollout_target="$(profile_rollout_target "$profile")" + body_compat_confirmed="$(profile_body_compat_confirmed "$profile")" + header_compat_confirmed="$(profile_header_compat_confirmed "$profile")" rustfs_log="${profile_dir}/rustfs.log" mkdir -p "${data_root}/disk1" "${data_root}/disk2" "${data_root}/disk3" "${data_root}/disk4" @@ -517,6 +560,9 @@ start_server() { 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_ROLLOUT="$rollout_target" + export RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED="$body_compat_confirmed" + export RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED="$header_compat_confirmed" export RUSTFS_GET_METADATA_EARLY_STOP_ENABLE="$(bool_from_on_off "$METADATA_EARLY_STOP")" export RUSTFS_GET_SHARD_LOCALITY_PREFERENCE_ENABLE="$(bool_from_on_off "$SHARD_LOCALITY_PREFERENCE")" export RUSTFS_GET_CODEC_STREAMING_MAX_INFLIGHT="$CODEC_MAX_INFLIGHT"