From a8e7cce5e13ef5f5a937c8d9fe19212be4bea1a3 Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 22 Jul 2026 08:15:41 +0800 Subject: [PATCH] feat: expose list read-dir amplification metrics (#5103) Record local read_dir entry counts and duration for live-walker ListObjects scans so wide root/prefix amplification can be measured below the cross-set merge layer. Add a focused LocalDisk scan_dir test showing a page limit of one still observes the whole parent directory enumeration, plus metric helper coverage. Co-authored-by: heihutu --- crates/ecstore/src/disk/local.rs | 104 +++++++++++++++++- crates/ecstore/src/test_metrics.rs | 15 +++ crates/io-metrics/src/lib.rs | 10 +- crates/io-metrics/src/list_objects_metrics.rs | 99 +++++++++++++++++ 4 files changed, 223 insertions(+), 5 deletions(-) diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 1ca209ee5..3b7064a49 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -5172,7 +5172,25 @@ impl LocalDisk { // larger `RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS` or the high-latency // drive-timeout profile; a streaming readdir rewrite is a separate, // higher-risk follow-up and is intentionally not done here. - let mut entries = match with_walk_stall_timeout(stall, self.list_dir("", &opts.bucket, ¤t, -1)).await { + let read_dir_started = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now); + let read_dir_result = with_walk_stall_timeout(stall, self.list_dir("", &opts.bucket, ¤t, -1)).await; + if let Some(started) = read_dir_started { + rustfs_io_metrics::record_list_objects_local_read_dir(rustfs_io_metrics::ListObjectsLocalReadDirObservation { + outcome: if read_dir_result.is_ok() { + rustfs_io_metrics::LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_OK + } else { + rustfs_io_metrics::LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_ERROR + }, + requested_count: -1, + returned_entries: read_dir_result.as_ref().map_or(0, Vec::len), + duration_ms: started.elapsed().as_secs_f64() * 1000.0, + is_root: current.trim_matches('/').is_empty(), + has_filter_prefix: !prefix.is_empty(), + has_forward: forward.is_some(), + }); + } + + let mut entries = match read_dir_result { Ok(res) => res, Err(e) => { if e != DiskError::VolumeNotFound && e != Error::FileNotFound { @@ -11963,6 +11981,90 @@ mod test { assert!(names.contains(&"quux/thud".to_string())); } + #[test] + #[serial_test::serial] + fn scan_dir_records_whole_parent_read_dir_before_page_limit() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime should be created"); + let recorder = crate::test_metrics::CapturingRecorder::default(); + let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled(); + rustfs_io_metrics::set_get_stage_metrics_enabled(true); + + metrics::with_local_recorder(&recorder, || { + runtime.block_on(async { + use rustfs_filemeta::MetacacheReader; + use tempfile::tempdir; + + let dir = tempdir().expect("tempdir should be created"); + let bucket = "test-bucket"; + let bucket_dir = dir.path().join(bucket); + const OBJECTS: usize = 12; + + for index in 0..OBJECTS { + let object_dir = bucket_dir.join(format!("object-{index:04}")); + fs::create_dir_all(&object_dir) + .await + .expect("object directory should be created"); + fs::write(object_dir.join(STORAGE_FORMAT_FILE), b"meta") + .await + .expect("object metadata should be written"); + } + + let endpoint = + Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + + let (reader, mut writer) = tokio::io::duplex(4096); + let mut out = MetacacheWriter::new(&mut writer); + let opts = WalkDirOptions { + bucket: bucket.to_string(), + base_dir: String::new(), + recursive: true, + limit: 1, + ..Default::default() + }; + let mut objs_returned = 0; + + disk.scan_dir(String::new(), String::new(), &opts, &mut out, &mut objs_returned, false, None) + .await + .expect("scan_dir should succeed"); + out.close().await.expect("metacache writer should close"); + drop(out); + drop(writer); + + let mut reader = MetacacheReader::new(reader); + let visible_objects = reader + .read_all() + .await + .expect("scan output should decode") + .into_iter() + .filter(|entry| !entry.metadata.is_empty()) + .count(); + + assert_eq!(visible_objects, 1); + assert_eq!(objs_returned, 1); + }); + }); + rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate); + + assert_eq!( + recorder.counter_value( + "rustfs_s3_list_objects_local_read_dir_total", + &[("outcome", "ok"), ("count_mode", "whole"), ("is_root", "true")] + ), + 1 + ); + assert_eq!( + recorder.histogram_values( + "rustfs_s3_list_objects_local_read_dir_entries", + &[("outcome", "ok"), ("count_mode", "whole")] + ), + vec![12.0] + ); + } + #[tokio::test] async fn test_scan_dir_reports_base_dir_object_metadata() { use rustfs_filemeta::MetacacheReader; diff --git a/crates/ecstore/src/test_metrics.rs b/crates/ecstore/src/test_metrics.rs index a9d20d7d4..94205ccf0 100644 --- a/crates/ecstore/src/test_metrics.rs +++ b/crates/ecstore/src/test_metrics.rs @@ -86,4 +86,19 @@ impl CapturingRecorder { .map(|(_, samples)| samples.0.lock().expect("samples should be lockable").len()) .sum() } + + pub(crate) fn histogram_values(&self, name: &str, labels: &[(&str, &str)]) -> Vec { + self.histograms + .lock() + .expect("histogram map should be lockable") + .iter() + .filter(|(key, _)| { + key.name() == name + && labels + .iter() + .all(|(lk, lv)| key.labels().any(|label| label.key() == *lk && label.value() == *lv)) + }) + .flat_map(|(_, samples)| samples.0.lock().expect("samples should be lockable").clone()) + .collect() + } } diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 161ac051c..eae6138dc 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -210,10 +210,12 @@ pub use io_metrics::{ record_io_scheduler_decision, record_load_level_change, record_queue_operation, record_starvation_event, }; pub use list_objects_metrics::{ - LIST_OBJECTS_GATHER_OUTCOME_INPUT_CLOSED, LIST_OBJECTS_GATHER_OUTCOME_LIMIT_REACHED, LIST_OBJECTS_SOURCE_WALKER, - ListObjectsGatherObservation, ListObjectsIndexPageObservation, init_list_objects_metrics, record_list_objects_gather, - record_list_objects_index_attempt, record_list_objects_index_fallback, record_list_objects_index_live_verify_failure, - record_list_objects_index_served, record_list_objects_merge, + LIST_OBJECTS_GATHER_OUTCOME_INPUT_CLOSED, LIST_OBJECTS_GATHER_OUTCOME_LIMIT_REACHED, + LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_ERROR, LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_OK, LIST_OBJECTS_SOURCE_WALKER, + ListObjectsGatherObservation, ListObjectsIndexPageObservation, ListObjectsLocalReadDirObservation, init_list_objects_metrics, + record_list_objects_gather, record_list_objects_index_attempt, record_list_objects_index_fallback, + record_list_objects_index_live_verify_failure, record_list_objects_index_served, record_list_objects_local_read_dir, + record_list_objects_merge, }; // Backpressure metrics exports diff --git a/crates/io-metrics/src/list_objects_metrics.rs b/crates/io-metrics/src/list_objects_metrics.rs index ce90fd017..c5283152e 100644 --- a/crates/io-metrics/src/list_objects_metrics.rs +++ b/crates/io-metrics/src/list_objects_metrics.rs @@ -42,6 +42,13 @@ const LIST_OBJECTS_INDEX_LIVE_VERIFY_FAILURE_TOTAL: &str = "rustfs_s3_list_objec const LIST_OBJECTS_INDEX_RETURNED_OBJECTS: &str = "rustfs_s3_list_objects_index_returned_objects"; const LIST_OBJECTS_INDEX_RETURNED_PREFIXES: &str = "rustfs_s3_list_objects_index_returned_prefixes"; const LIST_OBJECTS_INDEX_VERIFICATION_IO_AMPLIFICATION: &str = "rustfs_s3_list_objects_index_verification_io_amplification"; +const LIST_OBJECTS_LOCAL_READ_DIR_TOTAL: &str = "rustfs_s3_list_objects_local_read_dir_total"; +const LIST_OBJECTS_LOCAL_READ_DIR_DURATION_MS: &str = "rustfs_s3_list_objects_local_read_dir_duration_ms"; +const LIST_OBJECTS_LOCAL_READ_DIR_ENTRIES: &str = "rustfs_s3_list_objects_local_read_dir_entries"; +const LIST_OBJECTS_LOCAL_READ_DIR_LIMIT: &str = "rustfs_s3_list_objects_local_read_dir_limit"; + +pub const LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_OK: &str = "ok"; +pub const LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_ERROR: &str = "error"; #[derive(Clone, Copy, Debug, PartialEq)] pub struct ListObjectsGatherObservation { @@ -69,6 +76,17 @@ pub struct ListObjectsIndexPageObservation { pub is_truncated: bool, } +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct ListObjectsLocalReadDirObservation { + pub outcome: &'static str, + pub requested_count: i32, + pub returned_entries: usize, + pub duration_ms: f64, + pub is_root: bool, + pub has_filter_prefix: bool, + pub has_forward: bool, +} + #[inline(always)] fn bool_label(value: bool) -> &'static str { if value { "true" } else { "false" } @@ -84,6 +102,16 @@ fn limit_as_f64(value: i32) -> f64 { value.max(0) as f64 } +#[inline(always)] +fn count_mode_label(count: i32) -> &'static str { + if count < 0 { "whole" } else { "bounded" } +} + +#[inline(always)] +fn read_dir_count_as_f64(value: i32) -> f64 { + f64::from(value) +} + pub fn init_list_objects_metrics() { static METRICS_DESC_INIT: OnceLock<()> = OnceLock::new(); METRICS_DESC_INIT.get_or_init(|| { @@ -155,6 +183,22 @@ pub fn init_list_objects_metrics() { LIST_OBJECTS_INDEX_VERIFICATION_IO_AMPLIFICATION, "Ratio of live verification attempts to returned objects for opt-in ListObjects index serving." ); + describe_counter!( + LIST_OBJECTS_LOCAL_READ_DIR_TOTAL, + "Total number of local read_dir calls made while serving live-walker ListObjects pages." + ); + describe_histogram!( + LIST_OBJECTS_LOCAL_READ_DIR_DURATION_MS, + "Duration in milliseconds of local read_dir calls made while serving live-walker ListObjects pages." + ); + describe_histogram!( + LIST_OBJECTS_LOCAL_READ_DIR_ENTRIES, + "Number of immediate directory entries returned by local read_dir while serving live-walker ListObjects pages." + ); + describe_histogram!( + LIST_OBJECTS_LOCAL_READ_DIR_LIMIT, + "Requested local read_dir count used while serving live-walker ListObjects pages; negative counts mean whole-directory enumeration." + ); }); } @@ -324,6 +368,38 @@ pub fn record_list_objects_index_served(observation: ListObjectsIndexPageObserva .record(verification_io_amplification); } +pub fn record_list_objects_local_read_dir(observation: ListObjectsLocalReadDirObservation) { + if !get_stage_metrics_enabled() { + return; + } + + let count_mode = count_mode_label(observation.requested_count); + + counter!( + LIST_OBJECTS_LOCAL_READ_DIR_TOTAL, + "outcome" => observation.outcome, + "count_mode" => count_mode, + "is_root" => bool_label(observation.is_root), + "has_filter_prefix" => bool_label(observation.has_filter_prefix), + "has_forward" => bool_label(observation.has_forward), + ) + .increment(1); + histogram!( + LIST_OBJECTS_LOCAL_READ_DIR_DURATION_MS, + "outcome" => observation.outcome, + "count_mode" => count_mode, + ) + .record(observation.duration_ms); + histogram!( + LIST_OBJECTS_LOCAL_READ_DIR_ENTRIES, + "outcome" => observation.outcome, + "count_mode" => count_mode, + ) + .record(count_as_f64(observation.returned_entries)); + histogram!(LIST_OBJECTS_LOCAL_READ_DIR_LIMIT, "count_mode" => count_mode) + .record(read_dir_count_as_f64(observation.requested_count)); +} + #[cfg(test)] mod tests { use super::*; @@ -373,4 +449,27 @@ mod tests { }); record_list_objects_index_live_verify_failure("index_key_only", "read_error"); } + + #[test] + fn record_local_read_dir_observation_accepts_whole_directory_counts() { + init_list_objects_metrics(); + record_list_objects_local_read_dir(ListObjectsLocalReadDirObservation { + outcome: LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_OK, + requested_count: -1, + returned_entries: 4096, + duration_ms: 12.5, + is_root: true, + has_filter_prefix: false, + has_forward: false, + }); + record_list_objects_local_read_dir(ListObjectsLocalReadDirObservation { + outcome: LIST_OBJECTS_LOCAL_READ_DIR_OUTCOME_ERROR, + requested_count: -1, + returned_entries: 0, + duration_ms: 5000.0, + is_root: true, + has_filter_prefix: false, + has_forward: true, + }); + } }