diff --git a/crates/config/src/observability/mod.rs b/crates/config/src/observability/mod.rs index 9a1acc6c4..ca001fc8b 100644 --- a/crates/config/src/observability/mod.rs +++ b/crates/config/src/observability/mod.rs @@ -51,6 +51,11 @@ pub const ENV_OBS_PROFILING_EXPORT_ENABLED: &str = "RUSTFS_OBS_PROFILING_EXPORT_ /// PUT records multiple timers and histograms when attribution is active. pub const ENV_OBS_PUT_STAGE_METRICS_ENABLED: &str = "RUSTFS_OBS_PUT_STAGE_METRICS_ENABLED"; +/// Enables sampled in-process GetObject stage summaries without OTLP histograms. +pub const ENV_OBS_GET_STAGE_LOCAL_SUMMARY_ENABLED: &str = "RUSTFS_OBS_GET_STAGE_LOCAL_SUMMARY_ENABLED"; +pub const ENV_OBS_GET_STAGE_LOCAL_SUMMARY_SAMPLE_RATE: &str = "RUSTFS_OBS_GET_STAGE_LOCAL_SUMMARY_SAMPLE_RATE"; +pub const ENV_OBS_GET_STAGE_LOCAL_SUMMARY_INTERVAL_SECS: &str = "RUSTFS_OBS_GET_STAGE_LOCAL_SUMMARY_INTERVAL_SECS"; + pub const ENV_OBS_LOGGER_LEVEL: &str = "RUSTFS_OBS_LOGGER_LEVEL"; pub const ENV_OBS_LOG_STDOUT_ENABLED: &str = "RUSTFS_OBS_LOG_STDOUT_ENABLED"; pub const ENV_OBS_LOG_DIRECTORY: &str = "RUSTFS_OBS_LOG_DIRECTORY"; @@ -159,6 +164,15 @@ mod tests { assert_eq!(ENV_OBS_LOGS_EXPORT_ENABLED, "RUSTFS_OBS_LOGS_EXPORT_ENABLED"); assert_eq!(ENV_OBS_PROFILING_EXPORT_ENABLED, "RUSTFS_OBS_PROFILING_EXPORT_ENABLED"); assert_eq!(ENV_OBS_PUT_STAGE_METRICS_ENABLED, "RUSTFS_OBS_PUT_STAGE_METRICS_ENABLED"); + assert_eq!(ENV_OBS_GET_STAGE_LOCAL_SUMMARY_ENABLED, "RUSTFS_OBS_GET_STAGE_LOCAL_SUMMARY_ENABLED"); + assert_eq!( + ENV_OBS_GET_STAGE_LOCAL_SUMMARY_SAMPLE_RATE, + "RUSTFS_OBS_GET_STAGE_LOCAL_SUMMARY_SAMPLE_RATE" + ); + assert_eq!( + ENV_OBS_GET_STAGE_LOCAL_SUMMARY_INTERVAL_SECS, + "RUSTFS_OBS_GET_STAGE_LOCAL_SUMMARY_INTERVAL_SECS" + ); // Test log cleanup related env keys assert_eq!(ENV_OBS_LOG_MAX_TOTAL_SIZE_BYTES, "RUSTFS_OBS_LOG_MAX_TOTAL_SIZE_BYTES"); assert_eq!(ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, "RUSTFS_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES"); diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 93a3062c6..7bb2f66a6 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -49,8 +49,9 @@ #[macro_use] extern crate metrics; +use std::collections::HashMap; use std::sync::{ - Mutex, + LazyLock, Mutex, atomic::{AtomicBool, AtomicU64, Ordering}, }; @@ -61,6 +62,25 @@ use std::sync::{ /// Enabled only through an explicit runtime opt-in. static PUT_STAGE_METRICS_ENABLED: AtomicBool = AtomicBool::new(false); static GET_STAGE_METRICS_ENABLED: AtomicBool = AtomicBool::new(false); +static GET_STAGE_LOCAL_SUMMARY_ENABLED: AtomicBool = AtomicBool::new(false); +static GET_STAGE_LOCAL_SUMMARY_SAMPLE_RATE: AtomicU64 = AtomicU64::new(64); +static GET_STAGE_LOCAL_SUMMARY_SEQUENCE: AtomicU64 = AtomicU64::new(0); + +type GetStageLocalSummaryKey = (&'static str, &'static str, &'static str, &'static str); + +/// Sampled local aggregate for one bounded GetObject stage label set. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct GetStageLocalSummary { + pub path: &'static str, + pub stage: &'static str, + pub object_class: &'static str, + pub size_bucket: &'static str, + pub sampled_count: u64, + pub duration_nanoseconds: u64, +} + +static GET_STAGE_LOCAL_SUMMARY: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); /// Global switch for all remaining (non GET/PUT-stage) metric emission in this /// crate's free `record_*` functions — I/O scheduler, bytes-pool, zero-copy, @@ -87,6 +107,66 @@ pub fn set_get_stage_metrics_enabled(enabled: bool) { GET_STAGE_METRICS_ENABLED.store(enabled, Ordering::Relaxed); } +pub fn set_get_stage_local_summary_enabled(enabled: bool) { + GET_STAGE_LOCAL_SUMMARY_ENABLED.store(enabled, Ordering::Relaxed); +} + +pub fn set_get_stage_local_summary_sample_rate(sample_rate: u64) { + GET_STAGE_LOCAL_SUMMARY_SAMPLE_RATE.store(sample_rate.max(1), Ordering::Relaxed); +} + +pub fn get_stage_local_summary_enabled() -> bool { + GET_STAGE_LOCAL_SUMMARY_ENABLED.load(Ordering::Relaxed) +} + +pub fn get_stage_local_summary_sample_rate() -> u64 { + GET_STAGE_LOCAL_SUMMARY_SAMPLE_RATE.load(Ordering::Relaxed) +} + +pub fn take_get_stage_local_summary() -> Vec { + let Ok(mut summary) = GET_STAGE_LOCAL_SUMMARY.lock() else { + return Vec::new(); + }; + let mut rows = summary.drain().map(|(_, row)| row).collect::>(); + rows.sort_unstable_by_key(|row| (row.path, row.stage, row.object_class, row.size_bucket)); + rows +} + +fn record_get_stage_local_summary( + path: &'static str, + stage: &'static str, + object_class: &'static str, + size_bucket: &'static str, + duration_secs: f64, +) { + let sample_rate = get_stage_local_summary_sample_rate(); + if sample_rate > 1 && GET_STAGE_LOCAL_SUMMARY_SEQUENCE.fetch_add(1, Ordering::Relaxed) % sample_rate != 0 { + return; + } + + let Ok(duration) = std::time::Duration::try_from_secs_f64(duration_secs) else { + return; + }; + let Ok(duration_nanoseconds) = u64::try_from(duration.as_nanos()) else { + return; + }; + + let key = (path, stage, object_class, size_bucket); + let Ok(mut summary) = GET_STAGE_LOCAL_SUMMARY.lock() else { + return; + }; + let row = summary.entry(key).or_insert_with(|| GetStageLocalSummary { + path, + stage, + object_class, + size_bucket, + sampled_count: 0, + duration_nanoseconds: 0, + }); + row.sampled_count = row.sampled_count.saturating_add(1); + row.duration_nanoseconds = row.duration_nanoseconds.saturating_add(duration_nanoseconds); +} + /// Enable or disable general (non GET/PUT-stage) metric emission. /// /// Called once during startup, typically gated by `rustfs_obs::observability_metric_enabled()`. @@ -964,6 +1044,10 @@ pub fn record_get_object_io_state( /// Record GetObject phase duration for the current read path. #[inline(always)] pub fn record_get_object_stage_duration(path: &'static str, stage: &'static str, duration_secs: f64) { + if get_stage_local_summary_enabled() { + record_get_stage_local_summary(path, stage, "", "", duration_secs); + return; + } if !get_stage_metrics_enabled() { return; } @@ -979,6 +1063,10 @@ pub fn record_get_object_stage_duration_by_size( size_bucket: &'static str, duration_secs: f64, ) { + if get_stage_local_summary_enabled() { + record_get_stage_local_summary(path, stage, object_class, size_bucket, duration_secs); + return; + } if !get_stage_metrics_enabled() { return; } @@ -1872,6 +1960,10 @@ pub fn record_get_object_metadata_phase_duration(duration_secs: f64) { /// Record metadata phase duration with early-stop state label. #[inline(always)] pub fn record_get_object_metadata_phase_duration_with_early_stop(duration_secs: f64, early_stop_active: &'static str) { + if get_stage_local_summary_enabled() { + record_get_stage_local_summary("legacy_duplex", "metadata", early_stop_active, "", duration_secs); + return; + } if !get_stage_metrics_enabled() { return; } diff --git a/rustfs/src/startup_observability.rs b/rustfs/src/startup_observability.rs index 01ec050b8..114b1a808 100644 --- a/rustfs/src/startup_observability.rs +++ b/rustfs/src/startup_observability.rs @@ -13,12 +13,16 @@ // limitations under the License. use std::sync::Arc; +use std::time::Duration; use crate::init::{init_auto_tuner, init_update_check, print_server_info}; use crate::startup_runtime_sources; use crate::storage_api::startup::storage::{ECStore, init_compression_total_memory_from_backend}; +use tokio::io::AsyncWriteExt; use tokio_util::sync::CancellationToken; +const GET_STAGE_LOCAL_SUMMARY_FILE: &str = "/var/log/rustfs/get-stage-summary.log"; + pub(crate) async fn init_observability_runtime(store: Arc, ctx: CancellationToken) { print_server_info(); init_update_check(); @@ -27,6 +31,10 @@ pub(crate) async fn init_observability_runtime(store: Arc, ctx: Cancell let metrics_enabled = startup_runtime_sources::observability_metric_enabled(); configure_metric_gates(metrics_enabled); + if rustfs_io_metrics::get_stage_local_summary_enabled() { + init_get_stage_local_summary_runtime(ctx.clone()); + } + if metrics_enabled { // Load persisted compression stats into memory early, before any PUTs can occur. init_compression_total_memory_from_backend(store).await; @@ -37,21 +45,115 @@ pub(crate) async fn init_observability_runtime(store: Arc, ctx: Cancell } fn configure_metric_gates(metrics_enabled: bool) { + let local_summary_enabled = !metrics_enabled + && rustfs_utils::get_env_bool(rustfs_config::observability::ENV_OBS_GET_STAGE_LOCAL_SUMMARY_ENABLED, false); + let sample_rate = + rustfs_utils::get_env_u64(rustfs_config::observability::ENV_OBS_GET_STAGE_LOCAL_SUMMARY_SAMPLE_RATE, 64).max(1); + startup_runtime_sources::set_get_stage_local_summary_enabled(local_summary_enabled); + startup_runtime_sources::set_get_stage_local_summary_sample_rate(sample_rate); let put_stage_metrics_enabled = metrics_enabled && rustfs_utils::get_env_bool( rustfs_config::observability::ENV_OBS_PUT_STAGE_METRICS_ENABLED, rustfs_config::DEFAULT_OBS_PUT_STAGE_METRICS_ENABLED, ); startup_runtime_sources::set_put_stage_metrics_enabled(put_stage_metrics_enabled); - startup_runtime_sources::set_get_stage_metrics_enabled(metrics_enabled); + startup_runtime_sources::set_get_stage_metrics_enabled(metrics_enabled || local_summary_enabled); startup_runtime_sources::set_metrics_enabled(metrics_enabled); } +fn init_get_stage_local_summary_runtime(ctx: CancellationToken) { + let interval_secs = + rustfs_utils::get_env_u64(rustfs_config::observability::ENV_OBS_GET_STAGE_LOCAL_SUMMARY_INTERVAL_SECS, 15).max(1); + tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_secs(interval_secs)); + interval.tick().await; + loop { + tokio::select! { + _ = ctx.cancelled() => break, + _ = interval.tick() => { + let rows = rustfs_io_metrics::take_get_stage_local_summary(); + if rows.is_empty() { + continue; + } + let summary = rows.iter().map(|row| { + format!("{}|{}|{}|{}|{}|{}", row.path, row.stage, row.object_class, row.size_bucket, row.sampled_count, row.duration_nanoseconds) + }).collect::>().join(";"); + let timestamp_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(); + let line = format!( + "timestamp_ms={timestamp_ms}\tsample_rate={}\tstage_count={}\tstages={summary}\n", + rustfs_io_metrics::get_stage_local_summary_sample_rate(), + rows.len(), + ); + let Ok(mut file) = tokio::fs::OpenOptions::new() + .create(true) + .append(true) + .open(GET_STAGE_LOCAL_SUMMARY_FILE) + .await + else { + break; + }; + if file.write_all(line.as_bytes()).await.is_err() { + break; + } + } + } + } + }); +} + #[cfg(test)] mod tests { use super::*; const PUT_STAGE_ENV: &str = rustfs_config::observability::ENV_OBS_PUT_STAGE_METRICS_ENABLED; + const LOCAL_SUMMARY_ENV: &str = rustfs_config::observability::ENV_OBS_GET_STAGE_LOCAL_SUMMARY_ENABLED; + const LOCAL_SUMMARY_RATE_ENV: &str = rustfs_config::observability::ENV_OBS_GET_STAGE_LOCAL_SUMMARY_SAMPLE_RATE; + + #[test] + #[serial_test::serial] + fn get_stage_local_summary_can_run_without_metrics_export() { + let previous_metrics = rustfs_io_metrics::metrics_enabled(); + let previous_get_stages = rustfs_io_metrics::get_stage_metrics_enabled(); + let previous_local_summary = rustfs_io_metrics::get_stage_local_summary_enabled(); + let previous_sample_rate = rustfs_io_metrics::get_stage_local_summary_sample_rate(); + + temp_env::with_var(LOCAL_SUMMARY_ENV, Some("true"), || { + temp_env::with_var(LOCAL_SUMMARY_RATE_ENV, Some("1"), || { + configure_metric_gates(false); + assert!(!rustfs_io_metrics::metrics_enabled()); + assert!(rustfs_io_metrics::get_stage_metrics_enabled()); + assert!(rustfs_io_metrics::get_stage_local_summary_enabled()); + rustfs_io_metrics::take_get_stage_local_summary(); + + rustfs_io_metrics::record_get_object_stage_duration("legacy_duplex", "metadata", 0.001); + rustfs_io_metrics::record_get_object_stage_duration("legacy_duplex", "metadata", 0.003); + rustfs_io_metrics::record_get_object_stage_duration_by_size( + "legacy_duplex", + "inline_direct.decode", + "plain_single_part", + "le_4kib", + 0.002, + ); + + let rows = rustfs_io_metrics::take_get_stage_local_summary(); + let metadata = rows + .iter() + .find(|row| row.path == "legacy_duplex" && row.stage == "metadata") + .expect("local summary should retain sampled metadata observations"); + assert_eq!(metadata.sampled_count, 2); + assert_eq!(metadata.duration_nanoseconds, 4_000_000); + assert!(rows.iter().any(|row| row.size_bucket == "le_4kib")); + }); + }); + + startup_runtime_sources::set_metrics_enabled(previous_metrics); + startup_runtime_sources::set_get_stage_metrics_enabled(previous_get_stages); + startup_runtime_sources::set_get_stage_local_summary_enabled(previous_local_summary); + startup_runtime_sources::set_get_stage_local_summary_sample_rate(previous_sample_rate); + } #[test] #[serial_test::serial] diff --git a/rustfs/src/startup_runtime_sources.rs b/rustfs/src/startup_runtime_sources.rs index 90304d00f..4fd87dd7d 100644 --- a/rustfs/src/startup_runtime_sources.rs +++ b/rustfs/src/startup_runtime_sources.rs @@ -98,6 +98,14 @@ pub(crate) fn set_get_stage_metrics_enabled(enabled: bool) { rustfs_io_metrics::set_get_stage_metrics_enabled(enabled); } +pub(crate) fn set_get_stage_local_summary_enabled(enabled: bool) { + rustfs_io_metrics::set_get_stage_local_summary_enabled(enabled); +} + +pub(crate) fn set_get_stage_local_summary_sample_rate(sample_rate: u64) { + rustfs_io_metrics::set_get_stage_local_summary_sample_rate(sample_rate); +} + pub(crate) fn set_metrics_enabled(enabled: bool) { rustfs_io_metrics::set_metrics_enabled(enabled); }