From db477c26cf646b4081c57c4282eb53a21a027881 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 1 Aug 2026 18:03:30 +0800 Subject: [PATCH] fix(replication): harden backlog observability Add RAII guards for replication runtime backlog tickets so active worker and queue counters unwind on every terminal path. Expose node-local MRF pending, dropped, missed, and flush-failure metrics through the bucket replication Prometheus collector while keeping the existing current backlog and durable MRF gauges additive. Update durable MRF summary maintenance to aggregate incrementally during the persister loop, avoiding repeated full-entry scans on each successful flush. Co-Authored-By: heihutu --- crates/ecstore/src/api/mod.rs | 3 +- .../bucket/replication/replication_pool.rs | 550 +++++++++++++----- .../metrics/collectors/bucket_replication.rs | 100 +++- .../src/metrics/schema/bucket_replication.rs | 60 ++ crates/obs/src/metrics/stats_collector.rs | 6 + crates/obs/src/metrics/storage_api.rs | 94 ++- 6 files changed, 651 insertions(+), 162 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index b2af30c62..2f86d3ca9 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -173,7 +173,8 @@ pub mod bucket { pub mod replication { pub use crate::bucket::replication::replication_pool::{ - DurableMrfBacklogSummary, DurableMrfBucketBacklog, durable_mrf_backlog_summary_snapshot, + DurableMrfBacklogSummary, DurableMrfBucketBacklog, MrfBacklogObservabilitySummary, MrfBucketBacklogObservability, + durable_mrf_backlog_summary_snapshot, mrf_backlog_observability_snapshot, }; pub use crate::bucket::replication::{ BucketReplicationResyncStatus, BucketStats, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 245590977..67afcc5f1 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -53,6 +53,7 @@ use std::sync::LazyLock; use std::sync::RwLock as StdRwLock; use std::sync::atomic::AtomicI32; use std::sync::atomic::Ordering; +use std::time::Instant; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; use tokio::sync::Mutex; @@ -91,20 +92,46 @@ pub struct DurableMrfBacklogSummary { pub buckets: Vec, } +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct MrfBucketBacklogObservability { + pub bucket: String, + pub pending_count: u64, + pub pending_bytes: u64, + pub dropped_count: u64, + pub missed_count: u64, + pub flush_failure_count: u64, + pub last_flush_duration_millis: u64, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct MrfBacklogObservabilitySummary { + pub buckets: Vec, +} + static DURABLE_MRF_BACKLOG_SUMMARY: LazyLock> = LazyLock::new(|| StdRwLock::new(DurableMrfBacklogSummary::default())); +static MRF_BACKLOG_OBSERVABILITY: LazyLock> = + LazyLock::new(|| StdRwLock::new(MrfBacklogObservabilityTracker::default())); -fn durable_mrf_backlog_summary_from_sizes(entries: I) -> DurableMrfBacklogSummary -where - I: IntoIterator, -{ - let mut buckets = HashMap::::new(); - for (bucket_name, entry_size) in entries { +#[derive(Debug, Clone, Default)] +struct DurableMrfBacklogTracker { + available: bool, + buckets: HashMap, +} + +impl DurableMrfBacklogTracker { + fn add_entry(&mut self, bucket_name: String, entry_size: i64) { let Ok(size) = u64::try_from(entry_size) else { - return DurableMrfBacklogSummary::default(); + self.available = false; + self.buckets.clear(); + return; }; - let bucket = match buckets.entry(bucket_name) { + if !self.available { + return; + } + + let bucket = match self.buckets.entry(bucket_name) { Entry::Occupied(entry) => entry.into_mut(), Entry::Vacant(entry) => { let bucket = entry.key().clone(); @@ -118,12 +145,93 @@ where bucket.bytes = bucket.bytes.saturating_add(size); } - DurableMrfBacklogSummary { - available: true, - buckets: buckets.into_values().collect(), + fn into_summary(self) -> DurableMrfBacklogSummary { + if !self.available { + return DurableMrfBacklogSummary::default(); + } + + DurableMrfBacklogSummary { + available: true, + buckets: self.buckets.into_values().collect(), + } } } +#[derive(Debug, Clone, Default)] +struct MrfBacklogObservabilityTracker { + buckets: HashMap, +} + +impl MrfBacklogObservabilityTracker { + fn bucket_mut(&mut self, bucket_name: &str) -> &mut MrfBucketBacklogObservability { + match self.buckets.entry(bucket_name.to_string()) { + Entry::Occupied(entry) => entry.into_mut(), + Entry::Vacant(entry) => entry.insert(MrfBucketBacklogObservability { + bucket: bucket_name.to_string(), + ..Default::default() + }), + } + } + + fn add_pending(&mut self, entry: &MrfReplicateEntry) { + let Ok(size) = u64::try_from(entry.size) else { + return; + }; + let bucket = self.bucket_mut(&entry.bucket); + bucket.pending_count = bucket.pending_count.saturating_add(1); + bucket.pending_bytes = bucket.pending_bytes.saturating_add(size); + } + + fn flush_pending_entries<'a>(&mut self, entries: impl IntoIterator, duration_millis: u64) { + for entry in entries { + let Ok(size) = u64::try_from(entry.size) else { + continue; + }; + let bucket = self.bucket_mut(&entry.bucket); + bucket.pending_count = bucket.pending_count.saturating_sub(1); + bucket.pending_bytes = bucket.pending_bytes.saturating_sub(size); + bucket.last_flush_duration_millis = duration_millis; + } + } + + fn record_drop(&mut self, entry: &MrfReplicateEntry) { + let bucket = self.bucket_mut(&entry.bucket); + bucket.dropped_count = bucket.dropped_count.saturating_add(1); + } + + fn record_missed(&mut self, bucket_name: &str) { + let bucket = self.bucket_mut(bucket_name); + bucket.missed_count = bucket.missed_count.saturating_add(1); + } + + fn record_flush_failure(&mut self, duration_millis: u64) { + for bucket in self.buckets.values_mut().filter(|bucket| bucket.pending_count > 0) { + bucket.flush_failure_count = bucket.flush_failure_count.saturating_add(1); + bucket.last_flush_duration_millis = duration_millis; + } + } + + fn snapshot(&self) -> MrfBacklogObservabilitySummary { + MrfBacklogObservabilitySummary { + buckets: self.buckets.values().cloned().collect(), + } + } +} + +fn durable_mrf_backlog_summary_from_sizes(entries: I) -> DurableMrfBacklogSummary +where + I: IntoIterator, +{ + let mut tracker = DurableMrfBacklogTracker { + available: true, + ..Default::default() + }; + for (bucket_name, entry_size) in entries { + tracker.add_entry(bucket_name, entry_size); + } + tracker.into_summary() +} + fn set_durable_mrf_backlog_summary(summary: DurableMrfBacklogSummary) { match DURABLE_MRF_BACKLOG_SUMMARY.write() { Ok(mut guard) => *guard = summary, @@ -138,6 +246,40 @@ pub fn durable_mrf_backlog_summary_snapshot() -> DurableMrfBacklogSummary { } } +pub fn mrf_backlog_observability_snapshot() -> MrfBacklogObservabilitySummary { + match MRF_BACKLOG_OBSERVABILITY.read() { + Ok(guard) => guard.snapshot(), + Err(poisoned) => poisoned.into_inner().snapshot(), + } +} + +fn update_mrf_backlog_observability(mut update: impl FnMut(&mut MrfBacklogObservabilityTracker)) { + match MRF_BACKLOG_OBSERVABILITY.write() { + Ok(mut guard) => update(&mut guard), + Err(poisoned) => update(&mut poisoned.into_inner()), + } +} + +fn observe_mrf_pending(entry: &MrfReplicateEntry) { + update_mrf_backlog_observability(|tracker| tracker.add_pending(entry)); +} + +fn observe_mrf_pending_flushed(entries: &[MrfReplicateEntry], duration_millis: u64) { + update_mrf_backlog_observability(|tracker| tracker.flush_pending_entries(entries, duration_millis)); +} + +fn observe_mrf_drop(entry: &MrfReplicateEntry) { + update_mrf_backlog_observability(|tracker| tracker.record_drop(entry)); +} + +fn observe_mrf_missed(bucket: &str) { + update_mrf_backlog_observability(|tracker| tracker.record_missed(bucket)); +} + +fn observe_mrf_flush_failure(duration_millis: u64) { + update_mrf_backlog_observability(|tracker| tracker.record_flush_failure(duration_millis)); +} + fn durable_mrf_backlog_from_read(result: Result, EcstoreError>) -> DurableMrfBacklog { match result { Ok(data) => match decode_mrf_file(&data) { @@ -302,26 +444,8 @@ impl ReplicationPool { let handle = tokio::spawn(async move { let mut rx = rx; while let Some(operation) = rx.recv().await { - active_counter.fetch_add(1, Ordering::SeqCst); - - match operation { - ReplicationOperation::Object(obj_info) => { - let bucket = obj_info.bucket.clone(); - let size = obj_info.size; - let delete_marker = obj_info.delete_marker; - let op_type = obj_info.op_type; - replicate_object(*obj_info, storage.clone()).await; - stats.dec_q(&bucket, size, delete_marker, op_type); - } - ReplicationOperation::Delete(del_info) => { - let bucket = del_info.bucket.clone(); - let op_type = del_info.op_type; - replicate_delete(*del_info, storage.clone()).await; - stats.dec_q(&bucket, 0, true, op_type); - } - } - - active_counter.fetch_sub(1, Ordering::SeqCst); + let _active = ActiveWorkerGuard::new(active_counter.clone()); + process_replication_operation(operation, stats.clone(), storage.clone()).await; } }); @@ -379,30 +503,8 @@ impl ReplicationPool { let handle = tokio::spawn(async move { let mut rx = rx; while let Some(operation) = rx.recv().await { - active_counter.fetch_add(1, Ordering::SeqCst); - - match operation { - ReplicationOperation::Object(obj_info) => { - let bucket = obj_info.bucket.clone(); - let size = obj_info.size; - let delete_marker = obj_info.delete_marker; - let op_type = obj_info.op_type; - // Perform actual replication (placeholder) - replicate_object(*obj_info, storage.clone()).await; - - stats.dec_q(&bucket, size, delete_marker, op_type); - } - ReplicationOperation::Delete(del_info) => { - let bucket = del_info.bucket.clone(); - let op_type = del_info.op_type; - // Perform actual delete replication (placeholder) - replicate_delete(*del_info, storage.clone()).await; - - stats.dec_q(&bucket, 0, true, op_type); - } - } - - active_counter.fetch_sub(1, Ordering::SeqCst); + let _active = ActiveWorkerGuard::new(active_counter.clone()); + process_replication_operation(operation, stats.clone(), storage.clone()).await; } }); @@ -448,24 +550,8 @@ impl ReplicationPool { let operation = { mrf_rx.lock().await.recv().await }; let Some(operation) = operation else { break }; - active_counter.fetch_add(1, Ordering::SeqCst); - match operation { - ReplicationOperation::Object(obj_info) => { - let bucket = obj_info.bucket.clone(); - let size = obj_info.size; - let delete_marker = obj_info.delete_marker; - let op_type = obj_info.op_type; - replicate_object(*obj_info, storage.clone()).await; - stats.dec_q(&bucket, size, delete_marker, op_type); - } - ReplicationOperation::Delete(del_info) => { - let bucket = del_info.bucket.clone(); - let op_type = del_info.op_type; - replicate_delete(*del_info, storage.clone()).await; - stats.dec_q(&bucket, 0, true, op_type); - } - } - active_counter.fetch_sub(1, Ordering::SeqCst); + let _active = ActiveWorkerGuard::new(active_counter.clone()); + process_replication_operation(operation, stats.clone(), storage.clone()).await; } }); self.task_handles.lock().await.push(handle); @@ -902,6 +988,10 @@ impl ReplicationPool { // sustained failure storm can't grow it without limit. const MRF_PENDING_CAP: usize = 200_000; let mut pending: Vec = Vec::new(); + let mut durable_tracker = DurableMrfBacklogTracker { + available: true, + ..Default::default() + }; let mut flushed_len = 0usize; let mut dirty = false; let mut capped = false; @@ -915,6 +1005,7 @@ impl ReplicationPool { entry = rx.recv() => match entry { Some(e) => { if pending.len() >= MRF_PENDING_CAP { + observe_mrf_drop(&e); dec_mrf_entries(stats.as_ref(), std::slice::from_ref(&e)); if !capped { capped = true; @@ -927,16 +1018,19 @@ impl ReplicationPool { } continue; } + durable_tracker.add_entry(e.bucket.clone(), e.size); + observe_mrf_pending(&e); pending.push(e); dirty = true; // Flush eagerly once enough new entries have accumulated // since the last write (measured against the flushed // set, not the absolute length, so a large backlog is // not rewritten on every single add). - if pending.len() - flushed_len >= 1000 && flush_mrf_to_disk(&pending, &storage).await { - set_durable_mrf_backlog_summary(durable_mrf_backlog_summary_from_sizes( - pending.iter().map(|entry| (entry.bucket.clone(), entry.size)), - )); + 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()); + observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis); dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]); flushed_len = pending.len(); dirty = false; @@ -944,20 +1038,18 @@ impl ReplicationPool { } None => { // Channel closed (pool shutting down) — final flush. - if dirty && flush_mrf_to_disk(&pending, &storage).await { - set_durable_mrf_backlog_summary(durable_mrf_backlog_summary_from_sizes( - pending.iter().map(|entry| (entry.bucket.clone(), entry.size)), - )); + if dirty && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await { + set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary()); + observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis); dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]); } break; } }, _ = interval.tick() => { - if dirty && flush_mrf_to_disk(&pending, &storage).await { - set_durable_mrf_backlog_summary(durable_mrf_backlog_summary_from_sizes( - pending.iter().map(|entry| (entry.bucket.clone(), entry.size)), - )); + if dirty && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await { + set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary()); + observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis); dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]); flushed_len = pending.len(); dirty = false; @@ -977,30 +1069,8 @@ impl ReplicationPool { stats: Arc, ) { while let Some(operation) = rx.recv().await { - active_counter.fetch_add(1, Ordering::SeqCst); - - match operation { - ReplicationOperation::Object(obj_info) => { - let bucket = obj_info.bucket.clone(); - let size = obj_info.size; - let delete_marker = obj_info.delete_marker; - let op_type = obj_info.op_type; - // Perform actual replication (placeholder) - replicate_object(*obj_info, self.storage.clone()).await; - - stats.dec_q(&bucket, size, delete_marker, op_type); - } - ReplicationOperation::Delete(del_info) => { - let bucket = del_info.bucket.clone(); - let op_type = del_info.op_type; - // Perform actual delete replication (placeholder) - replicate_delete(*del_info, self.storage.clone()).await; - - stats.dec_q(&bucket, 0, true, op_type); - } - } - - active_counter.fetch_sub(1, Ordering::SeqCst); + let _active = ActiveWorkerGuard::new(active_counter.clone()); + process_replication_operation(operation, stats.clone(), self.storage.clone()).await; } } @@ -1013,26 +1083,8 @@ impl ReplicationPool { storage: Arc, ) { while let Some(operation) = rx.recv().await { - active_counter.fetch_add(1, Ordering::SeqCst); - - match operation { - ReplicationOperation::Object(obj_info) => { - let bucket = obj_info.bucket.clone(); - let size = obj_info.size; - let delete_marker = obj_info.delete_marker; - let op_type = obj_info.op_type; - replicate_object(*obj_info, storage.clone()).await; - stats.dec_q(&bucket, size, delete_marker, op_type); - } - ReplicationOperation::Delete(del_info) => { - let bucket = del_info.bucket.clone(); - let op_type = del_info.op_type; - replicate_delete(*del_info, storage.clone()).await; - stats.dec_q(&bucket, 0, true, op_type); - } - } - - active_counter.fetch_sub(1, Ordering::SeqCst); + let _active = ActiveWorkerGuard::new(active_counter.clone()); + process_replication_operation(operation, stats.clone(), storage.clone()).await; } } @@ -1044,27 +1096,8 @@ impl ReplicationPool { stats: Arc, ) { while let Some(operation) = rx.recv().await { - active_counter.fetch_add(1, Ordering::SeqCst); - - match operation { - ReplicationOperation::Object(obj_info) => { - let bucket = obj_info.bucket.clone(); - let size = obj_info.size; - let delete_marker = obj_info.delete_marker; - let op_type = obj_info.op_type; - replicate_object(obj_info.as_ref().clone(), self.storage.clone()).await; - - stats.dec_q(&bucket, size, delete_marker, op_type); - } - ReplicationOperation::Delete(del_info) => { - let bucket = del_info.bucket.clone(); - let op_type = del_info.op_type; - replicate_delete(*del_info, self.storage.clone()).await; - stats.dec_q(&bucket, 0, true, op_type); - } - } - - active_counter.fetch_sub(1, Ordering::SeqCst); + let _active = ActiveWorkerGuard::new(active_counter.clone()); + process_replication_operation(operation, stats.clone(), self.storage.clone()).await; } } @@ -1395,6 +1428,76 @@ impl ReplicationPool { } } +struct ActiveWorkerGuard { + counter: Arc, +} + +impl ActiveWorkerGuard { + fn new(counter: Arc) -> Self { + counter.fetch_add(1, Ordering::SeqCst); + Self { counter } + } +} + +impl Drop for ActiveWorkerGuard { + fn drop(&mut self) { + self.counter.fetch_sub(1, Ordering::SeqCst); + } +} + +struct ReplicationBacklogGuard { + stats: Arc, + bucket: String, + size: i64, + is_delete_marker: bool, + op_type: ReplicationType, +} + +impl ReplicationBacklogGuard { + fn for_object(stats: Arc, object: &ReplicateObjectInfo) -> Self { + Self { + stats, + bucket: object.bucket.clone(), + size: object.size, + is_delete_marker: object.delete_marker, + op_type: object.op_type, + } + } + + fn for_delete(stats: Arc, delete: &DeletedObjectReplicationInfo) -> Self { + Self { + stats, + bucket: delete.bucket.clone(), + size: 0, + is_delete_marker: true, + op_type: delete.op_type, + } + } +} + +impl Drop for ReplicationBacklogGuard { + fn drop(&mut self) { + self.stats.dec_q(&self.bucket, self.size, self.is_delete_marker, self.op_type); + } +} + +async fn process_replication_operation( + operation: ReplicationOperation, + stats: Arc, + storage: Arc, +) { + match operation { + ReplicationOperation::Object(obj_info) => { + let _backlog = ReplicationBacklogGuard::for_object(stats, obj_info.as_ref()); + replicate_object(*obj_info, storage).await; + } + ReplicationOperation::Delete(del_info) => { + let _backlog = ReplicationBacklogGuard::for_delete(stats, del_info.as_ref()); + replicate_delete(*del_info, storage).await; + } + } +} + async fn queue_mrf_save_entry( tx: &Sender, entry: MrfReplicateEntry, @@ -1414,6 +1517,7 @@ async fn queue_mrf_save_entry( queue_type = queue_type, "MRF save channel unavailable — replication failure entry could not be persisted for retry" ); + observe_mrf_missed(&entry.bucket); ReplicationQueueAdmission::Missed } @@ -1424,15 +1528,18 @@ fn dec_mrf_entries(stats: &ReplicationStats, entries: &[MrfReplicateEntry]) { } /// Encodes `entries` and overwrites the MRF persistence file. -/// Returns `true` on success; on failure logs the error and returns `false`. -/// Callers must NOT clear their in-memory buffer on `false` so the next tick +/// Returns the flush duration on success; on failure logs the error and returns `None`. +/// Callers must NOT clear their in-memory buffer on `None` so the next tick /// can retry — otherwise a transient storage error permanently drops the batch. -async fn flush_mrf_to_disk(entries: &[MrfReplicateEntry], storage: &Arc) -> bool { +async fn flush_mrf_to_disk(entries: &[MrfReplicateEntry], storage: &Arc) -> Option { + let started = Instant::now(); match encode_mrf_file(entries) { Ok(data) => { if let Err(e) = ReplicationConfigStore::save(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE, data).await { + let duration_millis = duration_millis_u64(started.elapsed()); + observe_mrf_flush_failure(duration_millis); warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, @@ -1440,11 +1547,12 @@ async fn flush_mrf_to_disk(entries: &[MrfReplicateEntry] error = %e, "Failed to flush MRF entries to disk" ); - return false; + return None; } - true + Some(duration_millis_u64(started.elapsed())) } Err(e) => { + observe_mrf_flush_failure(0); warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, @@ -1452,11 +1560,15 @@ async fn flush_mrf_to_disk(entries: &[MrfReplicateEntry] error = %e, "Failed to encode MRF entries for disk flush" ); - false + None } } } +fn duration_millis_u64(duration: std::time::Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + /// Load bucket resync metadata from disk async fn load_bucket_resync_metadata( bucket: &str, @@ -2664,6 +2776,132 @@ mod tests { assert_eq!(received.object, "second"); } + #[tokio::test] + async fn mrf_save_admission_records_missed_when_channel_is_closed() { + let (tx, rx) = mpsc::channel(1); + drop(rx); + let bucket = "mrf-missed-hook-bucket"; + + let admission = queue_mrf_save_entry( + &tx, + MrfReplicateEntry { + bucket: bucket.to_string(), + object: "missed".to_string(), + version_id: None, + retry_count: 1, + size: 1, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + }, + "test", + ) + .await; + + assert_eq!(admission, ReplicationQueueAdmission::Missed); + let snapshot = mrf_backlog_observability_snapshot(); + let bucket = snapshot + .buckets + .iter() + .find(|stats| stats.bucket == "mrf-missed-hook-bucket") + .expect("missed MRF admission should be observable"); + assert_eq!(bucket.missed_count, 1); + } + + #[tokio::test] + async fn mrf_flush_failure_keeps_pending_backlog_observable() { + let shared = empty_resync_shared_state(); + shared.fail_next_write.store(true, Ordering::SeqCst); + let storage = Arc::new(LoadResyncNodeStore::new("mrf-flush-failure", shared)); + let entry = MrfReplicateEntry { + bucket: "mrf-flush-failure-bucket".to_string(), + object: "pending".to_string(), + version_id: None, + retry_count: 1, + size: 2048, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + }; + observe_mrf_pending(&entry); + + let result = flush_mrf_to_disk(std::slice::from_ref(&entry), &storage).await; + + assert_eq!(result, None); + let snapshot = mrf_backlog_observability_snapshot(); + let bucket = snapshot + .buckets + .iter() + .find(|stats| stats.bucket == "mrf-flush-failure-bucket") + .expect("failed MRF flush should keep the bucket observable"); + assert_eq!(bucket.pending_count, 1); + assert_eq!(bucket.pending_bytes, 2048); + assert_eq!(bucket.flush_failure_count, 1); + } + + #[tokio::test] + async fn replication_backlog_guard_decrements_on_drop() { + let stats = Arc::new(ReplicationStats::new()); + stats.inc_q("guard-bucket", 256, false, ReplicationType::Object); + + { + let object = ReplicateObjectInfo { + bucket: "guard-bucket".to_string(), + size: 256, + op_type: ReplicationType::Object, + ..Default::default() + }; + let _guard = ReplicationBacklogGuard::for_object(stats.clone(), &object); + } + + 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); + } + + #[test] + fn mrf_observability_tracker_separates_pending_drop_miss_and_flush_failure() { + let first = MrfReplicateEntry { + bucket: "tracker-bucket".to_string(), + object: "first".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, + }; + let second = MrfReplicateEntry { + object: "second".to_string(), + size: 512, + ..first.clone() + }; + let mut tracker = MrfBacklogObservabilityTracker::default(); + + tracker.add_pending(&first); + tracker.add_pending(&second); + tracker.record_drop(&second); + tracker.record_missed("tracker-bucket"); + tracker.record_flush_failure(7); + tracker.flush_pending_entries([&first], 11); + + let snapshot = tracker.snapshot(); + let bucket = snapshot + .buckets + .iter() + .find(|stats| stats.bucket == "tracker-bucket") + .expect("tracker bucket should be present"); + assert_eq!(bucket.pending_count, 1); + assert_eq!(bucket.pending_bytes, 512); + assert_eq!(bucket.dropped_count, 1); + assert_eq!(bucket.missed_count, 1); + assert_eq!(bucket.flush_failure_count, 1); + assert_eq!(bucket.last_flush_duration_millis, 11); + } + #[test] fn auto_resume_resync_only_for_inflight_states() { assert!(should_auto_resume_resync(ResyncStatusType::ResyncPending)); diff --git a/crates/obs/src/metrics/collectors/bucket_replication.rs b/crates/obs/src/metrics/collectors/bucket_replication.rs index 32d7151fd..23891bc8e 100644 --- a/crates/obs/src/metrics/collectors/bucket_replication.rs +++ b/crates/obs/src/metrics/collectors/bucket_replication.rs @@ -20,6 +20,8 @@ use crate::metrics::schema::bucket_replication::{ 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_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, BUCKET_REPL_PROXIED_DELETE_TAGGING_REQUESTS_FAILURES_MD, 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, @@ -34,7 +36,7 @@ use crate::metrics::schema::bucket_replication::{ use std::borrow::Cow; const BASE_BUCKET_REPLICATION_METRICS_PER_BUCKET: usize = 25; -const BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET: usize = 5; +const BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET: usize = 11; #[derive(Debug, Clone, Default)] pub struct BucketReplicationTargetStats { @@ -91,6 +93,12 @@ pub(crate) struct BucketReplicationBacklogStats { pub(crate) durable_mrf_available: bool, pub(crate) durable_mrf_backlog_count: u64, pub(crate) durable_mrf_backlog_bytes: u64, + pub(crate) mrf_pending_count: u64, + pub(crate) mrf_pending_bytes: u64, + pub(crate) mrf_dropped_count: u64, + pub(crate) mrf_missed_count: u64, + pub(crate) mrf_flush_failures: u64, + pub(crate) mrf_last_flush_duration_millis: u64, } pub fn collect_bucket_replication_bandwidth_metrics(stats: &[BucketReplicationBandwidthStats]) -> Vec { @@ -309,6 +317,34 @@ pub(crate) fn collect_bucket_replication_backlog_metrics(stats: &[BucketReplicat PrometheusMetric::from_descriptor(&BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD, stat.durable_mrf_backlog_bytes as f64) .with_label(BUCKET_L, bucket_label), ); + let bucket_label: Cow<'static, str> = Cow::Owned(stat.bucket.clone()); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_PENDING_COUNT_MD, stat.mrf_pending_count as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_PENDING_BYTES_MD, stat.mrf_pending_bytes as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_DROPPED_COUNT_MD, stat.mrf_dropped_count as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_MISSED_COUNT_MD, stat.mrf_missed_count as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_FLUSH_FAILURES_MD, stat.mrf_flush_failures as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor( + &BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD, + stat.mrf_last_flush_duration_millis as f64, + ) + .with_label(BUCKET_L, bucket_label), + ); } metrics @@ -427,10 +463,16 @@ mod tests { durable_mrf_available: true, durable_mrf_backlog_count: 2, durable_mrf_backlog_bytes: 2048, + mrf_pending_count: 1, + mrf_pending_bytes: 512, + mrf_dropped_count: 3, + mrf_missed_count: 4, + mrf_flush_failures: 5, + mrf_last_flush_duration_millis: 6, }]; let metrics = collect_bucket_replication_backlog_metrics(&stats); - assert_eq!(metrics.len(), 5); + assert_eq!(metrics.len(), 11); let backlog_count_name = BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD.get_full_metric_name(); assert!(metrics.iter().any(|metric| { @@ -466,6 +508,48 @@ mod tests { && metric.value == 2048.0 && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") })); + + let pending_count_name = BUCKET_REPL_MRF_PENDING_COUNT_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == pending_count_name + && metric.value == 1.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let pending_bytes_name = BUCKET_REPL_MRF_PENDING_BYTES_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == pending_bytes_name + && metric.value == 512.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let dropped_count_name = BUCKET_REPL_MRF_DROPPED_COUNT_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == dropped_count_name + && metric.value == 3.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let missed_count_name = BUCKET_REPL_MRF_MISSED_COUNT_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == missed_count_name + && metric.value == 4.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let flush_failures_name = BUCKET_REPL_MRF_FLUSH_FAILURES_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == flush_failures_name + && metric.value == 5.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let flush_duration_name = BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == flush_duration_name + && metric.value == 6.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); } #[test] @@ -484,6 +568,12 @@ mod tests { durable_mrf_available: true, durable_mrf_backlog_count: 2, durable_mrf_backlog_bytes: 4096, + mrf_pending_count: 1, + mrf_pending_bytes: 2, + mrf_dropped_count: 3, + mrf_missed_count: 4, + mrf_flush_failures: 5, + mrf_last_flush_duration_millis: 6, }]; let metrics = collect_bucket_replication_backlog_metrics(&stats); @@ -493,6 +583,12 @@ mod tests { BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD.get_full_metric_name(), BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD.get_full_metric_name(), BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD.get_full_metric_name(), + BUCKET_REPL_MRF_PENDING_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_MRF_PENDING_BYTES_MD.get_full_metric_name(), + BUCKET_REPL_MRF_DROPPED_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_MRF_MISSED_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_MRF_FLUSH_FAILURES_MD.get_full_metric_name(), + BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD.get_full_metric_name(), ]; for name in backlog_names { diff --git a/crates/obs/src/metrics/schema/bucket_replication.rs b/crates/obs/src/metrics/schema/bucket_replication.rs index e029774ef..28237ecdf 100644 --- a/crates/obs/src/metrics/schema/bucket_replication.rs +++ b/crates/obs/src/metrics/schema/bucket_replication.rs @@ -38,6 +38,12 @@ const CURRENT_BACKLOG_BYTES: &str = "current_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 MRF_PENDING_COUNT: &str = "mrf_pending_count"; +const MRF_PENDING_BYTES: &str = "mrf_pending_bytes"; +const MRF_DROPPED_COUNT: &str = "mrf_dropped_count"; +const MRF_MISSED_COUNT: &str = "mrf_missed_count"; +const MRF_FLUSH_FAILURES: &str = "mrf_flush_failures"; +const MRF_LAST_FLUSH_DURATION_MILLIS: &str = "mrf_last_flush_duration_millis"; pub static BUCKET_REPL_LAST_HR_FAILED_BYTES_MD: LazyLock = LazyLock::new(|| { new_gauge_md( @@ -129,6 +135,60 @@ pub static BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD: LazyLock ) }); +pub static BUCKET_REPL_MRF_PENDING_COUNT_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(MRF_PENDING_COUNT), + "Current number of MRF entries waiting to be flushed to the durable recovery file for a bucket on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_MRF_PENDING_BYTES_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(MRF_PENDING_BYTES), + "Current bytes represented by MRF entries waiting to be flushed to the durable recovery file for a bucket on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_MRF_DROPPED_COUNT_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(MRF_DROPPED_COUNT), + "Total number of MRF entries dropped after the bounded pending backlog cap was reached for a bucket on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_MRF_MISSED_COUNT_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(MRF_MISSED_COUNT), + "Total number of MRF entries that could not be admitted to the save channel for a bucket on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_MRF_FLUSH_FAILURES_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(MRF_FLUSH_FAILURES), + "Total number of durable MRF flush failures observed while a bucket had pending MRF entries on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(MRF_LAST_FLUSH_DURATION_MILLIS), + "Duration in milliseconds of the last durable MRF flush that touched pending entries for a bucket on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + pub static BUCKET_REPL_PROXIED_DELETE_TAGGING_REQUESTS_TOTAL_MD: LazyLock = LazyLock::new(|| { new_counter_md( MetricName::ProxiedDeleteTaggingRequestsTotal, diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index f9040cc3c..170279e37 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -206,6 +206,12 @@ async fn obs_bucket_replication_stats_bundle() -> (Vec, durable_mrf_available: stats.durable_mrf_available, durable_mrf_backlog_count: stats.durable_mrf_backlog_count, durable_mrf_backlog_bytes: stats.durable_mrf_backlog_bytes, + mrf_pending_count: stats.mrf_pending_count, + mrf_pending_bytes: stats.mrf_pending_bytes, + mrf_dropped_count: stats.mrf_dropped_count, + 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, }); 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 55221d7aa..e1286ebe1 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, durable_mrf_backlog_summary_snapshot, get_global_replication_stats, + DurableMrfBucketBacklog, MrfBucketBacklogObservability, durable_mrf_backlog_summary_snapshot, get_global_replication_stats, + mrf_backlog_observability_snapshot, }; pub(crate) use rustfs_ecstore::api::capacity::{ get_total_usable_capacity as obs_get_total_usable_capacity, @@ -76,6 +77,12 @@ pub(crate) struct ObsBucketReplicationStatsSnapshot { pub(crate) durable_mrf_available: bool, pub(crate) durable_mrf_backlog_count: u64, pub(crate) durable_mrf_backlog_bytes: u64, + pub(crate) mrf_pending_count: u64, + pub(crate) mrf_pending_bytes: u64, + pub(crate) mrf_dropped_count: u64, + pub(crate) mrf_missed_count: u64, + pub(crate) mrf_flush_failures: u64, + pub(crate) mrf_last_flush_duration_millis: u64, pub(crate) targets: Vec, } @@ -152,6 +159,7 @@ fn bucket_replication_stats_snapshot_from_parts( proxy: ObsBucketReplicationProxySnapshot, durable_mrf_available: bool, durable_bucket: DurableMrfBucketBacklog, + mrf_observability: MrfBucketBacklogObservability, ) -> ObsBucketReplicationStatsSnapshot { ObsBucketReplicationStatsSnapshot { bucket, @@ -185,6 +193,12 @@ fn bucket_replication_stats_snapshot_from_parts( 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, targets: runtime.targets, } } @@ -196,7 +210,8 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec Vec>(); - let mut bucket_names = Vec::with_capacity(all_bucket_stats.len().saturating_add(durable_buckets.len())); + let mrf_observability = if replication_storage_available { + mrf_backlog_observability_snapshot() + } else { + Default::default() + }; + let mrf_observability_buckets = mrf_observability + .buckets + .into_iter() + .map(|bucket| (bucket.bucket.clone(), bucket)) + .collect::>(); + let mut bucket_names = Vec::with_capacity( + all_bucket_stats + .len() + .saturating_add(durable_buckets.len()) + .saturating_add(mrf_observability_buckets.len()), + ); bucket_names.extend(all_bucket_stats.keys().cloned()); bucket_names.extend( durable_buckets @@ -215,6 +245,12 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec Vec