From df988cfd8610a858a781b82354879c087dd2b2d5 Mon Sep 17 00:00:00 2001 From: Hauser Date: Thu, 8 Oct 2026 02:12:48 +0800 Subject: [PATCH] fix(obs): export bucket gauges from live snapshots (#8399) * fix(obs): export bucket gauges from live snapshots Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(obs): collapse snapshot update condition Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: zhi22915 --- crates/obs/src/metrics/scheduler.rs | 236 +++------------------ crates/obs/src/metrics/storage_snapshot.rs | 155 ++++++++++---- crates/obs/src/telemetry/mod.rs | 1 + crates/obs/src/telemetry/recorder.rs | 4 + 4 files changed, 149 insertions(+), 247 deletions(-) diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index e82ee8e31..ba3791fbc 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -110,9 +110,7 @@ use crate::metrics::schema::bucket_replication::{ use crate::metrics::schema::cluster_usage::{ BUCKET_LABEL as USAGE_BUCKET_LABEL, RANGE_LABEL as USAGE_RANGE_LABEL, USAGE_BUCKET_DELETE_MARKERS_COUNT_MD, USAGE_BUCKET_OBJECT_SIZE_DISTRIBUTION_MD, USAGE_BUCKET_OBJECT_VERSION_COUNT_DISTRIBUTION_MD, USAGE_BUCKET_OBJECTS_TOTAL_MD, - USAGE_BUCKET_QUOTA_TOTAL_BYTES_MD, USAGE_BUCKET_TOTAL_BYTES_MD, USAGE_BUCKET_VERSIONS_COUNT_MD, USAGE_BUCKETS_COUNT_MD, - USAGE_DELETE_MARKERS_COUNT_MD, USAGE_OBJECTS_COUNT_MD, USAGE_OBJECTS_DISTRIBUTION_MD, USAGE_SINCE_LAST_UPDATE_SECONDS_MD, - USAGE_TOTAL_BYTES_MD, USAGE_VERSIONS_COUNT_MD, USAGE_VERSIONS_DISTRIBUTION_MD, + USAGE_BUCKET_QUOTA_TOTAL_BYTES_MD, USAGE_BUCKET_TOTAL_BYTES_MD, USAGE_BUCKET_VERSIONS_COUNT_MD, }; use crate::metrics::schema::node_bucket::{BUCKET_OBJECTS_TOTAL_MD, BUCKET_QUOTA_BYTES_MD, BUCKET_USAGE_BYTES_MD}; use crate::metrics::schema::notification_target::{ @@ -150,7 +148,7 @@ use crate::metrics::stats_collector::{ }; use crate::metrics::storage_snapshot::StorageSnapshotMetrics; use crate::node_identity::{SERVER_LABEL, current_local_node_identity}; -use crate::telemetry::retire_metric_series; +use crate::telemetry::{process_global_meter, retire_metric_series}; use futures_util::FutureExt; use rustfs_audit::audit_target_metrics; use rustfs_config::METER_INTERVAL; @@ -731,6 +729,17 @@ fn storage_snapshot_metrics(scope: &'static str, collection_interval: Duration) ) } +fn bucket_metrics_snapshot(collection_interval: Duration) -> StorageSnapshotMetrics { + let export_interval = Duration::from_secs( + get_env_opt_u64(ENV_OBS_METER_INTERVAL) + .filter(|value| *value > 0) + .unwrap_or(METER_INTERVAL), + ); + let max_age = collection_interval.max(export_interval).saturating_mul(3); + let meter = process_global_meter().unwrap_or_else(|| opentelemetry::global::meter("rustfs.storage")); + StorageSnapshotMetrics::new_unscoped(meter, max_age) +} + fn metrics_interval(period: Duration, initial_delay: Duration) -> Interval { let mut interval = tokio::time::interval_at(Instant::now() + initial_delay, period); interval.set_missed_tick_behavior(MissedTickBehavior::Delay); @@ -850,26 +859,6 @@ fn bucket_live_keys(stats: &[crate::metrics::collectors::BucketStats]) -> HashSe stats.iter().map(|stat| stat.name.clone()).collect() } -fn bucket_observation_live_keys(stats: &[crate::metrics::collectors::BucketStats]) -> HashSet { - stats - .iter() - .filter(|stat| stat.size_bytes.is_some() || stat.objects_count.is_some()) - .map(|stat| stat.name.clone()) - .collect() -} - -fn bucket_observation_retire_keys( - previous_observations: &HashSet, - current_buckets: &HashSet, - current_observations: &HashSet, -) -> Vec { - previous_observations - .difference(current_observations) - .filter(|bucket| current_buckets.contains(*bucket)) - .cloned() - .collect() -} - fn collect_bucket_zero_tombstone_metrics(zero_tombstones: &HashMap) -> Vec { if zero_tombstones.is_empty() { return Vec::new(); @@ -892,14 +881,11 @@ fn collect_bucket_zero_tombstone_metrics(zero_tombstones: &HashMap, - observation_keys: HashSet, zero_tombstones: HashMap, } struct BucketSeriesUpdate { metrics: Vec, - retire_observations: Vec, - retire_buckets: Vec, } impl BucketSeriesState { @@ -910,10 +896,6 @@ impl BucketSeriesState { ) -> Option { let stats = stats?; let current_bucket_keys = bucket_live_keys(stats); - let current_observation_keys = bucket_observation_live_keys(stats); - let retire_observations = - bucket_observation_retire_keys(&self.observation_keys, ¤t_bucket_keys, ¤t_observation_keys); - self.observation_keys = current_observation_keys; update_series_zero_tombstones( &mut self.has_seen_snapshot, &mut self.live_keys, @@ -923,49 +905,11 @@ impl BucketSeriesState { ); let mut metrics = collect_bucket_metrics(stats); metrics.extend(collect_bucket_zero_tombstone_metrics(&self.zero_tombstones)); - let retire_buckets = expire_series_zero_tombstones(&mut self.zero_tombstones); - Some(BucketSeriesUpdate { - metrics, - retire_observations, - retire_buckets, - }) + let _ = expire_series_zero_tombstones(&mut self.zero_tombstones); + Some(BucketSeriesUpdate { metrics }) } } -fn retire_bucket_metric_series(bucket: &str) -> usize { - let bucket_label: Cow<'static, str> = Cow::Owned(bucket.to_string()); - let labels = [("bucket", bucket_label.clone())]; - retire_metric_series(&BUCKET_USAGE_BYTES_MD.get_full_metric_name(), &labels) - + retire_metric_series(&BUCKET_OBJECTS_TOTAL_MD.get_full_metric_name(), &labels) - + retire_metric_series(&BUCKET_QUOTA_BYTES_MD.get_full_metric_name(), &labels) -} - -fn retire_bucket_observation_metric_series(bucket: &str) -> usize { - let labels = [("bucket", Cow::Owned(bucket.to_string()))]; - retire_metric_series(&BUCKET_USAGE_BYTES_MD.get_full_metric_name(), &labels) - + retire_metric_series(&BUCKET_OBJECTS_TOTAL_MD.get_full_metric_name(), &labels) -} - -fn retire_cluster_usage_metric_series() -> usize { - let labels: [(&'static str, Cow<'static, str>); 0] = []; - [ - USAGE_SINCE_LAST_UPDATE_SECONDS_MD.get_full_metric_name(), - USAGE_TOTAL_BYTES_MD.get_full_metric_name(), - USAGE_OBJECTS_COUNT_MD.get_full_metric_name(), - USAGE_VERSIONS_COUNT_MD.get_full_metric_name(), - USAGE_DELETE_MARKERS_COUNT_MD.get_full_metric_name(), - USAGE_BUCKETS_COUNT_MD.get_full_metric_name(), - ] - .iter() - .map(|name| retire_metric_series(name, &labels)) - .sum() -} - -fn retire_cluster_usage_distribution_series(metric_name: String, range: &str) -> usize { - let labels = [(USAGE_RANGE_LABEL, Cow::Owned(range.to_string()))]; - retire_metric_series(&metric_name, &labels) -} - fn bucket_usage_live_keys(stats: &[crate::metrics::collectors::BucketUsageStats]) -> HashSet { stats.iter().map(|stat| stat.bucket.clone()).collect() } @@ -1043,24 +987,6 @@ fn collect_bucket_usage_zero_tombstone_metrics( zero_metrics } -fn retire_bucket_usage_metric_series(bucket: &str) -> usize { - let bucket_label: Cow<'static, str> = Cow::Owned(bucket.to_string()); - let labels = [(USAGE_BUCKET_LABEL, bucket_label.clone())]; - retire_metric_series(&USAGE_BUCKET_TOTAL_BYTES_MD.get_full_metric_name(), &labels) - + retire_metric_series(&USAGE_BUCKET_OBJECTS_TOTAL_MD.get_full_metric_name(), &labels) - + retire_metric_series(&USAGE_BUCKET_VERSIONS_COUNT_MD.get_full_metric_name(), &labels) - + retire_metric_series(&USAGE_BUCKET_DELETE_MARKERS_COUNT_MD.get_full_metric_name(), &labels) - + retire_metric_series(&USAGE_BUCKET_QUOTA_TOTAL_BYTES_MD.get_full_metric_name(), &labels) -} - -fn retire_bucket_usage_distribution_series(metric_name: String, bucket: &str, range: &str) -> usize { - let labels = [ - (USAGE_RANGE_LABEL, Cow::Owned(range.to_string())), - (USAGE_BUCKET_LABEL, Cow::Owned(bucket.to_string())), - ]; - retire_metric_series(&metric_name, &labels) -} - fn audit_target_live_keys(stats: &[AuditTargetRuntimeStats]) -> HashSet { stats .iter() @@ -1687,14 +1613,13 @@ pub fn init_metrics_runtime(token: CancellationToken) { let mut interval = metrics_interval(cluster_interval, stagger_duration(cluster_interval, 1, 3)); let tombstone_cycles = config.replication_bandwidth_zero_tombstone_cycles; let mut has_seen_bucket_usage_snapshot = false; + let mut bucket_usage_snapshot = bucket_metrics_snapshot(cluster_interval); let mut prev_bucket_usage_keys: HashSet = HashSet::new(); let mut bucket_usage_zero_tombstones: HashMap = HashMap::new(); let mut prev_bucket_usage_object_size_keys: HashSet = HashSet::new(); let mut bucket_usage_object_size_zero_tombstones: HashMap = HashMap::new(); let mut prev_bucket_usage_version_keys: HashSet = HashSet::new(); let mut bucket_usage_version_zero_tombstones: HashMap = HashMap::new(); - let mut prev_cluster_usage_object_size_keys: HashSet = HashSet::new(); - let mut prev_cluster_usage_version_keys: HashSet = HashSet::new(); loop { tokio::select! { _ = interval.tick() => { @@ -1704,6 +1629,7 @@ pub fn init_metrics_runtime(token: CancellationToken) { "supplementary_cluster_stats", async { let mut metrics = Vec::new(); + let mut usage_metrics_update = None; if let Some(stats) = collect_cluster_config_stats().await { metrics.extend(collect_cluster_config_metrics(&stats)); @@ -1713,32 +1639,10 @@ pub fn init_metrics_runtime(token: CancellationToken) { metrics.extend(collect_iam_metrics(&stats)); } + let usage_collection_started = std::time::Instant::now(); if let Some((cluster_usage, bucket_usage)) = collect_cluster_usage_metric_stats().await { - let current_cluster_usage_object_size_keys = cluster_usage - .object_size_distribution - .iter() - .map(|(range, _)| range.clone()) - .collect::>(); - for range in prev_cluster_usage_object_size_keys.difference(¤t_cluster_usage_object_size_keys) { - let _ = retire_cluster_usage_distribution_series( - USAGE_OBJECTS_DISTRIBUTION_MD.get_full_metric_name(), - range, - ); - } - prev_cluster_usage_object_size_keys = current_cluster_usage_object_size_keys; - let current_cluster_usage_version_keys = cluster_usage - .versions_distribution - .iter() - .map(|(range, _)| range.clone()) - .collect::>(); - for range in prev_cluster_usage_version_keys.difference(¤t_cluster_usage_version_keys) { - let _ = retire_cluster_usage_distribution_series( - USAGE_VERSIONS_DISTRIBUTION_MD.get_full_metric_name(), - range, - ); - } - prev_cluster_usage_version_keys = current_cluster_usage_version_keys; - metrics.extend(collect_cluster_usage_metrics(&cluster_usage)); + let mut usage_metrics = collect_cluster_usage_metrics(&cluster_usage); + usage_metrics.extend(collect_bucket_usage_metrics(&bucket_usage)); update_series_zero_tombstones( &mut has_seen_bucket_usage_snapshot, &mut prev_bucket_usage_keys, @@ -1760,64 +1664,21 @@ pub fn init_metrics_runtime(token: CancellationToken) { bucket_usage_version_live_keys(&bucket_usage), tombstone_cycles, ); - metrics.extend(collect_bucket_usage_metrics(&bucket_usage)); - metrics.extend(collect_bucket_usage_zero_tombstone_metrics( + usage_metrics.extend(collect_bucket_usage_zero_tombstone_metrics( &bucket_usage_zero_tombstones, &bucket_usage_object_size_zero_tombstones, &bucket_usage_version_zero_tombstones, )); - for bucket in expire_series_zero_tombstones(&mut bucket_usage_zero_tombstones) { - let _ = retire_bucket_usage_metric_series(&bucket); - } - for (bucket, range) in expire_series_zero_tombstones(&mut bucket_usage_object_size_zero_tombstones) { - let _ = retire_bucket_usage_distribution_series( - USAGE_BUCKET_OBJECT_SIZE_DISTRIBUTION_MD.get_full_metric_name(), - &bucket, - &range, - ); - } - for (bucket, range) in expire_series_zero_tombstones(&mut bucket_usage_version_zero_tombstones) { - let _ = retire_bucket_usage_distribution_series( - USAGE_BUCKET_OBJECT_VERSION_COUNT_DISTRIBUTION_MD.get_full_metric_name(), - &bucket, - &range, - ); - } - } else if has_seen_bucket_usage_snapshot { - let _ = retire_cluster_usage_metric_series(); - for range in prev_cluster_usage_object_size_keys.drain() { - let _ = retire_cluster_usage_distribution_series( - USAGE_OBJECTS_DISTRIBUTION_MD.get_full_metric_name(), - &range, - ); - } - for range in prev_cluster_usage_version_keys.drain() { - let _ = retire_cluster_usage_distribution_series( - USAGE_VERSIONS_DISTRIBUTION_MD.get_full_metric_name(), - &range, - ); - } - for bucket in prev_bucket_usage_keys.drain() { - let _ = retire_bucket_usage_metric_series(&bucket); - } - for (bucket, range) in prev_bucket_usage_object_size_keys.drain() { - let _ = retire_bucket_usage_distribution_series( - USAGE_BUCKET_OBJECT_SIZE_DISTRIBUTION_MD.get_full_metric_name(), - &bucket, - &range, - ); - } - for (bucket, range) in prev_bucket_usage_version_keys.drain() { - let _ = retire_bucket_usage_distribution_series( - USAGE_BUCKET_OBJECT_VERSION_COUNT_DISTRIBUTION_MD.get_full_metric_name(), - &bucket, - &range, - ); - } - bucket_usage_zero_tombstones.clear(); - bucket_usage_object_size_zero_tombstones.clear(); - bucket_usage_version_zero_tombstones.clear(); - has_seen_bucket_usage_snapshot = false; + let _ = expire_series_zero_tombstones(&mut bucket_usage_zero_tombstones); + let _ = expire_series_zero_tombstones(&mut bucket_usage_object_size_zero_tombstones); + let _ = expire_series_zero_tombstones(&mut bucket_usage_version_zero_tombstones); + usage_metrics_update = Some(usage_metrics); + } + + if let Some(usage_metrics) = usage_metrics_update + && let Err(error) = bucket_usage_snapshot.replace_collected(usage_metrics, usage_collection_started) + { + error!(event = EVENT_METRICS_RUNTIME_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_RUNTIME, collector = "supplementary_cluster_stats", result = "invalid_bucket_snapshot", error = %error, "bucket usage metrics snapshot rejected"); } if !metrics.is_empty() { @@ -1840,20 +1701,18 @@ pub fn init_metrics_runtime(token: CancellationToken) { let mut interval = metrics_interval(bucket_interval, Duration::ZERO); let tombstone_cycles = config.replication_bandwidth_zero_tombstone_cycles; let mut series_state = BucketSeriesState::default(); + let mut snapshot = bucket_metrics_snapshot(bucket_interval); loop { tokio::select! { _ = interval.tick() => { run_metrics_collector_tick(health, MetricsCollectorTaskId::BucketStats, "bucket_stats", async { + let collection_started = std::time::Instant::now(); let stats = collect_bucket_stats().await; let Some(update) = series_state.observe(stats.as_deref(), tombstone_cycles) else { return; }; - for bucket in update.retire_observations { - let _ = retire_bucket_observation_metric_series(&bucket); - } - report_metrics(&update.metrics); - for bucket in update.retire_buckets { - let _ = retire_bucket_metric_series(&bucket); + if let Err(error) = snapshot.replace_collected(update.metrics, collection_started) { + error!(event = EVENT_METRICS_RUNTIME_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_RUNTIME, collector = "bucket_stats", result = "invalid_snapshot", error = %error, "bucket metrics snapshot rejected"); } }).await; } @@ -3485,29 +3344,6 @@ mod tests { assert!(zero_tombstones.is_empty()); } - #[test] - fn bucket_observation_retirement_distinguishes_unknown_usage_from_deletion() { - let previous = HashSet::from(["bucket".to_string()]); - let unknown_stats = vec![crate::metrics::collectors::BucketStats { - name: "bucket".to_string(), - size_bytes: None, - objects_count: None, - quota_bytes: 1024, - }]; - let current_buckets = bucket_live_keys(&unknown_stats); - let current_observations = bucket_observation_live_keys(&unknown_stats); - - assert_eq!( - bucket_observation_retire_keys(&previous, ¤t_buckets, ¤t_observations), - vec!["bucket".to_string()], - "an existing bucket with unknown usage must retire its previous usage observations" - ); - assert!( - bucket_observation_retire_keys(&previous, &HashSet::new(), &HashSet::new()).is_empty(), - "a deleted bucket remains governed by the zero-tombstone lifecycle" - ); - } - #[test] fn unavailable_bucket_snapshot_preserves_metric_series_state() { let mut state = BucketSeriesState::default(); @@ -3520,12 +3356,10 @@ mod tests { assert!(state.observe(Some(&initial), 2).is_some()); let live_keys = state.live_keys.clone(); - let observation_keys = state.observation_keys.clone(); let zero_tombstones = state.zero_tombstones.clone(); assert!(state.observe(None, 2).is_none()); assert_eq!(state.live_keys, live_keys); - assert_eq!(state.observation_keys, observation_keys); assert_eq!(state.zero_tombstones, zero_tombstones); } diff --git a/crates/obs/src/metrics/storage_snapshot.rs b/crates/obs/src/metrics/storage_snapshot.rs index 4ba5f612a..8977403e9 100644 --- a/crates/obs/src/metrics/storage_snapshot.rs +++ b/crates/obs/src/metrics/storage_snapshot.rs @@ -60,34 +60,46 @@ pub(crate) struct StorageSnapshotMetrics { impl StorageSnapshotMetrics { pub(crate) fn new(meter: Meter, scope: &'static str, observer: String, max_age: Duration) -> Self { - let snapshot = Arc::new(RwLock::new(Snapshot::default())); let labels = vec![KeyValue::new(COLLECTION_SCOPE, scope), KeyValue::new(OBSERVER, observer)]; - let weak = Arc::downgrade(&snapshot); - let timestamp_labels = labels.clone(); - meter - .f64_observable_gauge(LAST_SUCCESS) - .with_description("Unix timestamp of the last successful storage snapshot collection") - .with_callback(move |observer| { - let Some(snapshot) = weak.upgrade() else { return }; - let Ok(snapshot) = snapshot.read() else { return }; - if snapshot.collected_at.is_some() { - observer.observe(snapshot.timestamp_seconds, ×tamp_labels); - } - }) - .build(); - let weak = Arc::downgrade(&snapshot); - let age_labels = labels.clone(); - meter - .f64_observable_gauge(MAX_AGE) - .with_description("Validity budget in seconds remaining after storage snapshot collection completed") - .with_callback(move |observer| { - let Some(snapshot) = weak.upgrade() else { return }; - let Ok(snapshot) = snapshot.read() else { return }; - if snapshot.collected_at.is_some() { - observer.observe(snapshot.remaining_age.as_secs_f64(), &age_labels); - } - }) - .build(); + Self::new_with_labels(meter, labels, max_age, true) + } + + /// Creates an unscoped snapshot for dynamic metric series whose membership + /// must match the latest authoritative collection. + pub(crate) fn new_unscoped(meter: Meter, max_age: Duration) -> Self { + Self::new_with_labels(meter, Vec::new(), max_age, false) + } + + fn new_with_labels(meter: Meter, labels: Vec, max_age: Duration, publish_freshness: bool) -> Self { + let snapshot = Arc::new(RwLock::new(Snapshot::default())); + if publish_freshness { + let weak = Arc::downgrade(&snapshot); + let timestamp_labels = labels.clone(); + meter + .f64_observable_gauge(LAST_SUCCESS) + .with_description("Unix timestamp of the last successful storage snapshot collection") + .with_callback(move |observer| { + let Some(snapshot) = weak.upgrade() else { return }; + let Ok(snapshot) = snapshot.read() else { return }; + if snapshot.collected_at.is_some() { + observer.observe(snapshot.timestamp_seconds, ×tamp_labels); + } + }) + .build(); + let weak = Arc::downgrade(&snapshot); + let age_labels = labels.clone(); + meter + .f64_observable_gauge(MAX_AGE) + .with_description("Validity budget in seconds remaining after storage snapshot collection completed") + .with_callback(move |observer| { + let Some(snapshot) = weak.upgrade() else { return }; + let Ok(snapshot) = snapshot.read() else { return }; + if snapshot.collected_at.is_some() { + observer.observe(snapshot.remaining_age.as_secs_f64(), &age_labels); + } + }) + .build(); + } Self { meter, snapshot, @@ -221,6 +233,7 @@ mod tests { #[derive(Clone, Debug)] struct ExportedPoint { + scope: String, name: String, value: f64, attributes: Vec, @@ -233,28 +246,33 @@ mod tests { async fn export(&self, metrics: &ResourceMetrics) -> OTelSdkResult { let mut exported = self.0.lock().unwrap(); exported.clear(); - for metric in metrics.scope_metrics().flat_map(|scope| scope.metrics()) { - match metric.data() { - AggregatedMetrics::F64(MetricData::Gauge(gauge)) => { - for point in gauge.data_points() { - exported.push(ExportedPoint { - name: metric.name().to_string(), - value: point.value(), - attributes: point.attributes().cloned().collect(), - }); + for scope in metrics.scope_metrics() { + let scope_name = scope.scope().name().to_string(); + for metric in scope.metrics() { + match metric.data() { + AggregatedMetrics::F64(MetricData::Gauge(gauge)) => { + for point in gauge.data_points() { + exported.push(ExportedPoint { + scope: scope_name.clone(), + name: metric.name().to_string(), + value: point.value(), + attributes: point.attributes().cloned().collect(), + }); + } } - } - AggregatedMetrics::U64(MetricData::Sum(sum)) => { - assert!(sum.is_monotonic()); - for point in sum.data_points() { - exported.push(ExportedPoint { - name: metric.name().to_string(), - value: point.value() as f64, - attributes: point.attributes().cloned().collect(), - }); + AggregatedMetrics::U64(MetricData::Sum(sum)) => { + assert!(sum.is_monotonic()); + for point in sum.data_points() { + exported.push(ExportedPoint { + scope: scope_name.clone(), + name: metric.name().to_string(), + value: point.value() as f64, + attributes: point.attributes().cloned().collect(), + }); + } } + data => panic!("unexpected storage aggregation: {data:?}"), } - data => panic!("unexpected storage aggregation: {data:?}"), } } Ok(()) @@ -282,6 +300,16 @@ mod tests { (provider, exporter, snapshot) } + fn setup_unscoped() -> (SdkMeterProvider, Exporter, StorageSnapshotMetrics) { + let exporter = Exporter::default(); + let reader = PeriodicReader::builder(exporter.clone()) + .with_interval(Duration::from_secs(3600)) + .build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let snapshot = StorageSnapshotMetrics::new_unscoped(provider.meter("service-test"), Duration::from_secs(30)); + (provider, exporter, snapshot) + } + fn drive(name: &'static str, kind: MetricType, id: &'static str, value: f64) -> PrometheusMetric { PrometheusMetric::new(name, kind, "test", value) .with_label("server", "node1:9000") @@ -326,6 +354,41 @@ mod tests { provider.shutdown().unwrap(); } + #[test] + fn actual_sdk_stops_exporting_deleted_bucket_quota_series_after_tombstone() { + let (provider, exporter, mut snapshot) = setup_unscoped(); + let bucket = "deleted-bucket"; + let quota_metrics = |value| { + vec![ + PrometheusMetric::from_descriptor(&super::super::schema::node_bucket::BUCKET_QUOTA_BYTES_MD, value) + .with_label("bucket", bucket), + PrometheusMetric::from_descriptor(&super::super::schema::cluster_usage::USAGE_BUCKET_QUOTA_TOTAL_BYTES_MD, value) + .with_label("bucket", bucket), + ] + }; + + snapshot.replace(quota_metrics(1024.0)).unwrap(); + assert_eq!( + flush(&provider, &exporter) + .iter() + .filter(|point| point.value == 1024.0) + .count(), + 2 + ); + assert!(flush(&provider, &exporter).iter().all(|point| point.scope == "service-test")); + + snapshot.replace(quota_metrics(0.0)).unwrap(); + let tombstones = flush(&provider, &exporter); + assert_eq!(tombstones.iter().filter(|point| point.value == 0.0).count(), 2); + + snapshot.replace(Vec::new()).unwrap(); + for _ in 0..2 { + let points = flush(&provider, &exporter); + assert!(points.is_empty(), "deleted bucket quota points must not remain in cumulative exports"); + } + provider.shutdown().unwrap(); + } + #[test] fn actual_sdk_preserves_counter_resets_instead_of_accumulating_old_absolute_values() { let (provider, exporter, mut snapshot) = setup(); diff --git a/crates/obs/src/telemetry/mod.rs b/crates/obs/src/telemetry/mod.rs index 051c52324..8ca372246 100644 --- a/crates/obs/src/telemetry/mod.rs +++ b/crates/obs/src/telemetry/mod.rs @@ -53,6 +53,7 @@ use crate::TelemetryError; use crate::config::OtelConfig; pub use filter::HTTP_SERVER_LOG_TARGET; pub use guard::OtelGuard; +pub(crate) use recorder::process_global_meter; pub use recorder::{Recorder, retire_metric_series}; use rustfs_config::observability::ENV_OBS_LOG_DIRECTORY; use rustfs_config::{DEFAULT_LOG_LEVEL, ENVIRONMENT, observability::DEFAULT_OBS_ENVIRONMENT_PRODUCTION}; diff --git a/crates/obs/src/telemetry/recorder.rs b/crates/obs/src/telemetry/recorder.rs index d8b9e51fe..eb40bfd05 100644 --- a/crates/obs/src/telemetry/recorder.rs +++ b/crates/obs/src/telemetry/recorder.rs @@ -226,6 +226,10 @@ fn remember_global_recorder(recorder: &Recorder) { let _ = GLOBAL_RECORDER.set(recorder.clone()); } +pub(crate) fn process_global_meter() -> Option { + GLOBAL_RECORDER.get().map(|recorder| recorder.meter.clone()) +} + pub(crate) fn install_process_global_recorder(recorder: Recorder) -> Result<(), metrics::SetRecorderError> { metrics::set_global_recorder(recorder.clone())?; remember_global_recorder(&recorder);