fix(replication): add resync metrics (#4408)

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<OffsetDateTime>,
+    now: OffsetDateTime,
+) -> Option<std::time::Duration> {
+    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<RwLock<HashMap<String, BucketReplicationResyncStatus>>>,
@@ -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(&current, &matching));
         assert!(!resync_state_accepts_update(&current, &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);
+    }
 }

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<Duration>) {
+        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();