feat(observability): add sampled local GetObject stage summaries

Add an opt-in GetObject stage aggregation path that works without the
OTLP metrics export. When RUSTFS_OBS_GET_STAGE_LOCAL_SUMMARY_ENABLED is
set, io-metrics keeps per-(path, stage, object class, size bucket)
sampled counts and duration totals in a bounded in-process map, and a
startup task flushes them periodically to
/var/log/rustfs/get-stage-summary.log, tunable via the sample rate and
interval env keys.
This commit is contained in:
Hauser
2026-09-29 18:34:19 +08:00
parent cb78742727
commit 6586da38fc
4 changed files with 218 additions and 2 deletions
+14
View File
@@ -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");
+93 -1
View File
@@ -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<Mutex<HashMap<GetStageLocalSummaryKey, GetStageLocalSummary>>> =
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<GetStageLocalSummary> {
let Ok(mut summary) = GET_STAGE_LOCAL_SUMMARY.lock() else {
return Vec::new();
};
let mut rows = summary.drain().map(|(_, row)| row).collect::<Vec<_>>();
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;
}
+103 -1
View File
@@ -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<ECStore>, ctx: CancellationToken) {
print_server_info();
init_update_check();
@@ -27,6 +31,10 @@ pub(crate) async fn init_observability_runtime(store: Arc<ECStore>, 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<ECStore>, 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::<Vec<_>>().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]
+8
View File
@@ -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);
}