gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(get): add limited opt-in rollout gates (#3963)

Browse Source 
* feat(bench): harden cooled get ab harness

* feat(get): add limited opt-in rollout gates

* fix(get): tighten rollout gate fallbacks
This commit is contained in:
houseme
2026-06-27 23:13:02 +08:00
committed by GitHub
parent 66ad138505
commit bf03ff2869
5 changed files with 544 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/diagnostics/get.rs
+18
View File
@@ -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");
crates/ecstore/src/set_disk/mod.rs
+295 -38
View File
@@ -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<bool> = 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<bool> = 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<bool> = 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<bool> = 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<bool> = 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<u32> = 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<u32> = 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<CodecStreamingDecodeEngine> {
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<HTTPRangeSpec>,
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<HTTPRangeSpec>,
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());
}
}
crates/ecstore/src/set_disk/read.rs
+168 -33
View File
@@ -48,6 +48,15 @@ const READ_REPAIR_HEAL_DEDUP_MAX_ENTRIES: usize = 4096;
static READ_REPAIR_HEAL_CACHE: OnceLock<RwLock<HashMap<ReadRepairHealCacheKey, Instant>>> = OnceLock::new();
pub(super) enum GetCodecStreamingReaderBuildOutcome {
Reader(Box<dyn AsyncRead + Unpin + Send + Sync>),
Fallback(GetCodecStreamingFallbackReason),
}
pub(super) fn codec_streaming_reader_setup_fallback_reason(missing_shards: usize) -> Option<GetCodecStreamingFallbackReason> {
(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<FileInfo>, Vec<Option<DiskStore>>)> {
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<FileInfo>,
fi: &FileInfo,
files: &[FileInfo],
disks: &[Option<DiskStore>],
set_index: usize,
pool_index: usize,
_set_index: usize,
_pool_index: usize,
skip_verify_bitrot: bool,
) -> Result<Box<dyn AsyncRead + Unpin + Send + Sync>> {
let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, &files, &fi);
) -> Result<GetCodecStreamingReaderBuildOutcome> {
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();
crates/io-metrics/src/lib.rs
+17
View File
@@ -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);
scripts/run_get_codec_streaming_smoke.sh
+46
View File
@@ -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"