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