From fbec33bd299bd4a72fae238e1b1a1ac9d4fe1b94 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 2 Aug 2026 01:31:50 +0800 Subject: [PATCH] Expose target-scoped durable MRF backlog metrics (#5584) * feat(replication): expose target durable mrf backlog Add target ARN attribution to durable MRF entries and surface target-scoped durable backlog metrics without changing existing bucket-only metric labels. Keep legacy MRF files bucket-only by defaulting missing targetARNs to an empty list, and expose target snapshots through an additive API so existing DurableMrfBacklogSummary callers remain source-compatible. Co-Authored-By: heihutu * feat(replication): expose runtime target backlog (#5586) Track runtime replication backlog by target ARN for regular, large, delete, and MRF admission paths while preserving the existing bucket-level backlog semantics. Add target-scoped current backlog metrics and merge them with durable target backlog snapshots for observability. Co-authored-by: heihutu --------- Co-authored-by: heihutu --- crates/ecstore/src/api/mod.rs | 19 +- crates/ecstore/src/bucket/replication/mod.rs | 2 +- .../bucket/replication/replication_pool.rs | 344 ++++++++++++++++-- .../bucket/replication/replication_state.rs | 231 +++++++++++- .../metrics/collectors/bucket_replication.rs | 116 +++++- crates/obs/src/metrics/collectors/mod.rs | 4 +- crates/obs/src/metrics/scheduler.rs | 158 +++++++- .../src/metrics/schema/bucket_replication.rs | 40 ++ crates/obs/src/metrics/stats_collector.rs | 21 +- crates/obs/src/metrics/storage_api.rs | 211 ++++++++--- crates/replication/src/delete.rs | 8 + crates/replication/src/filemeta.rs | 73 ++++ crates/replication/src/mrf.rs | 5 + rustfs/src/admin/handlers/replication.rs | 121 +++++- 14 files changed, 1241 insertions(+), 112 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 2393b3d16..5dd1d0c04 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -173,8 +173,9 @@ pub mod bucket { pub mod replication { pub use crate::bucket::replication::replication_pool::{ - DurableMrfBacklogSummary, DurableMrfBucketBacklog, MrfBacklogObservabilitySummary, MrfBucketBacklogObservability, - durable_mrf_backlog_summary_snapshot, mrf_backlog_observability_snapshot, + DurableMrfBacklogSummary, DurableMrfBucketBacklog, DurableMrfTargetBacklog, MrfBacklogObservabilitySummary, + MrfBucketBacklogObservability, durable_mrf_backlog_summary_snapshot, durable_mrf_target_backlog_snapshot, + mrf_backlog_observability_snapshot, }; pub use crate::bucket::replication::{ BucketReplicationResyncStatus, BucketStats, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, @@ -183,13 +184,13 @@ pub mod bucket { ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge, ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError, - ReplicationType, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, VersionPurgeStatusType, - delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool, - get_global_replication_stats, init_background_replication, read_durable_mrf_backlog, replication_state_to_filemeta, - replication_status_to_filemeta, replication_statuses_map, replication_target_arns, resync_start_conflict_id, - should_remove_replication_target, should_schedule_delete_replication, should_use_existing_delete_replication_info, - should_use_existing_delete_replication_source, unsupported_replication_config_field, - validate_replication_config_target_arns, version_purge_status_to_filemeta, + ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus, + VersionPurgeStatusType, delete_replication_state_from_config, delete_replication_version_id, + get_global_replication_pool, get_global_replication_stats, init_background_replication, read_durable_mrf_backlog, + replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, replication_target_arns, + resync_start_conflict_id, should_remove_replication_target, should_schedule_delete_replication, + should_use_existing_delete_replication_info, should_use_existing_delete_replication_source, + unsupported_replication_config_field, validate_replication_config_target_arns, version_purge_status_to_filemeta, }; } diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index eaa2e6ac4..1408d083c 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -78,7 +78,7 @@ pub use replication_queue_boundary::{ }; pub use replication_resync_boundary::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus}; pub use replication_scanner_bridge::ReplicationScannerBridge; -pub use replication_state::ReplicationStats; +pub use replication_state::{ReplicationStats, RuntimeReplicationTargetBacklog}; pub use replication_stats_boundary::BucketStats; pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage}; pub(crate) use replication_target_config_bridge::ReplicationTargetConfigBridge; diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 67afcc5f1..7dc058ff3 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -86,12 +86,26 @@ pub struct DurableMrfBucketBacklog { pub bytes: u64, } +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct DurableMrfTargetBacklog { + pub bucket: String, + pub target_arn: String, + pub count: u64, + pub bytes: u64, +} + #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct DurableMrfBacklogSummary { pub available: bool, pub buckets: Vec, } +#[derive(Debug, Clone, Default, PartialEq, Eq)] +struct DurableMrfBacklogSnapshot { + summary: DurableMrfBacklogSummary, + targets: Vec, +} + #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct MrfBucketBacklogObservability { pub bucket: String, @@ -110,6 +124,8 @@ pub struct MrfBacklogObservabilitySummary { static DURABLE_MRF_BACKLOG_SUMMARY: LazyLock> = LazyLock::new(|| StdRwLock::new(DurableMrfBacklogSummary::default())); +static DURABLE_MRF_TARGET_BACKLOG: LazyLock>> = + LazyLock::new(|| StdRwLock::new(Vec::new())); static MRF_BACKLOG_OBSERVABILITY: LazyLock> = LazyLock::new(|| StdRwLock::new(MrfBacklogObservabilityTracker::default())); @@ -117,13 +133,15 @@ static MRF_BACKLOG_OBSERVABILITY: LazyLock, + targets: HashMap<(String, String), DurableMrfTargetBacklog>, } impl DurableMrfBacklogTracker { - fn add_entry(&mut self, bucket_name: String, entry_size: i64) { - let Ok(size) = u64::try_from(entry_size) else { + fn add_entry(&mut self, entry: &MrfReplicateEntry) { + let Ok(size) = u64::try_from(entry.size) else { self.available = false; self.buckets.clear(); + self.targets.clear(); return; }; @@ -131,6 +149,7 @@ impl DurableMrfBacklogTracker { return; } + let bucket_name = entry.bucket.clone(); let bucket = match self.buckets.entry(bucket_name) { Entry::Occupied(entry) => entry.into_mut(), Entry::Vacant(entry) => { @@ -143,16 +162,39 @@ impl DurableMrfBacklogTracker { }; bucket.count = bucket.count.saturating_add(1); bucket.bytes = bucket.bytes.saturating_add(size); + + for target_arn in &entry.target_arns { + if target_arn.is_empty() { + continue; + } + let key = (entry.bucket.clone(), target_arn.clone()); + let target = match self.targets.entry(key) { + Entry::Occupied(entry) => entry.into_mut(), + Entry::Vacant(entry) => { + let (bucket, target_arn) = entry.key().clone(); + entry.insert(DurableMrfTargetBacklog { + bucket, + target_arn, + ..Default::default() + }) + } + }; + target.count = target.count.saturating_add(1); + target.bytes = target.bytes.saturating_add(size); + } } - fn into_summary(self) -> DurableMrfBacklogSummary { + fn into_snapshot(self) -> DurableMrfBacklogSnapshot { if !self.available { - return DurableMrfBacklogSummary::default(); + return DurableMrfBacklogSnapshot::default(); } - DurableMrfBacklogSummary { - available: true, - buckets: self.buckets.into_values().collect(), + DurableMrfBacklogSnapshot { + summary: DurableMrfBacklogSummary { + available: true, + buckets: self.buckets.into_values().collect(), + }, + targets: self.targets.into_values().collect(), } } } @@ -218,7 +260,21 @@ impl MrfBacklogObservabilityTracker { } } -fn durable_mrf_backlog_summary_from_sizes(entries: I) -> DurableMrfBacklogSummary +fn durable_mrf_backlog_summary_from_entries<'a>( + entries: impl IntoIterator, +) -> DurableMrfBacklogSnapshot { + let mut tracker = DurableMrfBacklogTracker { + available: true, + ..Default::default() + }; + for entry in entries { + tracker.add_entry(entry); + } + tracker.into_snapshot() +} + +#[cfg(test)] +fn durable_mrf_backlog_summary_from_sizes(entries: I) -> DurableMrfBacklogSnapshot where I: IntoIterator, { @@ -227,16 +283,38 @@ where ..Default::default() }; for (bucket_name, entry_size) in entries { - tracker.add_entry(bucket_name, entry_size); + tracker.add_entry(&MrfReplicateEntry { + bucket: bucket_name, + object: String::new(), + version_id: None, + retry_count: 0, + size: entry_size, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + target_arns: Vec::new(), + }); + } + tracker.into_snapshot() +} + +fn set_durable_mrf_backlog_snapshot(snapshot: DurableMrfBacklogSnapshot) { + match DURABLE_MRF_BACKLOG_SUMMARY.write() { + Ok(mut guard) => *guard = snapshot.summary, + Err(poisoned) => *poisoned.into_inner() = snapshot.summary, + } + match DURABLE_MRF_TARGET_BACKLOG.write() { + Ok(mut guard) => *guard = snapshot.targets, + Err(poisoned) => *poisoned.into_inner() = snapshot.targets, } - tracker.into_summary() } fn set_durable_mrf_backlog_summary(summary: DurableMrfBacklogSummary) { - match DURABLE_MRF_BACKLOG_SUMMARY.write() { - Ok(mut guard) => *guard = summary, - Err(poisoned) => *poisoned.into_inner() = summary, - } + set_durable_mrf_backlog_snapshot(DurableMrfBacklogSnapshot { + summary, + targets: Vec::new(), + }); } pub fn durable_mrf_backlog_summary_snapshot() -> DurableMrfBacklogSummary { @@ -246,6 +324,13 @@ pub fn durable_mrf_backlog_summary_snapshot() -> DurableMrfBacklogSummary { } } +pub fn durable_mrf_target_backlog_snapshot() -> Vec { + match DURABLE_MRF_TARGET_BACKLOG.read() { + Ok(guard) => guard.clone(), + Err(poisoned) => poisoned.into_inner().clone(), + } +} + pub fn mrf_backlog_observability_snapshot() -> MrfBacklogObservabilitySummary { match MRF_BACKLOG_OBSERVABILITY.read() { Ok(guard) => guard.snapshot(), @@ -675,6 +760,7 @@ impl ReplicationPool { /// Queues a replica task pub async fn queue_replica_task(&self, ri: ReplicateObjectInfo) -> ReplicationQueueAdmission { + let target_arns = ri.dsc.replicate_target_arns(); // If object is large, queue it to a static set of large workers if should_queue_large_object(ri.size) { use std::collections::hash_map::DefaultHasher; @@ -691,10 +777,12 @@ impl ReplicationPool { if let Some(worker) = lrg_workers.get(index) { self.stats.inc_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type); + self.stats.inc_target_q(&ri.bucket, &target_arns, ri.size); if worker.try_send(ReplicationOperation::Object(Box::new(ri.clone()))).is_ok() { return ReplicationQueueAdmission::Queued; } self.stats.dec_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type); + self.stats.dec_target_q(&ri.bucket, &target_arns, ri.size); // Try to add more workers if possible let max_l_workers = *self.max_l_workers.read().await; @@ -723,10 +811,12 @@ impl ReplicationPool { }; self.stats.inc_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type); + self.stats.inc_target_q(&ri.bucket, &target_arns, ri.size); if channel.try_send(ReplicationOperation::Object(Box::new(ri.clone()))).is_ok() { return ReplicationQueueAdmission::Queued; } self.stats.dec_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type); + self.stats.dec_target_q(&ri.bucket, &target_arns, ri.size); // Queue to MRF if all workers are busy. let admission = self.queue_mrf_save_admission(ri.to_mrf_entry(), "object").await; @@ -740,6 +830,11 @@ impl ReplicationPool { /// Queues a replica delete task pub async fn queue_replica_delete_task(&self, doi: DeletedObjectReplicationInfo) -> ReplicationQueueAdmission { + let target_arns = if doi.target_arn.is_empty() { + Vec::new() + } else { + vec![doi.target_arn.clone()] + }; let ch = self .worker_queue_channel(&doi.op_type, &doi.bucket, &doi.delete_object.object_name, 0) .await; @@ -749,10 +844,12 @@ impl ReplicationPool { }; self.stats.inc_q(&doi.bucket, 0, true, doi.op_type); + self.stats.inc_target_q(&doi.bucket, &target_arns, 0); if channel.try_send(ReplicationOperation::Delete(Box::new(doi.clone()))).is_ok() { return ReplicationQueueAdmission::Queued; } self.stats.dec_q(&doi.bucket, 0, true, doi.op_type); + self.stats.dec_target_q(&doi.bucket, &target_arns, 0); let admission = self.queue_mrf_save_admission(doi.to_mrf_entry(), "delete").await; @@ -771,9 +868,11 @@ impl ReplicationPool { let bucket = entry.bucket.clone(); let size = entry.size; let is_delete = matches!(entry.op, MrfOpKind::Delete); + let target_arns = entry.target_arns.clone(); let admission = queue_mrf_save_entry(&self.mrf_save_tx, entry, queue_type).await; if admission == ReplicationQueueAdmission::Queued { self.stats.inc_q(&bucket, size, is_delete, ReplicationType::Heal); + self.stats.inc_target_q(&bucket, &target_arns, size); } admission } @@ -827,9 +926,7 @@ impl ReplicationPool { return; } }; - set_durable_mrf_backlog_summary(durable_mrf_backlog_summary_from_sizes( - entries.iter().map(|entry| (entry.bucket.clone(), entry.size)), - )); + set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&entries)); let total = entries.len(); let mut queued_count = 0usize; @@ -1018,7 +1115,7 @@ impl ReplicationPool { } continue; } - durable_tracker.add_entry(e.bucket.clone(), e.size); + durable_tracker.add_entry(&e); observe_mrf_pending(&e); pending.push(e); dirty = true; @@ -1029,7 +1126,7 @@ impl ReplicationPool { if pending.len() - flushed_len >= 1000 && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await { - set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary()); + set_durable_mrf_backlog_snapshot(durable_tracker.clone().into_snapshot()); observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis); dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]); flushed_len = pending.len(); @@ -1039,7 +1136,7 @@ impl ReplicationPool { None => { // Channel closed (pool shutting down) — final flush. if dirty && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await { - set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary()); + set_durable_mrf_backlog_snapshot(durable_tracker.clone().into_snapshot()); observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis); dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]); } @@ -1048,7 +1145,7 @@ impl ReplicationPool { }, _ = interval.tick() => { if dirty && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await { - set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary()); + set_durable_mrf_backlog_snapshot(durable_tracker.clone().into_snapshot()); observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis); dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]); flushed_len = pending.len(); @@ -1451,6 +1548,7 @@ struct ReplicationBacklogGuard { size: i64, is_delete_marker: bool, op_type: ReplicationType, + target_arns: Vec, } impl ReplicationBacklogGuard { @@ -1461,16 +1559,23 @@ impl ReplicationBacklogGuard { size: object.size, is_delete_marker: object.delete_marker, op_type: object.op_type, + target_arns: object.dsc.replicate_target_arns(), } } fn for_delete(stats: Arc, delete: &DeletedObjectReplicationInfo) -> Self { + let target_arns = if delete.target_arn.is_empty() { + Vec::new() + } else { + vec![delete.target_arn.clone()] + }; Self { stats, bucket: delete.bucket.clone(), size: 0, is_delete_marker: true, op_type: delete.op_type, + target_arns, } } } @@ -1478,6 +1583,7 @@ impl ReplicationBacklogGuard { impl Drop for ReplicationBacklogGuard { fn drop(&mut self) { self.stats.dec_q(&self.bucket, self.size, self.is_delete_marker, self.op_type); + self.stats.dec_target_q(&self.bucket, &self.target_arns, self.size); } } @@ -1524,6 +1630,7 @@ async fn queue_mrf_save_entry( fn dec_mrf_entries(stats: &ReplicationStats, entries: &[MrfReplicateEntry]) { for entry in entries { stats.dec_q(&entry.bucket, entry.size, matches!(entry.op, MrfOpKind::Delete), ReplicationType::Heal); + stats.dec_target_q(&entry.bucket, &entry.target_arns, entry.size); } } @@ -1915,6 +2022,7 @@ async fn queue_replicate_deletes(batch: ReplicationHealResyncDeletes) -> Replica #[cfg(test)] mod tests { + use super::super::replication_filemeta_boundary::ReplicateTargetDecision; use super::super::replication_resync_boundary::{decode_mrf_file, encode_mrf_file, encode_resync_file}; use super::super::replication_storage_boundary::{ DeletedObject, FileInfo, GetObjectReader, HTTPRangeSpec, ListOperations, ObjectIO, ObjectOperations, PutObjReader, @@ -2260,6 +2368,22 @@ mod tests { (stats.replication_stats.q_stat.curr.count, stats.replication_stats.q_stat.curr.bytes) } + fn current_target_queue(pool: &ReplicationPool, bucket: &str, target_arn: &str) -> Option<(u64, u64)> { + pool.stats + .runtime_target_backlog_snapshot() + .into_iter() + .find(|target| target.bucket == bucket && target.target_arn == target_arn) + .map(|target| (target.count, target.bytes)) + } + + fn test_replicate_decision(target_arns: &[&str]) -> ReplicateDecision { + let mut decision = ReplicateDecision::default(); + for target_arn in target_arns { + decision.set(ReplicateTargetDecision::new((*target_arn).to_string(), true, false)); + } + decision + } + async fn wait_for_current_queue(pool: &ReplicationPool, bucket: &str, expected: (i64, i64)) { tokio::time::timeout(Duration::from_secs(10), async { loop { @@ -2293,6 +2417,35 @@ mod tests { assert_eq!(current_queue(&pool, "admission-bucket").await, (1, 4096)); } + #[tokio::test] + async fn regular_worker_admission_counts_target_backlog_before_receive() { + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await; + let (tx, _rx) = mpsc::channel(1); + pool.workers.write().await.push(tx); + + let admission = pool + .queue_replica_task(ReplicateObjectInfo { + bucket: "target-admission-bucket".to_string(), + name: "object".to_string(), + size: 4096, + op_type: ReplicationType::Object, + dsc: test_replicate_decision(&["arn:rustfs:replication:target-b", "arn:rustfs:replication:target-a"]), + ..Default::default() + }) + .await; + + assert_eq!(admission, ReplicationQueueAdmission::Queued); + assert_eq!(current_queue(&pool, "target-admission-bucket").await, (1, 4096)); + assert_eq!( + current_target_queue(&pool, "target-admission-bucket", "arn:rustfs:replication:target-a"), + Some((1, 4096)) + ); + assert_eq!( + current_target_queue(&pool, "target-admission-bucket", "arn:rustfs:replication:target-b"), + Some((1, 4096)) + ); + } + #[tokio::test] async fn large_worker_admission_counts_channel_backlog_before_receive() { let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await; @@ -2336,6 +2489,33 @@ mod tests { assert_eq!(current_queue(&pool, "delete-admission-bucket").await, (1, 0)); } + #[tokio::test] + async fn delete_admission_counts_target_backlog_before_receive() { + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await; + let (tx, _rx) = mpsc::channel(1); + pool.workers.write().await.push(tx); + + let admission = pool + .queue_replica_delete_task(DeletedObjectReplicationInfo { + bucket: "delete-target-admission-bucket".to_string(), + target_arn: "arn:rustfs:replication:target-a".to_string(), + delete_object: ReplicationDeletedObject { + object_name: "deleted-object".to_string(), + ..Default::default() + }, + op_type: ReplicationType::Delete, + ..Default::default() + }) + .await; + + assert_eq!(admission, ReplicationQueueAdmission::Queued); + assert_eq!(current_queue(&pool, "delete-target-admission-bucket").await, (1, 0)); + assert_eq!( + current_target_queue(&pool, "delete-target-admission-bucket", "arn:rustfs:replication:target-a"), + Some((1, 0)) + ); + } + #[tokio::test] async fn regular_worker_drains_current_backlog_after_processing() { let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await; @@ -2702,6 +2882,7 @@ mod tests { name: "fallback-object".to_string(), size: 2048, op_type: ReplicationType::Object, + dsc: test_replicate_decision(&["arn:rustfs:replication:target-a"]), ..Default::default() }) .await; @@ -2710,6 +2891,10 @@ mod tests { let queued = pool.stats.get_latest_replication_stats("runtime-backlog").await; assert_eq!(queued.replication_stats.q_stat.curr.count, 1); assert_eq!(queued.replication_stats.q_stat.curr.bytes, 2048); + assert_eq!( + current_target_queue(&pool, "runtime-backlog", "arn:rustfs:replication:target-a"), + Some((1, 2048)) + ); } #[test] @@ -2745,6 +2930,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: Vec::new(), }; let second = MrfReplicateEntry { object: "second".to_string(), @@ -2794,6 +2980,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: Vec::new(), }, "test", ) @@ -2824,6 +3011,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: Vec::new(), }; observe_mrf_pending(&entry); @@ -2845,12 +3033,14 @@ mod tests { async fn replication_backlog_guard_decrements_on_drop() { let stats = Arc::new(ReplicationStats::new()); stats.inc_q("guard-bucket", 256, false, ReplicationType::Object); + stats.inc_target_q("guard-bucket", &["arn:rustfs:replication:target-a".to_string()], 256); { let object = ReplicateObjectInfo { bucket: "guard-bucket".to_string(), size: 256, op_type: ReplicationType::Object, + dsc: test_replicate_decision(&["arn:rustfs:replication:target-a"]), ..Default::default() }; let _guard = ReplicationBacklogGuard::for_object(stats.clone(), &object); @@ -2859,6 +3049,30 @@ mod tests { let queued = stats.get_latest_replication_stats("guard-bucket").await; assert_eq!(queued.replication_stats.q_stat.curr.count, 0); assert_eq!(queued.replication_stats.q_stat.curr.bytes, 0); + assert!(stats.runtime_target_backlog_snapshot().is_empty()); + } + + #[test] + fn dec_mrf_entries_decrements_target_backlog() { + let stats = ReplicationStats::new(); + let entry = MrfReplicateEntry { + bucket: "mrf-target-drain-bucket".to_string(), + object: "object".to_string(), + version_id: None, + retry_count: 1, + size: 1024, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + target_arns: vec!["arn:rustfs:replication:target-a".to_string()], + }; + + stats.inc_q(&entry.bucket, entry.size, false, ReplicationType::Heal); + stats.inc_target_q(&entry.bucket, &entry.target_arns, entry.size); + dec_mrf_entries(&stats, std::slice::from_ref(&entry)); + + assert!(stats.runtime_target_backlog_snapshot().is_empty()); } #[test] @@ -2873,6 +3087,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: Vec::new(), }; let second = MrfReplicateEntry { object: "second".to_string(), @@ -2984,6 +3199,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: Vec::new(), }; let encoded = encode_mrf_file(std::slice::from_ref(&entry)).expect("encode"); @@ -3017,6 +3233,7 @@ mod tests { delete_marker_version_id: Some(dm_vid), delete_marker: true, delete_marker_mtime: Some(mtime_nanos), + target_arns: Vec::new(), }; let encoded = encode_mrf_file(std::slice::from_ref(&entry)).expect("encode"); @@ -3050,6 +3267,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: Vec::new(), }; let encoded = encode_mrf_file(&[entry]).expect("encode"); @@ -3078,6 +3296,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: Vec::new(), }, MrfReplicateEntry { bucket: "b".to_string(), @@ -3089,6 +3308,7 @@ mod tests { delete_marker_version_id: Some(del_dm_vid), delete_marker: true, delete_marker_mtime: None, + target_arns: Vec::new(), }, ]; @@ -3118,6 +3338,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: Vec::new(), }; assert_eq!(obj_entry.op, MrfOpKind::Object); @@ -3132,6 +3353,7 @@ mod tests { delete_marker_version_id: Some(Uuid::new_v4()), delete_marker: true, delete_marker_mtime: None, + target_arns: Vec::new(), }; assert_eq!(del_entry.op, MrfOpKind::Delete); @@ -3147,6 +3369,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: Vec::new(), }; assert_eq!(legacy_entry.op, MrfOpKind::Object, "legacy default must be Object"); } @@ -3196,6 +3419,7 @@ mod tests { // The "deleteMarkerMtime" key was absent in old files — #[serde(default)] must fill in // None so replay falls back to the current time (backlog#867 backward compatibility). assert_eq!(entry.delete_marker_mtime, None, "missing deleteMarkerMtime key must default to None"); + assert!(entry.target_arns.is_empty(), "old MRF entries must not be attributed to a target"); } #[test] @@ -3210,6 +3434,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: Vec::new(), }]; let encoded = encode_mrf_file(&entries).expect("durable MRF backlog should encode"); @@ -3226,9 +3451,10 @@ mod tests { #[test] fn durable_mrf_summary_aggregates_entries_by_bucket_for_obs() { - let summary = + let snapshot = durable_mrf_backlog_summary_from_sizes([("b1".to_string(), 1024), ("b1".to_string(), 512), ("b2".to_string(), 0)]); + let summary = snapshot.summary; assert!(summary.available); let buckets = summary .buckets @@ -3239,13 +3465,82 @@ mod tests { assert_eq!(buckets["b1"].bytes, 1536); assert_eq!(buckets["b2"].count, 1); assert_eq!(buckets["b2"].bytes, 0); + assert!(snapshot.targets.is_empty()); + } + + #[test] + fn durable_mrf_summary_aggregates_target_backlog_without_attributing_legacy_entries() { + let entries = vec![ + MrfReplicateEntry { + bucket: "b1".to_string(), + object: "object-a".to_string(), + version_id: None, + retry_count: 0, + size: 1024, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + target_arns: vec!["arn:target-a".to_string(), "arn:target-b".to_string()], + }, + MrfReplicateEntry { + bucket: "b1".to_string(), + object: "object-b".to_string(), + version_id: None, + retry_count: 0, + size: 512, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + target_arns: vec!["arn:target-a".to_string()], + }, + MrfReplicateEntry { + bucket: "b1".to_string(), + object: "legacy-object".to_string(), + version_id: None, + retry_count: 0, + size: 256, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + target_arns: Vec::new(), + }, + ]; + + let snapshot = durable_mrf_backlog_summary_from_entries(&entries); + + let summary = snapshot.summary; + assert!(summary.available); + let buckets = summary + .buckets + .into_iter() + .map(|bucket| (bucket.bucket.clone(), bucket)) + .collect::>(); + assert_eq!(buckets["b1"].count, 3); + assert_eq!(buckets["b1"].bytes, 1792); + + let targets = snapshot + .targets + .into_iter() + .map(|target| ((target.bucket.clone(), target.target_arn.clone()), target)) + .collect::>(); + let target_a = &targets[&("b1".to_string(), "arn:target-a".to_string())]; + assert_eq!(target_a.count, 2); + assert_eq!(target_a.bytes, 1536); + let target_b = &targets[&("b1".to_string(), "arn:target-b".to_string())]; + assert_eq!(target_b.count, 1); + assert_eq!(target_b.bytes, 1024); } #[test] fn durable_mrf_summary_marks_invalid_sizes_unavailable() { let invalid = durable_mrf_backlog_summary_from_sizes([("bucket".to_string(), -1)]); - assert!(!invalid.available); - assert!(invalid.buckets.is_empty()); + let summary = invalid.summary; + assert!(!summary.available); + assert!(summary.buckets.is_empty()); + assert!(invalid.targets.is_empty()); } #[test] @@ -3264,6 +3559,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: Vec::new(), }]) .expect("invalid persisted entry should still encode for boundary testing"); let invalid = durable_mrf_backlog_from_read(Ok(negative)); diff --git a/crates/ecstore/src/bucket/replication/replication_state.rs b/crates/ecstore/src/bucket/replication/replication_state.rs index dc218a66c..d16a507e5 100644 --- a/crates/ecstore/src/bucket/replication/replication_state.rs +++ b/crates/ecstore/src/bucket/replication/replication_state.rs @@ -22,9 +22,9 @@ use super::replication_stats_boundary::{ QueueCache, ReplicationMetricScope, SRMetricsSummary, XferStats, }; use super::runtime_boundary as runtime_sources; -use std::collections::HashMap; +use std::collections::{HashMap, hash_map::Entry}; use std::sync::atomic::{AtomicI64, Ordering}; -use std::sync::{Arc, Mutex as StdMutex}; +use std::sync::{Arc, LazyLock, Mutex as StdMutex, Weak}; use std::time::{Duration, SystemTime}; use tokio::sync::{Mutex, RwLock}; use tokio::time::interval; @@ -153,6 +153,83 @@ pub struct ReplicationStats { pub most_recent_stats: Arc>>, } +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct RuntimeReplicationTargetBacklog { + pub bucket: String, + pub target_arn: String, + pub count: u64, + pub bytes: u64, +} + +type TargetQueueKey = (String, String); +type TargetQueueCache = HashMap; + +struct TargetQueueCacheSlot { + owner: Weak>, + metrics: TargetQueueCache, +} + +impl TargetQueueCacheSlot { + fn new(owner: &Arc>) -> Self { + Self { + owner: Arc::downgrade(owner), + metrics: TargetQueueCache::default(), + } + } + + fn belongs_to(&self, owner: &Arc>) -> bool { + self.owner.upgrade().is_some_and(|current| Arc::ptr_eq(¤t, owner)) + } +} + +// Keep runtime target counters outside ReplicationStats to preserve its public struct shape. +static TARGET_QUEUE_CACHES: LazyLock>> = LazyLock::new(|| StdMutex::new(Vec::new())); + +fn i64_to_u64_floor_zero(value: i64) -> u64 { + u64::try_from(value.max(0)).unwrap_or(0) +} + +fn normalized_target_arns(target_arns: &[String]) -> Vec<&str> { + let mut target_arns = target_arns + .iter() + .map(String::as_str) + .filter(|target_arn| !target_arn.is_empty()) + .collect::>(); + target_arns.sort_unstable(); + target_arns.dedup(); + target_arns +} + +fn target_queue_cache_snapshot(cache: &TargetQueueCache) -> Vec { + cache + .iter() + .filter_map(|((bucket, target_arn), metric)| { + let count = i64_to_u64_floor_zero(metric.curr.get_current_count()); + let bytes = i64_to_u64_floor_zero(metric.curr.get_current_bytes()); + (count > 0 || bytes > 0).then(|| RuntimeReplicationTargetBacklog { + bucket: bucket.clone(), + target_arn: target_arn.clone(), + count, + bytes, + }) + }) + .collect() +} + +fn prune_stale_target_queue_caches(caches: &mut Vec) { + caches.retain(|slot| slot.owner.strong_count() > 0); +} + +fn with_target_queue_caches(f: impl FnOnce(&mut Vec) -> T) -> T { + match TARGET_QUEUE_CACHES.lock() { + Ok(mut caches) => f(&mut caches), + Err(poisoned) => { + let mut caches = poisoned.into_inner(); + f(&mut caches) + } + } +} + impl ReplicationStats { pub fn new() -> Self { Self { @@ -684,6 +761,68 @@ impl ReplicationStats { } } + pub(crate) fn inc_target_q(&self, bucket: &str, target_arns: &[String], size: i64) { + let target_arns = normalized_target_arns(target_arns); + if target_arns.is_empty() { + return; + } + with_target_queue_caches(|caches| { + prune_stale_target_queue_caches(caches); + let slot_index = match caches.iter().position(|slot| slot.belongs_to(&self.q_cache)) { + Some(index) => index, + None => { + caches.push(TargetQueueCacheSlot::new(&self.q_cache)); + caches.len() - 1 + } + }; + let slot = &mut caches[slot_index]; + let bucket = bucket.to_string(); + for target_arn in target_arns { + let metric = match slot.metrics.entry((bucket.clone(), target_arn.to_string())) { + Entry::Occupied(entry) => entry.into_mut(), + Entry::Vacant(entry) => entry.insert(InQueueMetric::default()), + }; + metric.curr.add_current(size, 1); + } + }); + } + + pub(crate) fn dec_target_q(&self, bucket: &str, target_arns: &[String], size: i64) { + let target_arns = normalized_target_arns(target_arns); + if target_arns.is_empty() { + return; + } + with_target_queue_caches(|caches| { + prune_stale_target_queue_caches(caches); + if let Some(slot) = caches.iter_mut().find(|slot| slot.belongs_to(&self.q_cache)) { + let bucket = bucket.to_string(); + for target_arn in target_arns { + if let Some(metric) = slot.metrics.get_mut(&(bucket.clone(), target_arn.to_string())) { + metric.curr.subtract_current(size, 1); + } + } + slot.metrics + .retain(|_, metric| metric.curr.get_current_count() > 0 || metric.curr.get_current_bytes() > 0); + } + }); + } + + pub fn runtime_target_backlog_snapshot(&self) -> Vec { + let mut snapshot = with_target_queue_caches(|caches| { + caches + .iter() + .find(|slot| slot.belongs_to(&self.q_cache)) + .map(|slot| target_queue_cache_snapshot(&slot.metrics)) + .unwrap_or_default() + }); + snapshot.sort_by(|left, right| { + left.bucket + .cmp(&right.bucket) + .then_with(|| left.target_arn.cmp(&right.target_arn)) + }); + snapshot + } + /// Increase proxy metrics pub async fn inc_proxy(&self, bucket: &str, api: &str, is_err: bool) { let mut p_cache = self.p_cache.lock().await; @@ -707,6 +846,94 @@ impl Default for ReplicationStats { mod tests { use super::*; + #[test] + fn runtime_target_backlog_snapshot_tracks_targets() { + let stats = ReplicationStats::new(); + stats.inc_target_q( + "photos", + &[ + "arn:rustfs:replication:target-b".to_string(), + "arn:rustfs:replication:target-a".to_string(), + "arn:rustfs:replication:target-a".to_string(), + ], + 1024, + ); + + let snapshot = stats.runtime_target_backlog_snapshot(); + + assert_eq!( + snapshot, + vec![ + RuntimeReplicationTargetBacklog { + bucket: "photos".to_string(), + target_arn: "arn:rustfs:replication:target-a".to_string(), + count: 1, + bytes: 1024, + }, + RuntimeReplicationTargetBacklog { + bucket: "photos".to_string(), + target_arn: "arn:rustfs:replication:target-b".to_string(), + count: 1, + bytes: 1024, + }, + ] + ); + } + + #[test] + fn runtime_target_backlog_ignores_empty_targets() { + let stats = ReplicationStats::new(); + stats.inc_target_q("photos", &["".to_string()], 1024); + + assert!(stats.runtime_target_backlog_snapshot().is_empty()); + } + + #[test] + fn runtime_target_backlog_is_scoped_to_stats_instance() { + let first = ReplicationStats::new(); + let second = ReplicationStats::new(); + first.inc_target_q("photos", &["arn:rustfs:replication:target-a".to_string()], 1024); + + assert!(second.runtime_target_backlog_snapshot().is_empty()); + assert_eq!(first.runtime_target_backlog_snapshot()[0].count, 1); + } + + #[test] + fn runtime_target_backlog_decrements_with_saturation() { + let stats = ReplicationStats::new(); + let target_arns = ["arn:rustfs:replication:target-a".to_string()]; + stats.inc_target_q("photos", &target_arns, 1024); + stats.dec_target_q("photos", &target_arns, 2048); + + assert!(stats.runtime_target_backlog_snapshot().is_empty()); + } + + #[test] + fn runtime_target_backlog_prunes_stale_sidecar_on_next_access() { + { + let stats = ReplicationStats::new(); + stats.inc_target_q("photos", &["arn:rustfs:replication:target-a".to_string()], 1024); + assert!( + TARGET_QUEUE_CACHES + .lock() + .expect("target queue cache mutex") + .iter() + .any(|slot| slot.owner.strong_count() > 0) + ); + } + + let stats = ReplicationStats::new(); + stats.inc_target_q("photos", &["arn:rustfs:replication:target-b".to_string()], 1024); + + assert!( + TARGET_QUEUE_CACHES + .lock() + .expect("target queue cache mutex") + .iter() + .all(|slot| slot.owner.strong_count() > 0) + ); + } + #[tokio::test] async fn test_replication_stats_new() { let stats = ReplicationStats::new(); diff --git a/crates/obs/src/metrics/collectors/bucket_replication.rs b/crates/obs/src/metrics/collectors/bucket_replication.rs index 23891bc8e..a3b764451 100644 --- a/crates/obs/src/metrics/collectors/bucket_replication.rs +++ b/crates/obs/src/metrics/collectors/bucket_replication.rs @@ -17,8 +17,10 @@ use crate::metrics::report::PrometheusMetric; use crate::metrics::schema::bucket_replication::{ 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_LAST_HR_FAILED_BYTES_MD, BUCKET_REPL_LAST_HR_FAILED_COUNT_MD, + BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD, BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD, + BUCKET_REPL_CURRENT_TARGET_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_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD, + BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD, BUCKET_REPL_LAST_HR_FAILED_BYTES_MD, BUCKET_REPL_LAST_HR_FAILED_COUNT_MD, BUCKET_REPL_LAST_MIN_FAILED_BYTES_MD, BUCKET_REPL_LAST_MIN_FAILED_COUNT_MD, BUCKET_REPL_LATENCY_MS_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, @@ -37,6 +39,7 @@ use std::borrow::Cow; const BASE_BUCKET_REPLICATION_METRICS_PER_BUCKET: usize = 25; const BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET: usize = 11; +const BUCKET_REPLICATION_BACKLOG_METRICS_PER_TARGET: usize = 4; #[derive(Debug, Clone, Default)] pub struct BucketReplicationTargetStats { @@ -99,6 +102,16 @@ pub(crate) struct BucketReplicationBacklogStats { pub(crate) mrf_missed_count: u64, pub(crate) mrf_flush_failures: u64, pub(crate) mrf_last_flush_duration_millis: u64, + pub(crate) target_backlogs: Vec, +} + +#[derive(Debug, Clone, Default)] +pub(crate) struct BucketReplicationTargetBacklogStats { + pub(crate) target_arn: String, + pub(crate) current_backlog_count: u64, + pub(crate) current_backlog_bytes: u64, + pub(crate) durable_mrf_backlog_count: u64, + pub(crate) durable_mrf_backlog_bytes: u64, } pub fn collect_bucket_replication_bandwidth_metrics(stats: &[BucketReplicationBandwidthStats]) -> Vec { @@ -290,7 +303,14 @@ pub(crate) fn collect_bucket_replication_backlog_metrics(stats: &[BucketReplicat return Vec::new(); } - let mut metrics = Vec::with_capacity(stats.len() * BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET); + let metric_count = stats + .iter() + .map(|stat| { + BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET + + stat.target_backlogs.len() * BUCKET_REPLICATION_BACKLOG_METRICS_PER_TARGET + }) + .sum(); + let mut metrics = Vec::with_capacity(metric_count); for stat in stats { let bucket_label: Cow<'static, str> = Cow::Owned(stat.bucket.clone()); @@ -345,6 +365,42 @@ pub(crate) fn collect_bucket_replication_backlog_metrics(stats: &[BucketReplicat ) .with_label(BUCKET_L, bucket_label), ); + for target in &stat.target_backlogs { + let bucket_label: Cow<'static, str> = Cow::Owned(stat.bucket.clone()); + let target_label: Cow<'static, str> = Cow::Owned(target.target_arn.clone()); + metrics.push( + PrometheusMetric::from_descriptor( + &BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD, + target.current_backlog_count as f64, + ) + .with_label(BUCKET_L, bucket_label.clone()) + .with_label(TARGET_ARN_L, target_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor( + &BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD, + target.current_backlog_bytes as f64, + ) + .with_label(BUCKET_L, bucket_label.clone()) + .with_label(TARGET_ARN_L, target_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor( + &BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD, + target.durable_mrf_backlog_count as f64, + ) + .with_label(BUCKET_L, bucket_label.clone()) + .with_label(TARGET_ARN_L, target_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor( + &BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD, + target.durable_mrf_backlog_bytes as f64, + ) + .with_label(BUCKET_L, bucket_label) + .with_label(TARGET_ARN_L, target_label), + ); + } } metrics @@ -469,10 +525,17 @@ mod tests { mrf_missed_count: 4, mrf_flush_failures: 5, mrf_last_flush_duration_millis: 6, + target_backlogs: vec![BucketReplicationTargetBacklogStats { + target_arn: "arn:rustfs:replication:target-a".to_string(), + current_backlog_count: 3, + current_backlog_bytes: 4096, + durable_mrf_backlog_count: 2, + durable_mrf_backlog_bytes: 2048, + }], }]; let metrics = collect_bucket_replication_backlog_metrics(&stats); - assert_eq!(metrics.len(), 11); + assert_eq!(metrics.len(), 15); let backlog_count_name = BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD.get_full_metric_name(); assert!(metrics.iter().any(|metric| { @@ -509,6 +572,50 @@ mod tests { && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") })); + let target_count_name = BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == target_count_name + && metric.value == 3.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + && metric + .labels + .iter() + .any(|(key, value)| *key == TARGET_ARN_L && value == "arn:rustfs:replication:target-a") + })); + + let target_bytes_name = BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == target_bytes_name + && metric.value == 4096.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + && metric + .labels + .iter() + .any(|(key, value)| *key == TARGET_ARN_L && value == "arn:rustfs:replication:target-a") + })); + + let durable_target_count_name = BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == durable_target_count_name + && metric.value == 2.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + && metric + .labels + .iter() + .any(|(key, value)| *key == TARGET_ARN_L && value == "arn:rustfs:replication:target-a") + })); + + let durable_target_bytes_name = BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == durable_target_bytes_name + && metric.value == 2048.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + && metric + .labels + .iter() + .any(|(key, value)| *key == TARGET_ARN_L && value == "arn:rustfs:replication:target-a") + })); + let pending_count_name = BUCKET_REPL_MRF_PENDING_COUNT_MD.get_full_metric_name(); assert!(metrics.iter().any(|metric| { metric.name == pending_count_name @@ -574,6 +681,7 @@ mod tests { mrf_missed_count: 4, mrf_flush_failures: 5, mrf_last_flush_duration_millis: 6, + target_backlogs: Vec::new(), }]; let metrics = collect_bucket_replication_backlog_metrics(&stats); diff --git a/crates/obs/src/metrics/collectors/mod.rs b/crates/obs/src/metrics/collectors/mod.rs index 43dbd2873..a674e1e5b 100644 --- a/crates/obs/src/metrics/collectors/mod.rs +++ b/crates/obs/src/metrics/collectors/mod.rs @@ -42,7 +42,9 @@ pub mod system_process; pub use audit::{AuditTargetStats, collect_audit_metrics}; pub use bucket::{BucketStats, collect_bucket_metrics}; -pub(crate) use bucket_replication::{BucketReplicationBacklogStats, collect_bucket_replication_backlog_metrics}; +pub(crate) use bucket_replication::{ + BucketReplicationBacklogStats, BucketReplicationTargetBacklogStats, collect_bucket_replication_backlog_metrics, +}; pub use bucket_replication::{ BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetStats, collect_bucket_replication_bandwidth_metrics, collect_bucket_replication_metrics, diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index a17430204..eddd04538 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -80,8 +80,10 @@ 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, 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_CURRENT_BACKLOG_COUNT_MD, BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD, + BUCKET_REPL_CURRENT_TARGET_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_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD, + BUCKET_REPL_DURABLE_MRF_TARGET_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, }; @@ -633,6 +635,17 @@ fn repl_backlog_live_keys(stats: &[BucketReplicationBacklogStats]) -> HashSet HashSet { + stats + .iter() + .flat_map(|stat| { + stat.target_backlogs + .iter() + .map(|target| (stat.bucket.clone(), target.target_arn.clone())) + }) + .collect() +} + fn update_series_zero_tombstones( has_seen_valid_snapshot: &mut bool, prev_live_keys: &mut HashSet, @@ -1060,6 +1073,40 @@ fn collect_repl_backlog_zero_tombstone_metrics(zero_tombstones: &HashMap) -> Vec { + if zero_tombstones.is_empty() { + return Vec::new(); + } + + let mut zero_metrics = Vec::with_capacity(zero_tombstones.len() * 4); + for (bucket, target_arn) in zero_tombstones.keys() { + let bucket_label: Cow<'static, str> = Cow::Owned(bucket.clone()); + let target_arn_label: Cow<'static, str> = Cow::Owned(target_arn.clone()); + zero_metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD, 0.0) + .with_label(BUCKET_L, bucket_label.clone()) + .with_label(TARGET_ARN_L, target_arn_label.clone()), + ); + zero_metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD, 0.0) + .with_label(BUCKET_L, bucket_label.clone()) + .with_label(TARGET_ARN_L, target_arn_label.clone()), + ); + zero_metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD, 0.0) + .with_label(BUCKET_L, bucket_label.clone()) + .with_label(TARGET_ARN_L, target_arn_label.clone()), + ); + zero_metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD, 0.0) + .with_label(BUCKET_L, bucket_label) + .with_label(TARGET_ARN_L, target_arn_label), + ); + } + + zero_metrics +} + fn retire_repl_bw_metric_series(bucket: &str, target_arn: &str) -> usize { let labels = [ (BUCKET_L, Cow::Owned(bucket.to_string())), @@ -1089,6 +1136,17 @@ fn retire_repl_backlog_metric_series(bucket: &str) -> usize { .sum() } +fn retire_repl_backlog_target_metric_series(bucket: &str, target_arn: &str) -> usize { + let labels = [ + (BUCKET_L, Cow::Owned(bucket.to_string())), + (TARGET_ARN_L, Cow::Owned(target_arn.to_string())), + ]; + retire_metric_series(&BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD.get_full_metric_name(), &labels) + + retire_metric_series(&BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD.get_full_metric_name(), &labels) + + retire_metric_series(&BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD.get_full_metric_name(), &labels) + + retire_metric_series(&BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD.get_full_metric_name(), &labels) +} + fn expire_repl_bw_zero_tombstones(monitor_available: bool, zero_tombstones: &mut HashMap) -> Vec { if !monitor_available { return Vec::new(); @@ -1105,6 +1163,17 @@ fn expire_repl_backlog_zero_tombstones(monitor_available: bool, zero_tombstones: expire_series_zero_tombstones(zero_tombstones) } +fn expire_repl_backlog_target_zero_tombstones( + target_metrics_available: bool, + zero_tombstones: &mut HashMap, +) -> Vec { + if !target_metrics_available { + return Vec::new(); + } + + expire_series_zero_tombstones(zero_tombstones) +} + /// Initialize all metrics collectors. /// /// This function spawns background tasks that periodically collect metrics @@ -1399,6 +1468,9 @@ pub fn init_metrics_runtime(token: CancellationToken) { 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; + let mut prev_backlog_target_live_keys: HashSet = HashSet::new(); + let mut backlog_target_zero_tombstones: HashMap = HashMap::new(); + let mut has_seen_valid_backlog_target_snapshot = false; loop { tokio::select! { _ = interval.tick() => { @@ -1429,6 +1501,9 @@ 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; + let durable_mrf_available = bucket_replication_backlog.iter().any(|stat| stat.durable_mrf_available); + let backlog_target_metrics_available = + durable_mrf_available || monitor_available || !bucket_replication_backlog.is_empty(); update_repl_backlog_zero_tombstones( monitor_available, &mut has_seen_valid_backlog_snapshot, @@ -1437,9 +1512,19 @@ pub fn init_metrics_runtime(token: CancellationToken) { repl_backlog_live_keys(&bucket_replication_backlog), repl_bw_zero_tombstone_cycles, ); + if backlog_target_metrics_available { + update_series_zero_tombstones( + &mut has_seen_valid_backlog_target_snapshot, + &mut prev_backlog_target_live_keys, + &mut backlog_target_zero_tombstones, + repl_backlog_target_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)); + metrics.extend(collect_repl_backlog_target_zero_tombstone_metrics(&backlog_target_zero_tombstones)); let replication = collect_replication_stats().await; metrics.extend(collect_replication_metrics(&replication)); report_metrics(&metrics); @@ -1451,6 +1536,12 @@ pub fn init_metrics_runtime(token: CancellationToken) { for bucket in expire_repl_backlog_zero_tombstones(monitor_available, &mut backlog_zero_tombstones) { let _ = retire_repl_backlog_metric_series(&bucket); } + for (bucket, target_arn) in expire_repl_backlog_target_zero_tombstones( + backlog_target_metrics_available, + &mut backlog_target_zero_tombstones, + ) { + let _ = retire_repl_backlog_target_metric_series(&bucket, &target_arn); + } }, ).await; } @@ -2259,6 +2350,69 @@ mod tests { assert_eq!(prev_live_keys, bucket_keys(&["photos"])); } + #[test] + fn repl_backlog_target_tombstones_zero_removed_targets_then_expire() { + let mut has_seen_valid_snapshot = false; + let mut prev_live_keys = HashSet::new(); + let mut zero_tombstones = HashMap::new(); + let key = repl_bw_key("photos", "arn:rustfs:replication:target-a"); + + update_series_zero_tombstones( + &mut has_seen_valid_snapshot, + &mut prev_live_keys, + &mut zero_tombstones, + repl_bw_keys(&[("photos", "arn:rustfs:replication:target-a")]), + 2, + ); + assert!(has_seen_valid_snapshot); + assert_eq!(prev_live_keys, repl_bw_keys(&[("photos", "arn:rustfs:replication:target-a")])); + assert!(zero_tombstones.is_empty()); + + update_series_zero_tombstones(&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_target_zero_tombstone_metrics(&zero_tombstones); + assert_eq!(metrics.len(), 4); + let expected_names = HashSet::from([ + BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD.get_full_metric_name(), + BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_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!(labels.get(TARGET_ARN_L).map(String::as_str), Some("arn:rustfs:replication:target-a")); + } + assert_eq!(actual_names, expected_names); + + let expired = expire_repl_backlog_target_zero_tombstones(true, &mut zero_tombstones); + assert!(expired.is_empty()); + assert_eq!(zero_tombstones.get(&key), Some(&1)); + + let expired = expire_repl_backlog_target_zero_tombstones(true, &mut zero_tombstones); + assert_eq!(expired, vec![key]); + assert!(zero_tombstones.is_empty()); + } + + #[test] + fn repl_backlog_target_tombstones_do_not_advance_when_target_metrics_unavailable() { + let key = repl_bw_key("videos", "arn:rustfs:replication:target-b"); + let mut zero_tombstones = HashMap::from([(key.clone(), 1)]); + + let expired = expire_repl_backlog_target_zero_tombstones(false, &mut zero_tombstones); + + assert!(expired.is_empty()); + assert_eq!(zero_tombstones.get(&key), Some(&1)); + } + #[test] fn bucket_tombstones_zero_removed_buckets_then_expire() { let mut has_seen_valid_snapshot = false; diff --git a/crates/obs/src/metrics/schema/bucket_replication.rs b/crates/obs/src/metrics/schema/bucket_replication.rs index 28237ecdf..823edd980 100644 --- a/crates/obs/src/metrics/schema/bucket_replication.rs +++ b/crates/obs/src/metrics/schema/bucket_replication.rs @@ -35,9 +35,13 @@ const RESYNC_CANCELED_TOTAL: &str = "resync_canceled_total"; const RESYNC_DURATION_MS_TOTAL: &str = "resync_duration_ms_total"; const CURRENT_BACKLOG_COUNT: &str = "current_backlog_count"; const CURRENT_BACKLOG_BYTES: &str = "current_backlog_bytes"; +const CURRENT_TARGET_BACKLOG_COUNT: &str = "current_target_backlog_count"; +const CURRENT_TARGET_BACKLOG_BYTES: &str = "current_target_backlog_bytes"; const DURABLE_MRF_AVAILABLE: &str = "durable_mrf_available"; const DURABLE_MRF_BACKLOG_COUNT: &str = "durable_mrf_backlog_count"; const DURABLE_MRF_BACKLOG_BYTES: &str = "durable_mrf_backlog_bytes"; +const DURABLE_MRF_TARGET_BACKLOG_COUNT: &str = "durable_mrf_target_backlog_count"; +const DURABLE_MRF_TARGET_BACKLOG_BYTES: &str = "durable_mrf_target_backlog_bytes"; const MRF_PENDING_COUNT: &str = "mrf_pending_count"; const MRF_PENDING_BYTES: &str = "mrf_pending_bytes"; const MRF_DROPPED_COUNT: &str = "mrf_dropped_count"; @@ -108,6 +112,24 @@ pub static BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD: LazyLock = La ) }); +pub static BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(CURRENT_TARGET_BACKLOG_COUNT), + "Current number of target-scoped operations admitted to in-memory replication worker queues for a bucket and target ARN", + &[BUCKET_L, TARGET_ARN_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(CURRENT_TARGET_BACKLOG_BYTES), + "Current bytes in target-scoped operations admitted to in-memory replication worker queues for a bucket and target ARN", + &[BUCKET_L, TARGET_ARN_L], + subsystems::BUCKET_REPLICATION, + ) +}); + pub static BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::from(DURABLE_MRF_AVAILABLE), @@ -135,6 +157,24 @@ pub static BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD: LazyLock ) }); +pub static BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(DURABLE_MRF_TARGET_BACKLOG_COUNT), + "Current number of target-scoped operations in the durable MRF backlog file for a bucket and target ARN on this node", + &[BUCKET_L, TARGET_ARN_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(DURABLE_MRF_TARGET_BACKLOG_BYTES), + "Current bytes in target-scoped operations in the durable MRF backlog file for a bucket and target ARN on this node", + &[BUCKET_L, TARGET_ARN_L], + subsystems::BUCKET_REPLICATION, + ) +}); + pub static BUCKET_REPL_MRF_PENDING_COUNT_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::from(MRF_PENDING_COUNT), diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 170279e37..7ab0a7311 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -21,11 +21,11 @@ //! and convert them to the Stats structs used by collectors. use crate::metrics::collectors::{ - BucketReplicationBacklogStats, BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetStats, - BucketStats, BucketUsageStats, ClusterConfigStats, ClusterHealthStats, ClusterStats, ClusterUsageStats, - CompressionClusterStats, CpuStats, DiskStats, DriveCountStats, DriveDetailedStats, ErasureSetStats, HostNetworkStats, - IamStats, IlmStats, MemoryStats, NetworkStats, ProcessStats, ProcessStatusType, ReplicationStats, ResourceStats, - ScannerStats, + BucketReplicationBacklogStats, BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetBacklogStats, + BucketReplicationTargetStats, BucketStats, BucketUsageStats, ClusterConfigStats, ClusterHealthStats, ClusterStats, + ClusterUsageStats, CompressionClusterStats, CpuStats, DiskStats, DriveCountStats, DriveDetailedStats, ErasureSetStats, + HostNetworkStats, IamStats, IlmStats, MemoryStats, NetworkStats, ProcessStats, ProcessStatusType, ReplicationStats, + ResourceStats, ScannerStats, }; use crate::metrics::runtime_sources::{ObsIlmRuntimeSnapshot, bucket_monitor_handle, iam_metrics_snapshot, ilm_runtime_snapshot}; use crate::metrics::{ @@ -212,6 +212,17 @@ async fn obs_bucket_replication_stats_bundle() -> (Vec, mrf_missed_count: stats.mrf_missed_count, mrf_flush_failures: stats.mrf_flush_failures, mrf_last_flush_duration_millis: stats.mrf_last_flush_duration_millis, + target_backlogs: stats + .target_backlogs + .iter() + .map(|target| BucketReplicationTargetBacklogStats { + target_arn: target.target_arn.clone(), + current_backlog_count: target.current_backlog_count, + current_backlog_bytes: target.current_backlog_bytes, + durable_mrf_backlog_count: target.durable_mrf_backlog_count, + durable_mrf_backlog_bytes: target.durable_mrf_backlog_bytes, + }) + .collect(), }); detail_stats.push(bucket_replication_detail_from_snapshot(stats)); } diff --git a/crates/obs/src/metrics/storage_api.rs b/crates/obs/src/metrics/storage_api.rs index e1286ebe1..e3801276e 100644 --- a/crates/obs/src/metrics/storage_api.rs +++ b/crates/obs/src/metrics/storage_api.rs @@ -18,7 +18,8 @@ use std::time::Duration; pub(crate) use rustfs_ecstore::api::bucket::bandwidth::monitor::Monitor as ObsBucketBandwidthMonitor; pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::get_quota_config as obs_get_quota_config; use rustfs_ecstore::api::bucket::replication::{ - DurableMrfBucketBacklog, MrfBucketBacklogObservability, durable_mrf_backlog_summary_snapshot, get_global_replication_stats, + DurableMrfBucketBacklog, DurableMrfTargetBacklog, MrfBucketBacklogObservability, RuntimeReplicationTargetBacklog, + durable_mrf_backlog_summary_snapshot, durable_mrf_target_backlog_snapshot, get_global_replication_stats, mrf_backlog_observability_snapshot, }; pub(crate) use rustfs_ecstore::api::capacity::{ @@ -44,6 +45,15 @@ pub(crate) struct ObsBucketReplicationTargetStatsSnapshot { pub(crate) latency_ms: f64, } +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct ObsBucketReplicationTargetBacklogSnapshot { + pub(crate) target_arn: String, + pub(crate) current_backlog_count: u64, + pub(crate) current_backlog_bytes: u64, + pub(crate) durable_mrf_backlog_count: u64, + pub(crate) durable_mrf_backlog_bytes: u64, +} + #[derive(Debug, Clone, PartialEq)] pub(crate) struct ObsBucketReplicationStatsSnapshot { pub(crate) bucket: String, @@ -84,6 +94,7 @@ pub(crate) struct ObsBucketReplicationStatsSnapshot { pub(crate) mrf_flush_failures: u64, pub(crate) mrf_last_flush_duration_millis: u64, pub(crate) targets: Vec, + pub(crate) target_backlogs: Vec, } #[derive(Debug, Clone, Default, PartialEq)] @@ -122,6 +133,15 @@ struct ObsBucketReplicationProxySnapshot { proxied_delete_tagging_requests_failures: u64, } +#[derive(Debug, Clone, Default, PartialEq)] +struct ObsBucketReplicationBacklogSnapshot { + durable_mrf_available: bool, + durable_bucket: DurableMrfBucketBacklog, + runtime_targets: Vec, + durable_targets: Vec, + mrf_observability: MrfBucketBacklogObservability, +} + #[derive(Debug, Clone, Default, PartialEq)] pub(crate) struct ObsReplicationSiteStatsSnapshot { pub(crate) average_active_workers: f64, @@ -157,10 +177,40 @@ fn bucket_replication_stats_snapshot_from_parts( bucket: String, runtime: ObsBucketReplicationRuntimeSnapshot, proxy: ObsBucketReplicationProxySnapshot, - durable_mrf_available: bool, - durable_bucket: DurableMrfBucketBacklog, - mrf_observability: MrfBucketBacklogObservability, + backlog: ObsBucketReplicationBacklogSnapshot, ) -> ObsBucketReplicationStatsSnapshot { + let mut target_backlogs = HashMap::with_capacity(backlog.runtime_targets.len().saturating_add(backlog.durable_targets.len())); + for target in backlog.runtime_targets { + let entry = + target_backlogs + .entry(target.target_arn.clone()) + .or_insert_with(|| ObsBucketReplicationTargetBacklogSnapshot { + target_arn: target.target_arn, + current_backlog_count: 0, + current_backlog_bytes: 0, + durable_mrf_backlog_count: 0, + durable_mrf_backlog_bytes: 0, + }); + entry.current_backlog_count = target.count; + entry.current_backlog_bytes = target.bytes; + } + for target in backlog.durable_targets { + let entry = + target_backlogs + .entry(target.target_arn.clone()) + .or_insert_with(|| ObsBucketReplicationTargetBacklogSnapshot { + target_arn: target.target_arn, + current_backlog_count: 0, + current_backlog_bytes: 0, + durable_mrf_backlog_count: 0, + durable_mrf_backlog_bytes: 0, + }); + entry.durable_mrf_backlog_count = target.count; + entry.durable_mrf_backlog_bytes = target.bytes; + } + let mut target_backlogs = target_backlogs.into_values().collect::>(); + target_backlogs.sort_by(|left, right| left.target_arn.cmp(&right.target_arn)); + ObsBucketReplicationStatsSnapshot { bucket, total_failed_bytes: runtime.total_failed_bytes, @@ -190,16 +240,17 @@ fn bucket_replication_stats_snapshot_from_parts( resync_duration_ms: runtime.resync_duration_ms, current_backlog_count: runtime.current_backlog_count, current_backlog_bytes: runtime.current_backlog_bytes, - durable_mrf_available, - durable_mrf_backlog_count: durable_bucket.count, - durable_mrf_backlog_bytes: durable_bucket.bytes, - mrf_pending_count: mrf_observability.pending_count, - mrf_pending_bytes: mrf_observability.pending_bytes, - mrf_dropped_count: mrf_observability.dropped_count, - mrf_missed_count: mrf_observability.missed_count, - mrf_flush_failures: mrf_observability.flush_failure_count, - mrf_last_flush_duration_millis: mrf_observability.last_flush_duration_millis, + durable_mrf_available: backlog.durable_mrf_available, + durable_mrf_backlog_count: backlog.durable_bucket.count, + durable_mrf_backlog_bytes: backlog.durable_bucket.bytes, + mrf_pending_count: backlog.mrf_observability.pending_count, + mrf_pending_bytes: backlog.mrf_observability.pending_bytes, + mrf_dropped_count: backlog.mrf_observability.dropped_count, + mrf_missed_count: backlog.mrf_observability.missed_count, + mrf_flush_failures: backlog.mrf_observability.flush_failure_count, + mrf_last_flush_duration_millis: backlog.mrf_observability.last_flush_duration_millis, targets: runtime.targets, + target_backlogs, } } @@ -222,6 +273,27 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec>(); + let mut durable_targets_by_bucket: HashMap> = HashMap::new(); + let durable_mrf_targets = if replication_storage_available && durable_mrf_available { + durable_mrf_target_backlog_snapshot() + } else { + Vec::new() + }; + for target in durable_mrf_targets { + durable_targets_by_bucket + .entry(target.bucket.clone()) + .or_default() + .push(target); + } + let mut runtime_targets_by_bucket: HashMap> = HashMap::new(); + if let Some(stats) = &stats { + for target in stats.runtime_target_backlog_snapshot() { + runtime_targets_by_bucket + .entry(target.bucket.clone()) + .or_default() + .push(target); + } + } let mrf_observability = if replication_storage_available { mrf_backlog_observability_snapshot() } else { @@ -236,7 +308,8 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec Vec Vec, + + #[serde(rename = "targetARNs", skip_serializing_if = "Vec::is_empty", default)] + pub target_arns: Vec, } fn retry_count_to_mrf(retry_count: u32) -> i32 { @@ -672,6 +675,18 @@ impl ReplicateDecision { } if result.is_empty() { None } else { Some(result) } } + + pub fn replicate_target_arns(&self) -> Vec { + let mut arns = self + .targets_map + .values() + .filter(|target| target.replicate && !target.arn.is_empty()) + .map(|target| target.arn.clone()) + .collect::>(); + arns.sort(); + arns.dedup(); + arns + } } impl fmt::Display for ReplicateDecision { @@ -799,6 +814,7 @@ impl ReplicationWorkerOperation for ReplicateObjectInfo { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: self.dsc.replicate_target_arns(), } } @@ -853,6 +869,7 @@ impl ReplicateObjectInfo { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: self.dsc.replicate_target_arns(), } } } @@ -994,6 +1011,62 @@ impl Default for ResyncDecision { mod tests { use super::*; + #[test] + fn replicate_decision_returns_sorted_unique_replicating_target_arns() { + let mut decision = ReplicateDecision::new(); + decision.set(ReplicateTargetDecision { + arn: "arn:target-b".to_string(), + replicate: true, + ..Default::default() + }); + decision.set(ReplicateTargetDecision { + arn: "arn:target-a".to_string(), + replicate: true, + ..Default::default() + }); + decision.set(ReplicateTargetDecision { + arn: "arn:target-c".to_string(), + replicate: false, + ..Default::default() + }); + decision.set(ReplicateTargetDecision { + arn: String::new(), + replicate: true, + ..Default::default() + }); + + assert_eq!( + decision.replicate_target_arns(), + vec!["arn:target-a".to_string(), "arn:target-b".to_string()] + ); + } + + #[test] + fn replicate_object_info_mrf_entry_carries_replicating_targets() { + let mut decision = ReplicateDecision::new(); + decision.set(ReplicateTargetDecision { + arn: "arn:target-a".to_string(), + replicate: true, + ..Default::default() + }); + decision.set(ReplicateTargetDecision { + arn: "arn:target-b".to_string(), + replicate: false, + ..Default::default() + }); + let info = ReplicateObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + size: 42, + dsc: decision, + ..Default::default() + }; + + let entry = info.to_mrf_entry(); + + assert_eq!(entry.target_arns, vec!["arn:target-a".to_string()]); + } + #[test] fn target_state_reads_resync_timestamp_from_target_reset_header_key() { let arn = "arn:rustfs:replication:us-east-1:target:bucket"; diff --git a/crates/replication/src/mrf.rs b/crates/replication/src/mrf.rs index 08c7c7095..2b9fff269 100644 --- a/crates/replication/src/mrf.rs +++ b/crates/replication/src/mrf.rs @@ -71,6 +71,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: vec!["arn:target-a".to_string(), "arn:target-b".to_string()], }, MrfReplicateEntry { bucket: "bucket-a".to_string(), @@ -82,6 +83,7 @@ mod tests { delete_marker_version_id: Some(del_vid), delete_marker: true, delete_marker_mtime: Some(1_705_312_200_123_456_789), + target_arns: vec!["arn:target-a".to_string()], }, ]; @@ -91,9 +93,11 @@ mod tests { assert_eq!(decoded.len(), 2); assert_eq!(decoded[0].version_id, Some(obj_vid)); assert_eq!(decoded[0].op, MrfOpKind::Object); + assert_eq!(decoded[0].target_arns, vec!["arn:target-a".to_string(), "arn:target-b".to_string()]); assert_eq!(decoded[0].delete_marker_mtime, None); assert_eq!(decoded[1].delete_marker_version_id, Some(del_vid)); assert_eq!(decoded[1].op, MrfOpKind::Delete); + assert_eq!(decoded[1].target_arns, vec!["arn:target-a".to_string()]); assert!(decoded[1].delete_marker); assert_eq!( decoded[1].delete_marker_mtime, @@ -129,6 +133,7 @@ mod tests { assert_eq!(decoded[0].retry_count, 2); assert_eq!(decoded[0].size, 100); assert_eq!(decoded[0].op, MrfOpKind::Object); + assert!(decoded[0].target_arns.is_empty()); // Old files lack the deleteMarkerMtime key; it must default to None so replay keeps the // pre-#867 fallback to the current time. assert_eq!(decoded[0].delete_marker_mtime, None); diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index b2e6202e4..8f71bdec6 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -835,6 +835,10 @@ struct MrfTargetBacklog { failed_count: i64, #[serde(rename = "FailedSize")] failed_size: i64, + #[serde(rename = "DurableCount")] + durable_count: i64, + #[serde(rename = "DurableSize")] + durable_size: i64, #[serde(rename = "ObservationScope")] observation_scope: &'static str, } @@ -842,7 +846,7 @@ struct MrfTargetBacklog { /// Response body for `GET /v3/replication/mrf`. /// /// Runtime failed/queued totals and the durable MRF recovery backlog are kept -/// separate because persisted MRF entries do not contain a target ARN. +/// separate because older persisted MRF entries do not contain a target ARN. #[derive(Debug, Serialize)] struct MrfResponse { #[serde(rename = "Bucket")] @@ -888,32 +892,60 @@ fn build_mrf_response( "partial_cluster" }; let mut targets: Vec = Vec::with_capacity(bucket_stats.replication_stats.stats.len()); + let mut targets_by_arn: HashMap = HashMap::with_capacity(bucket_stats.replication_stats.stats.len()); let mut total_failed_count: i64 = 0; let mut total_failed_size: i64 = 0; for (arn, stat) in &bucket_stats.replication_stats.stats { total_failed_count = total_failed_count.saturating_add(stat.failed.count); total_failed_size = total_failed_size.saturating_add(stat.failed.size); + targets_by_arn.insert(arn.clone(), targets.len()); targets.push(MrfTargetBacklog { arn: arn.clone(), failed_count: stat.failed.count, failed_size: stat.failed.size, + durable_count: 0, + durable_size: 0, observation_scope, }); } - targets.sort_by(|a, b| a.arn.cmp(&b.arn)); let queued = &bucket_stats.replication_stats.q_stat.curr; - let (durable_count, durable_size) = if durable.available { - durable - .entries - .iter() - .filter(|entry| entry.bucket == bucket) - .fold((0i64, 0i64), |(count, size), entry| { - (count.saturating_add(1), size.saturating_add(entry.size)) - }) + let (durable_count, durable_size, missing_target_arns) = if durable.available { + durable.entries.iter().filter(|entry| entry.bucket == bucket).fold( + (0i64, 0i64, false), + |(count, size, missing_target_arns), entry| { + let mut missing_target_arns = missing_target_arns; + for target_arn in &entry.target_arns { + if target_arn.is_empty() { + continue; + } + let index = if let Some(index) = targets_by_arn.get(target_arn).copied() { + index + } else { + targets_by_arn.insert(target_arn.clone(), targets.len()); + targets.push(MrfTargetBacklog { + arn: target_arn.clone(), + failed_count: 0, + failed_size: 0, + durable_count: 0, + durable_size: 0, + observation_scope, + }); + targets.len() - 1 + }; + targets[index].durable_count = targets[index].durable_count.saturating_add(1); + targets[index].durable_size = targets[index].durable_size.saturating_add(entry.size); + } + if entry.target_arns.is_empty() { + missing_target_arns = true; + } + (count.saturating_add(1), size.saturating_add(entry.size), missing_target_arns) + }, + ) } else { - (0, 0) + (0, 0, false) }; + targets.sort_by(|a, b| a.arn.cmp(&b.arn)); MrfResponse { bucket, @@ -930,7 +962,7 @@ fn build_mrf_response( durable_backlog_available: durable.available, durable_count, durable_size, - per_target_durable_entries_available: false, + per_target_durable_entries_available: durable.available && !missing_target_arns, } } @@ -940,8 +972,9 @@ fn build_mrf_response( /// /// Compatibility note: MinIO returns a stream of individual MRF entries. RustFS /// deliberately returns aggregate runtime and durable counters instead. -/// `PerObjectEntriesAvailable` and `PerTargetDurableEntriesAvailable` remain -/// false until an enumerable API and target-bearing durable format exist. +/// `PerObjectEntriesAvailable` remains false until an enumerable API exists. +/// `PerTargetDurableEntriesAvailable` is false when the durable backlog includes +/// older entries that cannot be attributed to a target. pub struct ReplicationMrfHandler {} #[async_trait::async_trait] @@ -1059,6 +1092,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: vec!["arn-a".to_string(), "arn-durable-only".to_string()], }, MrfReplicateEntry { bucket: "other-bucket".to_string(), @@ -1070,6 +1104,7 @@ mod tests { delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, + target_arns: Vec::new(), }, ], }; @@ -1087,7 +1122,64 @@ mod tests { assert_eq!(json["ClusterComplete"], false); assert_eq!(json["Targets"][0]["ObservationScope"], "partial_cluster"); assert_eq!(json["PerObjectEntriesAvailable"], false); + assert_eq!(json["PerTargetDurableEntriesAvailable"], true); + + let targets = json["Targets"].as_array().expect("targets should serialize as an array"); + let target_a = targets + .iter() + .find(|target| target["ARN"] == "arn-a") + .expect("runtime target should remain present"); + assert_eq!(target_a["FailedCount"], 3); + assert_eq!(target_a["FailedSize"], 900); + assert_eq!(target_a["DurableCount"], 1); + assert_eq!(target_a["DurableSize"], 250); + let target_durable_only = targets + .iter() + .find(|target| target["ARN"] == "arn-durable-only") + .expect("durable-only target should be surfaced"); + assert_eq!(target_durable_only["FailedCount"], 0); + assert_eq!(target_durable_only["FailedSize"], 0); + assert_eq!(target_durable_only["DurableCount"], 1); + assert_eq!(target_durable_only["DurableSize"], 250); + } + + #[test] + fn mrf_response_keeps_legacy_durable_entries_bucket_only() { + let mut stats = BucketStats::default(); + stats.replication_stats.provider_available = true; + stats.replication_stats.cluster_complete = true; + stats.replication_stats.observed_node_count = 1; + stats.replication_stats.expected_node_count = 1; + let durable = DurableMrfBacklog { + available: true, + entries: vec![MrfReplicateEntry { + bucket: "bucket-a".to_string(), + object: "legacy-object".to_string(), + version_id: None, + retry_count: 0, + size: 250, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + target_arns: Vec::new(), + }], + }; + + let response = build_mrf_response("bucket-a".to_string(), &stats, durable); + let json = serde_json::to_value(response).expect("MRF response should serialize"); + + assert_eq!(json["DurableBacklogAvailable"], true); + assert_eq!(json["DurableCount"], 1); + assert_eq!(json["DurableSize"], 250); assert_eq!(json["PerTargetDurableEntriesAvailable"], false); + assert_eq!( + json["Targets"] + .as_array() + .expect("targets should serialize as an array") + .len(), + 0 + ); } #[test] @@ -1115,6 +1207,7 @@ mod tests { assert_eq!(valid_empty_json["DurableBacklogAvailable"], true); assert_eq!(valid_empty_json["TotalFailedCount"], 0); assert_eq!(valid_empty_json["DurableCount"], 0); + assert_eq!(valid_empty_json["PerTargetDurableEntriesAvailable"], true); } #[test]