feat(ecstore): instrument rename sync tail metrics (#6205)

Add default-off PUT stage attribution for the rename_data sync tail so strict durability probes can split queue wait, fdatasync, directory fsync, rename, per-disk wait, and quorum wait without changing commit ordering or S3-visible behavior.

Co-authored-by: heihutu <heihutu@gmail.com>

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;
                         }

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<Semaphore>) -> 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<Semaphore>) -> 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<Path>, 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<Path>, 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<Semaphore>) -> io::Result<FileSyncAdmission> {
+    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<T: Send + 'stati
     global_permits: &Semaphore,
     operation: impl FnOnce() -> io::Result<T> + Send + 'static,
 ) -> io::Result<T> {
+    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<Path>,
         dst_file_path: impl AsRef<Path>,

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 {

crates/io-metrics/src/lib.rs

@@ -109,6 +109,17 @@ pub fn put_stage_timer() -> Option<std::time::Instant> {
     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::<HashSet<_>>();
+        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::<Vec<_>>()
+            })
+            .collect::<HashSet<_>>();
+        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");