From 9a7255540b7d9c98667f633cc40a29ae5db1de43 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 8 Jul 2026 15:01:28 +0800 Subject: [PATCH] fix(replication): add resync metrics (#4408) --- .../replication/replication_resyncer.rs | 50 ++++++++++- .../bucket/replication/replication_state.rs | 87 +++++++++++++++++++ .../metrics/collectors/bucket_replication.rs | 54 +++++++++++- .../src/metrics/schema/bucket_replication.rs | 50 +++++++++++ crates/obs/src/metrics/stats_collector.rs | 5 ++ crates/obs/src/metrics/storage_api.rs | 10 +++ crates/replication/src/stats.rs | 48 +++++++++- 7 files changed, 295 insertions(+), 9 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index fba2fba4e..74deade9e 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -214,6 +214,31 @@ async fn head_object_fallback( static RESYNC_WORKER_COUNT: usize = 10; +fn resync_status_duration( + status: ResyncStatusType, + start_time: Option, + now: OffsetDateTime, +) -> Option { + if !matches!( + status, + ResyncStatusType::ResyncCompleted | ResyncStatusType::ResyncFailed | ResyncStatusType::ResyncCanceled + ) { + return None; + } + + let millis = (now - start_time?).whole_milliseconds(); + if millis < 0 { + return None; + } + + let millis = if millis > i128::from(u64::MAX) { + u64::MAX + } else { + u64::try_from(millis).ok()? + }; + Some(std::time::Duration::from_millis(millis)) +} + #[derive(Debug)] pub struct ReplicationResyncer { pub status_map: Arc>>, @@ -252,7 +277,7 @@ impl ReplicationResyncer { where S: ReplicationObjectIO, { - let bucket_status = { + let (bucket_status, status_duration) = { let mut status_map = self.status_map.write().await; let now = OffsetDateTime::now_utc(); @@ -305,13 +330,17 @@ impl ReplicationResyncer { } state.resync_status = status; state.last_update = Some(now); + let status_duration = resync_status_duration(status, state.start_time, now); bucket_status.last_update = Some(now); - bucket_status.clone() + (bucket_status.clone(), status_duration) }; save_resync_status(&opts.bucket, &bucket_status, obj_layer).await?; + if let Some(stats) = runtime_sources::replication_stats() { + stats.record_resync_status(&opts.bucket, status, status_duration).await; + } Ok(()) } @@ -453,7 +482,6 @@ impl ReplicationResyncer { "Failed to update resync status" ); } - // TODO: Metrics } #[instrument(skip(cancellation_token, storage))] @@ -3616,4 +3644,20 @@ mod tests { assert!(resync_state_accepts_update(¤t, &matching)); assert!(!resync_state_accepts_update(¤t, &stale)); } + + #[test] + fn test_resync_status_duration_only_tracks_terminal_status() { + let start = match OffsetDateTime::from_unix_timestamp(1_700_000_000) { + Ok(start) => start, + Err(err) => panic!("valid test timestamp: {err}"), + }; + let end = start + time::Duration::seconds(2); + + assert_eq!( + resync_status_duration(ResyncStatusType::ResyncCompleted, Some(start), end), + Some(std::time::Duration::from_millis(2000)) + ); + assert_eq!(resync_status_duration(ResyncStatusType::ResyncStarted, Some(start), end), None); + assert_eq!(resync_status_duration(ResyncStatusType::ResyncFailed, None, end), None); + } } diff --git a/crates/ecstore/src/bucket/replication/replication_state.rs b/crates/ecstore/src/bucket/replication/replication_state.rs index 7799a6f4e..968c8eb86 100644 --- a/crates/ecstore/src/bucket/replication/replication_state.rs +++ b/crates/ecstore/src/bucket/replication/replication_state.rs @@ -14,6 +14,7 @@ use super::replication_error_boundary::Error; use super::replication_filemeta_boundary::{ReplicatedTargetInfo, ReplicationStatusType, ReplicationType}; +use super::replication_resync_boundary::ResyncStatusType; use super::replication_stats_boundary::{ ActiveWorkerStat, BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, ProxyMetric, ProxyStatsCache, QueueCache, SRMetricsSummary, XferStats, @@ -252,6 +253,12 @@ impl ReplicationStats { self.sr_stats.replica_count.fetch_add(1, Ordering::Relaxed); } + pub async fn record_resync_status(&self, bucket: &str, status: ResyncStatusType, duration: Option) { + let mut cache = self.cache.write().await; + let stats = cache.entry(bucket.to_string()).or_insert_with(BucketReplicationStats::new); + stats.record_resync_status(status, duration); + } + /// Site replication update replica statistics fn sr_update_replica_stat(&self, size: i64) { self.sr_stats.replica_size.fetch_add(size, Ordering::Relaxed); @@ -489,6 +496,26 @@ impl ReplicationStats { replica_count: tot_replica_count, replicated_size: tot_replicated_size, replicated_count: tot_replicated_count, + resync_started_count: bucket_stats + .iter() + .map(|stats| stats.replication_stats.resync_started_count) + .sum(), + resync_completed_count: bucket_stats + .iter() + .map(|stats| stats.replication_stats.resync_completed_count) + .sum(), + resync_failed_count: bucket_stats + .iter() + .map(|stats| stats.replication_stats.resync_failed_count) + .sum(), + resync_canceled_count: bucket_stats + .iter() + .map(|stats| stats.replication_stats.resync_canceled_count) + .sum(), + resync_duration_ms: bucket_stats + .iter() + .map(|stats| stats.replication_stats.resync_duration_ms) + .sum(), }; let qs = Default::default(); @@ -641,6 +668,29 @@ mod tests { assert_eq!(bucket_stats.replica_count, 1); } + #[tokio::test] + async fn test_record_resync_status_updates_bucket_stats() { + let stats = ReplicationStats::new(); + + stats + .record_resync_status("test-bucket", ResyncStatusType::ResyncStarted, None) + .await; + stats + .record_resync_status("test-bucket", ResyncStatusType::ResyncCompleted, Some(Duration::from_millis(1500))) + .await; + stats + .record_resync_status("test-bucket", ResyncStatusType::ResyncPending, Some(Duration::from_millis(500))) + .await; + + let bucket_stats = stats.get("test-bucket").await; + assert_eq!(bucket_stats.resync_started_count, 1); + assert_eq!(bucket_stats.resync_completed_count, 1); + assert_eq!(bucket_stats.resync_failed_count, 0); + assert_eq!(bucket_stats.resync_canceled_count, 0); + assert_eq!(bucket_stats.resync_duration_ms, 1500); + assert!(bucket_stats.has_replication_usage()); + } + #[tokio::test] async fn test_replication_stats_update() { let stats = ReplicationStats::new(); @@ -683,6 +733,43 @@ mod tests { assert!(all.contains_key("proxy-only-bucket")); } + #[tokio::test] + async fn test_calculate_bucket_replication_stats_merges_resync_metrics() { + let stats = ReplicationStats::new(); + let got = stats + .calculate_bucket_replication_stats( + "test-bucket", + vec![ + BucketStats { + replication_stats: BucketReplicationStats { + resync_started_count: 1, + resync_completed_count: 1, + resync_duration_ms: 1000, + ..Default::default() + }, + ..Default::default() + }, + BucketStats { + replication_stats: BucketReplicationStats { + resync_started_count: 2, + resync_failed_count: 1, + resync_canceled_count: 1, + resync_duration_ms: 2500, + ..Default::default() + }, + ..Default::default() + }, + ], + ) + .await; + + assert_eq!(got.replication_stats.resync_started_count, 3); + assert_eq!(got.replication_stats.resync_completed_count, 1); + assert_eq!(got.replication_stats.resync_failed_count, 1); + assert_eq!(got.replication_stats.resync_canceled_count, 1); + assert_eq!(got.replication_stats.resync_duration_ms, 3500); + } + #[test] fn test_sr_stats() { let sr_stats = SRStats::new(); diff --git a/crates/obs/src/metrics/collectors/bucket_replication.rs b/crates/obs/src/metrics/collectors/bucket_replication.rs index de763f252..6e777c802 100644 --- a/crates/obs/src/metrics/collectors/bucket_replication.rs +++ b/crates/obs/src/metrics/collectors/bucket_replication.rs @@ -24,12 +24,14 @@ use crate::metrics::schema::bucket_replication::{ BUCKET_REPL_PROXIED_GET_TAGGING_REQUESTS_TOTAL_MD, BUCKET_REPL_PROXIED_HEAD_REQUESTS_FAILURES_MD, BUCKET_REPL_PROXIED_HEAD_REQUESTS_TOTAL_MD, BUCKET_REPL_PROXIED_PUT_REQUESTS_FAILURES_MD, BUCKET_REPL_PROXIED_PUT_REQUESTS_TOTAL_MD, BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_FAILURES_MD, - BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_TOTAL_MD, BUCKET_REPL_SENT_BYTES_MD, BUCKET_REPL_SENT_COUNT_MD, - BUCKET_REPL_TOTAL_FAILED_BYTES_MD, BUCKET_REPL_TOTAL_FAILED_COUNT_MD, OPERATION_L, RANGE_L, TARGET_ARN_L, + BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_TOTAL_MD, BUCKET_REPL_RESYNC_CANCELED_TOTAL_MD, + BUCKET_REPL_RESYNC_COMPLETED_TOTAL_MD, BUCKET_REPL_RESYNC_DURATION_MS_TOTAL_MD, BUCKET_REPL_RESYNC_FAILED_TOTAL_MD, + BUCKET_REPL_RESYNC_STARTED_TOTAL_MD, BUCKET_REPL_SENT_BYTES_MD, BUCKET_REPL_SENT_COUNT_MD, BUCKET_REPL_TOTAL_FAILED_BYTES_MD, + BUCKET_REPL_TOTAL_FAILED_COUNT_MD, OPERATION_L, RANGE_L, TARGET_ARN_L, }; use std::borrow::Cow; -const BASE_BUCKET_REPLICATION_METRICS_PER_BUCKET: usize = 20; +const BASE_BUCKET_REPLICATION_METRICS_PER_BUCKET: usize = 25; #[derive(Debug, Clone, Default)] pub struct BucketReplicationTargetStats { @@ -70,6 +72,11 @@ pub struct BucketReplicationStats { pub proxied_get_tagging_requests_failures: u64, pub proxied_delete_tagging_requests_total: u64, pub proxied_delete_tagging_requests_failures: u64, + pub resync_started_count: u64, + pub resync_completed_count: u64, + pub resync_failed_count: u64, + pub resync_canceled_count: u64, + pub resync_duration_ms: u64, pub targets: Vec, } @@ -222,6 +229,26 @@ pub fn collect_bucket_replication_metrics(stats: &[BucketReplicationStats]) -> V ) .with_label(BUCKET_L, bucket_label.clone()), ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_RESYNC_STARTED_TOTAL_MD, stat.resync_started_count as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_RESYNC_COMPLETED_TOTAL_MD, stat.resync_completed_count as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_RESYNC_FAILED_TOTAL_MD, stat.resync_failed_count as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_RESYNC_CANCELED_TOTAL_MD, stat.resync_canceled_count as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_RESYNC_DURATION_MS_TOTAL_MD, stat.resync_duration_ms as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); for target in &stat.targets { let target_label: Cow<'static, str> = Cow::Owned(target.target_arn.clone()); @@ -266,6 +293,11 @@ mod tests { proxied_get_tagging_requests_failures: 0, proxied_delete_tagging_requests_total: 1, proxied_delete_tagging_requests_failures: 1, + resync_started_count: 2, + resync_completed_count: 1, + resync_failed_count: 1, + resync_canceled_count: 0, + resync_duration_ms: 1500, targets: vec![BucketReplicationTargetStats { target_arn: "arn:rustfs:replication:us-east-1:1:target".to_string(), bandwidth_limit_bytes_per_sec: 2048, @@ -275,7 +307,7 @@ mod tests { }]; let metrics = collect_bucket_replication_metrics(&stats); - assert_eq!(metrics.len(), 21); + assert_eq!(metrics.len(), 26); let sent_name = BUCKET_REPL_SENT_COUNT_MD.get_full_metric_name(); assert!(metrics.iter().any(|metric| { @@ -321,6 +353,20 @@ mod tests { && metric.value == 1.0 && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") })); + + let resync_started_name = BUCKET_REPL_RESYNC_STARTED_TOTAL_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == resync_started_name + && metric.value == 2.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let resync_duration_name = BUCKET_REPL_RESYNC_DURATION_MS_TOTAL_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == resync_duration_name + && metric.value == 1500.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); } #[test] diff --git a/crates/obs/src/metrics/schema/bucket_replication.rs b/crates/obs/src/metrics/schema/bucket_replication.rs index fed8afcde..1178327b5 100644 --- a/crates/obs/src/metrics/schema/bucket_replication.rs +++ b/crates/obs/src/metrics/schema/bucket_replication.rs @@ -28,6 +28,11 @@ pub const RANGE_L: &str = "range"; const PROXIED_PUT_REQUESTS_TOTAL: &str = "proxied_put_requests_total"; const PROXIED_PUT_REQUESTS_FAILURES: &str = "proxied_put_requests_failures"; +const RESYNC_STARTED_TOTAL: &str = "resync_started_total"; +const RESYNC_COMPLETED_TOTAL: &str = "resync_completed_total"; +const RESYNC_FAILED_TOTAL: &str = "resync_failed_total"; +const RESYNC_CANCELED_TOTAL: &str = "resync_canceled_total"; +const RESYNC_DURATION_MS_TOTAL: &str = "resync_duration_ms_total"; pub static BUCKET_REPL_LAST_HR_FAILED_BYTES_MD: LazyLock = LazyLock::new(|| { new_gauge_md( @@ -191,6 +196,51 @@ pub static BUCKET_REPL_SENT_COUNT_MD: LazyLock = LazyLock::new ) }); +pub static BUCKET_REPL_RESYNC_STARTED_TOTAL_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(RESYNC_STARTED_TOTAL), + "Total number of bucket replication resync runs started", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_RESYNC_COMPLETED_TOTAL_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(RESYNC_COMPLETED_TOTAL), + "Total number of bucket replication resync runs completed", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_RESYNC_FAILED_TOTAL_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(RESYNC_FAILED_TOTAL), + "Total number of bucket replication resync runs failed", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_RESYNC_CANCELED_TOTAL_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(RESYNC_CANCELED_TOTAL), + "Total number of bucket replication resync runs canceled", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_RESYNC_DURATION_MS_TOTAL_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(RESYNC_DURATION_MS_TOTAL), + "Total elapsed time of finished bucket replication resync runs in milliseconds", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + pub static BUCKET_REPL_TOTAL_FAILED_BYTES_MD: LazyLock = LazyLock::new(|| { new_counter_md( MetricName::TotalFailedBytes, diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index e715acfd7..dacf19af8 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -174,6 +174,11 @@ async fn obs_bucket_replication_detail_stats() -> Vec { proxied_get_tagging_requests_failures: stats.proxied_get_tagging_requests_failures, proxied_delete_tagging_requests_total: stats.proxied_delete_tagging_requests_total, proxied_delete_tagging_requests_failures: stats.proxied_delete_tagging_requests_failures, + resync_started_count: stats.resync_started_count, + resync_completed_count: stats.resync_completed_count, + resync_failed_count: stats.resync_failed_count, + resync_canceled_count: stats.resync_canceled_count, + resync_duration_ms: stats.resync_duration_ms, targets: stats .targets .into_iter() diff --git a/crates/obs/src/metrics/storage_api.rs b/crates/obs/src/metrics/storage_api.rs index 1b28f2e72..040b34cee 100644 --- a/crates/obs/src/metrics/storage_api.rs +++ b/crates/obs/src/metrics/storage_api.rs @@ -63,6 +63,11 @@ pub(crate) struct ObsBucketReplicationStatsSnapshot { pub(crate) proxied_get_tagging_requests_failures: u64, pub(crate) proxied_delete_tagging_requests_total: u64, pub(crate) proxied_delete_tagging_requests_failures: u64, + pub(crate) resync_started_count: u64, + pub(crate) resync_completed_count: u64, + pub(crate) resync_failed_count: u64, + pub(crate) resync_canceled_count: u64, + pub(crate) resync_duration_ms: u64, pub(crate) targets: Vec, } @@ -156,6 +161,11 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec bool { - self.stats.is_empty() && self.replica_size == 0 && self.replicated_size == 0 + self.stats.is_empty() + && self.replica_size == 0 + && self.replicated_size == 0 + && self.resync_started_count == 0 + && self.resync_completed_count == 0 + && self.resync_failed_count == 0 + && self.resync_canceled_count == 0 + && self.resync_duration_ms == 0 } pub fn has_replication_usage(&self) -> bool { - self.replica_size > 0 || self.replicated_size > 0 || !self.stats.is_empty() + self.replica_size > 0 + || self.replicated_size > 0 + || self.resync_started_count > 0 + || self.resync_completed_count > 0 + || self.resync_failed_count > 0 + || self.resync_canceled_count > 0 + || self.resync_duration_ms > 0 + || !self.stats.is_empty() } pub fn clone_stats(&self) -> Self { self.clone() } + + pub fn record_resync_status(&mut self, status: ResyncStatusType, duration: Option) { + match status { + ResyncStatusType::ResyncStarted => self.resync_started_count += 1, + ResyncStatusType::ResyncCompleted => self.resync_completed_count += 1, + ResyncStatusType::ResyncFailed => self.resync_failed_count += 1, + ResyncStatusType::ResyncCanceled => self.resync_canceled_count += 1, + ResyncStatusType::NoResync | ResyncStatusType::ResyncPending => return, + } + + if let Some(duration) = duration { + let duration_ms = duration.as_millis().min(I64_MAX_AS_U128); + if let Ok(duration_ms) = i64::try_from(duration_ms) { + self.resync_duration_ms = self.resync_duration_ms.saturating_add(duration_ms); + } + } + } } #[derive(Debug, Clone, Default, Serialize, Deserialize)]