gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-21 03:46:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into cxymds/fix-rebalance-multipart-retry

Browse Source 
# Conflicts:
#	crates/ecstore/src/set_disk/ops/object.rs
#	crates/ecstore/src/set_disk/read.rs
This commit is contained in:
马登山
2026-08-13 08:13:26 +08:00
parent 6cecf7d26f 65091aa6a8
commit fd727a8053
161 changed files with 8100 additions and 5463 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/set_disk/read.rs
+127 -69
View File
@@ -30,12 +30,12 @@ use crate::diagnostics::get::{
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_DIRECT_MEMORY,
GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE, GET_STAGE_METADATA_CACHE_LOOKUP,
GET_STAGE_METADATA_RESOLVE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP, GET_STAGE_READER_SETUP_DROP_PENDING,
GET_STAGE_READER_SETUP_SCHEDULE, GET_STAGE_READER_SETUP_WAIT_QUORUM, GET_STAGE_READER_TASK_BITROT_READER_INIT,
GET_STAGE_READER_TASK_FILE_OPEN, GET_STAGE_READER_TASK_READER_CONSTRUCTION, GetObjectFailureReason, classify_disk_error,
get_stage_timer_if_enabled, mark_get_object_downstream_closed, record_get_object_pipeline_failure,
record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled,
GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE,
GET_STAGE_METADATA_CACHE_LOOKUP, GET_STAGE_METADATA_RESOLVE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP,
GET_STAGE_READER_SETUP_DROP_PENDING, GET_STAGE_READER_SETUP_SCHEDULE, GET_STAGE_READER_SETUP_WAIT_QUORUM,
GET_STAGE_READER_TASK_BITROT_READER_INIT, GET_STAGE_READER_TASK_FILE_OPEN, GET_STAGE_READER_TASK_READER_CONSTRUCTION,
GetObjectFailureReason, classify_disk_error, get_stage_timer_if_enabled, mark_get_object_downstream_closed,
record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled,
};
use crate::erasure::coding::BitrotReader;
use crate::io_support::bitrot::{
@@ -116,9 +116,9 @@ impl SetDisks {
.then_some(GET_METADATA_CACHE_REASON_DIST_ERASURE)
}
async fn cached_get_object_fileinfo(&self, bucket: &str, object: &str) -> Option<GetObjectMetadataCacheEntry> {
async fn cached_get_object_fileinfo(&self, bucket: &str, object: &str) -> Option<Arc<GetObjectMetadataCacheEntry>> {
match self.lookup_cached_get_object_fileinfo(bucket, object).await {
MetadataCacheLookup::Hit(entry) => Some((*entry).clone()),
MetadataCacheLookup::Hit(entry) => Some(entry),
MetadataCacheLookup::Miss | MetadataCacheLookup::RejectedInsufficientQuorum => None,
}
}
@@ -180,9 +180,9 @@ impl SetDisks {
let key = GetObjectMetadataCacheKey::new(bucket, object, generation);
let entry = Arc::new(GetObjectMetadataCacheEntry {
created_at: Instant::now(),
fi: fi.clone(),
parts_metadata: parts_metadata.to_vec(),
online_disks: online_disks.to_vec(),
fi: Arc::new(fi.clone()),
parts_metadata: Arc::new(parts_metadata.to_vec()),
online_disks: Arc::new(online_disks.to_vec()),
read_quorum,
});
self.insert_get_object_metadata_cache_entry_after_insert(key, generation, entry, || {})
@@ -257,7 +257,7 @@ impl SetDisks {
opts: &ObjectOptions,
read_data: bool,
caller_allows_early_stop: bool,
) -> Result<(FileInfo, Vec<FileInfo>, Vec<Option<DiskStore>>)> {
) -> Result<GetObjectFileInfo> {
self.get_object_fileinfo_gated(bucket, object, opts, read_data, caller_allows_early_stop)
.await
}
@@ -274,7 +274,7 @@ impl SetDisks {
opts: &ObjectOptions,
read_data: bool,
allow_early_stop: bool,
) -> Result<(FileInfo, Vec<FileInfo>, Vec<Option<DiskStore>>)> {
) -> Result<GetObjectFileInfo> {
let vid = opts.version_id.clone().unwrap_or_default();
let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
@@ -300,7 +300,11 @@ impl SetDisks {
GET_STAGE_METADATA_CACHE_LOOKUP,
metadata_cache_lookup_start,
);
return Ok((cached.fi.clone(), cached.parts_metadata.clone(), cached.online_disks.clone()));
return Ok((
GetObjectMetadata::Shared(Arc::clone(&cached.fi)),
GetObjectMetadata::Shared(Arc::clone(&cached.parts_metadata)),
GetObjectMetadata::Shared(Arc::clone(&cached.online_disks)),
));
}
MetadataCacheLookup::Miss => {
rustfs_io_metrics::record_get_object_metadata_cache_decision(
@@ -349,7 +353,12 @@ impl SetDisks {
self.default_parity_count,
)
.await?;
metadata_fanout_diagnostics.record(GET_OBJECT_PATH_LEGACY_DUPLEX);
let metadata_metrics_path = if crate::bucket::utils::is_meta_bucketname(bucket) {
GET_OBJECT_PATH_INTERNAL_META
} else {
GET_OBJECT_PATH_LEGACY_DUPLEX
};
metadata_fanout_diagnostics.record(metadata_metrics_path);
let metadata_fanout_complete = metadata_fanout_diagnostics.total_responses() >= disks.len();
// warn!("get_object_fileinfo parts_metadata {:?}", &parts_metadata);
// warn!("get_object_fileinfo {}/{} errs {:?}", bucket, object, &errs);
@@ -396,7 +405,7 @@ impl SetDisks {
rustfs_utils::http::remove_str(&mut metadata.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS);
}
}
metadata_fanout_diagnostics.record_quorum_candidate_latency(GET_OBJECT_PATH_LEGACY_DUPLEX, fileinfo_selection_quorum);
metadata_fanout_diagnostics.record_quorum_candidate_latency(metadata_metrics_path, fileinfo_selection_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(
@@ -427,7 +436,11 @@ impl SetDisks {
// let online_disks: Vec<Option<DiskStore>> = op_online_disks.iter().filter(|v| v.is_some()).cloned().collect();
Ok((fi, parts_metadata, op_online_disks))
Ok((
GetObjectMetadata::Owned(fi),
GetObjectMetadata::Owned(parts_metadata),
GetObjectMetadata::Owned(op_online_disks),
))
}
#[hotpath::measure(impl_type = "SetDisks")]
@@ -2696,6 +2709,39 @@ mod metadata_cache_tests {
assert_eq!(cached.read_quorum, 0);
}
#[tokio::test]
async fn get_object_fileinfo_cache_hit_shares_cached_metadata() {
let set = new_metadata_cache_test_set().await;
let fi = valid_test_fileinfo("object");
let parts_metadata = vec![fi.clone()];
let online_disks = Vec::new();
let generation = set.get_object_metadata_cache_generation("bucket", "object");
set.cache_get_object_fileinfo(("bucket", "object"), generation, &fi, &parts_metadata, &online_disks, 0)
.await;
let cached = set
.cached_get_object_fileinfo("bucket", "object")
.await
.expect("fresh cache entry should be returned");
let (returned_fi, returned_parts_metadata, returned_online_disks) = set
.get_object_fileinfo("bucket", "object", &ObjectOptions::default(), true, false)
.await
.expect("cache-backed metadata lookup should succeed");
assert!(
matches!(returned_fi, GetObjectMetadata::Shared(ref value) if Arc::ptr_eq(value, &cached.fi)),
"cache hits must share FileInfo ownership"
);
assert!(
matches!(returned_parts_metadata, GetObjectMetadata::Shared(ref value) if Arc::ptr_eq(value, &cached.parts_metadata)),
"cache hits must share the metadata vector"
);
assert!(
matches!(returned_online_disks, GetObjectMetadata::Shared(ref value) if Arc::ptr_eq(value, &cached.online_disks)),
"cache hits must share the online-disk vector"
);
}
#[tokio::test]
async fn get_object_metadata_cache_rejects_deleted_and_invalid_fileinfo() {
let set = new_metadata_cache_test_set().await;
@@ -2735,9 +2781,9 @@ mod metadata_cache_tests {
),
Arc::new(GetObjectMetadataCacheEntry {
created_at: Instant::now(),
fi: fi.clone(),
parts_metadata: vec![fi],
online_disks: vec![None],
fi: Arc::new(fi.clone()),
parts_metadata: Arc::new(vec![fi]),
online_disks: Arc::new(vec![None]),
read_quorum: 1,
}),
)
@@ -2751,9 +2797,6 @@ mod metadata_cache_tests {
#[tokio::test]
async fn get_object_metadata_cache_rejects_stale_entries() {
// moka handles TTL expiry automatically via time_to_live(250ms).
// This test verifies that entries inserted with the cache API are retrievable
// while fresh, and that the cache API works correctly.
let set = new_metadata_cache_test_set().await;
let fi = valid_test_fileinfo("object");
@@ -2765,6 +2808,14 @@ mod metadata_cache_tests {
set.cached_get_object_fileinfo("bucket", "object").await.is_some(),
"freshly inserted entry should be returned"
);
tokio::time::timeout(GET_OBJECT_METADATA_CACHE_TTL + Duration::from_secs(1), async {
while set.cached_get_object_fileinfo("bucket", "object").await.is_some() {
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("metadata cache entry should expire after its TTL");
}
#[tokio::test]
@@ -2826,9 +2877,13 @@ mod metadata_cache_tests {
barrier.wait_until_paused().await;
set.invalidate_get_object_metadata_cache(bucket, object).await;
barrier.release();
read.await
let (fi, parts_metadata, online_disks) = read
.await
.expect("metadata read task should not panic")
.expect("metadata fanout should still return its selected FileInfo");
assert!(matches!(fi, GetObjectMetadata::Owned(_)));
assert!(matches!(parts_metadata, GetObjectMetadata::Owned(_)));
assert!(matches!(online_disks, GetObjectMetadata::Owned(_)));
assert!(
set.get_object_metadata_cache
@@ -2875,9 +2930,9 @@ mod metadata_cache_tests {
let key = GetObjectMetadataCacheKey::new("bucket", "object", generation);
let entry = Arc::new(GetObjectMetadataCacheEntry {
created_at: Instant::now(),
fi: fi.clone(),
parts_metadata: vec![fi],
online_disks: Vec::new(),
fi: Arc::new(fi.clone()),
parts_metadata: Arc::new(vec![fi]),
online_disks: Arc::new(Vec::new()),
read_quorum: 0,
});
@@ -2985,9 +3040,9 @@ mod metadata_cache_tests {
let entry = |fi: FileInfo| {
Arc::new(GetObjectMetadataCacheEntry {
created_at: Instant::now(),
parts_metadata: vec![fi.clone()],
fi,
online_disks: Vec::new(),
parts_metadata: Arc::new(vec![fi.clone()]),
fi: Arc::new(fi),
online_disks: Arc::new(Vec::new()),
read_quorum: 0,
})
};
@@ -3437,7 +3492,7 @@ mod tests {
);
assert_eq!(diagnostics.total_responses(), 9);
assert_eq!(diagnostics.valid_responses(), 1);
assert_eq!(diagnostics.error_responses(), 8);
assert_eq!(diagnostics.non_valid_responses(), 8);
}
#[test]
@@ -3452,7 +3507,7 @@ mod tests {
);
assert_eq!(diagnostics.ignored_responses(), 2);
assert_eq!(diagnostics.error_responses(), 3);
assert_eq!(diagnostics.non_valid_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);
@@ -3520,7 +3575,7 @@ mod tests {
assert_eq!(diagnostics.total_responses(), 3);
assert_eq!(diagnostics.valid_responses(), 3);
assert_eq!(diagnostics.error_responses(), 0);
assert_eq!(diagnostics.non_valid_responses(), 0);
assert!(
diagnostics
.observations
@@ -5496,33 +5551,36 @@ mod tests {
}
#[test]
fn rustfs_codec_streaming_uses_conservative_default_min_size() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)),
(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, None::<&str>),
(ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE, None::<&str>),
],
|| {
let below_threshold_fi = codec_streaming_test_fileinfo(512 * 1024, 1);
let below_threshold_object_info = codec_streaming_test_object_info(&below_threshold_fi);
assert_eq!(
codec_streaming_reader_gate_for_test(&None, &below_threshold_object_info, &below_threshold_fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize)
);
fn codec_streaming_default_min_size_meets_direct_memory_ceiling() {
for engine in [None, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)] {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, engine),
(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, None::<&str>),
(ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE, None::<&str>),
],
|| {
let below_threshold_fi = codec_streaming_test_fileinfo(128 * 1024 - 1, 1);
let below_threshold_object_info = codec_streaming_test_object_info(&below_threshold_fi);
assert_eq!(
codec_streaming_reader_gate_for_test(&None, &below_threshold_object_info, &below_threshold_fi, true)
.decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize)
);
let threshold_fi = codec_streaming_test_fileinfo(1_048_576, 1);
let threshold_object_info = codec_streaming_test_object_info(&threshold_fi);
assert_eq!(
codec_streaming_reader_gate_for_test(&None, &threshold_object_info, &threshold_fi, true).decision,
GetCodecStreamingDecision::Use
);
},
);
let threshold_fi = codec_streaming_test_fileinfo(128 * 1024, 1);
let threshold_object_info = codec_streaming_test_object_info(&threshold_fi);
assert_eq!(
codec_streaming_reader_gate_for_test(&None, &threshold_object_info, &threshold_fi, true).decision,
GetCodecStreamingDecision::Use
);
},
);
}
}
#[test]
@@ -5820,10 +5878,10 @@ mod tests {
(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")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>),
],
|| {
let fi = codec_streaming_test_fileinfo(1024, 1);
let fi = codec_streaming_test_fileinfo(128 * 1024, 1);
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
@@ -5844,10 +5902,10 @@ mod tests {
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("on")),
(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")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>),
],
|| {
let fi = codec_streaming_test_fileinfo(1024, 1);
let fi = codec_streaming_test_fileinfo(128 * 1024, 1);
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
@@ -5865,10 +5923,10 @@ mod tests {
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("false")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("on")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>),
],
|| {
let fi = codec_streaming_test_fileinfo(1024, 1);
let fi = codec_streaming_test_fileinfo(128 * 1024, 1);
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
@@ -5948,10 +6006,10 @@ mod tests {
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, Some("0")),
(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")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>),
],
|| {
let fi = codec_streaming_test_fileinfo(1024, 1);
let fi = codec_streaming_test_fileinfo(128 * 1024, 1);
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
@@ -5968,10 +6026,10 @@ mod tests {
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, Some("100")),
(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")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>),
],
|| {
let fi = codec_streaming_test_fileinfo(1024, 1);
let fi = codec_streaming_test_fileinfo(128 * 1024, 1);
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(