diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index 94392fa81..acc3a5e5e 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -1098,6 +1098,10 @@ pub(crate) async fn sync_dir_files_with_limiter(dir: impl AsRef, 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, 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(); diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index dcf5d11e3..784969051 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/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 { diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 785cea246..50a660383 100644 --- a/crates/io-metrics/src/lib.rs +++ b/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::>() + }) + .collect::>(); + 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::>() + }) + .collect::>(); + 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");