diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index d2ddae305..b2af30c62 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -172,6 +172,9 @@ pub mod bucket { } pub mod replication { + pub use crate::bucket::replication::replication_pool::{ + DurableMrfBacklogSummary, DurableMrfBucketBacklog, durable_mrf_backlog_summary_snapshot, + }; pub use crate::bucket::replication::{ BucketReplicationResyncStatus, BucketStats, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, REPLICATE_INCOMING_DELETE, ReplicateDecision, diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index f2972aae9..245590977 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -46,7 +46,11 @@ use super::replication_target_boundary::{ReplicationTargetStore, replication_obj use super::replication_versioning_boundary::ReplicationVersioningStore; use super::runtime_boundary as runtime_sources; use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str}; +use std::collections::HashMap; +use std::collections::hash_map::Entry; use std::sync::Arc; +use std::sync::LazyLock; +use std::sync::RwLock as StdRwLock; use std::sync::atomic::AtomicI32; use std::sync::atomic::Ordering; use time::OffsetDateTime; @@ -74,6 +78,66 @@ pub struct DurableMrfBacklog { pub entries: Vec, } +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct DurableMrfBucketBacklog { + pub bucket: String, + pub count: u64, + pub bytes: u64, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct DurableMrfBacklogSummary { + pub available: bool, + pub buckets: Vec, +} + +static DURABLE_MRF_BACKLOG_SUMMARY: LazyLock> = + LazyLock::new(|| StdRwLock::new(DurableMrfBacklogSummary::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 { + let Ok(size) = u64::try_from(entry_size) else { + return DurableMrfBacklogSummary::default(); + }; + + let bucket = match buckets.entry(bucket_name) { + Entry::Occupied(entry) => entry.into_mut(), + Entry::Vacant(entry) => { + let bucket = entry.key().clone(); + entry.insert(DurableMrfBucketBacklog { + bucket, + ..Default::default() + }) + } + }; + bucket.count = bucket.count.saturating_add(1); + bucket.bytes = bucket.bytes.saturating_add(size); + } + + DurableMrfBacklogSummary { + available: true, + buckets: buckets.into_values().collect(), + } +} + +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, + } +} + +pub fn durable_mrf_backlog_summary_snapshot() -> DurableMrfBacklogSummary { + match DURABLE_MRF_BACKLOG_SUMMARY.read() { + Ok(guard) => guard.clone(), + Err(poisoned) => poisoned.into_inner().clone(), + } +} + fn durable_mrf_backlog_from_read(result: Result, EcstoreError>) -> DurableMrfBacklog { match result { Ok(data) => match decode_mrf_file(&data) { @@ -233,6 +297,7 @@ impl ReplicationPool { let active_counter = self.active_lrg_workers.clone(); let storage = self.storage.clone(); + let stats = self.stats.clone(); let handle = tokio::spawn(async move { let mut rx = rx; @@ -241,10 +306,18 @@ impl ReplicationPool { 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); } } @@ -310,23 +383,22 @@ impl ReplicationPool { match operation { ReplicationOperation::Object(obj_info) => { - stats - .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) - .await; - + 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.as_ref().clone(), storage.clone()).await; + replicate_object(*obj_info, storage.clone()).await; - stats - .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) - .await; + stats.dec_q(&bucket, size, delete_marker, op_type); } ReplicationOperation::Delete(del_info) => { - stats.inc_q(&del_info.bucket, 0, true, del_info.op_type).await; + let bucket = del_info.bucket.clone(); + let op_type = del_info.op_type; // Perform actual delete replication (placeholder) - replicate_delete(del_info.as_ref().clone(), storage.clone()).await; + replicate_delete(*del_info, storage.clone()).await; - stats.dec_q(&del_info.bucket, 0, true, del_info.op_type).await; + stats.dec_q(&bucket, 0, true, op_type); } } @@ -379,16 +451,18 @@ impl ReplicationPool { active_counter.fetch_add(1, Ordering::SeqCst); match operation { ReplicationOperation::Object(obj_info) => { - stats - .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) - .await; - replicate_object(obj_info.as_ref().clone(), storage.clone()).await; - stats - .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) - .await; + 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); @@ -529,9 +603,13 @@ impl ReplicationPool { if !lrg_workers.is_empty() { let index = (hash as usize) % lrg_workers.len(); - if let Some(worker) = lrg_workers.get(index) - && worker.try_send(ReplicationOperation::Object(Box::new(ri.clone()))).is_err() - { + if let Some(worker) = lrg_workers.get(index) { + self.stats.inc_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type); + 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); + // Try to add more workers if possible let max_l_workers = *self.max_l_workers.read().await; let existing = lrg_workers.len(); @@ -539,17 +617,13 @@ impl ReplicationPool { drop(lrg_workers); // Queue to MRF if worker is busy. - let admission = - queue_mrf_save_admission(&self.mrf_save_tx, ri.to_mrf_entry(), &ri.bucket, &ri.name, "large_object") - .await; + let admission = self.queue_mrf_save_admission(ri.to_mrf_entry(), "large_object").await; if let Some(resize) = resize { self.resize_lrg_workers(resize.new_count, resize.existing_count).await; } return admission; } - - return ReplicationQueueAdmission::Queued; } return ReplicationQueueAdmission::Missed; } @@ -562,12 +636,14 @@ impl ReplicationPool { return ReplicationQueueAdmission::Missed; }; + self.stats.inc_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type); 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); // Queue to MRF if all workers are busy. - let admission = queue_mrf_save_admission(&self.mrf_save_tx, ri.to_mrf_entry(), &ri.bucket, &ri.name, "object").await; + let admission = self.queue_mrf_save_admission(ri.to_mrf_entry(), "object").await; // Try to scale up workers based on priority self.apply_queue_backpressure("object", true, "Replication queue is backpressured") @@ -586,18 +662,13 @@ impl ReplicationPool { return ReplicationQueueAdmission::Missed; }; + self.stats.inc_q(&doi.bucket, 0, true, doi.op_type); 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); - let admission = queue_mrf_save_admission( - &self.mrf_save_tx, - doi.to_mrf_entry(), - &doi.bucket, - &doi.delete_object.object_name, - "delete", - ) - .await; + let admission = self.queue_mrf_save_admission(doi.to_mrf_entry(), "delete").await; self.apply_queue_backpressure("delete", false, "Replication delete queue is backpressured") .await; @@ -607,7 +678,18 @@ impl ReplicationPool { /// Queues an MRF save operation async fn queue_mrf_save(&self, entry: MrfReplicateEntry) { - let _ = queue_mrf_save_admission(&self.mrf_save_tx, entry, "", "", "mrf_worker").await; + let _ = self.queue_mrf_save_admission(entry, "mrf_worker").await; + } + + async fn queue_mrf_save_admission(&self, entry: MrfReplicateEntry, queue_type: &'static str) -> ReplicationQueueAdmission { + let bucket = entry.bucket.clone(); + let size = entry.size; + let is_delete = matches!(entry.op, MrfOpKind::Delete); + 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); + } + admission } /// Starts the MRF processor — one-shot at startup. @@ -622,7 +704,13 @@ impl ReplicationPool { let handle = tokio::spawn(async move { let data = match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await { Ok(d) => d, - Err(EcstoreError::ConfigNotFound) => return, // no file yet — normal on first start + Err(EcstoreError::ConfigNotFound) => { + set_durable_mrf_backlog_summary(DurableMrfBacklogSummary { + available: true, + buckets: Vec::new(), + }); + return; + } Err(e) => { warn!( component = LOG_COMPONENT_ECSTORE, @@ -653,6 +741,9 @@ impl ReplicationPool { return; } }; + set_durable_mrf_backlog_summary(durable_mrf_backlog_summary_from_sizes( + entries.iter().map(|entry| (entry.bucket.clone(), entry.size)), + )); let total = entries.len(); let mut queued_count = 0usize; @@ -765,6 +856,11 @@ impl ReplicationPool { error = %e, "Failed to clear MRF recovery file after replay — entries may be replayed again on next restart" ); + } else { + set_durable_mrf_backlog_summary(DurableMrfBacklogSummary { + available: true, + buckets: Vec::new(), + }); } if queued_count > 0 { @@ -793,6 +889,7 @@ impl ReplicationPool { return; }; let storage = self.storage.clone(); + let stats = self.stats.clone(); let handle = tokio::spawn(async move { // The on-disk MRF file is a restart-recovery backstop: entries are @@ -818,6 +915,7 @@ impl ReplicationPool { entry = rx.recv() => match entry { Some(e) => { if pending.len() >= MRF_PENDING_CAP { + dec_mrf_entries(stats.as_ref(), std::slice::from_ref(&e)); if !capped { capped = true; warn!( @@ -836,20 +934,31 @@ impl ReplicationPool { // 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)), + )); + dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]); flushed_len = pending.len(); dirty = false; } } None => { // Channel closed (pool shutting down) — final flush. - if dirty { - flush_mrf_to_disk(&pending, &storage).await; + 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)), + )); + 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)), + )); + dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]); flushed_len = pending.len(); dirty = false; } @@ -872,24 +981,22 @@ impl ReplicationPool { match operation { ReplicationOperation::Object(obj_info) => { - stats - .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) - .await; - + 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.as_ref().clone(), self.storage.clone()).await; + replicate_object(*obj_info, self.storage.clone()).await; - stats - .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) - .await; + stats.dec_q(&bucket, size, delete_marker, op_type); } ReplicationOperation::Delete(del_info) => { - stats.inc_q(&del_info.bucket, 0, true, del_info.op_type).await; - + let bucket = del_info.bucket.clone(); + let op_type = del_info.op_type; // Perform actual delete replication (placeholder) - replicate_delete(del_info.as_ref().clone(), self.storage.clone()).await; + replicate_delete(*del_info, self.storage.clone()).await; - stats.dec_q(&del_info.bucket, 0, true, del_info.op_type).await; + stats.dec_q(&bucket, 0, true, op_type); } } @@ -898,16 +1005,30 @@ impl ReplicationPool { } /// Worker function for handling large object replication operations - async fn add_large_worker(&self, mut rx: Receiver, active_counter: Arc, storage: Arc) { + async fn add_large_worker( + &self, + mut rx: Receiver, + active_counter: Arc, + stats: Arc, + 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); } } @@ -927,18 +1048,19 @@ impl ReplicationPool { match operation { ReplicationOperation::Object(obj_info) => { - stats - .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) - .await; - + 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(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) - .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); } } @@ -1273,29 +1395,34 @@ impl ReplicationPool { } } -async fn queue_mrf_save_admission( +async fn queue_mrf_save_entry( tx: &Sender, entry: MrfReplicateEntry, - bucket: &str, - object: &str, queue_type: &'static str, ) -> ReplicationQueueAdmission { - if tx.send(entry).await.is_ok() { + let Err(error) = tx.send(entry).await else { return ReplicationQueueAdmission::Queued; - } + }; + let entry = error.0; warn!( event = EVENT_REPLICATION_MRF_QUEUE_UNAVAILABLE, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, - bucket = %bucket, - object = %object, + bucket = %entry.bucket, + object = %entry.object, queue_type = queue_type, "MRF save channel unavailable — replication failure entry could not be persisted for retry" ); ReplicationQueueAdmission::Missed } +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); + } +} + /// 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 @@ -2016,6 +2143,147 @@ mod tests { }) } + async fn current_queue(pool: &ReplicationPool, bucket: &str) -> (i64, i64) { + let stats = pool.stats.get_latest_replication_stats(bucket).await; + (stats.replication_stats.q_stat.curr.count, stats.replication_stats.q_stat.curr.bytes) + } + + async fn wait_for_current_queue(pool: &ReplicationPool, bucket: &str, expected: (i64, i64)) { + tokio::time::timeout(Duration::from_secs(10), async { + loop { + if current_queue(pool, bucket).await == expected { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("replication queue should reach the expected state"); + } + + #[tokio::test] + async fn regular_worker_admission_counts_channel_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: "admission-bucket".to_string(), + name: "object".to_string(), + size: 4096, + op_type: ReplicationType::Object, + ..Default::default() + }) + .await; + + assert_eq!(admission, ReplicationQueueAdmission::Queued); + assert_eq!(current_queue(&pool, "admission-bucket").await, (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; + let (tx, _rx) = mpsc::channel(1); + pool.lrg_workers.write().await.push(tx); + let size = 128 * 1024 * 1024; + + let admission = pool + .queue_replica_task(ReplicateObjectInfo { + bucket: "large-admission-bucket".to_string(), + name: "large-object".to_string(), + size, + op_type: ReplicationType::Object, + ..Default::default() + }) + .await; + + assert_eq!(admission, ReplicationQueueAdmission::Queued); + assert_eq!(current_queue(&pool, "large-admission-bucket").await, (1, size)); + } + + #[tokio::test] + async fn delete_admission_counts_channel_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-admission-bucket".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-admission-bucket").await, (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; + pool.resize_workers(1, 0).await; + + let admission = pool + .queue_replica_task(ReplicateObjectInfo { + bucket: "regular-drain-bucket".to_string(), + name: "object".to_string(), + size: 4096, + op_type: ReplicationType::Object, + ..Default::default() + }) + .await; + + assert_eq!(admission, ReplicationQueueAdmission::Queued); + wait_for_current_queue(&pool, "regular-drain-bucket", (0, 0)).await; + } + + #[tokio::test] + async fn large_worker_drains_current_backlog_after_processing() { + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await; + pool.resize_lrg_workers(1, 0).await; + let size = 128 * 1024 * 1024; + + let admission = pool + .queue_replica_task(ReplicateObjectInfo { + bucket: "large-drain-bucket".to_string(), + name: "large-object".to_string(), + size, + op_type: ReplicationType::Object, + ..Default::default() + }) + .await; + + assert_eq!(admission, ReplicationQueueAdmission::Queued); + wait_for_current_queue(&pool, "large-drain-bucket", (0, 0)).await; + } + + #[tokio::test] + async fn regular_delete_worker_drains_current_backlog_after_processing() { + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await; + pool.resize_workers(1, 0).await; + + let admission = pool + .queue_replica_delete_task(DeletedObjectReplicationInfo { + bucket: "delete-drain-bucket".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); + wait_for_current_queue(&pool, "delete-drain-bucket", (0, 0)).await; + } + fn load_resync_test_metadata() -> Vec { let mut status = BucketReplicationResyncStatus::new(); status.targets_map.insert( @@ -2301,6 +2569,37 @@ mod tests { assert_eq!(admission, ReplicationQueueAdmission::Missed); } + #[tokio::test] + async fn queue_replica_task_counts_mrf_pending_backlog_when_worker_queue_is_full() { + let shared = empty_resync_shared_state(); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared))).await; + let (tx, _rx) = mpsc::channel(1); + tx.try_send(ReplicationOperation::Object(Box::new(ReplicateObjectInfo { + bucket: "runtime-backlog".to_string(), + name: "already-buffered".to_string(), + size: 1, + op_type: ReplicationType::Object, + ..Default::default() + }))) + .expect("test setup should fill the worker queue"); + pool.workers.write().await.push(tx); + + let admission = pool + .queue_replica_task(ReplicateObjectInfo { + bucket: "runtime-backlog".to_string(), + name: "fallback-object".to_string(), + size: 2048, + op_type: ReplicationType::Object, + ..Default::default() + }) + .await; + + assert_eq!(admission, ReplicationQueueAdmission::Queued); + 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); + } + #[test] fn replicate_object_info_from_object_info_preserves_ssec_checksum() { let checksum = bytes::Bytes::from_static(b"ssec-checksum"); @@ -2342,7 +2641,7 @@ mod tests { tx.try_send(first).expect("first MRF entry should fill the test channel"); - let admission = queue_mrf_save_admission(&tx, second, "bucket", "second", "test"); + let admission = queue_mrf_save_entry(&tx, second, "test"); tokio::pin!(admission); assert!( @@ -2687,6 +2986,30 @@ mod tests { assert!(missing_file.entries.is_empty()); } + #[test] + fn durable_mrf_summary_aggregates_entries_by_bucket_for_obs() { + let summary = + durable_mrf_backlog_summary_from_sizes([("b1".to_string(), 1024), ("b1".to_string(), 512), ("b2".to_string(), 0)]); + + assert!(summary.available); + let buckets = summary + .buckets + .into_iter() + .map(|bucket| (bucket.bucket.clone(), bucket)) + .collect::>(); + assert_eq!(buckets["b1"].count, 2); + assert_eq!(buckets["b1"].bytes, 1536); + assert_eq!(buckets["b2"].count, 1); + assert_eq!(buckets["b2"].bytes, 0); + } + + #[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()); + } + #[test] fn durable_mrf_snapshot_marks_corrupt_or_invalid_data_unavailable() { let corrupt = durable_mrf_backlog_from_read(Ok(vec![0, 1, 2])); diff --git a/crates/ecstore/src/bucket/replication/replication_state.rs b/crates/ecstore/src/bucket/replication/replication_state.rs index 11a5e99e7..dc218a66c 100644 --- a/crates/ecstore/src/bucket/replication/replication_state.rs +++ b/crates/ecstore/src/bucket/replication/replication_state.rs @@ -23,8 +23,8 @@ use super::replication_stats_boundary::{ }; use super::runtime_boundary as runtime_sources; use std::collections::HashMap; -use std::sync::Arc; use std::sync::atomic::{AtomicI64, Ordering}; +use std::sync::{Arc, Mutex as StdMutex}; use std::time::{Duration, SystemTime}; use tokio::sync::{Mutex, RwLock}; use tokio::time::interval; @@ -143,7 +143,7 @@ pub struct ReplicationStats { // Active worker statistics pub workers: Arc>, // Queue statistics cache - pub q_cache: Arc>, + pub q_cache: Arc>, // Proxy statistics cache pub p_cache: Arc>, // MRF backlog statistics (simplified) @@ -158,7 +158,7 @@ impl ReplicationStats { Self { sr_stats: Arc::new(SRStats::new()), workers: Arc::new(Mutex::new(ActiveWorkerStat::new())), - q_cache: Arc::new(Mutex::new(QueueCache::new())), + q_cache: Arc::new(StdMutex::new(QueueCache::new())), p_cache: Arc::new(Mutex::new(ProxyStatsCache::new())), mrf_stats: HashMap::new(), cache: Arc::new(RwLock::new(HashMap::new())), @@ -198,8 +198,9 @@ impl ReplicationStats { let mut interval = interval(Duration::from_secs(2)); loop { interval.tick().await; - let mut cache = q_cache_clone.lock().await; - cache.update(); + if let Ok(mut cache) = q_cache_clone.lock() { + cache.update(); + } } }); } @@ -391,12 +392,13 @@ impl ReplicationStats { drop(cache); { - let q_cache = self.q_cache.lock().await; - for (bucket, queue_stats) in &q_cache.bucket_stats { - let bucket_stats = result.entry(bucket.clone()).or_insert_with(BucketReplicationStats::new); - bucket_stats.q_stat = queue_stats.snapshot(); - bucket_stats.mark_node_local_provider_available(); - bucket_stats.queue_scope = ReplicationMetricScope::NodeLocal; + if let Ok(q_cache) = self.q_cache.lock() { + for (bucket, queue_stats) in &q_cache.bucket_stats { + let bucket_stats = result.entry(bucket.clone()).or_insert_with(BucketReplicationStats::new); + bucket_stats.q_stat = queue_stats.snapshot(); + bucket_stats.mark_node_local_provider_available(); + bucket_stats.queue_scope = ReplicationMetricScope::NodeLocal; + } } } @@ -429,8 +431,11 @@ impl ReplicationStats { let boot_time = SystemTime::UNIX_EPOCH; // simplified implementation let uptime = SystemTime::now().duration_since(boot_time).unwrap_or_default().as_secs() as i64; - let q_cache = self.q_cache.lock().await; - let queued = q_cache.get_site_stats(); + let queued = self + .q_cache + .lock() + .map(|q_cache| q_cache.get_site_stats()) + .unwrap_or_default(); let p_cache = self.p_cache.lock().await; let proxied = p_cache.get_site_stats(); @@ -633,8 +638,9 @@ impl ReplicationStats { drop(cache); { - let q_cache = self.q_cache.lock().await; - if let Some(queue_stats) = q_cache.bucket_stats.get(bucket) { + if let Ok(q_cache) = self.q_cache.lock() + && let Some(queue_stats) = q_cache.bucket_stats.get(bucket) + { replication_stats.q_stat = queue_stats.snapshot(); } } @@ -665,31 +671,17 @@ impl ReplicationStats { } /// Increase queue statistics - pub async fn inc_q(&self, bucket: &str, size: i64, _is_delete_repl: bool, _op_type: ReplicationType) { - let mut q_cache = self.q_cache.lock().await; - let stats = q_cache - .bucket_stats - .entry(bucket.to_string()) - .or_insert_with(InQueueMetric::default); - stats.curr.now_bytes.fetch_add(size, Ordering::Relaxed); - stats.curr.now_count.fetch_add(1, Ordering::Relaxed); - - q_cache.sr_queue_stats.curr.now_bytes.fetch_add(size, Ordering::Relaxed); - q_cache.sr_queue_stats.curr.now_count.fetch_add(1, Ordering::Relaxed); + pub fn inc_q(&self, bucket: &str, size: i64, _is_delete_repl: bool, _op_type: ReplicationType) { + if let Ok(mut q_cache) = self.q_cache.lock() { + q_cache.inc(bucket, size); + } } /// Decrease queue statistics - pub async fn dec_q(&self, bucket: &str, size: i64, _is_del_marker: bool, _op_type: ReplicationType) { - let mut q_cache = self.q_cache.lock().await; - let stats = q_cache - .bucket_stats - .entry(bucket.to_string()) - .or_insert_with(InQueueMetric::default); - stats.curr.now_bytes.fetch_sub(size, Ordering::Relaxed); - stats.curr.now_count.fetch_sub(1, Ordering::Relaxed); - - q_cache.sr_queue_stats.curr.now_bytes.fetch_sub(size, Ordering::Relaxed); - q_cache.sr_queue_stats.curr.now_count.fetch_sub(1, Ordering::Relaxed); + pub fn dec_q(&self, bucket: &str, size: i64, _is_del_marker: bool, _op_type: ReplicationType) { + if let Ok(mut q_cache) = self.q_cache.lock() { + q_cache.dec(bucket, size); + } } /// Increase proxy metrics @@ -801,14 +793,14 @@ mod tests { async fn latest_stats_include_queue_until_drained() { let stats = ReplicationStats::new(); - stats.inc_q("queued-bucket", 4096, false, ReplicationType::Object).await; + stats.inc_q("queued-bucket", 4096, false, ReplicationType::Object); let queued = stats.get_latest_replication_stats("queued-bucket").await; assert!(queued.replication_stats.provider_available); assert_eq!(queued.replication_stats.q_stat.curr.count, 1); assert_eq!(queued.replication_stats.q_stat.curr.bytes, 4096); assert_eq!(queued.replication_stats.queue_scope, ReplicationMetricScope::NodeLocal); - stats.dec_q("queued-bucket", 4096, false, ReplicationType::Object).await; + stats.dec_q("queued-bucket", 4096, false, ReplicationType::Object); let drained = stats.get_latest_replication_stats("queued-bucket").await; assert_eq!(drained.replication_stats.q_stat.curr.count, 0); assert_eq!(drained.replication_stats.q_stat.curr.bytes, 0); @@ -911,7 +903,7 @@ mod tests { for _ in 0..32 { let stats = Arc::clone(&stats); tasks.push(tokio::spawn(async move { - stats.inc_q("concurrent-bucket", 7, false, ReplicationType::Object).await; + stats.inc_q("concurrent-bucket", 7, false, ReplicationType::Object); })); } for task in tasks { diff --git a/crates/kms/src/backends/snapshots/rustfs_kms__backends__aws__tests__aws_backend_capabilities.snap b/crates/kms/src/backends/snapshots/rustfs_kms__backends__aws__tests__aws_backend_capabilities.snap index a6f50d1e1..ccf635599 100644 --- a/crates/kms/src/backends/snapshots/rustfs_kms__backends__aws__tests__aws_backend_capabilities.snap +++ b/crates/kms/src/backends/snapshots/rustfs_kms__backends__aws__tests__aws_backend_capabilities.snap @@ -10,5 +10,6 @@ expression: capabilities_snapshot(backend.capabilities()) "physical_delete": false, "rotate": true, "schedule_deletion": true, + "update_key_metadata": false, "versioning": true } diff --git a/crates/obs/src/metrics/collectors/bucket_replication.rs b/crates/obs/src/metrics/collectors/bucket_replication.rs index 6e777c802..32d7151fd 100644 --- a/crates/obs/src/metrics/collectors/bucket_replication.rs +++ b/crates/obs/src/metrics/collectors/bucket_replication.rs @@ -16,22 +16,25 @@ 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_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_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, BUCKET_REPL_PROXIED_HEAD_REQUESTS_FAILURES_MD, - BUCKET_REPL_PROXIED_HEAD_REQUESTS_TOTAL_MD, BUCKET_REPL_PROXIED_PUT_REQUESTS_FAILURES_MD, - BUCKET_REPL_PROXIED_PUT_REQUESTS_TOTAL_MD, BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_FAILURES_MD, - BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_TOTAL_MD, BUCKET_REPL_RESYNC_CANCELED_TOTAL_MD, - BUCKET_REPL_RESYNC_COMPLETED_TOTAL_MD, BUCKET_REPL_RESYNC_DURATION_MS_TOTAL_MD, BUCKET_REPL_RESYNC_FAILED_TOTAL_MD, - BUCKET_REPL_RESYNC_STARTED_TOTAL_MD, BUCKET_REPL_SENT_BYTES_MD, BUCKET_REPL_SENT_COUNT_MD, BUCKET_REPL_TOTAL_FAILED_BYTES_MD, - BUCKET_REPL_TOTAL_FAILED_COUNT_MD, OPERATION_L, RANGE_L, TARGET_ARN_L, + 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_LAST_MIN_FAILED_BYTES_MD, BUCKET_REPL_LAST_MIN_FAILED_COUNT_MD, BUCKET_REPL_LATENCY_MS_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, + BUCKET_REPL_PROXIED_HEAD_REQUESTS_FAILURES_MD, BUCKET_REPL_PROXIED_HEAD_REQUESTS_TOTAL_MD, + BUCKET_REPL_PROXIED_PUT_REQUESTS_FAILURES_MD, BUCKET_REPL_PROXIED_PUT_REQUESTS_TOTAL_MD, + BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_FAILURES_MD, BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_TOTAL_MD, + BUCKET_REPL_RESYNC_CANCELED_TOTAL_MD, BUCKET_REPL_RESYNC_COMPLETED_TOTAL_MD, BUCKET_REPL_RESYNC_DURATION_MS_TOTAL_MD, + BUCKET_REPL_RESYNC_FAILED_TOTAL_MD, BUCKET_REPL_RESYNC_STARTED_TOTAL_MD, BUCKET_REPL_SENT_BYTES_MD, + BUCKET_REPL_SENT_COUNT_MD, BUCKET_REPL_TOTAL_FAILED_BYTES_MD, BUCKET_REPL_TOTAL_FAILED_COUNT_MD, OPERATION_L, RANGE_L, + TARGET_ARN_L, }; use std::borrow::Cow; const BASE_BUCKET_REPLICATION_METRICS_PER_BUCKET: usize = 25; +const BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET: usize = 5; #[derive(Debug, Clone, Default)] pub struct BucketReplicationTargetStats { @@ -80,6 +83,16 @@ pub struct BucketReplicationStats { pub targets: Vec, } +#[derive(Debug, Clone, Default)] +pub(crate) struct BucketReplicationBacklogStats { + pub(crate) bucket: String, + pub(crate) current_backlog_count: u64, + pub(crate) current_backlog_bytes: u64, + pub(crate) durable_mrf_available: bool, + pub(crate) durable_mrf_backlog_count: u64, + pub(crate) durable_mrf_backlog_bytes: u64, +} + pub fn collect_bucket_replication_bandwidth_metrics(stats: &[BucketReplicationBandwidthStats]) -> Vec { if stats.is_empty() { return Vec::new(); @@ -249,7 +262,6 @@ pub fn collect_bucket_replication_metrics(stats: &[BucketReplicationStats]) -> V PrometheusMetric::from_descriptor(&BUCKET_REPL_RESYNC_DURATION_MS_TOTAL_MD, stat.resync_duration_ms as f64) .with_label(BUCKET_L, bucket_label.clone()), ); - for target in &stat.targets { let target_label: Cow<'static, str> = Cow::Owned(target.target_arn.clone()); metrics.push( @@ -265,6 +277,43 @@ pub fn collect_bucket_replication_metrics(stats: &[BucketReplicationStats]) -> V metrics } +pub(crate) fn collect_bucket_replication_backlog_metrics(stats: &[BucketReplicationBacklogStats]) -> Vec { + if stats.is_empty() { + return Vec::new(); + } + + let mut metrics = Vec::with_capacity(stats.len() * BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET); + for stat in stats { + let bucket_label: Cow<'static, str> = Cow::Owned(stat.bucket.clone()); + + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD, stat.current_backlog_count as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_CURRENT_BACKLOG_BYTES_MD, stat.current_backlog_bytes as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor( + &BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD, + if stat.durable_mrf_available { 1.0 } else { 0.0 }, + ) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD, stat.durable_mrf_backlog_count as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD, stat.durable_mrf_backlog_bytes as f64) + .with_label(BUCKET_L, bucket_label), + ); + } + + metrics +} + #[cfg(test)] mod tests { use super::*; @@ -369,6 +418,56 @@ mod tests { })); } + #[test] + fn test_collect_bucket_replication_backlog_metrics() { + let stats = vec![BucketReplicationBacklogStats { + bucket: "b1".to_string(), + current_backlog_count: 3, + current_backlog_bytes: 4096, + durable_mrf_available: true, + durable_mrf_backlog_count: 2, + durable_mrf_backlog_bytes: 2048, + }]; + + let metrics = collect_bucket_replication_backlog_metrics(&stats); + assert_eq!(metrics.len(), 5); + + let backlog_count_name = BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == backlog_count_name + && metric.value == 3.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let backlog_bytes_name = BUCKET_REPL_CURRENT_BACKLOG_BYTES_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == backlog_bytes_name + && metric.value == 4096.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let durable_available_name = BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == durable_available_name + && metric.value == 1.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let durable_count_name = BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == durable_count_name + && metric.value == 2.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let durable_bytes_name = BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == durable_bytes_name + && metric.value == 2048.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + } + #[test] fn test_collect_bucket_replication_metrics_empty() { let stats: Vec = Vec::new(); @@ -376,6 +475,41 @@ mod tests { assert!(metrics.is_empty()); } + #[test] + fn backlog_metrics_are_bucket_scoped_without_target_labels() { + let stats = vec![BucketReplicationBacklogStats { + bucket: "scope-bucket".to_string(), + current_backlog_count: 5, + current_backlog_bytes: 8192, + durable_mrf_available: true, + durable_mrf_backlog_count: 2, + durable_mrf_backlog_bytes: 4096, + }]; + + let metrics = collect_bucket_replication_backlog_metrics(&stats); + let backlog_names = [ + BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_CURRENT_BACKLOG_BYTES_MD.get_full_metric_name(), + 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(), + ]; + + for name in backlog_names { + let metric = metrics + .iter() + .find(|metric| metric.name == name) + .expect("backlog metric should be emitted"); + assert!( + metric + .labels + .iter() + .any(|(key, value)| *key == BUCKET_L && value == "scope-bucket") + ); + assert!(!metric.labels.iter().any(|(key, _)| *key == TARGET_ARN_L)); + } + } + #[test] fn test_collect_bucket_replication_bandwidth_metrics() { let stats = vec![BucketReplicationBandwidthStats { diff --git a/crates/obs/src/metrics/collectors/mod.rs b/crates/obs/src/metrics/collectors/mod.rs index 22f1c8858..43dbd2873 100644 --- a/crates/obs/src/metrics/collectors/mod.rs +++ b/crates/obs/src/metrics/collectors/mod.rs @@ -42,6 +42,7 @@ 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 use bucket_replication::{ BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetStats, collect_bucket_replication_bandwidth_metrics, collect_bucket_replication_metrics, diff --git a/crates/obs/src/metrics/mod.rs b/crates/obs/src/metrics/mod.rs index 30730bbed..af2c4b8ab 100644 --- a/crates/obs/src/metrics/mod.rs +++ b/crates/obs/src/metrics/mod.rs @@ -31,9 +31,9 @@ pub use scheduler::{ metrics_runtime_controller_snapshot, metrics_runtime_status_snapshot, }; pub(crate) use storage_api::metrics::{ - BucketOperations, BucketOptions, ObsBucketBandwidthMonitor, ObsEcstoreResult, ObsStore, StorageAdminApi, - obs_bucket_replication_stats_snapshot, obs_expiry_state_handle, obs_get_global_bucket_monitor, obs_get_quota_config, - obs_get_total_usable_capacity, obs_get_total_usable_capacity_free, obs_is_disk_compression_enabled, + BucketOperations, BucketOptions, ObsBucketBandwidthMonitor, ObsBucketReplicationStatsSnapshot, ObsEcstoreResult, ObsStore, + StorageAdminApi, obs_bucket_replication_stats_snapshot, obs_expiry_state_handle, obs_get_global_bucket_monitor, + obs_get_quota_config, obs_get_total_usable_capacity, obs_get_total_usable_capacity_free, obs_is_disk_compression_enabled, obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_replication_site_stats_snapshot, obs_resolve_object_store_handle, obs_transition_state_handle, }; diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index 189c0c04d..41b6a75a2 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -35,6 +35,7 @@ use crate::metrics::collectors::{ ProcessMemoryStats, collect_audit_metrics, collect_bucket_metrics, + collect_bucket_replication_backlog_metrics, collect_bucket_replication_bandwidth_metrics, collect_bucket_replication_metrics, collect_bucket_usage_metrics, @@ -94,7 +95,7 @@ use crate::metrics::schema::notification_target::{ }; use crate::metrics::schema::system_process::{PROCESS_EXECUTABLE_NAME_LABEL, PROCESS_PID_LABEL}; use crate::metrics::stats_collector::{ - ProcessMetricBundle, collect_bucket_replication_bandwidth_stats, collect_bucket_replication_detail_stats, + ProcessMetricBundle, collect_bucket_replication_bandwidth_stats, collect_bucket_replication_stats_bundle, collect_bucket_stats, collect_cluster_and_health_stats, collect_cluster_config_stats, collect_cluster_usage_metric_stats, collect_compression_cluster_stats, collect_disk_and_system_drive_stats, collect_erasure_set_stats, collect_host_network_stats, collect_iam_stats, collect_ilm_metric_stats, collect_internode_network_stats, @@ -1348,8 +1349,9 @@ pub fn init_metrics_runtime(token: CancellationToken) { // Phase-1 action: force zero for removed keys during tombstone cycles. metrics.extend(collect_repl_bw_zero_tombstone_metrics(&zero_tombstones)); - let bucket_replication = collect_bucket_replication_detail_stats().await; + let (bucket_replication, bucket_replication_backlog) = collect_bucket_replication_stats_bundle().await; metrics.extend(collect_bucket_replication_metrics(&bucket_replication)); + metrics.extend(collect_bucket_replication_backlog_metrics(&bucket_replication_backlog)); let replication = collect_replication_stats().await; metrics.extend(collect_replication_metrics(&replication)); report_metrics(&metrics); diff --git a/crates/obs/src/metrics/schema/bucket_replication.rs b/crates/obs/src/metrics/schema/bucket_replication.rs index 1178327b5..e029774ef 100644 --- a/crates/obs/src/metrics/schema/bucket_replication.rs +++ b/crates/obs/src/metrics/schema/bucket_replication.rs @@ -33,6 +33,11 @@ const RESYNC_COMPLETED_TOTAL: &str = "resync_completed_total"; const RESYNC_FAILED_TOTAL: &str = "resync_failed_total"; 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 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"; pub static BUCKET_REPL_LAST_HR_FAILED_BYTES_MD: LazyLock = LazyLock::new(|| { new_gauge_md( @@ -79,6 +84,51 @@ pub static BUCKET_REPL_LATENCY_MS_MD: LazyLock = LazyLock::new ) }); +pub static BUCKET_REPL_CURRENT_BACKLOG_BYTES_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(CURRENT_BACKLOG_BYTES), + "Current number of bytes admitted to the in-memory replication worker queues for a bucket", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(CURRENT_BACKLOG_COUNT), + "Current number of objects admitted to the in-memory replication worker queues for a bucket", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(DURABLE_MRF_AVAILABLE), + "Whether the durable MRF backlog snapshot is available for this bucket on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(DURABLE_MRF_BACKLOG_COUNT), + "Current number of objects in the durable MRF backlog file for a bucket on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(DURABLE_MRF_BACKLOG_BYTES), + "Current bytes in the durable MRF backlog file 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/schema/replication.rs b/crates/obs/src/metrics/schema/replication.rs index 6b857aa5b..db1563255 100644 --- a/crates/obs/src/metrics/schema/replication.rs +++ b/crates/obs/src/metrics/schema/replication.rs @@ -128,7 +128,7 @@ pub static REPLICATION_MAX_DATA_TRANSFER_RATE_MD: LazyLock = L pub static REPLICATION_RECENT_BACKLOG_COUNT_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::ReplicationRecentBacklogCount, - "Total number of objects currently in replication backlog (failed plus queued)", + "Legacy replication backlog indicator: failed target objects plus objects currently queued on this node", &[], subsystems::REPLICATION, ) diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 7f14aa599..f9040cc3c 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -21,17 +21,18 @@ //! and convert them to the Stats structs used by collectors. use crate::metrics::collectors::{ - 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, 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::{ - BucketOperations, BucketOptions, ObsEcstoreResult, ObsStore, StorageAdminApi, obs_bucket_replication_stats_snapshot, - obs_get_quota_config, obs_get_total_usable_capacity, obs_get_total_usable_capacity_free, - obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_replication_site_stats_snapshot, - obs_resolve_object_store_handle, + BucketOperations, BucketOptions, ObsBucketReplicationStatsSnapshot, ObsEcstoreResult, ObsStore, StorageAdminApi, + obs_bucket_replication_stats_snapshot, obs_get_quota_config, obs_get_total_usable_capacity, + obs_get_total_usable_capacity_free, obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, + obs_replication_site_stats_snapshot, obs_resolve_object_store_handle, }; use crate::node_identity::current_local_node_identity; use chrono::Utc; @@ -192,49 +193,65 @@ async fn obs_ilm_runtime_snapshot() -> ObsIlmRuntimeSnapshot { ilm_runtime_snapshot().await } -async fn obs_bucket_replication_detail_stats() -> Vec { - obs_bucket_replication_stats_snapshot() - .await - .into_iter() - .map(|stats| BucketReplicationStats { - bucket: stats.bucket, - total_failed_bytes: stats.total_failed_bytes, - total_failed_count: stats.total_failed_count, - last_min_failed_bytes: stats.last_min_failed_bytes, - last_min_failed_count: stats.last_min_failed_count, - last_hour_failed_bytes: stats.last_hour_failed_bytes, - last_hour_failed_count: stats.last_hour_failed_count, - sent_bytes: stats.sent_bytes, - sent_count: stats.sent_count, - proxied_get_requests_total: stats.proxied_get_requests_total, - proxied_get_requests_failures: stats.proxied_get_requests_failures, - proxied_head_requests_total: stats.proxied_head_requests_total, - proxied_head_requests_failures: stats.proxied_head_requests_failures, - proxied_put_requests_total: stats.proxied_put_requests_total, - proxied_put_requests_failures: stats.proxied_put_requests_failures, - proxied_put_tagging_requests_total: stats.proxied_put_tagging_requests_total, - proxied_put_tagging_requests_failures: stats.proxied_put_tagging_requests_failures, - proxied_get_tagging_requests_total: stats.proxied_get_tagging_requests_total, - proxied_get_tagging_requests_failures: stats.proxied_get_tagging_requests_failures, - proxied_delete_tagging_requests_total: stats.proxied_delete_tagging_requests_total, - proxied_delete_tagging_requests_failures: stats.proxied_delete_tagging_requests_failures, - resync_started_count: stats.resync_started_count, - resync_completed_count: stats.resync_completed_count, - resync_failed_count: stats.resync_failed_count, - resync_canceled_count: stats.resync_canceled_count, - resync_duration_ms: stats.resync_duration_ms, - targets: stats - .targets - .into_iter() - .map(|target| BucketReplicationTargetStats { - target_arn: target.target_arn, - bandwidth_limit_bytes_per_sec: target.bandwidth_limit_bytes_per_sec, - current_bandwidth_bytes_per_sec: target.current_bandwidth_bytes_per_sec, - latency_ms: target.latency_ms, - }) - .collect(), - }) - .collect() +async fn obs_bucket_replication_stats_bundle() -> (Vec, Vec) { + let snapshots = obs_bucket_replication_stats_snapshot().await; + let mut detail_stats = Vec::with_capacity(snapshots.len()); + let mut backlog_stats = Vec::with_capacity(snapshots.len()); + + for stats in snapshots { + backlog_stats.push(BucketReplicationBacklogStats { + bucket: stats.bucket.clone(), + current_backlog_count: stats.current_backlog_count, + current_backlog_bytes: stats.current_backlog_bytes, + 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, + }); + detail_stats.push(bucket_replication_detail_from_snapshot(stats)); + } + + (detail_stats, backlog_stats) +} + +fn bucket_replication_detail_from_snapshot(stats: ObsBucketReplicationStatsSnapshot) -> BucketReplicationStats { + BucketReplicationStats { + bucket: stats.bucket, + total_failed_bytes: stats.total_failed_bytes, + total_failed_count: stats.total_failed_count, + last_min_failed_bytes: stats.last_min_failed_bytes, + last_min_failed_count: stats.last_min_failed_count, + last_hour_failed_bytes: stats.last_hour_failed_bytes, + last_hour_failed_count: stats.last_hour_failed_count, + sent_bytes: stats.sent_bytes, + sent_count: stats.sent_count, + proxied_get_requests_total: stats.proxied_get_requests_total, + proxied_get_requests_failures: stats.proxied_get_requests_failures, + proxied_head_requests_total: stats.proxied_head_requests_total, + proxied_head_requests_failures: stats.proxied_head_requests_failures, + proxied_put_requests_total: stats.proxied_put_requests_total, + proxied_put_requests_failures: stats.proxied_put_requests_failures, + proxied_put_tagging_requests_total: stats.proxied_put_tagging_requests_total, + proxied_put_tagging_requests_failures: stats.proxied_put_tagging_requests_failures, + proxied_get_tagging_requests_total: stats.proxied_get_tagging_requests_total, + proxied_get_tagging_requests_failures: stats.proxied_get_tagging_requests_failures, + proxied_delete_tagging_requests_total: stats.proxied_delete_tagging_requests_total, + proxied_delete_tagging_requests_failures: stats.proxied_delete_tagging_requests_failures, + resync_started_count: stats.resync_started_count, + resync_completed_count: stats.resync_completed_count, + resync_failed_count: stats.resync_failed_count, + resync_canceled_count: stats.resync_canceled_count, + resync_duration_ms: stats.resync_duration_ms, + targets: stats + .targets + .into_iter() + .map(|target| BucketReplicationTargetStats { + target_arn: target.target_arn, + bandwidth_limit_bytes_per_sec: target.bandwidth_limit_bytes_per_sec, + current_bandwidth_bytes_per_sec: target.current_bandwidth_bytes_per_sec, + latency_ms: target.latency_ms, + }) + .collect(), + } } async fn obs_site_replication_stats() -> ReplicationStats { @@ -526,7 +543,16 @@ pub fn collect_bucket_replication_bandwidth_stats() -> Vec Vec { - obs_bucket_replication_detail_stats().await + obs_bucket_replication_stats_snapshot() + .await + .into_iter() + .map(bucket_replication_detail_from_snapshot) + .collect() +} + +pub(crate) async fn collect_bucket_replication_stats_bundle() -> (Vec, Vec) +{ + obs_bucket_replication_stats_bundle().await } /// Collect site-level replication stats from the global replication runtime. diff --git a/crates/obs/src/metrics/storage_api.rs b/crates/obs/src/metrics/storage_api.rs index a1fed693e..55221d7aa 100644 --- a/crates/obs/src/metrics/storage_api.rs +++ b/crates/obs/src/metrics/storage_api.rs @@ -12,11 +12,14 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::HashMap; 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::get_global_replication_stats; +use rustfs_ecstore::api::bucket::replication::{ + DurableMrfBucketBacklog, durable_mrf_backlog_summary_snapshot, get_global_replication_stats, +}; pub(crate) use rustfs_ecstore::api::capacity::{ get_total_usable_capacity as obs_get_total_usable_capacity, get_total_usable_capacity_free as obs_get_total_usable_capacity_free, @@ -68,9 +71,50 @@ pub(crate) struct ObsBucketReplicationStatsSnapshot { pub(crate) resync_failed_count: u64, pub(crate) resync_canceled_count: u64, pub(crate) resync_duration_ms: u64, + pub(crate) current_backlog_count: u64, + pub(crate) current_backlog_bytes: u64, + pub(crate) durable_mrf_available: bool, + pub(crate) durable_mrf_backlog_count: u64, + pub(crate) durable_mrf_backlog_bytes: u64, pub(crate) targets: Vec, } +#[derive(Debug, Clone, Default, PartialEq)] +struct ObsBucketReplicationRuntimeSnapshot { + total_failed_bytes: u64, + total_failed_count: u64, + last_min_failed_bytes: u64, + last_min_failed_count: u64, + last_hour_failed_bytes: u64, + last_hour_failed_count: u64, + sent_bytes: u64, + sent_count: u64, + resync_started_count: u64, + resync_completed_count: u64, + resync_failed_count: u64, + resync_canceled_count: u64, + resync_duration_ms: u64, + current_backlog_count: u64, + current_backlog_bytes: u64, + targets: Vec, +} + +#[derive(Debug, Clone, Default, PartialEq)] +struct ObsBucketReplicationProxySnapshot { + proxied_get_requests_total: u64, + proxied_get_requests_failures: u64, + proxied_head_requests_total: u64, + proxied_head_requests_failures: u64, + proxied_put_requests_total: u64, + proxied_put_requests_failures: u64, + proxied_put_tagging_requests_total: u64, + proxied_put_tagging_requests_failures: u64, + proxied_get_tagging_requests_total: u64, + proxied_get_tagging_requests_failures: u64, + proxied_delete_tagging_requests_total: u64, + proxied_delete_tagging_requests_failures: u64, +} + #[derive(Debug, Clone, Default, PartialEq)] pub(crate) struct ObsReplicationSiteStatsSnapshot { pub(crate) average_active_workers: f64, @@ -102,59 +146,85 @@ fn replication_backlog_count(failed_counts: impl Iterator, queued_co failed_backlog.saturating_add(i64_to_u64_floor_zero(queued_count)) } +fn bucket_replication_stats_snapshot_from_parts( + bucket: String, + runtime: ObsBucketReplicationRuntimeSnapshot, + proxy: ObsBucketReplicationProxySnapshot, + durable_mrf_available: bool, + durable_bucket: DurableMrfBucketBacklog, +) -> ObsBucketReplicationStatsSnapshot { + ObsBucketReplicationStatsSnapshot { + bucket, + total_failed_bytes: runtime.total_failed_bytes, + total_failed_count: runtime.total_failed_count, + last_min_failed_bytes: runtime.last_min_failed_bytes, + last_min_failed_count: runtime.last_min_failed_count, + last_hour_failed_bytes: runtime.last_hour_failed_bytes, + last_hour_failed_count: runtime.last_hour_failed_count, + sent_bytes: runtime.sent_bytes, + sent_count: runtime.sent_count, + proxied_get_requests_total: proxy.proxied_get_requests_total, + proxied_get_requests_failures: proxy.proxied_get_requests_failures, + proxied_head_requests_total: proxy.proxied_head_requests_total, + proxied_head_requests_failures: proxy.proxied_head_requests_failures, + proxied_put_requests_total: proxy.proxied_put_requests_total, + proxied_put_requests_failures: proxy.proxied_put_requests_failures, + proxied_put_tagging_requests_total: proxy.proxied_put_tagging_requests_total, + proxied_put_tagging_requests_failures: proxy.proxied_put_tagging_requests_failures, + proxied_get_tagging_requests_total: proxy.proxied_get_tagging_requests_total, + proxied_get_tagging_requests_failures: proxy.proxied_get_tagging_requests_failures, + proxied_delete_tagging_requests_total: proxy.proxied_delete_tagging_requests_total, + proxied_delete_tagging_requests_failures: proxy.proxied_delete_tagging_requests_failures, + resync_started_count: runtime.resync_started_count, + resync_completed_count: runtime.resync_completed_count, + resync_failed_count: runtime.resync_failed_count, + resync_canceled_count: runtime.resync_canceled_count, + 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, + targets: runtime.targets, + } +} + pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec { - let Some(stats) = get_global_replication_stats() else { - return Vec::new(); + let stats = get_global_replication_stats(); + let all_bucket_stats = if let Some(stats) = &stats { + stats.get_all().await + } else { + HashMap::new() }; + let durable_mrf_summary = if obs_resolve_object_store_handle().is_some() { + durable_mrf_backlog_summary_snapshot() + } else { + Default::default() + }; + let durable_mrf_available = durable_mrf_summary.available; + let durable_buckets = durable_mrf_summary + .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())); + bucket_names.extend(all_bucket_stats.keys().cloned()); + bucket_names.extend( + durable_buckets + .keys() + .filter(|bucket| !all_bucket_stats.contains_key(*bucket)) + .cloned(), + ); + let mut buckets = Vec::with_capacity(bucket_names.len()); - let all_bucket_stats = stats.get_all().await; - let mut buckets = Vec::with_capacity(all_bucket_stats.len()); - - for (bucket, bucket_stats) in all_bucket_stats { - let proxy = stats.get_proxy_stats(&bucket).await; - let mut total_failed_bytes = 0u64; - let mut total_failed_count = 0u64; - let mut last_min_failed_bytes = 0u64; - let mut last_min_failed_count = 0u64; - let mut last_hour_failed_bytes = 0u64; - let mut last_hour_failed_count = 0u64; - let mut sent_bytes = 0u64; - let mut sent_count = 0u64; - let mut targets = Vec::with_capacity(bucket_stats.stats.len()); - - for (target_arn, target_stats) in bucket_stats.stats { - total_failed_bytes = total_failed_bytes.saturating_add(i64_to_u64_floor_zero(target_stats.fail_stats.size)); - total_failed_count = total_failed_count.saturating_add(i64_to_u64_floor_zero(target_stats.fail_stats.count)); - - let last_min = target_stats.fail_stats.recent_since(Duration::from_secs(60)); - last_min_failed_bytes = last_min_failed_bytes.saturating_add(i64_to_u64_floor_zero(last_min.size)); - last_min_failed_count = last_min_failed_count.saturating_add(i64_to_u64_floor_zero(last_min.count)); - - let last_hour = target_stats.fail_stats.recent_since(Duration::from_secs(60 * 60)); - last_hour_failed_bytes = last_hour_failed_bytes.saturating_add(i64_to_u64_floor_zero(last_hour.size)); - last_hour_failed_count = last_hour_failed_count.saturating_add(i64_to_u64_floor_zero(last_hour.count)); - - sent_bytes = sent_bytes.saturating_add(i64_to_u64_floor_zero(target_stats.replicated_size)); - sent_count = sent_count.saturating_add(i64_to_u64_floor_zero(target_stats.replicated_count)); - - targets.push(ObsBucketReplicationTargetStatsSnapshot { - target_arn, - bandwidth_limit_bytes_per_sec: i64_to_u64_floor_zero(target_stats.bandwidth_limit_bytes_per_sec), - current_bandwidth_bytes_per_sec: target_stats.current_bandwidth_bytes_per_sec, - latency_ms: target_stats.latency.curr, - }); - } - - buckets.push(ObsBucketReplicationStatsSnapshot { - bucket, - total_failed_bytes, - total_failed_count, - last_min_failed_bytes, - last_min_failed_count, - last_hour_failed_bytes, - last_hour_failed_count, - sent_bytes, - sent_count, + for bucket in bucket_names { + let bucket_stats = all_bucket_stats.get(&bucket); + let proxy = if let Some(stats) = &stats { + stats.get_proxy_stats(&bucket).await + } else { + Default::default() + }; + let proxy = ObsBucketReplicationProxySnapshot { proxied_get_requests_total: i64_to_u64_floor_zero(proxy.get_total), proxied_get_requests_failures: i64_to_u64_floor_zero(proxy.get_failed), proxied_head_requests_total: i64_to_u64_floor_zero(proxy.head_total), @@ -167,13 +237,67 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec i64 { self.now_bytes.load(Ordering::Relaxed) } @@ -250,6 +260,14 @@ impl InQueueStats { } } +fn saturating_atomic_sub(value: &AtomicI64, delta: i64) { + if delta == 0 { + return; + } + + let _ = value.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(delta).max(0))); +} + #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct InQueueMetric { pub curr: InQueueStats, @@ -359,6 +377,18 @@ impl QueueCache { Self::default() } + pub fn inc(&mut self, bucket: &str, size: i64) { + let stats = self.bucket_stats.entry(bucket.to_string()).or_default(); + stats.curr.add_current(size, 1); + self.sr_queue_stats.curr.add_current(size, 1); + } + + pub fn dec(&mut self, bucket: &str, size: i64) { + let stats = self.bucket_stats.entry(bucket.to_string()).or_default(); + stats.curr.subtract_current(size, 1); + self.sr_queue_stats.curr.subtract_current(size, 1); + } + pub fn update(&mut self) { let observed_at = Instant::now(); self.sr_queue_stats.observe(observed_at); @@ -808,12 +838,10 @@ mod tests { #[test] fn in_queue_metric_observe_updates_rolling_stats() { let mut metric = InQueueMetric::default(); - metric.curr.now_bytes.store(128, Ordering::Relaxed); - metric.curr.now_count.store(4, Ordering::Relaxed); + metric.curr.add_current(128, 4); metric.observe(Instant::now()); - metric.curr.now_bytes.store(256, Ordering::Relaxed); - metric.curr.now_count.store(6, Ordering::Relaxed); + metric.curr.add_current(128, 2); metric.observe(Instant::now()); assert_eq!(metric.curr.bytes, 256); @@ -824,6 +852,21 @@ mod tests { assert_eq!(metric.last_minute.count, 5); } + #[test] + fn queue_cache_decrement_saturates_at_zero() { + let mut cache = QueueCache::new(); + cache.inc("bucket", 128); + cache.dec("bucket", 512); + cache.dec("bucket", 1); + + let bucket = cache.get_bucket_stats("bucket"); + let site = cache.get_site_stats(); + assert_eq!(bucket.curr.count, 0); + assert_eq!(bucket.curr.bytes, 0); + assert_eq!(site.curr.count, 0); + assert_eq!(site.curr.bytes, 0); + } + #[test] fn fail_stats_recent_since_tracks_windows() { let mut stats = FailStats::default();