From 621fcb93c7d7a8adb28a1902737e67fb85d66210 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 20 Aug 2026 16:59:37 +0800 Subject: [PATCH] feat(ecstore): expose fdatasync group wait metrics (#6299) Add PUT-stage diagnostics for file fdatasync group commit wait time, per-group outstanding depth, and rename disk completion position. These metrics keep the existing default-off PUT stage gate and do not change group commit scheduling or quorum behavior. Co-authored-by: heihutu --- crates/ecstore/src/disk/os.rs | 48 ++++++- .../src/set_disk/core/io_primitives.rs | 33 ++++- crates/io-metrics/src/lib.rs | 126 ++++++++++++++++++ 3 files changed, 198 insertions(+), 9 deletions(-) diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index d32e8466c..da8487fc9 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -23,6 +23,7 @@ use std::{ io, path::{Component, Path, PathBuf}, sync::{Arc, LazyLock, Weak}, + time::Instant, }; use tokio::fs; use tokio::sync::{ @@ -766,6 +767,8 @@ type FileFdatasyncGroupKey = usize; struct FileFdatasyncWaiter { files: Vec, + enqueued_at: Option, + wait_role: &'static str, result_tx: oneshot::Sender, } @@ -799,6 +802,7 @@ struct FileFdatasyncGroup { #[derive(Default)] struct FileFdatasyncGroupInner { worker_running: bool, + pending_files: usize, pending: VecDeque, } @@ -865,11 +869,30 @@ impl FileFdatasyncGroupCommit { }; let file_count = files.len(); let mut group_state = group.inner.lock(); - group_state.pending.push_back(FileFdatasyncWaiter { files, result_tx }); let start_worker = !group_state.worker_running; + let wait_role = if start_worker { + rustfs_io_metrics::PUT_RENAME_FDATASYNC_GROUP_WAIT_ROLE_LEADER + } else { + rustfs_io_metrics::PUT_RENAME_FDATASYNC_GROUP_WAIT_ROLE_FOLLOWER + }; + group_state.pending.push_back(FileFdatasyncWaiter { + files, + enqueued_at: rustfs_io_metrics::put_stage_timer(), + wait_role, + result_tx, + }); + group_state.pending_files += file_count; if start_worker { group_state.worker_running = true; } + rustfs_io_metrics::record_put_rename_fdatasync_group_outstanding( + rustfs_io_metrics::PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_ENQUEUE_WAITERS, + group_state.pending.len(), + ); + rustfs_io_metrics::record_put_rename_fdatasync_group_outstanding( + rustfs_io_metrics::PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_ENQUEUE_FILES, + group_state.pending_files, + ); registry.total_waiters += 1; registry.total_files += file_count; drop(group_state); @@ -911,9 +934,11 @@ async fn run_file_fdatasync_group_worker(group: Arc) { #[cfg(test)] file_sync_probe::run_before_group_batch(); tokio::task::yield_now().await; - let batch: Vec = { + let (batch, batch_file_count): (Vec, usize) = { let mut group_state = group.inner.lock(); - group_state.pending.drain(..).collect() + let batch_file_count = group_state.pending_files; + group_state.pending_files = 0; + (group_state.pending.drain(..).collect(), batch_file_count) }; if batch.is_empty() { let mut group_state = group.inner.lock(); @@ -923,8 +948,23 @@ async fn run_file_fdatasync_group_worker(group: Arc) { return; } + rustfs_io_metrics::record_put_rename_fdatasync_group_outstanding( + rustfs_io_metrics::PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_BATCH_WAITERS, + batch.len(), + ); + rustfs_io_metrics::record_put_rename_fdatasync_group_outstanding( + rustfs_io_metrics::PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_BATCH_FILES, + batch_file_count, + ); + for waiter in &batch { + if let Some(enqueued_at) = waiter.enqueued_at { + rustfs_io_metrics::record_put_rename_fdatasync_group_wait( + waiter.wait_role, + enqueued_at.elapsed().as_secs_f64() * 1000.0, + ); + } + } let batch_files: Vec = batch.iter().flat_map(|waiter| waiter.files.iter().cloned()).collect(); - let batch_file_count = batch_files.len(); #[cfg(test)] file_sync_probe::record_group_batch(batch_file_count); rustfs_io_metrics::record_put_rename_fdatasync_batch( diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index e70d4f353..98219f1f2 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -74,7 +74,10 @@ use std::{ collections::{HashMap, HashSet, VecDeque}, future::Future, pin::Pin, - sync::OnceLock, + sync::{ + OnceLock, + atomic::{AtomicUsize, Ordering}, + }, task::{Context, Poll}, time::{Duration, Instant}, }; @@ -3368,6 +3371,8 @@ impl SetDisks { // preserving slot-indexed quorum and convergence accounting without a // scheduler task for every disk. let fanout = tokio::spawn(async move { + let successful_rename_completion_rank = + rustfs_io_metrics::put_stage_metrics_enabled().then(|| Arc::new(AtomicUsize::new(0))); let futures = fanout_disks .into_iter() .zip(fanout_file_infos.iter()) @@ -3377,6 +3382,7 @@ impl SetDisks { let src_object = fanout_src_object.clone(); let dst_object = fanout_dst_object.clone(); let dst_bucket = fanout_dst_bucket.clone(); + let successful_rename_completion_rank = successful_rename_completion_rank.clone(); std::panic::AssertUnwindSafe(async move { // Test-only introspection guard: counts this operation as @@ -3409,10 +3415,27 @@ impl SetDisks { let result = disk .rename_data_borrowed(&src_bucket, &src_object, file_info, &dst_bucket, &dst_object) .await; - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DISK_WAIT, - disk_wait_started, - ); + if let Some(disk_wait_started) = disk_wait_started { + let duration_ms = disk_wait_started.elapsed().as_secs_f64() * 1000.0; + rustfs_io_metrics::record_put_object_stage_duration( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DISK_WAIT, + duration_ms, + ); + let position = if result.is_ok() { + let rank = successful_rename_completion_rank + .as_ref() + .map(|rank| rank.fetch_add(1, Ordering::Relaxed) + 1) + .unwrap_or(1); + if rank <= write_quorum { + rustfs_io_metrics::PUT_RENAME_DISK_WAIT_COMPLETION_POSITION_QUORUM_FIRST + } else { + rustfs_io_metrics::PUT_RENAME_DISK_WAIT_COMPLETION_POSITION_QUORUM_TAIL + } + } else { + rustfs_io_metrics::PUT_RENAME_DISK_WAIT_COMPLETION_POSITION_ERROR + }; + rustfs_io_metrics::record_put_rename_disk_wait_completion(position, duration_ms); + } result }) .catch_unwind() diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 50a660383..3aa15c382 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -122,11 +122,20 @@ pub const PUT_STAGE_SET_DISK_RENAME_RENAME_SYSCALL: &str = "set_disk_rename_rena pub const PUT_RENAME_FDATASYNC_BATCH_MODE_SERIAL: &str = "serial"; pub const PUT_RENAME_FDATASYNC_BATCH_MODE_PARALLEL: &str = "parallel"; +pub const PUT_RENAME_FDATASYNC_GROUP_WAIT_ROLE_LEADER: &str = "leader"; +pub const PUT_RENAME_FDATASYNC_GROUP_WAIT_ROLE_FOLLOWER: &str = "follower"; +pub const PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_ENQUEUE_WAITERS: &str = "enqueue_waiters"; +pub const PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_ENQUEUE_FILES: &str = "enqueue_files"; +pub const PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_BATCH_WAITERS: &str = "batch_waiters"; +pub const PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_BATCH_FILES: &str = "batch_files"; pub const PUT_RENAME_QUORUM_FANOUT_STATE_SCHEDULED: &str = "scheduled"; pub const PUT_RENAME_QUORUM_FANOUT_STATE_WRITE_QUORUM: &str = "write_quorum"; pub const PUT_RENAME_QUORUM_FANOUT_STATE_SUCCESS: &str = "success"; pub const PUT_RENAME_QUORUM_FANOUT_STATE_ERROR: &str = "error"; pub const PUT_RENAME_QUORUM_FANOUT_STATE_PANIC: &str = "panic"; +pub const PUT_RENAME_DISK_WAIT_COMPLETION_POSITION_QUORUM_FIRST: &str = "quorum_first"; +pub const PUT_RENAME_DISK_WAIT_COMPLETION_POSITION_QUORUM_TAIL: &str = "quorum_tail"; +pub const PUT_RENAME_DISK_WAIT_COMPLETION_POSITION_ERROR: &str = "error"; #[inline(always)] pub fn get_stage_metrics_enabled() -> bool { @@ -2066,6 +2075,30 @@ pub fn record_put_rename_fdatasync_batch(mode: &'static str, files: usize) { histogram!("rustfs_s3_put_object_rename_fdatasync_batch_files", "mode" => mode).record(put_stage_count_value(files)); } +#[inline(always)] +pub fn record_put_rename_fdatasync_group_wait(role: &'static str, duration_ms: f64) { + if !put_stage_metrics_enabled() { + return; + } + histogram!("rustfs_s3_put_object_rename_fdatasync_group_wait_ms", "role" => role).record(duration_ms); +} + +#[inline(always)] +pub fn record_put_rename_fdatasync_group_outstanding(state: &'static str, count: usize) { + if !put_stage_metrics_enabled() { + return; + } + histogram!("rustfs_s3_put_object_rename_fdatasync_group_outstanding", "state" => state).record(put_stage_count_value(count)); +} + +#[inline(always)] +pub fn record_put_rename_disk_wait_completion(position: &'static str, duration_ms: f64) { + if !put_stage_metrics_enabled() { + return; + } + histogram!("rustfs_s3_put_object_rename_disk_wait_completion_ms", "position" => position).record(duration_ms); +} + #[inline(always)] pub fn record_put_rename_quorum_wait_fanout( scheduled: usize, @@ -3176,10 +3209,22 @@ mod tests { metrics::with_local_recorder(&recorder, || { set_put_stage_metrics_enabled(false); record_put_rename_fdatasync_batch(PUT_RENAME_FDATASYNC_BATCH_MODE_SERIAL, 2); + record_put_rename_fdatasync_group_wait(PUT_RENAME_FDATASYNC_GROUP_WAIT_ROLE_LEADER, 1.0); + record_put_rename_fdatasync_group_outstanding(PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_ENQUEUE_WAITERS, 2); + record_put_rename_disk_wait_completion(PUT_RENAME_DISK_WAIT_COMPLETION_POSITION_QUORUM_FIRST, 3.0); record_put_rename_quorum_wait_fanout(4, 3, 3, 1, 0); set_put_stage_metrics_enabled(true); record_put_rename_fdatasync_batch(PUT_RENAME_FDATASYNC_BATCH_MODE_PARALLEL, 9); + record_put_rename_fdatasync_group_wait(PUT_RENAME_FDATASYNC_GROUP_WAIT_ROLE_LEADER, 1.0); + record_put_rename_fdatasync_group_wait(PUT_RENAME_FDATASYNC_GROUP_WAIT_ROLE_FOLLOWER, 2.0); + record_put_rename_fdatasync_group_outstanding(PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_ENQUEUE_WAITERS, 2); + record_put_rename_fdatasync_group_outstanding(PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_ENQUEUE_FILES, 4); + record_put_rename_fdatasync_group_outstanding(PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_BATCH_WAITERS, 3); + record_put_rename_fdatasync_group_outstanding(PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_BATCH_FILES, 6); + record_put_rename_disk_wait_completion(PUT_RENAME_DISK_WAIT_COMPLETION_POSITION_QUORUM_FIRST, 3.0); + record_put_rename_disk_wait_completion(PUT_RENAME_DISK_WAIT_COMPLETION_POSITION_QUORUM_TAIL, 4.0); + record_put_rename_disk_wait_completion(PUT_RENAME_DISK_WAIT_COMPLETION_POSITION_ERROR, 5.0); record_put_rename_quorum_wait_fanout(4, 3, 3, 1, 0); set_put_stage_metrics_enabled(false); }); @@ -3205,6 +3250,87 @@ mod tests { let quorum_samples = histogram_samples(&rows, "rustfs_s3_put_object_rename_quorum_wait_fanout_disks"); assert_eq!(quorum_samples, vec![0.0, 1.0, 3.0, 3.0, 4.0]); + assert_eq!( + histogram_samples(&rows, "rustfs_s3_put_object_rename_fdatasync_group_wait_ms"), + vec![1.0, 2.0] + ); + assert_eq!( + histogram_samples(&rows, "rustfs_s3_put_object_rename_fdatasync_group_outstanding"), + vec![2.0, 3.0, 4.0, 6.0] + ); + assert_eq!( + histogram_samples(&rows, "rustfs_s3_put_object_rename_disk_wait_completion_ms"), + vec![3.0, 4.0, 5.0] + ); + let group_wait_roles = rows + .iter() + .filter(|(composite, _, _, _)| { + composite.kind() == MetricKind::Histogram + && composite.key().name() == "rustfs_s3_put_object_rename_fdatasync_group_wait_ms" + }) + .flat_map(|(composite, _, _, _)| { + composite + .key() + .labels() + .filter(|label| label.key() == "role") + .map(|label| label.value().to_string()) + .collect::>() + }) + .collect::>(); + assert_eq!( + group_wait_roles, + HashSet::from([ + PUT_RENAME_FDATASYNC_GROUP_WAIT_ROLE_LEADER.to_string(), + PUT_RENAME_FDATASYNC_GROUP_WAIT_ROLE_FOLLOWER.to_string(), + ]) + ); + let group_outstanding_states = rows + .iter() + .filter(|(composite, _, _, _)| { + composite.kind() == MetricKind::Histogram + && composite.key().name() == "rustfs_s3_put_object_rename_fdatasync_group_outstanding" + }) + .flat_map(|(composite, _, _, _)| { + composite + .key() + .labels() + .filter(|label| label.key() == "state") + .map(|label| label.value().to_string()) + .collect::>() + }) + .collect::>(); + assert_eq!( + group_outstanding_states, + HashSet::from([ + PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_ENQUEUE_WAITERS.to_string(), + PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_ENQUEUE_FILES.to_string(), + PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_BATCH_WAITERS.to_string(), + PUT_RENAME_FDATASYNC_GROUP_OUTSTANDING_STATE_BATCH_FILES.to_string(), + ]) + ); + let disk_wait_positions = rows + .iter() + .filter(|(composite, _, _, _)| { + composite.kind() == MetricKind::Histogram + && composite.key().name() == "rustfs_s3_put_object_rename_disk_wait_completion_ms" + }) + .flat_map(|(composite, _, _, _)| { + composite + .key() + .labels() + .filter(|label| label.key() == "position") + .map(|label| label.value().to_string()) + .collect::>() + }) + .collect::>(); + assert_eq!( + disk_wait_positions, + HashSet::from([ + PUT_RENAME_DISK_WAIT_COMPLETION_POSITION_QUORUM_FIRST.to_string(), + PUT_RENAME_DISK_WAIT_COMPLETION_POSITION_QUORUM_TAIL.to_string(), + PUT_RENAME_DISK_WAIT_COMPLETION_POSITION_ERROR.to_string(), + ]) + ); let quorum_states = rows .iter() .filter(|(composite, _, _, _)| {