diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 1932744e8..a71f2c55b 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -9264,17 +9264,27 @@ impl DiskAPI for LocalDisk { // accept that window (documented in docs/operations/durability-modes.md). if durability.syncs_commit_metadata() && let Some(parent) = dst_file_path.parent() - && let Err(err) = os::fsync_dir(parent).await { - rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) - .map_err(to_file_error)?; - // The commit rename changed the dst part inodes before this fsync - // failed and rolled them back; drop any fd cached during that - // window so readers re-open the restored inode (rustfs/backlog#1177). - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; + let fsync_started = rustfs_io_metrics::put_stage_timer(); + if let Err(err) = os::fsync_dir(parent).await { + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, + fsync_started, + ); + rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) + .map_err(to_file_error)?; + // The commit rename changed the dst part inodes before this fsync + // failed and rolled them back; drop any fd cached during that + // window so readers re-open the restored inode (rustfs/backlog#1177). + for part_path in &invalidate_part_paths { + self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; + } + return Err(to_file_error(err).into()); } - return Err(to_file_error(err).into()); + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, + fsync_started, + ); } // First PUT of an object creates its directory (and any missing prefix @@ -9293,7 +9303,12 @@ impl DiskAPI for LocalDisk { if !dir.starts_with(&dst_volume_dir) { break; } + let fsync_started = rustfs_io_metrics::put_stage_timer(); if let Err(err) = os::fsync_dir(dir).await { + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, + fsync_started, + ); rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) .map_err(to_file_error)?; // Same post-commit rollback window as above — drop cached @@ -9304,6 +9319,10 @@ impl DiskAPI for LocalDisk { } return Err(to_file_error(err).into()); } + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, + fsync_started, + ); if dir == dst_volume_dir.as_path() { break; } @@ -9532,10 +9551,21 @@ impl DiskAPI for LocalDisk { } if let Some(admission) = file_sync_admission.as_ref() && let Some(backup_parent) = backup_path.parent() - && let Err(err) = - os::fsync_dir_with_namespace_file_sync_limit(backup_parent, mutation_lease.clone(), admission).await { - return Err(DiskError::from(to_file_error(err))); + let fsync_started = rustfs_io_metrics::put_stage_timer(); + if let Err(err) = + os::fsync_dir_with_namespace_file_sync_limit(backup_parent, mutation_lease.clone(), admission).await + { + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC, + fsync_started, + ); + return Err(DiskError::from(to_file_error(err))); + } + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC, + fsync_started, + ); } local_rollback_path = None; } @@ -9573,11 +9603,22 @@ impl DiskAPI for LocalDisk { // Persist the commit rename's directory entry across power loss. if let Some(admission) = file_sync_admission.as_ref() && let Some(dst_parent) = dst_file_path.parent() - && let Err(err) = - os::fsync_dir_with_namespace_file_sync_limit(dst_parent, mutation_lease.clone(), admission).await { - rollback_inline_metadata_commit_std(&dst_file_path, rollback_data_dir, local_rollback_path.as_deref())?; - return Err(err); + let fsync_started = rustfs_io_metrics::put_stage_timer(); + if let Err(err) = + os::fsync_dir_with_namespace_file_sync_limit(dst_parent, mutation_lease.clone(), admission).await + { + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, + fsync_started, + ); + rollback_inline_metadata_commit_std(&dst_file_path, rollback_data_dir, local_rollback_path.as_deref())?; + return Err(err); + } + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, + fsync_started, + ); } // Same power-loss gap as the non-inline path (rustfs/backlog#922 @@ -9595,9 +9636,14 @@ impl DiskAPI for LocalDisk { if !ancestor_dir.starts_with(&dst_volume_dir) { break; } + let fsync_started = rustfs_io_metrics::put_stage_timer(); if let Err(err) = os::fsync_dir_with_namespace_file_sync_limit(ancestor_dir, mutation_lease.clone(), admission).await { + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, + fsync_started, + ); rollback_inline_metadata_commit_std( &dst_file_path, rollback_data_dir, @@ -9605,6 +9651,10 @@ impl DiskAPI for LocalDisk { )?; return Err(err); } + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, + fsync_started, + ); if ancestor_dir == dst_volume_dir.as_path() { break; } diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index f5b1c5f66..decb36964 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -343,6 +343,7 @@ pub(crate) async fn acquire_rename_data_mutation_lease( /// this order uniform prevents one slow disk from reserving global capacity /// while it waits for its own concurrency slot. async fn acquire_file_sync_permits(disk_permits: Arc) -> io::Result<(OwnedSemaphorePermit, SemaphorePermit<'static>)> { + let wait_started = rustfs_io_metrics::put_stage_timer(); let disk_permit = disk_permits .acquire_owned() .await @@ -351,6 +352,10 @@ async fn acquire_file_sync_permits(disk_permits: Arc) -> io::Result<( .acquire() .await .map_err(|_| io::Error::other("global file sync concurrency limiter closed"))?; + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_FILE_SYNC_PERMIT_WAIT, + wait_started, + ); Ok((disk_permit, global_permit)) } @@ -551,9 +556,19 @@ pub(crate) fn sync_file(path: &Path) -> io::Result<()> { file.sync_data() } +fn sync_file_with_put_stage_metric(path: &Path) -> io::Result<()> { + let sync_started = rustfs_io_metrics::put_stage_timer(); + let result = sync_file(path); + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_FILE_FDATASYNC, + sync_started, + ); + result +} + fn sync_files(paths: &[PathBuf]) -> io::Result<()> { for path in paths { - sync_file(path)?; + sync_file_with_put_stage_metric(path)?; } Ok(()) } @@ -599,7 +614,13 @@ pub(crate) async fn sync_dir_files_with_limiter(dir: impl AsRef, disk_perm let files = regular_files(&scan_dir)?; if files.len() < PARALLEL_FILE_SYNC_THRESHOLD { sync_files(&files)?; - fsync_dir_std(scan_dir)?; + let fsync_started = rustfs_io_metrics::put_stage_timer(); + let result = fsync_dir_std(scan_dir); + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_SRC_DIR_FSYNC, + fsync_started, + ); + result?; return Ok(None); } Ok::<_, io::Error>(Some(files)) @@ -612,10 +633,19 @@ pub(crate) async fn sync_dir_files_with_limiter(dir: impl AsRef, disk_perm futures::stream::iter(files.into_iter().map(Ok::<_, io::Error>)) .try_for_each_concurrent(MAX_PARALLEL_FILE_SYNCS, |path| { let disk_permits = disk_permits.clone(); - async move { run_file_sync_blocking(disk_permits, move || sync_file(&path)).await } + async move { run_file_sync_blocking(disk_permits, move || sync_file_with_put_stage_metric(&path)).await } }) .await?; - run_file_sync_blocking(disk_permits, move || fsync_dir_std(dir)).await + run_file_sync_blocking(disk_permits, move || { + let fsync_started = rustfs_io_metrics::put_stage_timer(); + let result = fsync_dir_std(dir); + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_SRC_DIR_FSYNC, + fsync_started, + ); + result + }) + .await } /// Check if the given disk path is the root disk. @@ -1174,10 +1204,15 @@ pub(crate) struct FileSyncAdmission { } pub(crate) async fn acquire_file_sync_admission(disk_permits: Arc) -> io::Result { + let wait_started = rustfs_io_metrics::put_stage_timer(); let disk_permit = disk_permits .acquire_owned() .await .map_err(|_| io::Error::other("disk file sync concurrency limiter closed"))?; + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_FILE_SYNC_PERMIT_WAIT, + wait_started, + ); Ok(FileSyncAdmission { disk_permit: Arc::new(disk_permit), }) @@ -1200,10 +1235,15 @@ async fn run_blocking_namespace_file_sync_operation_with_global io::Result + Send + 'static, ) -> io::Result { + let wait_started = rustfs_io_metrics::put_stage_timer(); let global_permit = global_permits .acquire() .await .map_err(|_| io::Error::other("global file sync concurrency limiter closed"))?; + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_GLOBAL_FILE_SYNC_PERMIT_WAIT, + wait_started, + ); let disk_permit = admission.disk_permit.clone(); let result = tokio::task::spawn_blocking(move || { let _lease = lease; @@ -1420,7 +1460,13 @@ fn rename_into_existing_parent( use rustix::fs::{Mode, OFlags, open, renameat}; let Some(parent_guard) = parent_guard else { - return super::fs::rename_std(src_file_path, dst_file_path); + let rename_started = rustfs_io_metrics::put_stage_timer(); + let result = super::fs::rename_std(src_file_path, dst_file_path); + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_RENAME_SYSCALL, + rename_started, + ); + return result; }; let src_parent = src_file_path .parent() @@ -1441,7 +1487,13 @@ fn rename_into_existing_parent( .last() .ok_or_else(|| io::Error::other("rename destination parent guard is empty"))?; - renameat(&src_parent, src_name, dst_parent, dst_name).map_err(io::Error::from) + let rename_started = rustfs_io_metrics::put_stage_timer(); + let result = renameat(&src_parent, src_name, dst_parent, dst_name).map_err(io::Error::from); + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_RENAME_SYSCALL, + rename_started, + ); + result } #[cfg(windows)] @@ -2890,6 +2942,7 @@ pub fn is_dir_not_empty_error(err: &io::Error) -> bool { #[cfg(test)] mod tests { use super::*; + use crate::test_metrics::CapturingRecorder; use std::sync::Mutex; use std::time::Duration; use tempfile::tempdir; @@ -2910,6 +2963,42 @@ mod tests { PublicationRoot::new(&common).expect("test publication root should open") } + #[test] + #[serial_test::serial(file_sync_metrics)] + fn sync_file_with_put_stage_metric_records_fdatasync_only_when_enabled() { + let previous_gate = rustfs_io_metrics::put_stage_metrics_enabled(); + rustfs_io_metrics::set_put_stage_metrics_enabled(false); + let dir = tempdir().expect("temp dir should be created"); + let path = dir.path().join("part.1"); + std::fs::write(&path, b"payload").expect("test file should be written"); + let recorder = CapturingRecorder::default(); + + metrics::with_local_recorder(&recorder, || { + sync_file_with_put_stage_metric(&path).expect("disabled metric sync_file should succeed"); + assert_eq!( + recorder.histogram_sample_count("rustfs_s3_put_object_stage_duration_ms"), + 0, + "disabled PUT stage metrics must not emit fdatasync samples" + ); + + rustfs_io_metrics::set_put_stage_metrics_enabled(true); + sync_file_with_put_stage_metric(&path).expect("enabled metric sync_file should succeed"); + rustfs_io_metrics::set_put_stage_metrics_enabled(false); + }); + + assert_eq!( + recorder + .histogram_values( + "rustfs_s3_put_object_stage_duration_ms", + &[("stage", rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_FILE_FDATASYNC)] + ) + .len(), + 1, + "enabled PUT stage metrics must emit one fdatasync sample" + ); + rustfs_io_metrics::set_put_stage_metrics_enabled(previous_gate); + } + async fn rename_all( src_file_path: impl AsRef, dst_file_path: impl AsRef, diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 7af25a64c..dcf5d11e3 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -3389,8 +3389,15 @@ impl SetDisks { // A no-op immediately-ready future in production. Self::rename_fanout_barrier(&dst_object, i, rename_fanout_barrier_phase::RENAME).await; - disk.rename_data_borrowed(&src_bucket, &src_object, file_info, &dst_bucket, &dst_object) - .await + let disk_wait_started = rustfs_io_metrics::put_stage_timer(); + 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, + ); + result }) .catch_unwind() }); @@ -3403,7 +3410,13 @@ impl SetDisks { let mut cleanup_data_dirs = vec![None; disk_count]; let mut old_current_sizes = vec![None; disk_count]; - let (results, mut file_infos) = fanout.await.map_err(|_| DiskError::Unexpected)?; + let quorum_wait_started = rustfs_io_metrics::put_stage_timer(); + let fanout_result = fanout.await; + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_QUORUM_WAIT, + quorum_wait_started, + ); + let (results, mut file_infos) = fanout_result.map_err(|_| DiskError::Unexpected)?; for (idx, result) in results.iter().enumerate() { match result { diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 3f021772d..86b38ed5a 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -109,6 +109,17 @@ pub fn put_stage_timer() -> Option { put_stage_metrics_enabled().then(std::time::Instant::now) } +pub const PUT_STAGE_SET_DISK_RENAME_QUORUM_WAIT: &str = "set_disk_rename_quorum_wait"; +pub const PUT_STAGE_SET_DISK_RENAME_DISK_WAIT: &str = "set_disk_rename_disk_wait"; +pub const PUT_STAGE_SET_DISK_RENAME_FILE_SYNC_PERMIT_WAIT: &str = "set_disk_rename_file_sync_permit_wait"; +pub const PUT_STAGE_SET_DISK_RENAME_GLOBAL_FILE_SYNC_PERMIT_WAIT: &str = "set_disk_rename_global_file_sync_permit_wait"; +pub const PUT_STAGE_SET_DISK_RENAME_FILE_FDATASYNC: &str = "set_disk_rename_file_fdatasync"; +pub const PUT_STAGE_SET_DISK_RENAME_SRC_DIR_FSYNC: &str = "set_disk_rename_src_dir_fsync"; +pub const PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC: &str = "set_disk_rename_dst_dir_fsync"; +pub const PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC: &str = "set_disk_rename_backup_dir_fsync"; +pub const PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC: &str = "set_disk_rename_ancestor_dir_fsync"; +pub const PUT_STAGE_SET_DISK_RENAME_RENAME_SYSCALL: &str = "set_disk_rename_rename_syscall"; + #[inline(always)] pub fn get_stage_metrics_enabled() -> bool { GET_STAGE_METRICS_ENABLED.load(Ordering::Relaxed) @@ -2618,6 +2629,7 @@ mod tests { use super::*; use metrics_util::MetricKind; use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + use std::collections::HashSet; use std::sync::{Arc, Barrier, Mutex}; // Serialize tests that mutate the process-global PUT_STAGE_METRICS_ENABLED flag. @@ -2861,6 +2873,63 @@ mod tests { set_put_stage_metrics_enabled(false); } + #[test] + fn put_stage_sync_tail_labels_are_static_and_gated() { + let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let stages = [ + PUT_STAGE_SET_DISK_RENAME_QUORUM_WAIT, + PUT_STAGE_SET_DISK_RENAME_DISK_WAIT, + PUT_STAGE_SET_DISK_RENAME_FILE_SYNC_PERMIT_WAIT, + PUT_STAGE_SET_DISK_RENAME_GLOBAL_FILE_SYNC_PERMIT_WAIT, + PUT_STAGE_SET_DISK_RENAME_FILE_FDATASYNC, + PUT_STAGE_SET_DISK_RENAME_SRC_DIR_FSYNC, + PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, + PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC, + PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, + PUT_STAGE_SET_DISK_RENAME_RENAME_SYSCALL, + ]; + let unique = stages.iter().copied().collect::>(); + assert_eq!(unique.len(), stages.len()); + assert!( + stages + .iter() + .all(|stage| stage.starts_with("set_disk_rename_") && !stage.contains('/') && !stage.contains('{')) + ); + + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + set_put_stage_metrics_enabled(false); + for stage in stages { + record_put_object_stage_duration(stage, 1.0); + } + set_put_stage_metrics_enabled(true); + for stage in stages { + record_put_object_stage_duration(stage, 1.0); + } + set_put_stage_metrics_enabled(false); + }); + + let recorded = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(composite, _, _, _)| { + composite.kind() == MetricKind::Histogram && composite.key().name() == "rustfs_s3_put_object_stage_duration_ms" + }) + .flat_map(|(composite, _, _, _)| { + composite + .key() + .labels() + .filter(|label| label.key() == "stage") + .map(|label| label.value().to_string()) + .collect::>() + }) + .collect::>(); + assert_eq!(recorded.len(), stages.len()); + assert!(stages.iter().all(|stage| recorded.contains(*stage))); + } + #[test] fn test_put_object_diagnostic_buckets() { assert_eq!(put_object_size_bucket(0), "unknown");