From 29dedcc7fd430325207668eef891bbbaa1648995 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 1 Aug 2026 21:21:41 +0800 Subject: [PATCH] fix(obs): retire replication backlog metric series (#5574) Emit zero tombstones for removed bucket-level replication backlog series and retire the corresponding metric descriptors after the configured tombstone window, matching the existing replication bandwidth lifecycle. Co-authored-by: heihutu --- crates/obs/src/metrics/scheduler.rs | 243 +++++++++++++++++++++++++++- 1 file changed, 242 insertions(+), 1 deletion(-) diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index 41b6a75a2..a17430204 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -25,6 +25,7 @@ use crate::metrics::collectors::{ AuditTargetStats, + BucketReplicationBacklogStats, BucketReplicationBandwidthStats, NotificationStats, NotificationTargetStats, @@ -78,7 +79,11 @@ use crate::metrics::report::{PrometheusMetric, report_metrics}; use crate::metrics::runtime_sources::bucket_monitor_available; use crate::metrics::schema::audit::{AUDIT_FAILED_MESSAGES_MD, AUDIT_TARGET_QUEUE_LENGTH_MD, AUDIT_TOTAL_MESSAGES_MD}; use crate::metrics::schema::bucket_replication::{ - BUCKET_L, BUCKET_REPL_BANDWIDTH_CURRENT_MD, BUCKET_REPL_BANDWIDTH_LIMIT_MD, TARGET_ARN_L, + BUCKET_L, BUCKET_REPL_BANDWIDTH_CURRENT_MD, BUCKET_REPL_BANDWIDTH_LIMIT_MD, BUCKET_REPL_CURRENT_BACKLOG_BYTES_MD, + BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD, BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD, BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD, + BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD, BUCKET_REPL_MRF_DROPPED_COUNT_MD, BUCKET_REPL_MRF_FLUSH_FAILURES_MD, + BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD, BUCKET_REPL_MRF_MISSED_COUNT_MD, BUCKET_REPL_MRF_PENDING_BYTES_MD, + BUCKET_REPL_MRF_PENDING_COUNT_MD, TARGET_ARN_L, }; use crate::metrics::schema::cluster::{CLUSTER_BUCKETS_TOTAL_MD, CLUSTER_OBJECTS_TOTAL_MD}; use crate::metrics::schema::cluster_usage::{ @@ -624,6 +629,10 @@ fn repl_bw_live_keys(stats: &[BucketReplicationBandwidthStats]) -> HashSet HashSet { + stats.iter().map(|s| s.bucket.clone()).collect() +} + fn update_series_zero_tombstones( has_seen_valid_snapshot: &mut bool, prev_live_keys: &mut HashSet, @@ -986,6 +995,30 @@ fn update_repl_bw_zero_tombstones( *has_seen_valid_snapshot = true; } +fn update_repl_backlog_zero_tombstones( + monitor_available: bool, + has_seen_valid_snapshot: &mut bool, + prev_live_keys: &mut HashSet, + zero_tombstones: &mut HashMap, + current_live_keys: HashSet, + tombstone_cycles: u8, +) { + if !monitor_available { + for key in ¤t_live_keys { + zero_tombstones.remove(key); + } + return; + } + + update_series_zero_tombstones( + has_seen_valid_snapshot, + prev_live_keys, + zero_tombstones, + current_live_keys, + tombstone_cycles, + ); +} + fn collect_repl_bw_zero_tombstone_metrics(zero_tombstones: &HashMap) -> Vec { if zero_tombstones.is_empty() { return Vec::new(); @@ -1012,6 +1045,21 @@ fn collect_repl_bw_zero_tombstone_metrics(zero_tombstones: &HashMap) -> Vec { + if zero_tombstones.is_empty() { + return Vec::new(); + } + + let stats = zero_tombstones + .keys() + .map(|bucket| BucketReplicationBacklogStats { + bucket: bucket.clone(), + ..Default::default() + }) + .collect::>(); + collect_bucket_replication_backlog_metrics(&stats) +} + fn retire_repl_bw_metric_series(bucket: &str, target_arn: &str) -> usize { let labels = [ (BUCKET_L, Cow::Owned(bucket.to_string())), @@ -1021,6 +1069,26 @@ fn retire_repl_bw_metric_series(bucket: &str, target_arn: &str) -> usize { + retire_metric_series(&BUCKET_REPL_BANDWIDTH_CURRENT_MD.get_full_metric_name(), &labels) } +fn retire_repl_backlog_metric_series(bucket: &str) -> usize { + let labels = [(BUCKET_L, Cow::Owned(bucket.to_string()))]; + [ + BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_CURRENT_BACKLOG_BYTES_MD.get_full_metric_name(), + BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD.get_full_metric_name(), + BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD.get_full_metric_name(), + BUCKET_REPL_MRF_PENDING_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_MRF_PENDING_BYTES_MD.get_full_metric_name(), + BUCKET_REPL_MRF_DROPPED_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_MRF_MISSED_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_MRF_FLUSH_FAILURES_MD.get_full_metric_name(), + BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD.get_full_metric_name(), + ] + .iter() + .map(|name| retire_metric_series(name, &labels)) + .sum() +} + fn expire_repl_bw_zero_tombstones(monitor_available: bool, zero_tombstones: &mut HashMap) -> Vec { if !monitor_available { return Vec::new(); @@ -1029,6 +1097,14 @@ fn expire_repl_bw_zero_tombstones(monitor_available: bool, zero_tombstones: &mut expire_series_zero_tombstones(zero_tombstones) } +fn expire_repl_backlog_zero_tombstones(monitor_available: bool, zero_tombstones: &mut HashMap) -> Vec { + if !monitor_available { + return Vec::new(); + } + + expire_series_zero_tombstones(zero_tombstones) +} + /// Initialize all metrics collectors. /// /// This function spawns background tasks that periodically collect metrics @@ -1320,6 +1396,9 @@ pub fn init_metrics_runtime(token: CancellationToken) { let mut prev_live_keys: HashSet = HashSet::new(); let mut zero_tombstones: HashMap = HashMap::new(); let mut has_seen_valid_snapshot = false; + let mut prev_backlog_live_keys: HashSet = HashSet::new(); + let mut backlog_zero_tombstones: HashMap = HashMap::new(); + let mut has_seen_valid_backlog_snapshot = false; loop { tokio::select! { _ = interval.tick() => { @@ -1350,8 +1429,17 @@ pub fn init_metrics_runtime(token: CancellationToken) { metrics.extend(collect_repl_bw_zero_tombstone_metrics(&zero_tombstones)); let (bucket_replication, bucket_replication_backlog) = collect_bucket_replication_stats_bundle().await; + update_repl_backlog_zero_tombstones( + monitor_available, + &mut has_seen_valid_backlog_snapshot, + &mut prev_backlog_live_keys, + &mut backlog_zero_tombstones, + repl_backlog_live_keys(&bucket_replication_backlog), + repl_bw_zero_tombstone_cycles, + ); metrics.extend(collect_bucket_replication_metrics(&bucket_replication)); metrics.extend(collect_bucket_replication_backlog_metrics(&bucket_replication_backlog)); + metrics.extend(collect_repl_backlog_zero_tombstone_metrics(&backlog_zero_tombstones)); let replication = collect_replication_stats().await; metrics.extend(collect_replication_metrics(&replication)); report_metrics(&metrics); @@ -1360,6 +1448,9 @@ pub fn init_metrics_runtime(token: CancellationToken) { for (bucket, target_arn) in expire_repl_bw_zero_tombstones(monitor_available, &mut zero_tombstones) { let _ = retire_repl_bw_metric_series(&bucket, &target_arn); } + for bucket in expire_repl_backlog_zero_tombstones(monitor_available, &mut backlog_zero_tombstones) { + let _ = retire_repl_backlog_metric_series(&bucket); + } }, ).await; } @@ -1795,6 +1886,14 @@ mod tests { .collect() } + fn bucket_key(bucket: &str) -> BucketKey { + bucket.to_string() + } + + fn bucket_keys(keys: &[&str]) -> HashSet { + keys.iter().map(|bucket| bucket_key(bucket)).collect() + } + #[test] fn metrics_runtime_status_reports_disabled_state() { let snapshot = build_metrics_runtime_status_snapshot(false, false, fixed_metrics_runtime_config(), false); @@ -2018,6 +2117,148 @@ mod tests { assert_eq!(zero_tombstones.get(&repl_bw_key("videos", "arn:rustfs:replication:target-b")), Some(&1)); } + #[test] + fn repl_backlog_tombstones_zero_removed_buckets_then_expire() { + let mut has_seen_valid_snapshot = false; + let mut prev_live_keys = HashSet::new(); + let mut zero_tombstones = HashMap::new(); + let key = bucket_key("photos"); + + update_repl_backlog_zero_tombstones( + true, + &mut has_seen_valid_snapshot, + &mut prev_live_keys, + &mut zero_tombstones, + bucket_keys(&["photos"]), + 2, + ); + assert!(has_seen_valid_snapshot); + assert_eq!(prev_live_keys, bucket_keys(&["photos"])); + assert!(zero_tombstones.is_empty()); + + update_repl_backlog_zero_tombstones( + true, + &mut has_seen_valid_snapshot, + &mut prev_live_keys, + &mut zero_tombstones, + HashSet::new(), + 2, + ); + assert_eq!(zero_tombstones.get(&key), Some(&2)); + + let metrics = collect_repl_backlog_zero_tombstone_metrics(&zero_tombstones); + assert_eq!(metrics.len(), 11); + + let expected_names = HashSet::from([ + BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_CURRENT_BACKLOG_BYTES_MD.get_full_metric_name(), + BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD.get_full_metric_name(), + BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD.get_full_metric_name(), + BUCKET_REPL_MRF_PENDING_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_MRF_PENDING_BYTES_MD.get_full_metric_name(), + BUCKET_REPL_MRF_DROPPED_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_MRF_MISSED_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_MRF_FLUSH_FAILURES_MD.get_full_metric_name(), + BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD.get_full_metric_name(), + ]); + let mut actual_names = HashSet::new(); + for metric in metrics { + actual_names.insert(metric.name.to_string()); + assert_eq!(metric.value, 0.0); + + let labels = metric + .labels + .into_iter() + .map(|(key, value)| (key, value.to_string())) + .collect::>(); + assert_eq!(labels.get(BUCKET_L).map(String::as_str), Some("photos")); + } + assert_eq!(actual_names, expected_names); + + let expired = expire_repl_backlog_zero_tombstones(true, &mut zero_tombstones); + assert!(expired.is_empty()); + assert_eq!(zero_tombstones.get(&key), Some(&1)); + + let expired = expire_repl_backlog_zero_tombstones(true, &mut zero_tombstones); + assert_eq!(expired, vec![key]); + assert!(zero_tombstones.is_empty()); + } + + #[test] + fn repl_backlog_tombstones_stop_zeroing_when_bucket_becomes_live_again() { + let mut has_seen_valid_snapshot = false; + let mut prev_live_keys = HashSet::new(); + let mut zero_tombstones = HashMap::new(); + let live_keys = bucket_keys(&["photos"]); + + update_repl_backlog_zero_tombstones( + true, + &mut has_seen_valid_snapshot, + &mut prev_live_keys, + &mut zero_tombstones, + live_keys.clone(), + 3, + ); + update_repl_backlog_zero_tombstones( + true, + &mut has_seen_valid_snapshot, + &mut prev_live_keys, + &mut zero_tombstones, + HashSet::new(), + 3, + ); + assert_eq!(zero_tombstones.get(&bucket_key("photos")), Some(&3)); + + update_repl_backlog_zero_tombstones( + true, + &mut has_seen_valid_snapshot, + &mut prev_live_keys, + &mut zero_tombstones, + live_keys.clone(), + 3, + ); + + assert!(zero_tombstones.is_empty()); + assert_eq!(prev_live_keys, live_keys); + } + + #[test] + fn repl_backlog_tombstones_do_not_advance_when_monitor_unavailable() { + let mut has_seen_valid_snapshot = true; + let mut prev_live_keys = bucket_keys(&["photos"]); + let mut zero_tombstones = HashMap::from([(bucket_key("videos"), 1)]); + + update_repl_backlog_zero_tombstones( + false, + &mut has_seen_valid_snapshot, + &mut prev_live_keys, + &mut zero_tombstones, + HashSet::new(), + 3, + ); + + assert!(has_seen_valid_snapshot); + assert_eq!(prev_live_keys, bucket_keys(&["photos"])); + assert_eq!(zero_tombstones.get(&bucket_key("videos")), Some(&1)); + + let expired = expire_repl_backlog_zero_tombstones(false, &mut zero_tombstones); + assert!(expired.is_empty()); + assert_eq!(zero_tombstones.get(&bucket_key("videos")), Some(&1)); + + update_repl_backlog_zero_tombstones( + false, + &mut has_seen_valid_snapshot, + &mut prev_live_keys, + &mut zero_tombstones, + bucket_keys(&["videos"]), + 3, + ); + + assert!(zero_tombstones.is_empty()); + assert_eq!(prev_live_keys, bucket_keys(&["photos"])); + } + #[test] fn bucket_tombstones_zero_removed_buckets_then_expire() { let mut has_seen_valid_snapshot = false;