diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index 29213edeb..4a2afbbe7 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -786,6 +786,9 @@ pub struct Metrics { scanner_expiry_queue_missed: AtomicU64, scanner_expiry_queued_total: AtomicU64, scanner_expiry_missed_total: AtomicU64, + scanner_expiry_blocked_total: AtomicU64, + scanner_expiry_not_enqueued_total: AtomicU64, + scanner_expiry_delete_failed_total: AtomicU64, scanner_transition_queue_capacity: AtomicU64, scanner_transition_queued: AtomicU64, scanner_transition_active: AtomicU64, @@ -1051,6 +1054,12 @@ pub struct ScannerLifecycleExpirySnapshot { pub queue_missed: u64, pub scanner_queued: u64, pub scanner_missed: u64, + #[serde(default)] + pub scanner_blocked: u64, + #[serde(default)] + pub scanner_not_enqueued: u64, + #[serde(default)] + pub delete_failed: u64, } #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] @@ -1755,6 +1764,9 @@ impl Metrics { scanner_expiry_queue_missed: AtomicU64::new(0), scanner_expiry_queued_total: AtomicU64::new(0), scanner_expiry_missed_total: AtomicU64::new(0), + scanner_expiry_blocked_total: AtomicU64::new(0), + scanner_expiry_not_enqueued_total: AtomicU64::new(0), + scanner_expiry_delete_failed_total: AtomicU64::new(0), scanner_transition_queue_capacity: AtomicU64::new(0), scanner_transition_queued: AtomicU64::new(0), scanner_transition_active: AtomicU64::new(0), @@ -1985,9 +1997,18 @@ impl Metrics { self.scanner_expiry_queued_total.fetch_add(count, Ordering::Relaxed); } else { self.scanner_expiry_missed_total.fetch_add(count, Ordering::Relaxed); + self.scanner_expiry_not_enqueued_total.fetch_add(count, Ordering::Relaxed); } } + pub fn record_scanner_expiry_blocked(&self, count: u64) { + self.scanner_expiry_blocked_total.fetch_add(count, Ordering::Relaxed); + } + + pub fn record_scanner_expiry_delete_failed(&self, count: u64) { + self.scanner_expiry_delete_failed_total.fetch_add(count, Ordering::Relaxed); + } + pub fn record_scanner_transition_enqueue_result(&self, count: u64, queued: bool) { self.record_scanner_ilm_enqueue_result(count, queued); if queued { @@ -2850,6 +2871,9 @@ impl Metrics { queue_missed: self.scanner_expiry_queue_missed.load(Ordering::Relaxed), scanner_queued: self.scanner_expiry_queued_total.load(Ordering::Relaxed), scanner_missed: self.scanner_expiry_missed_total.load(Ordering::Relaxed), + scanner_blocked: self.scanner_expiry_blocked_total.load(Ordering::Relaxed), + scanner_not_enqueued: self.scanner_expiry_not_enqueued_total.load(Ordering::Relaxed), + delete_failed: self.scanner_expiry_delete_failed_total.load(Ordering::Relaxed), }; m.lifecycle_transition = ScannerLifecycleTransitionSnapshot { current_queue_capacity: self.scanner_transition_queue_capacity.load(Ordering::Relaxed), @@ -3385,6 +3409,8 @@ mod tests { }); metrics.record_scanner_expiry_enqueue_result(6, true); metrics.record_scanner_expiry_enqueue_result(2, false); + metrics.record_scanner_expiry_blocked(4); + metrics.record_scanner_expiry_delete_failed(1); let report = metrics.report().await; @@ -3395,6 +3421,9 @@ mod tests { assert_eq!(report.lifecycle_expiry.queue_missed, 3); assert_eq!(report.lifecycle_expiry.scanner_queued, 6); assert_eq!(report.lifecycle_expiry.scanner_missed, 2); + assert_eq!(report.lifecycle_expiry.scanner_blocked, 4); + assert_eq!(report.lifecycle_expiry.scanner_not_enqueued, 2); + assert_eq!(report.lifecycle_expiry.delete_failed, 1); } #[tokio::test] diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index caebd465d..313128480 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -212,6 +212,18 @@ fn record_scanner_lifecycle_enqueue_result(src: &LcEventSrc, count: u64, queued: } } +fn record_scanner_lifecycle_expiry_blocked(src: &LcEventSrc, count: u64) { + if matches!(src, LcEventSrc::Scanner) { + global_metrics().record_scanner_expiry_blocked(count); + } +} + +fn record_scanner_lifecycle_expiry_delete_failed(src: &LcEventSrc, count: u64) { + if matches!(src, LcEventSrc::Scanner) { + global_metrics().record_scanner_expiry_delete_failed(count); + } +} + fn record_scanner_transition_enqueue_result(src: &LcEventSrc, count: u64, queued: bool) { if matches!(src, LcEventSrc::Scanner) { global_metrics().record_scanner_transition_enqueue_result(count, queued); @@ -845,6 +857,7 @@ impl ExpiryState { if deleted { trace.emit(EVENT_LIFECYCLE_DELETE_COMPLETED, "delete_completed", None); } else { + record_scanner_lifecycle_expiry_delete_failed(&v.src, 1); trace.emit( EVENT_LIFECYCLE_DELETE_FAILED, "delete_failed", @@ -3495,6 +3508,7 @@ async fn enqueue_expiry_for_existing_object_group( } }; if blocked_by_replication { + record_scanner_lifecycle_expiry_blocked(context.src, 1); continue; } apply_existing_object_expiry(context.api.clone(), object, event, context.src).await; @@ -5504,6 +5518,7 @@ mod tests { assert_eq!(state.stats.missed_tasks(), 1); let after = global_metrics().report().await.lifecycle_expiry; assert!(after.scanner_missed >= before.scanner_missed.saturating_add(1)); + assert!(after.scanner_not_enqueued >= before.scanner_not_enqueued.saturating_add(1)); let observed = observed.lock().expect("observability test events should not poison"); assert!(observed.contains(&(EVENT_LIFECYCLE_EXPIRED_DETECTED, "detected", None))); assert!(observed.contains(&(EVENT_LIFECYCLE_NOT_ENQUEUED, "not_enqueued", Some("worker_unavailable")))); @@ -5607,6 +5622,8 @@ mod tests { assert!(!second); assert_eq!(state.stats.pending_tasks(), 1); assert_eq!(state.stats.missed_tasks(), 1); + let after = global_metrics().report().await.lifecycle_expiry; + assert!(after.scanner_not_enqueued >= 1); let observed = observed.lock().expect("observability test events should not poison"); assert_eq!( observed diff --git a/crates/ecstore/src/services/metrics_realtime.rs b/crates/ecstore/src/services/metrics_realtime.rs index 451c67f47..f85169483 100644 --- a/crates/ecstore/src/services/metrics_realtime.rs +++ b/crates/ecstore/src/services/metrics_realtime.rs @@ -192,6 +192,9 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo queue_missed: metrics.lifecycle_expiry.queue_missed, scanner_queued: metrics.lifecycle_expiry.scanner_queued, scanner_missed: metrics.lifecycle_expiry.scanner_missed, + scanner_blocked: metrics.lifecycle_expiry.scanner_blocked, + scanner_not_enqueued: metrics.lifecycle_expiry.scanner_not_enqueued, + delete_failed: metrics.lifecycle_expiry.delete_failed, }, usage_freshness: MadminScannerUsageFreshnessSnapshot { dirty_pending_buckets: metrics.usage_freshness.dirty_pending_buckets, @@ -646,6 +649,9 @@ mod test { queue_missed: 3, scanner_queued: 6, scanner_missed: 2, + scanner_blocked: 4, + scanner_not_enqueued: 2, + delete_failed: 1, }, lifecycle_transition: rustfs_common::metrics::ScannerLifecycleTransitionSnapshot { current_queue_capacity: 16, @@ -672,6 +678,9 @@ mod test { assert_eq!(scanner.lifecycle_expiry.queue_missed, 3); assert_eq!(scanner.lifecycle_expiry.scanner_queued, 6); assert_eq!(scanner.lifecycle_expiry.scanner_missed, 2); + assert_eq!(scanner.lifecycle_expiry.scanner_blocked, 4); + assert_eq!(scanner.lifecycle_expiry.scanner_not_enqueued, 2); + assert_eq!(scanner.lifecycle_expiry.delete_failed, 1); assert_eq!(scanner.lifecycle_transition.current_queue_capacity, 16); assert_eq!(scanner.lifecycle_transition.current_queued, 5); assert_eq!(scanner.lifecycle_transition.current_active, 2); diff --git a/crates/madmin/src/metrics.rs b/crates/madmin/src/metrics.rs index f186922f1..4910f569d 100644 --- a/crates/madmin/src/metrics.rs +++ b/crates/madmin/src/metrics.rs @@ -468,6 +468,12 @@ pub struct ScannerLifecycleExpirySnapshot { pub scanner_queued: u64, #[serde(rename = "scanner_missed", default)] pub scanner_missed: u64, + #[serde(rename = "scanner_blocked", default)] + pub scanner_blocked: u64, + #[serde(rename = "scanner_not_enqueued", default)] + pub scanner_not_enqueued: u64, + #[serde(rename = "delete_failed", default)] + pub delete_failed: u64, } impl ScannerLifecycleExpirySnapshot { @@ -479,6 +485,9 @@ impl ScannerLifecycleExpirySnapshot { self.queue_missed = self.queue_missed.saturating_add(other.queue_missed); self.scanner_queued = self.scanner_queued.saturating_add(other.scanner_queued); self.scanner_missed = self.scanner_missed.saturating_add(other.scanner_missed); + self.scanner_blocked = self.scanner_blocked.saturating_add(other.scanner_blocked); + self.scanner_not_enqueued = self.scanner_not_enqueued.saturating_add(other.scanner_not_enqueued); + self.delete_failed = self.delete_failed.saturating_add(other.delete_failed); } } @@ -1508,6 +1517,9 @@ mod tests { queue_missed: 3, scanner_queued: 5, scanner_missed: 2, + scanner_blocked: 4, + scanner_not_enqueued: 1, + delete_failed: 1, }, lifecycle_transition: ScannerLifecycleTransitionSnapshot { current_queue_capacity: 8, @@ -1541,6 +1553,9 @@ mod tests { queue_missed: 2, scanner_queued: 6, scanner_missed: 4, + scanner_blocked: 7, + scanner_not_enqueued: 5, + delete_failed: 2, }, lifecycle_transition: ScannerLifecycleTransitionSnapshot { current_queue_capacity: 4, @@ -1567,6 +1582,9 @@ mod tests { assert_eq!(scanner.lifecycle_expiry.queue_missed, 5); assert_eq!(scanner.lifecycle_expiry.scanner_queued, 11); assert_eq!(scanner.lifecycle_expiry.scanner_missed, 6); + assert_eq!(scanner.lifecycle_expiry.scanner_blocked, 11); + assert_eq!(scanner.lifecycle_expiry.scanner_not_enqueued, 6); + assert_eq!(scanner.lifecycle_expiry.delete_failed, 3); assert_eq!(scanner.lifecycle_transition.current_queue_capacity, 12); assert_eq!(scanner.lifecycle_transition.current_queued, 5); assert_eq!(scanner.lifecycle_transition.current_active, 3);