feat(ecstore): expose rename sync tail metrics (#6257)

Add default-off PUT stage helpers for fdatasync batch shape and rename quorum fanout shape so #925 follow-up probes can distinguish shard sync batching opportunities from fanout convergence.

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

crates/ecstore/src/disk/os.rs

@@ -1098,6 +1098,10 @@ pub(crate) async fn sync_dir_files_with_limiter(dir: impl AsRef<Path>, disk_perm
     let files = run_file_sync_blocking(disk_permits.clone(), move || {
         let files = regular_files(&scan_dir)?;
         if files.len() < PARALLEL_FILE_SYNC_THRESHOLD {
+            rustfs_io_metrics::record_put_rename_fdatasync_batch(
+                rustfs_io_metrics::PUT_RENAME_FDATASYNC_BATCH_MODE_SERIAL,
+                files.len(),
+            );
             sync_files(&files)?;
             let fsync_started = rustfs_io_metrics::put_stage_timer();
             let result = fsync_dir_std(scan_dir);
@@ -1115,6 +1119,10 @@ pub(crate) async fn sync_dir_files_with_limiter(dir: impl AsRef<Path>, disk_perm
     let Some(files) = files else {
         return Ok(());
     };
+    rustfs_io_metrics::record_put_rename_fdatasync_batch(
+        rustfs_io_metrics::PUT_RENAME_FDATASYNC_BATCH_MODE_PARALLEL,
+        files.len(),
+    );
     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();

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -3417,6 +3417,25 @@ impl SetDisks {
             quorum_wait_started,
         );
         let (results, mut file_infos) = fanout_result.map_err(|_| DiskError::Unexpected)?;
+        if rustfs_io_metrics::put_stage_metrics_enabled() {
+            let mut fanout_success = 0;
+            let mut fanout_error = 0;
+            let mut fanout_panic = 0;
+            for result in &results {
+                match result {
+                    Ok(Ok(_)) => fanout_success += 1,
+                    Ok(Err(_)) => fanout_error += 1,
+                    Err(_) => fanout_panic += 1,
+                }
+            }
+            rustfs_io_metrics::record_put_rename_quorum_wait_fanout(
+                results.len(),
+                write_quorum,
+                fanout_success,
+                fanout_error,
+                fanout_panic,
+            );
+        }
 
         for (idx, result) in results.iter().enumerate() {
             match result {

crates/io-metrics/src/lib.rs

@@ -120,6 +120,14 @@ pub const PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC: &str = "set_disk_rename_ba
 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";
 
+pub const PUT_RENAME_FDATASYNC_BATCH_MODE_SERIAL: &str = "serial";
+pub const PUT_RENAME_FDATASYNC_BATCH_MODE_PARALLEL: &str = "parallel";
+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";
+
 #[inline(always)]
 pub fn get_stage_metrics_enabled() -> bool {
     GET_STAGE_METRICS_ENABLED.load(Ordering::Relaxed)
@@ -2042,6 +2050,44 @@ pub fn record_put_object_stage_duration_from(stage: &'static str, started_at: Op
     }
 }
 
+#[inline(always)]
+fn put_stage_count_value(value: usize) -> f64 {
+    match u32::try_from(value) {
+        Ok(value) => f64::from(value),
+        Err(_) => f64::from(u32::MAX),
+    }
+}
+
+#[inline(always)]
+pub fn record_put_rename_fdatasync_batch(mode: &'static str, files: usize) {
+    if !put_stage_metrics_enabled() {
+        return;
+    }
+    histogram!("rustfs_s3_put_object_rename_fdatasync_batch_files", "mode" => mode).record(put_stage_count_value(files));
+}
+
+#[inline(always)]
+pub fn record_put_rename_quorum_wait_fanout(
+    scheduled: usize,
+    write_quorum: usize,
+    success: usize,
+    error: usize,
+    panicked: usize,
+) {
+    if !put_stage_metrics_enabled() {
+        return;
+    }
+    for (state, count) in [
+        (PUT_RENAME_QUORUM_FANOUT_STATE_SCHEDULED, scheduled),
+        (PUT_RENAME_QUORUM_FANOUT_STATE_WRITE_QUORUM, write_quorum),
+        (PUT_RENAME_QUORUM_FANOUT_STATE_SUCCESS, success),
+        (PUT_RENAME_QUORUM_FANOUT_STATE_ERROR, error),
+        (PUT_RENAME_QUORUM_FANOUT_STATE_PANIC, panicked),
+    ] {
+        histogram!("rustfs_s3_put_object_rename_quorum_wait_fanout_disks", "state" => state).record(put_stage_count_value(count));
+    }
+}
+
 /// Record generic internal operation stage duration (non-PUT paths).
 /// Use this for metacache walks, listing, lifecycle, and other background
 /// operations that are NOT part of the PUT object hot path.
@@ -3122,6 +3168,70 @@ mod tests {
         assert!(stages.iter().all(|stage| recorded.contains(*stage)));
     }
 
+    #[test]
+    fn put_rename_code_level_metrics_are_static_and_gated() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        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_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_quorum_wait_fanout(4, 3, 3, 1, 0);
+            set_put_stage_metrics_enabled(false);
+        });
+
+        let rows = snapshotter.snapshot().into_vec();
+        assert_eq!(histogram_samples(&rows, "rustfs_s3_put_object_rename_fdatasync_batch_files"), vec![9.0]);
+        let batch_modes = rows
+            .iter()
+            .filter(|(composite, _, _, _)| {
+                composite.kind() == MetricKind::Histogram
+                    && composite.key().name() == "rustfs_s3_put_object_rename_fdatasync_batch_files"
+            })
+            .flat_map(|(composite, _, _, _)| {
+                composite
+                    .key()
+                    .labels()
+                    .filter(|label| label.key() == "mode")
+                    .map(|label| label.value().to_string())
+                    .collect::<Vec<_>>()
+            })
+            .collect::<HashSet<_>>();
+        assert_eq!(batch_modes, HashSet::from([PUT_RENAME_FDATASYNC_BATCH_MODE_PARALLEL.to_string()]));
+
+        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]);
+        let quorum_states = rows
+            .iter()
+            .filter(|(composite, _, _, _)| {
+                composite.kind() == MetricKind::Histogram
+                    && composite.key().name() == "rustfs_s3_put_object_rename_quorum_wait_fanout_disks"
+            })
+            .flat_map(|(composite, _, _, _)| {
+                composite
+                    .key()
+                    .labels()
+                    .filter(|label| label.key() == "state")
+                    .map(|label| label.value().to_string())
+                    .collect::<Vec<_>>()
+            })
+            .collect::<HashSet<_>>();
+        assert_eq!(
+            quorum_states,
+            HashSet::from([
+                PUT_RENAME_QUORUM_FANOUT_STATE_SCHEDULED.to_string(),
+                PUT_RENAME_QUORUM_FANOUT_STATE_WRITE_QUORUM.to_string(),
+                PUT_RENAME_QUORUM_FANOUT_STATE_SUCCESS.to_string(),
+                PUT_RENAME_QUORUM_FANOUT_STATE_ERROR.to_string(),
+                PUT_RENAME_QUORUM_FANOUT_STATE_PANIC.to_string(),
+            ])
+        );
+    }
+
     #[test]
     fn test_put_object_diagnostic_buckets() {
         assert_eq!(put_object_size_bucket(0), "unknown");