fix(obs): add proxied PUT replication metrics (#3020)

fix(obs): add proxied put replication metrics

crates/obs/src/metrics/collectors/bucket_replication.rs

@@ -22,12 +22,15 @@ use crate::metrics::schema::bucket_replication::{
     BUCKET_REPL_PROXIED_DELETE_TAGGING_REQUESTS_TOTAL_MD, BUCKET_REPL_PROXIED_GET_REQUESTS_FAILURES_MD,
     BUCKET_REPL_PROXIED_GET_REQUESTS_TOTAL_MD, BUCKET_REPL_PROXIED_GET_TAGGING_REQUESTS_FAILURES_MD,
     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_TAGGING_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,
 };
 use std::borrow::Cow;
 
+const BASE_BUCKET_REPLICATION_METRICS_PER_BUCKET: usize = 20;
+
 #[derive(Debug, Clone, Default)]
 pub struct BucketReplicationTargetStats {
     pub target_arn: String,
@@ -59,6 +62,8 @@ pub struct BucketReplicationStats {
     pub proxied_get_requests_failures: u64,
     pub proxied_head_requests_total: u64,
     pub proxied_head_requests_failures: u64,
+    pub proxied_put_requests_total: u64,
+    pub proxied_put_requests_failures: u64,
     pub proxied_put_tagging_requests_total: u64,
     pub proxied_put_tagging_requests_failures: u64,
     pub proxied_get_tagging_requests_total: u64,
@@ -99,7 +104,11 @@ pub fn collect_bucket_replication_metrics(stats: &[BucketReplicationStats]) -> V
         return Vec::new();
     }
 
-    let mut metrics = Vec::new();
+    let metric_count = stats
+        .iter()
+        .map(|stat| BASE_BUCKET_REPLICATION_METRICS_PER_BUCKET + stat.targets.len())
+        .sum();
+    let mut metrics = Vec::with_capacity(metric_count);
     for stat in stats {
         let bucket_label: Cow<'static, str> = Cow::Owned(stat.bucket.clone());
 
@@ -160,6 +169,17 @@ pub fn collect_bucket_replication_metrics(stats: &[BucketReplicationStats]) -> V
             )
             .with_label(BUCKET_L, bucket_label.clone()),
         );
+        metrics.push(
+            PrometheusMetric::from_descriptor(&BUCKET_REPL_PROXIED_PUT_REQUESTS_TOTAL_MD, stat.proxied_put_requests_total as f64)
+                .with_label(BUCKET_L, bucket_label.clone()),
+        );
+        metrics.push(
+            PrometheusMetric::from_descriptor(
+                &BUCKET_REPL_PROXIED_PUT_REQUESTS_FAILURES_MD,
+                stat.proxied_put_requests_failures as f64,
+            )
+            .with_label(BUCKET_L, bucket_label.clone()),
+        );
         metrics.push(
             PrometheusMetric::from_descriptor(
                 &BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_TOTAL_MD,
@@ -238,6 +258,8 @@ mod tests {
             proxied_get_requests_failures: 1,
             proxied_head_requests_total: 4,
             proxied_head_requests_failures: 0,
+            proxied_put_requests_total: 6,
+            proxied_put_requests_failures: 2,
             proxied_put_tagging_requests_total: 3,
             proxied_put_tagging_requests_failures: 1,
             proxied_get_tagging_requests_total: 2,
@@ -253,7 +275,7 @@ mod tests {
         }];
 
         let metrics = collect_bucket_replication_metrics(&stats);
-        assert_eq!(metrics.len(), 19);
+        assert_eq!(metrics.len(), 21);
 
         let sent_name = BUCKET_REPL_SENT_COUNT_MD.get_full_metric_name();
         assert!(metrics.iter().any(|metric| {
@@ -262,6 +284,20 @@ mod tests {
                 && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
         }));
 
+        let put_total_name = BUCKET_REPL_PROXIED_PUT_REQUESTS_TOTAL_MD.get_full_metric_name();
+        assert!(metrics.iter().any(|metric| {
+            metric.name == put_total_name
+                && metric.value == 6.0
+                && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
+        }));
+
+        let put_failures_name = BUCKET_REPL_PROXIED_PUT_REQUESTS_FAILURES_MD.get_full_metric_name();
+        assert!(metrics.iter().any(|metric| {
+            metric.name == put_failures_name
+                && metric.value == 2.0
+                && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
+        }));
+
         let latency_name = BUCKET_REPL_LATENCY_MS_MD.get_full_metric_name();
         assert!(metrics.iter().any(|metric| {
             metric.name == latency_name

crates/obs/src/metrics/schema/bucket_replication.rs

@@ -26,6 +26,9 @@ pub const TARGET_ARN_L: &str = "target_arn";
 /// Replication range
 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";
+
 pub static BUCKET_REPL_LAST_HR_FAILED_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_gauge_md(
         MetricName::LastHourFailedBytes,
@@ -143,6 +146,24 @@ pub static BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_FAILURES_MD: LazyLock<Metric
     )
 });
 
+pub static BUCKET_REPL_PROXIED_PUT_REQUESTS_FAILURES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::from(PROXIED_PUT_REQUESTS_FAILURES),
+        "Number of failures in PUT requests proxied to replication target",
+        &[BUCKET_L],
+        subsystems::BUCKET_REPLICATION,
+    )
+});
+
+pub static BUCKET_REPL_PROXIED_PUT_REQUESTS_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::from(PROXIED_PUT_REQUESTS_TOTAL),
+        "Number of PUT requests proxied to replication target",
+        &[BUCKET_L],
+        subsystems::BUCKET_REPLICATION,
+    )
+});
+
 pub static BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_counter_md(
         MetricName::ProxiedPutTaggingRequestsTotal,

crates/obs/src/metrics/stats_collector.rs

@@ -355,6 +355,8 @@ pub async fn collect_bucket_replication_detail_stats() -> Vec<BucketReplicationS
             proxied_get_requests_failures: proxy.get_failed.max(0) as u64,
             proxied_head_requests_total: proxy.head_total.max(0) as u64,
             proxied_head_requests_failures: proxy.head_failed.max(0) as u64,
+            proxied_put_requests_total: proxy.put_total.max(0) as u64,
+            proxied_put_requests_failures: proxy.put_failed.max(0) as u64,
             proxied_put_tagging_requests_total: proxy.put_tag_total.max(0) as u64,
             proxied_put_tagging_requests_failures: proxy.put_tag_failed.max(0) as u64,
             proxied_get_tagging_requests_total: proxy.get_tag_total.max(0) as u64,