diff --git a/crates/common/src/globals.rs b/crates/common/src/globals.rs index 2b8197e1d..cc0bacd12 100644 --- a/crates/common/src/globals.rs +++ b/crates/common/src/globals.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -#![allow(non_upper_case_globals)] // FIXME - use chrono::{DateTime, Utc}; use std::collections::HashMap; use std::sync::LazyLock; diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index c9f360a62..a09925f54 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -1416,6 +1416,10 @@ fn spawn_tier_free_version_recovery_once(api: Arc) { ); } Err(err) => { + rustfs_io_metrics::record_stage_duration( + "lifecycle_free_version_recovery_failed", + started_at.elapsed().as_secs_f64() * 1000.0, + ); warn!( event = EVENT_LIFECYCLE_WORKER_STATE, component = LOG_COMPONENT_ECSTORE, diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index ca2c4ec5a..678e2b9fc 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -184,9 +184,19 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d let mut need_fallback = false; let mut last_err = None; if let Some(disk) = opdisk { + let primary_walk_started = std::time::Instant::now(); match disk.walk_dir(wakl_opts, &mut wr).await { - Ok(_res) => {} + Ok(_res) => { + rustfs_io_metrics::record_stage_duration( + "metacache_walk_dir_primary", + primary_walk_started.elapsed().as_secs_f64() * 1000.0, + ); + } Err(err) => { + rustfs_io_metrics::record_stage_duration( + "metacache_walk_dir_primary_failed", + primary_walk_started.elapsed().as_secs_f64() * 1000.0, + ); warn!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, @@ -237,6 +247,7 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d return Err(err); }; + let fallback_walk_started = std::time::Instant::now(); match disk .as_ref() .walk_dir( @@ -256,10 +267,18 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d .await { Ok(_r) => { + rustfs_io_metrics::record_stage_duration( + "metacache_walk_dir_fallback", + fallback_walk_started.elapsed().as_secs_f64() * 1000.0, + ); need_fallback = false; last_err = None; } Err(err) => { + rustfs_io_metrics::record_stage_duration( + "metacache_walk_dir_fallback_failed", + fallback_walk_started.elapsed().as_secs_f64() * 1000.0, + ); error!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, @@ -538,7 +557,9 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d Ok(()) }); + let merge_started = std::time::Instant::now(); if let Err(err) = revjob.await.map_err(std::io::Error::other)? { + rustfs_io_metrics::record_stage_duration("metacache_merge_failed", merge_started.elapsed().as_secs_f64() * 1000.0); error!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, @@ -556,6 +577,7 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d return Err(err); } + rustfs_io_metrics::record_stage_duration("metacache_merge", merge_started.elapsed().as_secs_f64() * 1000.0); // The merge consumer can finish successfully before every producer finishes // (for example after reaching EOF quorum while a tolerated drive is stalled, diff --git a/crates/ecstore/src/erasure_coding/encode.rs b/crates/ecstore/src/erasure_coding/encode.rs index 5d74fcf7a..0d4ee1886 100644 --- a/crates/ecstore/src/erasure_coding/encode.rs +++ b/crates/ecstore/src/erasure_coding/encode.rs @@ -31,6 +31,10 @@ const ENV_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES: &str = "RUSTFS_ERASURE_ENCOD const DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES: usize = 32 * 1024 * 1024; const DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BLOCKS: usize = 32; +/// Cached value of `RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES` env var. +/// Read once at first use via `OnceLock` to avoid per-encode syscall. +static CACHED_MAX_INFLIGHT_BYTES: std::sync::OnceLock = std::sync::OnceLock::new(); + fn encode_channel_capacity(expanded_block_bytes: usize, max_inflight_bytes: usize) -> usize { if expanded_block_bytes == 0 { return 1; @@ -273,10 +277,12 @@ impl Erasure { // Bound queued encoded blocks by memory budget to avoid per-request spikes. let expanded_block_bytes = self.shard_size().saturating_mul(self.total_shard_count()); - let max_inflight_bytes = rustfs_utils::get_env_usize( - ENV_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES, - DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES, - ); + let max_inflight_bytes = *CACHED_MAX_INFLIGHT_BYTES.get_or_init(|| { + rustfs_utils::get_env_usize( + ENV_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES, + DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES, + ) + }); let inflight_blocks = encode_channel_capacity(expanded_block_bytes, max_inflight_bytes); let (tx, mut rx) = mpsc::channel::>(inflight_blocks); @@ -443,7 +449,7 @@ mod tests { ))]; let erasure = Arc::new(Erasure::new(1, 0, 16)); - let reader = tokio::io::BufReader::new(std::io::Cursor::new(b"small payload".to_vec())); + let reader = tokio::io::BufReader::new(Cursor::new(b"small payload".to_vec())); let (_reader, written) = erasure.encode(reader, &mut writers, 1).await.unwrap(); assert_eq!(written, b"small payload".len()); @@ -482,7 +488,7 @@ mod tests { ))]; let erasure = Arc::new(Erasure::new(1, 0, 0)); - let reader = tokio::io::BufReader::new(std::io::Cursor::new(b"payload".to_vec())); + let reader = tokio::io::BufReader::new(Cursor::new(b"payload".to_vec())); let err = erasure .encode(reader, &mut writers, 1) .await @@ -505,7 +511,7 @@ mod tests { ))]; let erasure = Arc::new(Erasure::new(1, 0, 16)); - let reader = tokio::io::BufReader::new(std::io::Cursor::new(Vec::::new())); + let reader = tokio::io::BufReader::new(Cursor::new(Vec::::new())); let (_reader, total) = erasure.encode_inline_small(reader, &mut writers, 1).await.unwrap(); assert_eq!(total, 0); @@ -537,7 +543,7 @@ mod tests { let payload = b"hello inline small"; let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE)); - let reader = tokio::io::BufReader::new(std::io::Cursor::new(payload.to_vec())); + let reader = tokio::io::BufReader::new(Cursor::new(payload.to_vec())); let (_reader, total) = erasure.encode_inline_small(reader, &mut writers, DATA_SHARDS).await.unwrap(); assert_eq!(total, payload.len()); @@ -569,7 +575,7 @@ mod tests { let payload = b"hello single block"; let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE)); - let reader = tokio::io::BufReader::new(std::io::Cursor::new(payload.to_vec())); + let reader = tokio::io::BufReader::new(Cursor::new(payload.to_vec())); let (_reader, total) = erasure .encode_single_block_non_inline(reader, &mut writers, DATA_SHARDS) .await @@ -603,7 +609,7 @@ mod tests { let payload = vec![1u8; BLOCK_SIZE + 1]; let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE)); - let reader = tokio::io::BufReader::new(std::io::Cursor::new(payload)); + let reader = tokio::io::BufReader::new(Cursor::new(payload)); let err = erasure .encode_single_block_non_inline(reader, &mut writers, DATA_SHARDS) .await diff --git a/crates/ecstore/src/erasure_coding/erasure.rs b/crates/ecstore/src/erasure_coding/erasure.rs index 69a186d6f..d8919cdbb 100644 --- a/crates/ecstore/src/erasure_coding/erasure.rs +++ b/crates/ecstore/src/erasure_coding/erasure.rs @@ -394,7 +394,7 @@ impl Erasure { encoder, legacy_encoder, uses_legacy, - _id: Uuid::new_v4(), + _id: Uuid::nil(), // Unused in hot paths; avoid CSPRNG syscall } } @@ -631,15 +631,11 @@ impl Erasure { ) -> Result where R: AsyncRead + Send + Sync + Unpin, - F: FnMut(std::io::Result>) -> Fut + Send, - Fut: std::future::Future> + Send, + F: FnMut(io::Result>) -> Fut + Send, + Fut: Future> + Send, { if self.block_size == 0 { - on_block(Err(std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "erasure block_size must be non-zero", - ))) - .await?; + on_block(Err(io::Error::new(io::ErrorKind::InvalidInput, "erasure block_size must be non-zero"))).await?; return Ok(0); } @@ -662,7 +658,7 @@ impl Erasure { { Ok(result) => result, Err(err) => { - on_block(Err(std::io::Error::other(format!("EC encode task failed: {err}")))).await?; + on_block(Err(io::Error::other(format!("EC encode task failed: {err}")))).await?; break; } }; @@ -673,7 +669,7 @@ impl Erasure { warn!("encode_stream_callback_async read unexpected ok"); break; } - Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => { + Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => { warn!("encode_stream_callback_async read unexpected eof"); break; } @@ -690,7 +686,6 @@ impl Erasure { #[cfg(test)] mod tests { - use super::*; fn optional_shards(shards: &[Bytes]) -> Vec>> { @@ -1140,7 +1135,7 @@ mod tests { assert_eq!(total, 0); let observed = observed.lock().unwrap(); let (kind, message) = observed.as_ref().expect("callback should be invoked once"); - assert_eq!(*kind, std::io::ErrorKind::InvalidInput); + assert_eq!(*kind, io::ErrorKind::InvalidInput); assert!(message.contains("block_size")); } diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 88e9783be..2ad5c4bce 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -137,6 +137,8 @@ const LOG_SUBSYSTEM_SET_DISK: &str = "set_disk"; const EVENT_SET_DISK_MULTIPART: &str = "set_disk_multipart"; const EVENT_SET_DISK_WRITE: &str = "set_disk_write"; const EVENT_SET_DISK_HEAL: &str = "set_disk_heal"; +const EVENT_SET_DISK_COMMIT_TAIL_SLOW: &str = "set_disk_commit_tail_slow"; +const SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS: u128 = 5_000; use crate::rio::{EtagResolvable, HashReader, HashReaderMut, TryGetIndex as _}; @@ -804,6 +806,16 @@ enum SmallWritePath { Pipeline, } +impl SmallWritePath { + fn metric_label(&self) -> &'static str { + match self { + SmallWritePath::Inline => "write_inline", + SmallWritePath::SingleBlockNonInline => "write_single_block_non_inline", + SmallWritePath::Pipeline => "write_pipeline", + } + } +} + fn classify_small_write_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> SmallWritePath { if should_use_inline_small_fast_path(is_inline_buffer, object_size, block_size) { SmallWritePath::Inline @@ -1061,6 +1073,7 @@ impl ObjectIO for SetDisks { let shard_file_size = erasure.shard_file_size(data.size()); let shard_size = erasure.shard_size(); + let writer_setup_stage_start = Instant::now(); let writer_futs: Vec<_> = shuffle_disks .iter() .map(|disk_op| { @@ -1107,6 +1120,10 @@ impl ObjectIO for SetDisks { writers.push(w); errors.push(e); } + rustfs_io_metrics::record_put_object_stage_duration( + "set_disk_writer_setup", + writer_setup_stage_start.elapsed().as_secs_f64() * 1000.0, + ); let nil_count = errors.iter().filter(|&e| e.is_none()).count(); if nil_count < write_quorum { @@ -1135,7 +1152,9 @@ impl ObjectIO for SetDisks { ); let write_path = classify_small_write_path(is_inline_buffer, data.size(), fi.erasure.block_size); + rustfs_io_metrics::record_put_object_path(write_path.metric_label()); + let encode_stage_start = Instant::now(); let (reader, w_size) = match write_path { SmallWritePath::Inline => match Arc::new(erasure) .encode_inline_small(stream, &mut writers, write_quorum) @@ -1165,6 +1184,10 @@ impl ObjectIO for SetDisks { } }, }; + rustfs_io_metrics::record_put_object_stage_duration( + "set_disk_encode", + encode_stage_start.elapsed().as_secs_f64() * 1000.0, + ); let _ = mem::replace(&mut data.stream, reader); // if let Err(err) = close_bitrot_writers(&mut writers).await { @@ -1259,6 +1282,7 @@ impl ObjectIO for SetDisks { object_lock_guard = Some(self.acquire_write_lock_diag("put_object_commit", bucket, object).await?); } + let rename_stage_start = Instant::now(); let (online_disks, _, op_old_dir, cleanup_disks) = Self::rename_data( &shuffle_disks, RUSTFS_META_TMP_BUCKET, @@ -1269,10 +1293,52 @@ impl ObjectIO for SetDisks { write_quorum, ) .await?; + rustfs_io_metrics::record_put_object_stage_duration( + "set_disk_rename", + rename_stage_start.elapsed().as_secs_f64() * 1000.0, + ); + let rename_stage_ms = rename_stage_start.elapsed().as_millis(); + if rename_stage_ms >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS { + warn!( + event = EVENT_SET_DISK_COMMIT_TAIL_SLOW, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + stage = "rename_data", + bucket = %bucket, + object = %object, + tmp_dir = %tmp_dir, + duration_ms = rename_stage_ms as u64, + write_quorum, + state = "slow", + "SetDisk commit tail stage is slow" + ); + } if let Some(old_dir) = op_old_dir { + let cleanup_stage_start = Instant::now(); self.commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), write_quorum) .await?; + rustfs_io_metrics::record_put_object_stage_duration( + "set_disk_old_data_cleanup", + cleanup_stage_start.elapsed().as_secs_f64() * 1000.0, + ); + let cleanup_stage_ms = cleanup_stage_start.elapsed().as_millis(); + if cleanup_stage_ms >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS { + warn!( + event = EVENT_SET_DISK_COMMIT_TAIL_SLOW, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + stage = "commit_rename_data_dir", + bucket = %bucket, + object = %object, + tmp_dir = %tmp_dir, + old_dir = %old_dir, + duration_ms = cleanup_stage_ms as u64, + write_quorum, + state = "slow", + "SetDisk commit tail stage is slow" + ); + } } drop(object_lock_guard); // drop object lock guard to release the lock @@ -1307,6 +1373,23 @@ impl ObjectIO for SetDisks { ); } + let total_commit_tail_ms = rename_stage_start.elapsed().as_millis(); + if total_commit_tail_ms >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS { + warn!( + event = EVENT_SET_DISK_COMMIT_TAIL_SLOW, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + stage = "put_object_commit_tail", + bucket = %bucket, + object = %object, + tmp_dir = %tmp_dir, + duration_ms = total_commit_tail_ms as u64, + write_quorum, + state = "slow", + "SetDisk commit tail is slow" + ); + } + Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended)) } .await; diff --git a/crates/ecstore/src/store_list_objects.rs b/crates/ecstore/src/store_list_objects.rs index a21aca20b..a5757f3ac 100644 --- a/crates/ecstore/src/store_list_objects.rs +++ b/crates/ecstore/src/store_list_objects.rs @@ -1158,6 +1158,7 @@ impl ECStore { tokio::spawn(async move { merge_entry_channels(rx, inputs, merge_tx, 1).await }.instrument(tracing::Span::current())); + let walk_started = std::time::Instant::now(); let walk_results = join_all(futures).await; let mut errs = Vec::new(); for walk_result in walk_results { @@ -1166,6 +1167,10 @@ impl ECStore { Err(err) => errs.push(Some(err.into())), } } + rustfs_io_metrics::record_stage_duration( + "store_list_objects_walk_internal", + walk_started.elapsed().as_secs_f64() * 1000.0, + ); let result = walk_result_from_set_errors(&errs); if let Err(err) = &result { diff --git a/crates/io-core/src/pool.rs b/crates/io-core/src/pool.rs index a8d7577fb..ff46e0384 100644 --- a/crates/io-core/src/pool.rs +++ b/crates/io-core/src/pool.rs @@ -423,23 +423,31 @@ impl PoolTier { }) } - /// Return a buffer to the pool for reuse. + /// Return a buffer to the pool for reuse without ever blocking the caller. fn return_buffer(&self, buffer: BytesMut) { - let mut available = self.available_buffers.lock().unwrap_or_else(|e| e.into_inner()); - // Limit the size of the pool to prevent unbounded growth - if available.len() < self.max_buffers { - available.push(buffer); - if let Some(ref metrics) = *self.metrics.lock().unwrap_or_else(|e| e.into_inner()) { + let mut buffer = Some(buffer); + + if let Ok(mut available) = self.available_buffers.try_lock() + && available.len() < self.max_buffers + { + available.push(buffer.take().expect("buffer should be present until returned")); + if let Ok(metrics) = self.metrics.try_lock() + && let Some(metrics) = metrics.as_ref() + { metrics.available_buffers.fetch_add(1, Ordering::Relaxed); } - } else { + } + + if let Some(buffer) = buffer { let released_bytes = buffer.capacity() as u64; self.tier_current_allocated_bytes .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { Some(current.saturating_sub(released_bytes)) }) .ok(); - if let Some(ref metrics) = *self.metrics.lock().unwrap_or_else(|e| e.into_inner()) { + if let Ok(metrics) = self.metrics.try_lock() + && let Some(metrics) = metrics.as_ref() + { metrics .current_allocated_bytes .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { @@ -448,7 +456,6 @@ impl PoolTier { .ok(); } } - // If pool is full, buffer is dropped and memory is freed rustfs_io_metrics::record_bytes_pool_allocated(self.name, self.tier_current_allocated_bytes.load(Ordering::Relaxed)); } } @@ -458,8 +465,6 @@ impl Drop for PooledBuffer { // buffer moves it exactly once into the pool when a tier still owns it. #[allow(unsafe_code)] fn drop(&mut self) { - // Return buffer to pool if tier reference exists. - // Otherwise, drop the standalone fallback buffer normally. let buffer = unsafe { ManuallyDrop::take(&mut self.buffer) }; if let Some(ref tier) = self.tier { tier.return_buffer(buffer); diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index ff742c69f..4ba7b43a7 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -49,7 +49,30 @@ #[macro_use] extern crate metrics; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + +/// Global switch for detailed per-stage PUT metrics (path label, stage durations). +/// When `false`, `record_put_object_path` and `record_put_object_stage_duration` +/// become no-ops, and callers can skip the `Instant::now()` syscalls entirely. +/// +/// Set to `true` during startup when OTEL metric export is enabled. +static PUT_STAGE_METRICS_ENABLED: AtomicBool = AtomicBool::new(false); + +/// Enable or disable detailed per-stage PUT metrics. +/// +/// Called once during startup, typically gated by `rustfs_obs::observability_metric_enabled()`. +pub fn set_put_stage_metrics_enabled(enabled: bool) { + PUT_STAGE_METRICS_ENABLED.store(enabled, Ordering::Relaxed); +} + +/// Returns `true` if detailed per-stage PUT metrics are enabled. +/// +/// Callers should check this before calling `Instant::now()` for stage timing +/// to avoid unnecessary syscalls when metrics are disabled. +#[inline(always)] +pub fn put_stage_metrics_enabled() -> bool { + PUT_STAGE_METRICS_ENABLED.load(Ordering::Relaxed) +} // Public modules pub mod adaptive_ttl; @@ -409,9 +432,9 @@ pub fn record_get_object(duration_ms: f64, size_bytes: i64) { /// /// * `duration_ms` - Operation duration in milliseconds /// * `size_bytes` - Object size in bytes -/// * `zero_copy_enabled` - Whether zero-copy was enabled for this operation +/// * `zero_copy_eligible` - Whether the request was eligible for a zero-copy path #[inline(always)] -pub fn record_put_object(duration_ms: f64, size_bytes: i64, zero_copy_enabled: bool) { +pub fn record_put_object(duration_ms: f64, size_bytes: i64, zero_copy_eligible: bool) { counter!("rustfs_s3_put_object_total").increment(1); histogram!("rustfs_s3_put_object_duration_ms").record(duration_ms); @@ -419,11 +442,37 @@ pub fn record_put_object(duration_ms: f64, size_bytes: i64, zero_copy_enabled: b histogram!("rustfs_s3_put_object_size_bytes").record(size_bytes as f64); } - if zero_copy_enabled { + if zero_copy_eligible { + // Backward-compatible alias for historical dashboards. counter!("rustfs_s3_put_object_zero_copy_enabled_total").increment(1); + counter!("rustfs_s3_put_object_zero_copy_eligible_total").increment(1); } } +#[inline(always)] +pub fn record_put_object_path(path: &'static str) { + if !put_stage_metrics_enabled() { + return; + } + counter!("rustfs_s3_put_object_path_total", "path" => path).increment(1); +} + +#[inline(always)] +pub fn record_put_object_stage_duration(stage: &'static str, duration_ms: f64) { + if !put_stage_metrics_enabled() { + return; + } + histogram!("rustfs_s3_put_object_stage_duration_ms", "stage" => stage).record(duration_ms); +} + +/// 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. +#[inline(always)] +pub fn record_stage_duration(stage: &'static str, duration_ms: f64) { + histogram!("rustfs_internal_stage_duration_ms", "stage" => stage).record(duration_ms); +} + /// Record ListObjects operation metrics. /// /// # Arguments @@ -788,6 +837,10 @@ pub fn record_io_latency_p99(latency_ms: f64) { #[cfg(test)] mod tests { use super::*; + use std::sync::Mutex; + + // Serialize tests that mutate the process-global PUT_STAGE_METRICS_ENABLED flag. + static METRICS_FLAG_LOCK: Mutex<()> = Mutex::new(()); #[test] fn test_record_zero_copy_read() { @@ -824,6 +877,36 @@ mod tests { record_put_object(100.0, 512, false); } + #[test] + fn test_record_put_object_path_and_stage() { + let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + set_put_stage_metrics_enabled(true); + record_put_object_path("small_eager"); + record_put_object_path("write_inline"); + record_put_object_stage_duration("ingress_prepare", 12.5); + record_put_object_stage_duration("set_disk_encode", 8.0); + set_put_stage_metrics_enabled(false); + } + + #[test] + fn test_put_stage_metrics_disabled_by_default() { + let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + set_put_stage_metrics_enabled(false); + // These should be no-ops (no panic, no recording) + record_put_object_path("small_eager"); + record_put_object_stage_duration("set_disk_encode", 5.0); + // Still disabled + assert!(!put_stage_metrics_enabled()); + } + + #[test] + fn test_record_stage_duration_generic() { + // Generic stage duration should always record (no gating flag) + record_stage_duration("metacache_walk_dir_primary", 15.0); + record_stage_duration("store_list_objects_walk_internal", 8.5); + record_stage_duration("lifecycle_free_version_recovery_failed", 120.0); + } + #[test] fn test_record_list_objects() { record_list_objects(50.0, 100, false); diff --git a/crates/object-capacity/src/scan.rs b/crates/object-capacity/src/scan.rs index cacd224ad..ed371e722 100644 --- a/crates/object-capacity/src/scan.rs +++ b/crates/object-capacity/src/scan.rs @@ -556,6 +556,7 @@ async fn get_dir_size_async(path: &Path) -> Result Result S3Result> { @@ -374,6 +382,33 @@ impl AsyncRead for ExtractArchiveEtagReader { } } +struct PooledBufferReader { + buffer: PooledBuffer, + len: usize, + pos: usize, +} + +impl PooledBufferReader { + fn new(buffer: PooledBuffer, len: usize) -> Self { + Self { buffer, len, pos: 0 } + } +} + +impl AsyncRead for PooledBufferReader { + fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + if self.pos >= self.len { + return Poll::Ready(Ok(())); + } + + let remaining = self.len - self.pos; + let to_read = remaining.min(buf.remaining()); + buf.put_slice(&self.buffer[self.pos..self.pos + to_read]); + self.pos += to_read; + + Poll::Ready(Ok(())) + } +} + /// Determine if zero-copy write should be used for this PutObject operation. /// /// Zero-copy is beneficial for large objects without encryption or compression. @@ -429,6 +464,105 @@ fn should_use_zero_copy(size: i64, headers: &HeaderMap) -> bool { true } +fn has_put_sse_request_headers(headers: &HeaderMap) -> bool { + headers.get(AMZ_SERVER_SIDE_ENCRYPTION).is_some() + || headers.get(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM).is_some() + || headers.get(AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID).is_some() +} + +fn should_use_small_eager_put_path( + size: i64, + headers: &HeaderMap, + server_side_encryption_requested: bool, + should_compress: bool, + is_extract: bool, +) -> bool { + const SMALL_EAGER_PUT_MAX_SIZE: i64 = 1024 * 1024; + + if is_extract || should_compress || server_side_encryption_requested { + return false; + } + + if size <= 0 || size > SMALL_EAGER_PUT_MAX_SIZE { + return false; + } + + if has_put_sse_request_headers(headers) { + return false; + } + + if request_uses_aws_chunked(headers) && decoded_content_length_from_headers(headers).ok().flatten().is_none() { + return false; + } + + true +} + +/// Objects at or below this size bypass BytesPool and use direct allocation. +/// This avoids Small-tier Mutex contention under high concurrency for tiny objects +/// where the allocation cost is negligible (≤4KiB memcpy). +const POOL_BYPASS_MAX_SIZE: usize = 4 * 1024; + +async fn read_small_put_body_exact_pooled(mut body: R, size: usize, pool: &BytesPool) -> S3Result +where + R: AsyncRead + Unpin, +{ + let mut buf = pool.acquire_buffer(size).await; + buf.resize(size, 0); + let mut filled = 0; + + while filled < size { + let read = tokio::io::AsyncReadExt::read(&mut body, &mut buf[filled..size]) + .await + .map_err(|err| ApiError::from(StorageError::other(err.to_string())))?; + if read == 0 { + return Err(s3_error!(IncompleteBody)); + } + filled += read; + } + + let mut extra = [0u8; 1]; + let extra_read = tokio::io::AsyncReadExt::read(&mut body, &mut extra) + .await + .map_err(|err| ApiError::from(StorageError::other(err.to_string())))?; + if extra_read != 0 { + return Err(s3_error!(UnexpectedContent)); + } + + Ok(buf) +} + +/// Read small PUT body into a directly-allocated buffer, bypassing BytesPool. +/// Used for objects ≤4KiB where pool contention under high concurrency +/// outweighs the allocation cost. +async fn read_small_put_body_exact_direct(mut body: R, size: usize) -> S3Result>> +where + R: AsyncRead + Unpin, +{ + let mut buf = vec![0u8; size]; + let mut filled = 0; + + while filled < size { + let read = tokio::io::AsyncReadExt::read(&mut body, &mut buf[filled..size]) + .await + .map_err(|err| ApiError::from(StorageError::other(err.to_string())))?; + if read == 0 { + return Err(s3_error!(IncompleteBody)); + } + filled += read; + } + + let mut extra = [0u8; 1]; + let extra_read = tokio::io::AsyncReadExt::read(&mut body, &mut extra) + .await + .map_err(|err| ApiError::from(StorageError::other(err.to_string())))?; + if extra_read != 0 { + return Err(s3_error!(UnexpectedContent)); + } + + Ok(std::io::Cursor::new(buf)) +} + fn object_seek_support_threshold() -> usize { static OBJECT_SEEK_SUPPORT_THRESHOLD: OnceLock = OnceLock::new(); *OBJECT_SEEK_SUPPORT_THRESHOLD.get_or_init(|| { @@ -1814,10 +1948,21 @@ impl DefaultObjectUsecase { return Err(s3_error!(UnexpectedContent)); } + let ingress_stage_start = std::time::Instant::now(); + let should_compress = is_disk_compressible(&req.headers, &key) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64; + let server_side_encryption_requested = + server_side_encryption.is_some() || sse_customer_algorithm.is_some() || ssekms_key_id.is_some(); + // Apply adaptive buffer sizing based on file size for optimal streaming performance. // Uses workload profile configuration (enabled by default) to select appropriate buffer size. // Buffer sizes range from 32KB to 4MB depending on file size and configured workload profile. - let buffer_size = get_buffer_size_opt_in(size); + // Concurrency-aware adjustment reduces buffer size under high concurrency to lower memory pressure. + // TODO: get_concurrency_aware_buffer_size reads ACTIVE_GET_REQUESTS (GET concurrency tracker), + // not PUT concurrency. Under pure PUT load the counter stays zero so buffers never shrink; + // unrelated GET load can shrink PUT buffers instead. Fix by adding ACTIVE_PUT_REQUESTS + + // PutObjectGuard and using PUT concurrency here. See PR #3514 review comment. + let base_buffer_size = get_buffer_size_opt_in(size); + let buffer_size = get_concurrency_aware_buffer_size(size, base_buffer_size); // Detect zero-copy opportunity before encryption/compression decisions // Zero-copy is beneficial for large unencrypted, uncompressed objects @@ -1830,10 +1975,15 @@ impl DefaultObjectUsecase { debug!("Zero-copy write enabled for {} byte object (bucket={}, key={})", size, bucket, key); } - let body = tokio::io::BufReader::with_capacity( - buffer_size, - StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))), - ); + let use_small_eager_put_path = + should_use_small_eager_put_path(size, &req.headers, server_side_encryption_requested, should_compress, false); + let put_path = if should_compress { + "stream_compressed" + } else if use_small_eager_put_path { + "small_eager" + } else { + "streaming" + }; let store = get_validated_store(&bucket).await?; @@ -1955,9 +2105,12 @@ impl DefaultObjectUsecase { let mut sha256hex = get_content_sha256_with_query(&req.headers, req.uri.query()); - let should_compress = is_disk_compressible(&req.headers, &key) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64; let mut write_plan = WritePlan::new(); let mut reader = if should_compress { + let body = tokio::io::BufReader::with_capacity( + buffer_size, + StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))), + ); let algorithm = CompressionAlgorithm::default(); insert_str( &mut metadata, @@ -1985,7 +2138,35 @@ impl DefaultObjectUsecase { write_plan = write_plan.with_compression(algorithm); hrd } else { - HashReader::from_stream(body, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)? + if use_small_eager_put_path { + if (actual_size as usize) <= POOL_BYPASS_MAX_SIZE { + // Bypass BytesPool for very small objects to avoid Small-tier + // Mutex contention under high concurrency. Direct allocation + // for ≤4KiB is negligible cost. + let eager_body = read_small_put_body_exact_direct( + StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))), + actual_size as usize, + ) + .await?; + HashReader::from_stream(eager_body, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)? + } else { + let pool = get_concurrency_manager().bytes_pool(); + let eager_body = read_small_put_body_exact_pooled( + StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))), + actual_size as usize, + pool.as_ref(), + ) + .await?; + let eager_reader = PooledBufferReader::new(eager_body, actual_size as usize); + HashReader::from_stream(eager_reader, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)? + } + } else { + let body = tokio::io::BufReader::with_capacity( + buffer_size, + StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))), + ); + HashReader::from_stream(body, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)? + } }; if size >= 0 { @@ -1995,6 +2176,11 @@ impl DefaultObjectUsecase { opts.want_checksum = reader.checksum(); } + rustfs_io_metrics::record_put_object_path(put_path); + rustfs_io_metrics::record_put_object_stage_duration( + "ingress_prepare", + ingress_stage_start.elapsed().as_secs_f64() * 1000.0, + ); let mut helper = OperationHelper::new(&req, event_name, S3Operation::PutObject); let ssekms_context = extract_ssekms_context_from_headers(&req.headers)?; @@ -2038,6 +2224,11 @@ impl DefaultObjectUsecase { let mt2 = metadata.clone(); opts.user_defined.extend(metadata); + let request_context = req.extensions.get::().cloned(); + let request_id = request_context + .as_ref() + .map(|ctx| ctx.request_id.clone()) + .unwrap_or_else(|| request_context::RequestContext::fallback().request_id); let repoptions = get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts.clone()); @@ -2052,13 +2243,76 @@ impl DefaultObjectUsecase { ); } + let store_put_watchdog = tokio_util::sync::CancellationToken::new(); + spawn_traced({ + let store_put_watchdog = store_put_watchdog.clone(); + let request_id = request_id.clone(); + let bucket = bucket.clone(); + let key = key.clone(); + let put_path = put_path.to_string(); + async move { + tokio::select! { + _ = store_put_watchdog.cancelled() => {} + _ = tokio::time::sleep(PUT_OBJECT_STORE_WARN_THRESHOLD) => { + warn!( + target: "rustfs::app::object_usecase", + event = EVENT_PUT_OBJECT_STORE_INFLIGHT_SLOW, + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_OBJECT, + request_id = %request_id, + bucket = %bucket, + key = %key, + put_path = %put_path, + object_size = actual_size, + threshold_ms = PUT_OBJECT_STORE_WARN_THRESHOLD.as_millis() as u64, + state = "store_put_pending", + "PutObject store write remains in flight" + ); + } + } + } + }); + let obj_info = match store .put_object(&bucket, &key, &mut reader, &opts) .await .map_err(ApiError::from) { - Ok(obj_info) => obj_info, + Ok(obj_info) => { + store_put_watchdog.cancel(); + debug!( + target: "rustfs::app::object_usecase", + event = EVENT_PUT_OBJECT_STORE_RETURNED, + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_OBJECT, + request_id = %request_id, + bucket = %bucket, + key = %key, + put_path = put_path, + object_size = actual_size, + duration_ms = start_time.elapsed().as_millis() as u64, + result = "success", + "PutObject store write returned" + ); + obj_info + } Err(err) => { + store_put_watchdog.cancel(); + warn!( + target: "rustfs::app::object_usecase", + event = EVENT_PUT_OBJECT_STORE_RETURNED, + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_OBJECT, + request_id = %request_id, + bucket = %bucket, + key = %key, + put_path = put_path, + object_size = actual_size, + duration_ms = start_time.elapsed().as_millis() as u64, + result = "error", + error = %err, + "PutObject store write returned" + ); let result: S3Result> = Err(err.into()); let _ = helper.complete(&result); return result; @@ -5056,6 +5310,86 @@ mod tests { assert!(!should_use_zero_copy(2 * 1024 * 1024, &headers)); } + #[test] + fn should_use_small_eager_put_path_allows_up_to_1mb() { + let headers = HeaderMap::new(); + + assert!(should_use_small_eager_put_path(1024, &headers, false, false, false)); + assert!(should_use_small_eager_put_path(1024 * 1024, &headers, false, false, false)); + assert!(!should_use_small_eager_put_path(1024 * 1024 + 1, &headers, false, false, false)); + } + + #[test] + fn should_use_small_eager_put_path_rejects_sse_requests() { + let headers = HeaderMap::new(); + + assert!(!should_use_small_eager_put_path(1024, &headers, true, false, false)); + } + + #[test] + fn should_use_small_eager_put_path_rejects_compressible_objects() { + let headers = HeaderMap::new(); + + assert!(!should_use_small_eager_put_path(1024, &headers, false, true, false)); + } + + #[test] + fn should_use_small_eager_put_path_rejects_extract_requests() { + let headers = HeaderMap::new(); + + assert!(!should_use_small_eager_put_path(1024, &headers, false, false, true)); + } + + #[test] + fn should_use_small_eager_put_path_rejects_large_or_empty_objects() { + let headers = HeaderMap::new(); + + assert!(!should_use_small_eager_put_path(0, &headers, false, false, false)); + assert!(!should_use_small_eager_put_path(1024 * 1024 + 1, &headers, false, false, false)); + } + + #[tokio::test] + async fn read_small_put_body_exact_pooled_reads_exact_bytes() { + let pool = get_concurrency_manager().bytes_pool(); + let body = std::io::Cursor::new(b"hello".to_vec()); + + let buffer = read_small_put_body_exact_pooled(body, 5, pool.as_ref()) + .await + .expect("pooled exact read should succeed"); + + assert_eq!(&buffer[..5], b"hello"); + } + + #[tokio::test] + async fn read_small_put_body_exact_pooled_rejects_short_body() { + let pool = get_concurrency_manager().bytes_pool(); + let body = std::io::Cursor::new(b"hell".to_vec()); + + let err = match read_small_put_body_exact_pooled(body, 5, pool.as_ref()).await { + Ok(_) => panic!("short pooled body should fail"), + Err(err) => err, + }; + + assert_eq!(err.code(), &S3ErrorCode::IncompleteBody); + } + + #[tokio::test] + async fn pooled_buffer_reader_keeps_buffer_alive_until_consumed() { + use tokio::io::AsyncReadExt; + + let pool = get_concurrency_manager().bytes_pool(); + let body = std::io::Cursor::new(b"hello".to_vec()); + let buffer = read_small_put_body_exact_pooled(body, 5, pool.as_ref()) + .await + .expect("pooled exact read should succeed"); + let mut reader = PooledBufferReader::new(buffer, 5); + let mut out = Vec::new(); + + reader.read_to_end(&mut out).await.expect("pooled reader should be readable"); + + assert_eq!(out, b"hello"); + } + #[test] fn should_use_zero_copy_allows_large_unencrypted_binary_objects() { let mut headers = HeaderMap::new(); diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 3b55858af..e5f0b1f18 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -275,13 +275,31 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc Result<()> { - socket.set_reuse_address(true)?; + if let Err(e) = socket.set_reuse_address(true) { + debug!( + event = "socket_option_unavailable", + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_STARTUP, + option = "SO_REUSEADDR", + error = %e, + "Socket option is unavailable" + ); + } // Set the socket to non-blocking before passing it to Tokio. socket.set_nonblocking(true)?; // 1. Disable Nagle algorithm: Critical for 4KB Payload, achieving ultra-low latency - socket.set_tcp_nodelay(true)?; + if let Err(e) = socket.set_tcp_nodelay(true) { + debug!( + event = "socket_option_unavailable", + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_STARTUP, + option = "TCP_NODELAY", + error = %e, + "Socket option is unavailable" + ); + } // 2. Enable SO_REUSEPORT for better multi-core scalability on supported platforms #[cfg(all(unix, not(target_os = "solaris"), not(target_os = "illumos")))] @@ -297,11 +315,40 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc String { let path = uri.path(); @@ -322,9 +327,39 @@ where fn call(&mut self, req: HttpRequest) -> Self::Future { let context = RequestLogContext::from_request(&req); let mut inner = self.inner.clone(); + let watchdog = CancellationToken::new(); + let watchdog_context = context.clone(); + + spawn_traced({ + let watchdog = watchdog.clone(); + async move { + tokio::select! { + _ = watchdog.cancelled() => {} + _ = tokio::time::sleep(HTTP_REQUEST_INFLIGHT_WARN_THRESHOLD) => { + warn!( + event = HTTP_REQUEST_INFLIGHT_SLOW_EVENT, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_HTTP, + request_id = %watchdog_context.request_id, + trace_id = %watchdog_context.trace_id.as_deref().unwrap_or("unknown"), + span_id = %watchdog_context.span_id.as_deref().unwrap_or("unknown"), + peer_addr = %watchdog_context.peer_addr, + method = %watchdog_context.method, + uri = %watchdog_context.uri, + duration_ms = watchdog_context.duration_ms(), + active_requests = active_http_requests(), + threshold_ms = HTTP_REQUEST_INFLIGHT_WARN_THRESHOLD.as_millis() as u64, + state = "response_pending", + "HTTP request remains in flight" + ); + } + } + } + }); Box::pin(async move { let result = inner.call(req).await; + watchdog.cancel(); match &result { Ok(response) => context.log_response(response), Err(error) => context.log_failure(error), @@ -3054,8 +3089,8 @@ mod tests { assert_eq!(redact_sensitive_uri_query(&uri), "/rustfs/admin/v3/users?token=not-a-download-token"); } - #[test] - fn request_logging_layer_emits_single_completion_event_with_standard_fields() { + #[tokio::test] + async fn request_logging_layer_emits_single_completion_event_with_standard_fields() { let writer = SharedWriter::default(); let captured = writer.buffer.clone(); let subscriber = Registry::default().with( @@ -3067,25 +3102,25 @@ mod tests { .with_writer(writer), ); - tracing::subscriber::with_default(subscriber, || { - let mut service = tower::ServiceBuilder::new() - .layer(RequestContextLayer) - .layer(RequestLoggingLayer) - .service(StatusService::new(StatusCode::OK)); + let _guard = tracing::subscriber::set_default(subscriber); - let mut request: Request> = Request::builder() - .method(Method::GET) - .uri("/bucket/object.txt") - .header("x-request-id", "req-123") - .body(Full::from(Bytes::new())) - .expect("request"); - request - .extensions_mut() - .insert(RemoteAddr("127.0.0.1:9000".parse().expect("socket addr"))); + let mut service = tower::ServiceBuilder::new() + .layer(RequestContextLayer) + .layer(RequestLoggingLayer) + .service(StatusService::new(StatusCode::OK)); - let response = futures::executor::block_on(service.call(request)).expect("response"); - assert_eq!(response.status(), StatusCode::OK); - }); + let mut request: Request> = Request::builder() + .method(Method::GET) + .uri("/bucket/object.txt") + .header("x-request-id", "req-123") + .body(Full::from(Bytes::new())) + .expect("request"); + request + .extensions_mut() + .insert(RemoteAddr("127.0.0.1:9000".parse().expect("socket addr"))); + + let response = service.call(request).await.expect("response"); + assert_eq!(response.status(), StatusCode::OK); let output = String::from_utf8(captured.lock().expect("captured logs").clone()).expect("utf8 logs"); assert_eq!(output.matches("HTTP request completed").count(), 1, "{output}"); @@ -3110,8 +3145,8 @@ mod tests { assert!(output.contains("duration_ms"), "{output}"); } - #[test] - fn request_logging_layer_uses_request_context_trace_fields() { + #[tokio::test] + async fn request_logging_layer_uses_request_context_trace_fields() { let writer = SharedWriter::default(); let captured = writer.buffer.clone(); let subscriber = Registry::default().with( @@ -3123,28 +3158,28 @@ mod tests { .with_writer(writer), ); - tracing::subscriber::with_default(subscriber, || { - let mut service = RequestLoggingLayer.layer(StatusService::new(StatusCode::INTERNAL_SERVER_ERROR)); + let _guard = tracing::subscriber::set_default(subscriber); - let mut request = Request::builder() - .method(Method::GET) - .uri("/bucket/object.txt") - .body(()) - .expect("request"); - request.extensions_mut().insert(RequestContext { - request_id: "req-ctx".to_string(), - x_amz_request_id: "amz-ctx".to_string(), - trace_id: Some("trace-ctx".to_string()), - span_id: Some("span-ctx".to_string()), - start_time: Instant::now(), - }); - request - .extensions_mut() - .insert(RemoteAddr("127.0.0.1:9000".parse().expect("socket addr"))); + let mut service = RequestLoggingLayer.layer(StatusService::new(StatusCode::INTERNAL_SERVER_ERROR)); - let response = futures::executor::block_on(service.call(request)).expect("response"); - assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + let mut request = Request::builder() + .method(Method::GET) + .uri("/bucket/object.txt") + .body(()) + .expect("request"); + request.extensions_mut().insert(RequestContext { + request_id: "req-ctx".to_string(), + x_amz_request_id: "amz-ctx".to_string(), + trace_id: Some("trace-ctx".to_string()), + span_id: Some("span-ctx".to_string()), + start_time: Instant::now(), }); + request + .extensions_mut() + .insert(RemoteAddr("127.0.0.1:9000".parse().expect("socket addr"))); + + let response = service.call(request).await.expect("response"); + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); let output = String::from_utf8(captured.lock().expect("captured logs").clone()).expect("utf8 logs"); assert!(output.contains("http_request_completed"), "{output}"); diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index 93538629a..44a62b149 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -355,6 +355,7 @@ async fn init_observability_runtime(ctx: CancellationToken) { crate::allocator_reclaim::init_allocator_reclaim(ctx.clone()); if rustfs_obs::observability_metric_enabled() { + rustfs_io_metrics::set_put_stage_metrics_enabled(true); init_metrics_runtime(ctx.clone()); crate::memory_observability::init_memory_observability(ctx.clone()); init_auto_tuner(ctx).await; diff --git a/scripts/run_object_batch_bench_enhanced.sh b/scripts/run_object_batch_bench_enhanced.sh index 57f80f07d..efbfc3c11 100755 --- a/scripts/run_object_batch_bench_enhanced.sh +++ b/scripts/run_object_batch_bench_enhanced.sh @@ -202,47 +202,56 @@ trim() { to_bps() { local human="$1" + local number unit factor if [[ "$human" == "N/A" || -z "$human" ]]; then echo "N/A" return fi - awk -v v="$human" ' - function abs(x){return x<0?-x:x} - BEGIN{ - if (match(v, /^([0-9]+(\.[0-9]+)?)\s*(GiB\/s|MiB\/s|KiB\/s|GB\/s|MB\/s|KB\/s|B\/s)$/, m)) { - n=m[1]; u=m[3]; - if (u=="GiB/s") f=1024*1024*1024; - else if (u=="MiB/s") f=1024*1024; - else if (u=="KiB/s") f=1024; - else if (u=="GB/s") f=1000*1000*1000; - else if (u=="MB/s") f=1000*1000; - else if (u=="KB/s") f=1000; - else f=1; - printf "%.6f\n", n*f; - } else { - print "N/A"; - } - }' + + # Normalize "123MiB/s" → "123 MiB/s" when no space separator exists. + human="$(echo "$human" | sed -E 's/^([0-9]+(\.[0-9]+)?)([A-Za-zµ])/\1 \3/')" + number="$(echo "$human" | awk '{print $1}')" + unit="$(echo "$human" | awk '{print $2}')" + case "$unit" in + GiB/s) factor=1073741824 ;; + MiB/s) factor=1048576 ;; + KiB/s) factor=1024 ;; + GB/s) factor=1000000000 ;; + MB/s) factor=1000000 ;; + KB/s) factor=1000 ;; + B/s) factor=1 ;; + *) + echo "N/A" + return + ;; + esac + + awk -v n="$number" -v f="$factor" 'BEGIN { printf "%.6f\n", n * f }' } to_ms() { local human="$1" + local number unit factor if [[ "$human" == "N/A" || -z "$human" ]]; then echo "N/A" return fi - awk -v v="$human" ' - BEGIN{ - if (match(v, /^([0-9]+(\.[0-9]+)?)\s*(ms|us|µs|s)$/, m)) { - n=m[1]; u=m[3]; - if (u=="s") f=1000; - else if (u=="ms") f=1; - else f=0.001; - printf "%.6f\n", n*f; - } else { - print "N/A"; - } - }' + + # Normalize "50ms" → "50 ms" when no space separator exists. + human="$(echo "$human" | sed -E 's/^([0-9]+(\.[0-9]+)?)([A-Za-zµ])/\1 \3/')" + number="$(echo "$human" | awk '{print $1}')" + unit="$(echo "$human" | awk '{print $2}')" + case "$unit" in + s) factor=1000 ;; + ms) factor=1 ;; + us|µs) factor=0.001 ;; + *) + echo "N/A" + return + ;; + esac + + awk -v n="$number" -v f="$factor" 'BEGIN { printf "%.6f\n", n * f }' } extract_first() { @@ -255,20 +264,24 @@ extract_metrics() { local log_file="$1" local throughput reqps latency - throughput="$(extract_first '[0-9]+(\.[0-9]+)?\s*(GiB/s|MiB/s|KiB/s|GB/s|MB/s|KB/s|B/s)' "$log_file")" - reqps="$(extract_first '[0-9]+(\.[0-9]+)?\s*(req/s|ops/s|requests/s)' "$log_file")" - latency="$(extract_first '[0-9]+(\.[0-9]+)?\s*(ms|us|µs|s)\s*(avg|mean)' "$log_file")" + throughput="$(extract_first '[0-9]+(\.[0-9]+)?[[:space:]]*(GiB/s|MiB/s|KiB/s|GB/s|MB/s|KB/s|B/s)' "$log_file")" + reqps="$(extract_first '[0-9]+(\.[0-9]+)?[[:space:]]*(obj/s|req/s|ops/s|requests/s)' "$log_file")" + latency="$(rg -o 'Reqs:[[:space:]]+Avg:[[:space:]]+[0-9]+(\.[0-9]+)?(ms|us|µs|s)' "$log_file" | head -n1 | sed -E 's/^Reqs:[[:space:]]+Avg:[[:space:]]+//')" if [[ -z "$latency" ]]; then - latency="$(extract_first '[0-9]+(\.[0-9]+)?\s*(ms|us|µs|s)' "$log_file")" + latency="$(extract_first '[0-9]+(\.[0-9]+)?[[:space:]]*(ms|us|µs|s)' "$log_file")" fi throughput="$(trim "${throughput:-N/A}")" reqps="$(trim "${reqps:-N/A}")" latency="$(trim "${latency:-N/A}")" - # Keep only " " for latency if suffix avg/mean exists. - latency="$(echo "$latency" | awk '{print $1" "$2}')" + # Normalize to " " shape for downstream conversion helpers. + if [[ "$latency" =~ ^([0-9]+(\.[0-9]+)?)(ms|us|µs|s)$ ]]; then + latency="${BASH_REMATCH[1]} ${BASH_REMATCH[3]}" + else + latency="$(echo "$latency" | awk '{print $1" "$2}')" + fi reqps_num="$(echo "$reqps" | awk '{print $1}')" echo "$throughput,${reqps_num:-N/A},$latency" @@ -329,7 +342,7 @@ run_one_attempt() { printf '\n' echo "dry run" > "$log_file" else - if ! "${cmd[@]}" 2>&1 | tee "$log_file"; then + if ! "${cmd[@]}" 2>&1 | tee "$log_file" >&2; then status="failed" fi fi @@ -357,7 +370,7 @@ run_one_attempt() { printf '\n' echo "dry run" > "$log_file" else - if ! "${cmd[@]}" 2>&1 | tee "$log_file"; then + if ! "${cmd[@]}" 2>&1 | tee "$log_file" >&2; then status="failed" fi fi