diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index 5af6d5396..142a678d9 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -363,6 +363,17 @@ impl ScannerSourceWorkUpdate { } } + pub const fn failed(count: u64) -> Self { + Self { + checked: 0, + queued: 0, + executed: 0, + failed: count, + skipped: 0, + missed: 0, + } + } + pub const fn missed(count: u64) -> Self { Self { checked: 0, @@ -638,6 +649,13 @@ pub struct Metrics { scanner_ilm_actions: AtomicU64, scanner_lifecycle_expiry_actions: AtomicU64, scanner_lifecycle_transition_actions: AtomicU64, + scanner_expiry_queue_capacity: AtomicU64, + scanner_expiry_queued: AtomicU64, + scanner_expiry_active: AtomicU64, + scanner_expiry_workers: AtomicU64, + scanner_expiry_queue_missed: AtomicU64, + scanner_expiry_queued_total: AtomicU64, + scanner_expiry_missed_total: AtomicU64, scanner_transition_queue_capacity: AtomicU64, scanner_transition_queued: AtomicU64, scanner_transition_active: AtomicU64, @@ -645,6 +663,7 @@ pub struct Metrics { scanner_transition_queue_full: AtomicU64, scanner_transition_queue_send_timeout: AtomicU64, scanner_transition_compensation_scheduled: AtomicU64, + scanner_transition_compensation_pending: AtomicU64, scanner_transition_compensation_running: AtomicU64, scanner_transition_queued_total: AtomicU64, scanner_transition_missed_total: AtomicU64, @@ -823,6 +842,26 @@ impl Default for ScannerPacingPressureSnapshot { } } +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ScannerLifecycleExpiryStateUpdate { + pub queue_capacity: u64, + pub queued: u64, + pub active: u64, + pub workers: u64, + pub queue_missed: u64, +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct ScannerLifecycleExpirySnapshot { + pub current_queue_capacity: u64, + pub current_queued: u64, + pub current_active: u64, + pub current_workers: u64, + pub queue_missed: u64, + pub scanner_queued: u64, + pub scanner_missed: u64, +} + #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct ScannerLifecycleTransitionStateUpdate { pub queue_capacity: u64, @@ -832,6 +871,7 @@ pub struct ScannerLifecycleTransitionStateUpdate { pub queue_full: u64, pub queue_send_timeout: u64, pub compensation_scheduled: u64, + pub compensation_pending: u64, pub compensation_running: u64, } @@ -844,6 +884,7 @@ pub struct ScannerLifecycleTransitionSnapshot { pub queue_full: u64, pub queue_send_timeout: u64, pub compensation_scheduled: u64, + pub compensation_pending: u64, pub compensation_running: u64, pub scanner_queued: u64, pub scanner_missed: u64, @@ -966,6 +1007,8 @@ pub struct ScannerMetricsReport { #[serde(default)] pub pacing_pressure: ScannerPacingPressureSnapshot, #[serde(default)] + pub lifecycle_expiry: ScannerLifecycleExpirySnapshot, + #[serde(default)] pub lifecycle_transition: ScannerLifecycleTransitionSnapshot, #[serde(default)] pub maintenance_control: ScannerMaintenanceControlSnapshot, @@ -1136,6 +1179,9 @@ const SCANNER_MAINTENANCE_REASON_ACTIVE_WORK: &str = "active_work"; const SCANNER_MAINTENANCE_REASON_QUEUED_WORK: &str = "queued_work"; const SCANNER_MAINTENANCE_REASON_PARTIAL_CYCLE: &str = "partial_cycle"; const SCANNER_MAINTENANCE_REASON_MISSED_WORK: &str = "missed_work"; +const SCANNER_MAINTENANCE_REASON_EXPIRY_QUEUE_BACKLOG: &str = "expiry_queue_backlog"; +const SCANNER_MAINTENANCE_REASON_TRANSITION_FAILED: &str = "transition_failed"; +const SCANNER_MAINTENANCE_REASON_TRANSITION_COMPENSATION_BACKLOG: &str = "transition_compensation_backlog"; const SCANNER_MAINTENANCE_REASON_TRANSITION_QUEUE_BACKLOG: &str = "transition_queue_backlog"; const SCANNER_MAINTENANCE_REASON_TRANSITION_QUEUE_FULL: &str = "transition_queue_full"; @@ -1167,6 +1213,15 @@ fn scanner_lifecycle_transition_backlog(metrics: &ScannerMetricsReport) -> u64 { .lifecycle_transition .current_queued .saturating_add(metrics.lifecycle_transition.current_active) + .saturating_add(metrics.lifecycle_transition.compensation_pending) + .saturating_add(metrics.lifecycle_transition.compensation_running) +} + +fn scanner_lifecycle_expiry_backlog(metrics: &ScannerMetricsReport) -> u64 { + metrics + .lifecycle_expiry + .current_queued + .saturating_add(metrics.lifecycle_expiry.current_active) } fn scanner_source_maintenance_state( @@ -1178,6 +1233,14 @@ fn scanner_source_maintenance_state( return (SCANNER_MAINTENANCE_STATE_BLOCKED, SCANNER_MAINTENANCE_REASON_MISSED_WORK, current.missed); } + if source == ScannerWorkSource::Lifecycle && current.failed > 0 { + return ( + SCANNER_MAINTENANCE_STATE_BLOCKED, + SCANNER_MAINTENANCE_REASON_TRANSITION_FAILED, + current.failed, + ); + } + if source == ScannerWorkSource::Lifecycle { let transition_backlog = scanner_lifecycle_transition_backlog(metrics); if metrics.lifecycle_transition.current_queue_capacity > 0 @@ -1200,6 +1263,27 @@ fn scanner_source_maintenance_state( } if source == ScannerWorkSource::Lifecycle { + let expiry_backlog = scanner_lifecycle_expiry_backlog(metrics); + if expiry_backlog > 0 { + return ( + SCANNER_MAINTENANCE_STATE_ACTIVE, + SCANNER_MAINTENANCE_REASON_EXPIRY_QUEUE_BACKLOG, + expiry_backlog, + ); + } + + let compensation_backlog = metrics + .lifecycle_transition + .compensation_pending + .saturating_add(metrics.lifecycle_transition.compensation_running); + if compensation_backlog > 0 { + return ( + SCANNER_MAINTENANCE_STATE_ACTIVE, + SCANNER_MAINTENANCE_REASON_TRANSITION_COMPENSATION_BACKLOG, + compensation_backlog, + ); + } + let transition_backlog = scanner_lifecycle_transition_backlog(metrics); if transition_backlog > 0 { return ( @@ -1418,6 +1502,13 @@ impl Metrics { scanner_ilm_actions: AtomicU64::new(0), scanner_lifecycle_expiry_actions: AtomicU64::new(0), scanner_lifecycle_transition_actions: AtomicU64::new(0), + scanner_expiry_queue_capacity: AtomicU64::new(0), + scanner_expiry_queued: AtomicU64::new(0), + scanner_expiry_active: AtomicU64::new(0), + scanner_expiry_workers: AtomicU64::new(0), + scanner_expiry_queue_missed: AtomicU64::new(0), + scanner_expiry_queued_total: AtomicU64::new(0), + scanner_expiry_missed_total: AtomicU64::new(0), scanner_transition_queue_capacity: AtomicU64::new(0), scanner_transition_queued: AtomicU64::new(0), scanner_transition_active: AtomicU64::new(0), @@ -1425,6 +1516,7 @@ impl Metrics { scanner_transition_queue_full: AtomicU64::new(0), scanner_transition_queue_send_timeout: AtomicU64::new(0), scanner_transition_compensation_scheduled: AtomicU64::new(0), + scanner_transition_compensation_pending: AtomicU64::new(0), scanner_transition_compensation_running: AtomicU64::new(0), scanner_transition_queued_total: AtomicU64::new(0), scanner_transition_missed_total: AtomicU64::new(0), @@ -1622,6 +1714,15 @@ impl Metrics { } } + pub fn record_scanner_expiry_enqueue_result(&self, count: u64, queued: bool) { + self.record_scanner_ilm_enqueue_result(count, queued); + if queued { + self.scanner_expiry_queued_total.fetch_add(count, Ordering::Relaxed); + } else { + self.scanner_expiry_missed_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 { @@ -1631,6 +1732,15 @@ impl Metrics { } } + pub fn record_scanner_lifecycle_expiry_state(&self, state: ScannerLifecycleExpiryStateUpdate) { + self.scanner_expiry_queue_capacity + .store(state.queue_capacity, Ordering::Relaxed); + self.scanner_expiry_queued.store(state.queued, Ordering::Relaxed); + self.scanner_expiry_active.store(state.active, Ordering::Relaxed); + self.scanner_expiry_workers.store(state.workers, Ordering::Relaxed); + self.scanner_expiry_queue_missed.store(state.queue_missed, Ordering::Relaxed); + } + pub fn record_scanner_lifecycle_transition_state(&self, state: ScannerLifecycleTransitionStateUpdate) { self.scanner_transition_queue_capacity .store(state.queue_capacity, Ordering::Relaxed); @@ -1642,6 +1752,8 @@ impl Metrics { .store(state.queue_send_timeout, Ordering::Relaxed); self.scanner_transition_compensation_scheduled .store(state.compensation_scheduled, Ordering::Relaxed); + self.scanner_transition_compensation_pending + .store(state.compensation_pending, Ordering::Relaxed); self.scanner_transition_compensation_running .store(state.compensation_running, Ordering::Relaxed); } @@ -1652,6 +1764,10 @@ impl Metrics { pub fn record_scanner_transition_failed(&self, count: u64) { self.scanner_transition_failed.fetch_add(count, Ordering::Relaxed); + self.record_scanner_source_failed(ScannerWorkSource::Lifecycle, count); + if !self.current_scan_cycle_work_active.load(Ordering::Relaxed) { + self.record_last_cycle_scanner_source_work(ScannerWorkSource::Lifecycle, ScannerSourceWorkUpdate::failed(count)); + } } pub fn record_scanner_checkpoint_set(&self, version: u16, resume_after: impl Into, reason: impl Into) { @@ -1696,6 +1812,12 @@ impl Metrics { } } + fn record_last_cycle_scanner_source_work(&self, source: ScannerWorkSource, work: ScannerSourceWorkUpdate) { + if let Some(counters) = self.last_scan_cycle_source_work.get(source.index()) { + counters.add(work); + } + } + pub fn record_scanner_source_checked(&self, source: ScannerWorkSource, count: u64) { self.record_scanner_source_work(source, ScannerSourceWorkUpdate::checked(count)); } @@ -1708,6 +1830,10 @@ impl Metrics { self.record_scanner_source_work(source, ScannerSourceWorkUpdate::executed(count)); } + pub fn record_scanner_source_failed(&self, source: ScannerWorkSource, count: u64) { + self.record_scanner_source_work(source, ScannerSourceWorkUpdate::failed(count)); + } + pub fn record_scanner_source_missed(&self, source: ScannerWorkSource, count: u64) { self.record_scanner_source_work(source, ScannerSourceWorkUpdate::missed(count)); } @@ -2261,6 +2387,15 @@ impl Metrics { }) }) .collect(); + m.lifecycle_expiry = ScannerLifecycleExpirySnapshot { + current_queue_capacity: self.scanner_expiry_queue_capacity.load(Ordering::Relaxed), + current_queued: self.scanner_expiry_queued.load(Ordering::Relaxed), + current_active: self.scanner_expiry_active.load(Ordering::Relaxed), + current_workers: self.scanner_expiry_workers.load(Ordering::Relaxed), + 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), + }; m.lifecycle_transition = ScannerLifecycleTransitionSnapshot { current_queue_capacity: self.scanner_transition_queue_capacity.load(Ordering::Relaxed), current_queued: self.scanner_transition_queued.load(Ordering::Relaxed), @@ -2269,6 +2404,7 @@ impl Metrics { queue_full: self.scanner_transition_queue_full.load(Ordering::Relaxed), queue_send_timeout: self.scanner_transition_queue_send_timeout.load(Ordering::Relaxed), compensation_scheduled: self.scanner_transition_compensation_scheduled.load(Ordering::Relaxed), + compensation_pending: self.scanner_transition_compensation_pending.load(Ordering::Relaxed), compensation_running: self.scanner_transition_compensation_running.load(Ordering::Relaxed), scanner_queued: self.scanner_transition_queued_total.load(Ordering::Relaxed), scanner_missed: self.scanner_transition_missed_total.load(Ordering::Relaxed), @@ -2645,6 +2781,7 @@ mod tests { queue_full: 3, queue_send_timeout: 1, compensation_scheduled: 2, + compensation_pending: 3, compensation_running: 1, }); metrics.record_scanner_transition_enqueue_result(6, true); @@ -2661,6 +2798,7 @@ mod tests { assert_eq!(report.lifecycle_transition.queue_full, 3); assert_eq!(report.lifecycle_transition.queue_send_timeout, 1); assert_eq!(report.lifecycle_transition.compensation_scheduled, 2); + assert_eq!(report.lifecycle_transition.compensation_pending, 3); assert_eq!(report.lifecycle_transition.compensation_running, 1); assert_eq!(report.lifecycle_transition.scanner_queued, 6); assert_eq!(report.lifecycle_transition.scanner_missed, 2); @@ -2668,6 +2806,30 @@ mod tests { assert_eq!(report.lifecycle_transition.failed, 1); } + #[tokio::test] + async fn report_includes_scanner_lifecycle_expiry_status() { + let metrics = Metrics::new(); + metrics.record_scanner_lifecycle_expiry_state(ScannerLifecycleExpiryStateUpdate { + queue_capacity: 16, + queued: 5, + active: 2, + workers: 4, + queue_missed: 3, + }); + metrics.record_scanner_expiry_enqueue_result(6, true); + metrics.record_scanner_expiry_enqueue_result(2, false); + + let report = metrics.report().await; + + assert_eq!(report.lifecycle_expiry.current_queue_capacity, 16); + assert_eq!(report.lifecycle_expiry.current_queued, 5); + assert_eq!(report.lifecycle_expiry.current_active, 2); + assert_eq!(report.lifecycle_expiry.current_workers, 4); + assert_eq!(report.lifecycle_expiry.queue_missed, 3); + assert_eq!(report.lifecycle_expiry.scanner_queued, 6); + assert_eq!(report.lifecycle_expiry.scanner_missed, 2); + } + #[tokio::test] async fn report_derives_scanner_maintenance_control_status() { let metrics = Metrics::new(); @@ -2725,6 +2887,129 @@ mod tests { metrics.finish_scan_cycle_work(start); } + #[tokio::test] + async fn report_marks_transition_failures_as_blocked_lifecycle_control() { + let metrics = Metrics::new(); + let _start = metrics.start_scan_cycle_work(); + + metrics.record_scanner_transition_failed(2); + let report = metrics.report().await; + + let lifecycle_work = report + .current_cycle_source_work + .iter() + .find(|work| work.source == ScannerWorkSource::Lifecycle.as_str()) + .expect("current lifecycle source work should be visible"); + assert_eq!(lifecycle_work.failed, 2); + + let lifecycle = report + .maintenance_control + .sources + .iter() + .find(|source| source.source == ScannerWorkSource::Lifecycle.as_str()) + .expect("lifecycle source control should be visible"); + assert_eq!(report.maintenance_control.primary_control, "blocked_source"); + assert_eq!(lifecycle.state, "blocked"); + assert_eq!(lifecycle.reason, "transition_failed"); + assert_eq!(lifecycle.backlog, 2); + } + + #[tokio::test] + async fn report_marks_off_cycle_transition_failures_as_blocked_lifecycle_control() { + let metrics = Metrics::new(); + let start = metrics.start_scan_cycle_work(); + metrics.finish_scan_cycle_work(start); + + metrics.record_scanner_transition_failed(1); + let report = metrics.report().await; + + assert!(report.current_cycle_source_work.is_empty()); + let lifecycle_work = report + .last_cycle_source_work + .iter() + .find(|work| work.source == ScannerWorkSource::Lifecycle.as_str()) + .expect("last-cycle lifecycle source work should be visible"); + assert_eq!(lifecycle_work.failed, 1); + + let lifecycle = report + .maintenance_control + .sources + .iter() + .find(|source| source.source == ScannerWorkSource::Lifecycle.as_str()) + .expect("lifecycle source control should be visible"); + assert_eq!(report.maintenance_control.primary_control, "blocked_source"); + assert_eq!(lifecycle.state, "blocked"); + assert_eq!(lifecycle.reason, "transition_failed"); + assert_eq!(lifecycle.backlog, 1); + } + + #[tokio::test] + async fn report_marks_expiry_queue_backlog_as_lifecycle_control() { + let metrics = Metrics::new(); + metrics.record_scanner_lifecycle_expiry_state(ScannerLifecycleExpiryStateUpdate { + queued: 2, + active: 1, + ..Default::default() + }); + + let report = metrics.report().await; + + let lifecycle = report + .maintenance_control + .sources + .iter() + .find(|source| source.source == ScannerWorkSource::Lifecycle.as_str()) + .expect("lifecycle source control should be visible"); + assert_eq!(report.maintenance_control.primary_control, "active_source"); + assert_eq!(lifecycle.state, "active"); + assert_eq!(lifecycle.reason, "expiry_queue_backlog"); + assert_eq!(lifecycle.backlog, 3); + } + + #[tokio::test] + async fn report_marks_running_transition_compensation_as_lifecycle_backlog() { + let metrics = Metrics::new(); + metrics.record_scanner_lifecycle_transition_state(ScannerLifecycleTransitionStateUpdate { + compensation_running: 1, + ..Default::default() + }); + + let report = metrics.report().await; + + let lifecycle = report + .maintenance_control + .sources + .iter() + .find(|source| source.source == ScannerWorkSource::Lifecycle.as_str()) + .expect("lifecycle source control should be visible"); + assert_eq!(report.maintenance_control.primary_control, "active_source"); + assert_eq!(lifecycle.state, "active"); + assert_eq!(lifecycle.reason, "transition_compensation_backlog"); + assert_eq!(lifecycle.backlog, 1); + } + + #[tokio::test] + async fn report_marks_pending_transition_compensation_as_lifecycle_backlog() { + let metrics = Metrics::new(); + metrics.record_scanner_lifecycle_transition_state(ScannerLifecycleTransitionStateUpdate { + compensation_pending: 2, + ..Default::default() + }); + + let report = metrics.report().await; + + let lifecycle = report + .maintenance_control + .sources + .iter() + .find(|source| source.source == ScannerWorkSource::Lifecycle.as_str()) + .expect("lifecycle source control should be visible"); + assert_eq!(report.maintenance_control.primary_control, "active_source"); + assert_eq!(lifecycle.state, "active"); + assert_eq!(lifecycle.reason, "transition_compensation_backlog"); + assert_eq!(lifecycle.backlog, 2); + } + #[tokio::test] async fn report_uses_last_cycle_source_work_for_maintenance_control_between_cycles() { let metrics = Metrics::new(); diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 7c167352e..4332e6787 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -45,7 +45,9 @@ use futures::Future; use http::HeaderMap; use lazy_static::lazy_static; use rustfs_common::heal_channel::rep_has_active_rules; -use rustfs_common::metrics::{IlmAction, Metrics, ScannerLifecycleTransitionStateUpdate, global_metrics}; +use rustfs_common::metrics::{ + IlmAction, Metrics, ScannerLifecycleExpiryStateUpdate, ScannerLifecycleTransitionStateUpdate, global_metrics, +}; use rustfs_config::{ DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_QUEUE_SEND_TIMEOUT_MS, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX, DEFAULT_TRANSITION_WORKERS_CAP, ENV_TRANSITION_QUEUE_CAPACITY, ENV_TRANSITION_QUEUE_SEND_TIMEOUT_MS, ENV_TRANSITION_WORKERS, @@ -110,6 +112,7 @@ const ENV_STALE_UPLOADS_CLEANUP_INTERVAL: &str = "RUSTFS_API_STALE_UPLOADS_CLEAN const DEFAULT_STALE_UPLOADS_EXPIRY: StdDuration = StdDuration::from_secs(24 * 60 * 60); const DEFAULT_STALE_UPLOADS_CLEANUP_INTERVAL: StdDuration = StdDuration::from_secs(6 * 60 * 60); const DATE_EXPIRY_EXISTING_OBJECTS_GRACE_SECS: i64 = 5; +const EXPIRY_WORKER_QUEUE_CAPACITY: usize = 1000; lazy_static! { pub static ref GLOBAL_ExpiryState: Arc> = ExpiryState::new(); @@ -157,7 +160,7 @@ fn is_immediate_transition_source(src: &LcEventSrc) -> bool { fn record_scanner_lifecycle_enqueue_result(src: &LcEventSrc, count: u64, queued: bool) { if matches!(src, LcEventSrc::Scanner) { - global_metrics().record_scanner_ilm_enqueue_result(count, queued); + global_metrics().record_scanner_expiry_enqueue_result(count, queued); } } @@ -261,6 +264,8 @@ struct ExpiryStats { missed_expiry_tasks: AtomicI64, missed_freevers_tasks: AtomicI64, missed_tier_journal_tasks: AtomicI64, + pending_tasks: AtomicI64, + active_tasks: AtomicI64, workers: AtomicI64, } @@ -278,9 +283,91 @@ impl ExpiryStats { self.missed_tier_journal_tasks.load(Ordering::SeqCst) } + pub fn pending_tasks(&self) -> i64 { + self.pending_tasks.load(Ordering::SeqCst) + } + + pub fn active_tasks(&self) -> i64 { + self.active_tasks.load(Ordering::SeqCst) + } + fn num_workers(&self) -> i64 { self.workers.load(Ordering::SeqCst) } + + fn add_nonnegative(counter: &AtomicI64, delta: i64) { + let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_add(delta).max(0))); + } + + fn increment_missed_expiry_tasks(&self) { + Self::add_nonnegative(&self.missed_expiry_tasks, 1); + } + + fn increment_missed_freevers_tasks(&self) { + Self::add_nonnegative(&self.missed_freevers_tasks, 1); + } + + fn increment_missed_tier_journal_tasks(&self) { + Self::add_nonnegative(&self.missed_tier_journal_tasks, 1); + } + + fn increment_pending_tasks(&self) { + Self::add_nonnegative(&self.pending_tasks, 1); + } + + fn decrement_pending_tasks(&self) { + Self::add_nonnegative(&self.pending_tasks, -1); + } + + fn increment_active_tasks(&self) { + Self::add_nonnegative(&self.active_tasks, 1); + } + + fn decrement_active_tasks(&self) { + Self::add_nonnegative(&self.active_tasks, -1); + } + + fn increment_workers(&self) { + Self::add_nonnegative(&self.workers, 1); + } + + fn decrement_workers(&self) { + Self::add_nonnegative(&self.workers, -1); + } + + fn scanner_expiry_state_update(&self) -> ScannerLifecycleExpiryStateUpdate { + let workers = nonnegative_i64_to_u64(self.num_workers()); + ScannerLifecycleExpiryStateUpdate { + queue_capacity: workers.saturating_mul(usize_to_u64_saturated(EXPIRY_WORKER_QUEUE_CAPACITY)), + queued: nonnegative_i64_to_u64(self.pending_tasks()), + active: nonnegative_i64_to_u64(self.active_tasks()), + workers, + queue_missed: nonnegative_i64_to_u64(self.missed_tasks()), + } + } + + fn record_scanner_expiry_state(&self) { + global_metrics().record_scanner_lifecycle_expiry_state(self.scanner_expiry_state_update()); + } +} + +struct ExpiryActiveTask { + stats: Arc, +} + +impl ExpiryActiveTask { + fn begin(stats: Arc) -> Self { + stats.increment_active_tasks(); + stats.record_scanner_expiry_state(); + Self { stats } + } +} + +impl Drop for ExpiryActiveTask { + fn drop(&mut self) { + self.stats.decrement_active_tasks(); + self.stats.record_scanner_expiry_state(); + } } pub trait ExpiryOp: 'static { @@ -334,7 +421,7 @@ impl ExpiryOp for NewerNoncurrentTask { pub struct ExpiryState { tasks_tx: Vec>>, tasks_rx: Vec>>>>, - stats: Option, + stats: Arc, } impl ExpiryState { @@ -343,41 +430,43 @@ impl ExpiryState { Arc::new(RwLock::new(Self { tasks_tx: vec![], tasks_rx: vec![], - stats: Some(ExpiryStats { + stats: Arc::new(ExpiryStats { missed_expiry_tasks: AtomicI64::new(0), missed_freevers_tasks: AtomicI64::new(0), missed_tier_journal_tasks: AtomicI64::new(0), + pending_tasks: AtomicI64::new(0), + active_tasks: AtomicI64::new(0), workers: AtomicI64::new(0), }), })) } - pub async fn pending_tasks(&self) -> usize { - let rxs = &self.tasks_rx; - if rxs.is_empty() { - return 0; + pub fn pending_tasks(&self) -> usize { + usize::try_from(self.stats.pending_tasks().max(0)).unwrap_or(usize::MAX) + } + + async fn send_expiry_task(&self, wrkr: Sender>, task: ExpiryOpType) -> bool { + self.stats.increment_pending_tasks(); + let queued = wrkr.send(Some(task)).await.is_ok(); + if !queued { + self.stats.decrement_pending_tasks(); } - let mut tasks = 0; - for rx in rxs.iter() { - tasks += rx.lock().await.len(); - } - tasks + queued } pub async fn enqueue_tier_journal_entry(&mut self, je: &Jentry) -> Result<(), std::io::Error> { let wrkr = self.get_worker_ch(je.op_hash()); if wrkr.is_none() { - *self.stats.as_mut().expect("stats lock").missed_tier_journal_tasks.get_mut() += 1; + self.stats.increment_missed_tier_journal_tasks(); + self.stats.record_scanner_expiry_state(); return Ok(()); } let wrkr = wrkr.expect("worker channel should exist after None check"); - select! { - //_ -> GlobalContext.Done() => () - _ = wrkr.send(Some(Box::new(je.clone()))) => (), - else => { - *self.stats.as_mut().expect("stats lock").missed_tier_journal_tasks.get_mut() += 1; - } + let queued = self.send_expiry_task(wrkr, Box::new(je.clone())).await; + if !queued { + self.stats.increment_missed_tier_journal_tasks(); } + self.stats.record_scanner_expiry_state(); Ok(()) } @@ -385,17 +474,16 @@ impl ExpiryState { let task = FreeVersionTask(oi); let wrkr = self.get_worker_ch(task.op_hash()); if wrkr.is_none() { - *self.stats.as_mut().expect("stats lock").missed_freevers_tasks.get_mut() += 1; + self.stats.increment_missed_freevers_tasks(); + self.stats.record_scanner_expiry_state(); return; } let wrkr = wrkr.expect("worker channel should exist after None check"); - select! { - //_ -> GlobalContext.Done() => {} - _ = wrkr.send(Some(Box::new(task))) => (), - else => { - *self.stats.as_mut().expect("stats lock").missed_freevers_tasks.get_mut() += 1; - } + let queued = self.send_expiry_task(wrkr, Box::new(task)).await; + if !queued { + self.stats.increment_missed_freevers_tasks(); } + self.stats.record_scanner_expiry_state(); } pub async fn enqueue_by_days(&mut self, oi: &ObjectInfo, event: &lifecycle::Event, src: &LcEventSrc) -> bool { @@ -406,16 +494,18 @@ impl ExpiryState { }; let wrkr = self.get_worker_ch(task.op_hash()); if wrkr.is_none() { - *self.stats.as_mut().expect("stats lock").missed_expiry_tasks.get_mut() += 1; + self.stats.increment_missed_expiry_tasks(); record_scanner_lifecycle_enqueue_result(src, 1, false); + self.stats.record_scanner_expiry_state(); return false; } let wrkr = wrkr.expect("worker channel should exist after None check"); - let queued = wrkr.send(Some(Box::new(task))).await.is_ok(); + let queued = self.send_expiry_task(wrkr, Box::new(task)).await; if !queued { - *self.stats.as_mut().expect("stats lock").missed_expiry_tasks.get_mut() += 1; + self.stats.increment_missed_expiry_tasks(); } record_scanner_lifecycle_enqueue_result(src, 1, queued); + self.stats.record_scanner_expiry_state(); queued } @@ -438,16 +528,18 @@ impl ExpiryState { }; let wrkr = self.get_worker_ch(task.op_hash()); if wrkr.is_none() { - *self.stats.as_mut().expect("stats lock").missed_expiry_tasks.get_mut() += 1; + self.stats.increment_missed_expiry_tasks(); record_scanner_lifecycle_enqueue_result(src, version_count, false); + self.stats.record_scanner_expiry_state(); return false; } let wrkr = wrkr.expect("worker channel should exist after None check"); - let queued = wrkr.send(Some(Box::new(task))).await.is_ok(); + let queued = self.send_expiry_task(wrkr, Box::new(task)).await; if !queued { - *self.stats.as_mut().expect("stats lock").missed_expiry_tasks.get_mut() += 1; + self.stats.increment_missed_expiry_tasks(); } record_scanner_lifecycle_enqueue_result(src, version_count, queued); + self.stats.record_scanner_expiry_state(); queued } @@ -459,7 +551,8 @@ impl ExpiryState { } pub fn increment_missed_tier_journal_tasks(&mut self) { - *self.stats.as_mut().expect("stats lock").missed_tier_journal_tasks.get_mut() += 1; + self.stats.increment_missed_tier_journal_tasks(); + self.stats.record_scanner_expiry_state(); } pub async fn resize_workers(n: usize, api: Arc) { @@ -470,16 +563,17 @@ impl ExpiryState { let mut state = GLOBAL_ExpiryState.write().await; while state.tasks_tx.len() < n { - let (tx, rx) = mpsc::channel(1000); + let (tx, rx) = mpsc::channel(EXPIRY_WORKER_QUEUE_CAPACITY); let api = api.clone(); let rx = Arc::new(tokio::sync::Mutex::new(rx)); + let stats = Arc::clone(&state.stats); state.tasks_tx.push(tx); state.tasks_rx.push(rx.clone()); - *state.stats.as_mut().expect("stats lock").workers.get_mut() += 1; + state.stats.increment_workers(); tokio::spawn(async move { let mut rx = rx.lock().await; //let mut expiry_state = GLOBAL_ExpiryState.read().await; - ExpiryState::worker(&mut rx, api).await; + ExpiryState::worker(&mut rx, api, stats).await; }); } @@ -489,12 +583,13 @@ impl ExpiryState { worker.send(None).await.unwrap_or(()); state.tasks_tx.remove(l - 1); state.tasks_rx.remove(l - 1); - *state.stats.as_mut().expect("stats lock").workers.get_mut() -= 1; + state.stats.decrement_workers(); l -= 1; } + state.stats.record_scanner_expiry_state(); } - pub async fn worker(rx: &mut Receiver>, api: Arc) { + async fn worker(rx: &mut Receiver>, api: Arc, stats: Arc) { let cancel_token = crate::global::get_background_services_cancel_token().unwrap_or_else(|| { static FALLBACK: std::sync::OnceLock = std::sync::OnceLock::new(); FALLBACK.get_or_init(tokio_util::sync::CancellationToken::new) @@ -525,6 +620,8 @@ impl ExpiryState { return; } let v = v.expect("received None after None check"); + stats.decrement_pending_tasks(); + let _active_task = ExpiryActiveTask::begin(Arc::clone(&stats)); if v.as_any().is::() { let v = v.as_any().downcast_ref::().expect("ExpiryTask downcast failed"); //debug!("lifecycle expiry worker received task: {:?}", v.obj_info); @@ -803,6 +900,7 @@ impl TransitionState { queue_full: nonnegative_i64_to_u64(Self::counter_value(&self.queue_full_tasks)), queue_send_timeout: nonnegative_i64_to_u64(Self::counter_value(&self.queue_send_timeout_tasks)), compensation_scheduled: nonnegative_i64_to_u64(Self::counter_value(&self.compensation_scheduled_tasks)), + compensation_pending: self.compensation_pending_tasks(), compensation_running: nonnegative_i64_to_u64(Self::counter_value(&self.compensation_running_tasks)), } } @@ -1002,6 +1100,13 @@ impl TransitionState { Self::counter_value(&self.compensation_scheduled_tasks) } + pub fn compensation_pending_tasks(&self) -> u64 { + match self.compensation_buckets.lock() { + Ok(scheduled) => usize_to_u64_saturated(scheduled.len()), + Err(poisoned) => usize_to_u64_saturated(poisoned.into_inner().len()), + } + } + pub fn compensation_running_tasks(&self) -> i64 { Self::counter_value(&self.compensation_running_tasks) } @@ -2655,8 +2760,13 @@ mod tests { let queued = state.enqueue_by_days(&object, &event, &LcEventSrc::Scanner).await; assert!(!queued); - let stats = state.stats.as_ref().expect("expiry stats should exist"); - assert_eq!(stats.missed_tasks(), 1); + assert_eq!(state.stats.missed_tasks(), 1); + let expiry = global_metrics().report().await.lifecycle_expiry; + assert_eq!(expiry.current_queue_capacity, 0); + assert_eq!(expiry.current_queued, 0); + assert_eq!(expiry.current_active, 0); + assert_eq!(expiry.current_workers, 0); + assert_eq!(expiry.queue_missed, 1); } #[tokio::test] @@ -3060,6 +3170,17 @@ mod tests { assert_eq!(state.compensation_scheduled_tasks(), 1); } + #[tokio::test(flavor = "current_thread")] + async fn scanner_transition_state_reports_compensation_pending_buckets() { + let state = TransitionState::new_with_capacity(1); + + assert_eq!(state.scanner_transition_state_update().compensation_pending, 0); + state.compensation_buckets.lock().unwrap().insert("bucket-a".to_string()); + assert_eq!(state.scanner_transition_state_update().compensation_pending, 1); + state.compensation_buckets.lock().unwrap().insert("bucket-a".to_string()); + assert_eq!(state.scanner_transition_state_update().compensation_pending, 1); + } + #[tokio::test] #[serial] async fn transition_state_init_honors_runtime_configured_worker_count() { diff --git a/crates/ecstore/src/metrics_realtime.rs b/crates/ecstore/src/metrics_realtime.rs index c91b150f3..a7d263bb4 100644 --- a/crates/ecstore/src/metrics_realtime.rs +++ b/crates/ecstore/src/metrics_realtime.rs @@ -19,6 +19,7 @@ use rustfs_io_metrics::internode_metrics::global_internode_metrics; use rustfs_madmin::metrics::{ DiskIOStats, DiskMetric, LastMinute as MadminLastMinute, NetDevLine, NetMetrics, RPCMetrics, RealtimeMetrics, ScannerCheckpointReport as MadminScannerCheckpointReport, + ScannerLifecycleExpirySnapshot as MadminScannerLifecycleExpirySnapshot, ScannerLifecycleTransitionSnapshot as MadminScannerLifecycleTransitionSnapshot, ScannerMaintenanceControlSnapshot as MadminScannerMaintenanceControlSnapshot, ScannerMaintenanceSourceSnapshot as MadminScannerMaintenanceSourceSnapshot, ScannerMetrics as MadminScannerMetrics, @@ -173,12 +174,22 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo queue_full: metrics.lifecycle_transition.queue_full, queue_send_timeout: metrics.lifecycle_transition.queue_send_timeout, compensation_scheduled: metrics.lifecycle_transition.compensation_scheduled, + compensation_pending: metrics.lifecycle_transition.compensation_pending, compensation_running: metrics.lifecycle_transition.compensation_running, scanner_queued: metrics.lifecycle_transition.scanner_queued, scanner_missed: metrics.lifecycle_transition.scanner_missed, completed: metrics.lifecycle_transition.completed, failed: metrics.lifecycle_transition.failed, }, + lifecycle_expiry: MadminScannerLifecycleExpirySnapshot { + current_queue_capacity: metrics.lifecycle_expiry.current_queue_capacity, + current_queued: metrics.lifecycle_expiry.current_queued, + current_active: metrics.lifecycle_expiry.current_active, + current_workers: metrics.lifecycle_expiry.current_workers, + queue_missed: metrics.lifecycle_expiry.queue_missed, + scanner_queued: metrics.lifecycle_expiry.scanner_queued, + scanner_missed: metrics.lifecycle_expiry.scanner_missed, + }, maintenance_control: MadminScannerMaintenanceControlSnapshot { primary_control: metrics.maintenance_control.primary_control, sources: metrics @@ -566,6 +577,15 @@ mod test { current_cycle_lifecycle_transition_actions: 3, last_cycle_lifecycle_expiry_actions: 5, last_cycle_lifecycle_transition_actions: 7, + lifecycle_expiry: rustfs_common::metrics::ScannerLifecycleExpirySnapshot { + current_queue_capacity: 16, + current_queued: 5, + current_active: 2, + current_workers: 4, + queue_missed: 3, + scanner_queued: 6, + scanner_missed: 2, + }, lifecycle_transition: rustfs_common::metrics::ScannerLifecycleTransitionSnapshot { current_queue_capacity: 16, current_queued: 5, @@ -574,6 +594,7 @@ mod test { queue_full: 3, queue_send_timeout: 1, compensation_scheduled: 2, + compensation_pending: 3, compensation_running: 1, scanner_queued: 6, scanner_missed: 2, @@ -583,6 +604,13 @@ mod test { ..Default::default() }); + assert_eq!(scanner.lifecycle_expiry.current_queue_capacity, 16); + assert_eq!(scanner.lifecycle_expiry.current_queued, 5); + assert_eq!(scanner.lifecycle_expiry.current_active, 2); + assert_eq!(scanner.lifecycle_expiry.current_workers, 4); + 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_transition.current_queue_capacity, 16); assert_eq!(scanner.lifecycle_transition.current_queued, 5); assert_eq!(scanner.lifecycle_transition.current_active, 2); @@ -590,6 +618,7 @@ mod test { assert_eq!(scanner.lifecycle_transition.queue_full, 3); assert_eq!(scanner.lifecycle_transition.queue_send_timeout, 1); assert_eq!(scanner.lifecycle_transition.compensation_scheduled, 2); + assert_eq!(scanner.lifecycle_transition.compensation_pending, 3); assert_eq!(scanner.lifecycle_transition.compensation_running, 1); assert_eq!(scanner.lifecycle_transition.scanner_queued, 6); assert_eq!(scanner.lifecycle_transition.scanner_missed, 2); diff --git a/crates/madmin/src/metrics.rs b/crates/madmin/src/metrics.rs index afdc586eb..3fc9259a5 100644 --- a/crates/madmin/src/metrics.rs +++ b/crates/madmin/src/metrics.rs @@ -357,6 +357,36 @@ impl ScannerPacingPressureSnapshot { } } +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct ScannerLifecycleExpirySnapshot { + #[serde(rename = "current_queue_capacity", default)] + pub current_queue_capacity: u64, + #[serde(rename = "current_queued", default)] + pub current_queued: u64, + #[serde(rename = "current_active", default)] + pub current_active: u64, + #[serde(rename = "current_workers", default)] + pub current_workers: u64, + #[serde(rename = "queue_missed", default)] + pub queue_missed: u64, + #[serde(rename = "scanner_queued", default)] + pub scanner_queued: u64, + #[serde(rename = "scanner_missed", default)] + pub scanner_missed: u64, +} + +impl ScannerLifecycleExpirySnapshot { + fn merge(&mut self, other: &Self) { + self.current_queue_capacity = self.current_queue_capacity.saturating_add(other.current_queue_capacity); + self.current_queued = self.current_queued.saturating_add(other.current_queued); + self.current_active = self.current_active.saturating_add(other.current_active); + self.current_workers = self.current_workers.saturating_add(other.current_workers); + 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); + } +} + #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] pub struct ScannerLifecycleTransitionSnapshot { #[serde(rename = "current_queue_capacity", default)] @@ -373,6 +403,8 @@ pub struct ScannerLifecycleTransitionSnapshot { pub queue_send_timeout: u64, #[serde(rename = "compensation_scheduled", default)] pub compensation_scheduled: u64, + #[serde(rename = "compensation_pending", default)] + pub compensation_pending: u64, #[serde(rename = "compensation_running", default)] pub compensation_running: u64, #[serde(rename = "scanner_queued", default)] @@ -394,6 +426,7 @@ impl ScannerLifecycleTransitionSnapshot { self.queue_full = self.queue_full.saturating_add(other.queue_full); self.queue_send_timeout = self.queue_send_timeout.saturating_add(other.queue_send_timeout); self.compensation_scheduled = self.compensation_scheduled.saturating_add(other.compensation_scheduled); + self.compensation_pending = self.compensation_pending.saturating_add(other.compensation_pending); self.compensation_running = self.compensation_running.saturating_add(other.compensation_running); self.scanner_queued = self.scanner_queued.saturating_add(other.scanner_queued); self.scanner_missed = self.scanner_missed.saturating_add(other.scanner_missed); @@ -524,6 +557,8 @@ pub struct ScannerMetrics { pub partial_cycles_by_source: Vec, #[serde(rename = "pacing_pressure", default)] pub pacing_pressure: ScannerPacingPressureSnapshot, + #[serde(rename = "lifecycle_expiry", default)] + pub lifecycle_expiry: ScannerLifecycleExpirySnapshot, #[serde(rename = "lifecycle_transition", default)] pub lifecycle_transition: ScannerLifecycleTransitionSnapshot, #[serde(rename = "maintenance_control", default)] @@ -599,6 +634,7 @@ impl ScannerMetrics { self.active_scan_paths = self.active_scan_paths.saturating_add(other.active_scan_paths); self.oldest_active_path_age_seconds = self.oldest_active_path_age_seconds.max(other.oldest_active_path_age_seconds); self.pacing_pressure.merge(&other.pacing_pressure); + self.lifecycle_expiry.merge(&other.lifecycle_expiry); self.lifecycle_transition.merge(&other.lifecycle_transition); self.maintenance_control.merge(&other.maintenance_control); self.current_set_scan_concurrency_limit = self @@ -1354,6 +1390,15 @@ mod tests { current_cycle_lifecycle_transition_actions: 3, last_cycle_lifecycle_expiry_actions: 5, last_cycle_lifecycle_transition_actions: 7, + lifecycle_expiry: ScannerLifecycleExpirySnapshot { + current_queue_capacity: 8, + current_queued: 2, + current_active: 1, + current_workers: 2, + queue_missed: 3, + scanner_queued: 5, + scanner_missed: 2, + }, lifecycle_transition: ScannerLifecycleTransitionSnapshot { current_queue_capacity: 8, current_queued: 2, @@ -1362,6 +1407,7 @@ mod tests { queue_full: 3, queue_send_timeout: 1, compensation_scheduled: 1, + compensation_pending: 2, compensation_running: 1, scanner_queued: 5, scanner_missed: 2, @@ -1377,6 +1423,15 @@ mod tests { current_cycle_lifecycle_transition_actions: 13, last_cycle_lifecycle_expiry_actions: 17, last_cycle_lifecycle_transition_actions: 19, + lifecycle_expiry: ScannerLifecycleExpirySnapshot { + current_queue_capacity: 4, + current_queued: 3, + current_active: 2, + current_workers: 1, + queue_missed: 2, + scanner_queued: 6, + scanner_missed: 4, + }, lifecycle_transition: ScannerLifecycleTransitionSnapshot { current_queue_capacity: 4, current_queued: 3, @@ -1385,6 +1440,7 @@ mod tests { queue_full: 2, queue_send_timeout: 3, compensation_scheduled: 4, + compensation_pending: 3, compensation_running: 0, scanner_queued: 6, scanner_missed: 4, @@ -1394,6 +1450,13 @@ mod tests { ..Default::default() }); + assert_eq!(scanner.lifecycle_expiry.current_queue_capacity, 12); + assert_eq!(scanner.lifecycle_expiry.current_queued, 5); + assert_eq!(scanner.lifecycle_expiry.current_active, 3); + assert_eq!(scanner.lifecycle_expiry.current_workers, 3); + 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_transition.current_queue_capacity, 12); assert_eq!(scanner.lifecycle_transition.current_queued, 5); assert_eq!(scanner.lifecycle_transition.current_active, 3); @@ -1401,6 +1464,7 @@ mod tests { assert_eq!(scanner.lifecycle_transition.queue_full, 5); assert_eq!(scanner.lifecycle_transition.queue_send_timeout, 4); assert_eq!(scanner.lifecycle_transition.compensation_scheduled, 5); + assert_eq!(scanner.lifecycle_transition.compensation_pending, 5); assert_eq!(scanner.lifecycle_transition.compensation_running, 1); assert_eq!(scanner.lifecycle_transition.scanner_queued, 11); assert_eq!(scanner.lifecycle_transition.scanner_missed, 6); diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 624c22819..77e7b5122 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -870,7 +870,7 @@ pub async fn collect_cluster_usage_metric_stats() -> Option<(ClusterUsageStats, /// Collect ILM metrics from the current lifecycle runtime state. pub async fn collect_ilm_metric_stats() -> Option { - let expiry_pending_tasks = GLOBAL_ExpiryState.read().await.pending_tasks().await as u64; + let expiry_pending_tasks = GLOBAL_ExpiryState.read().await.pending_tasks() as u64; let transition_active_tasks = GLOBAL_TransitionState.active_tasks().max(0) as u64; let transition_pending_tasks = GLOBAL_TransitionState.pending_tasks() as u64; let transition_missed_immediate_tasks = GLOBAL_TransitionState.missed_immediate_tasks().max(0) as u64; diff --git a/docs/operations/scanner-benchmark-runbook.md b/docs/operations/scanner-benchmark-runbook.md index 04ddf0343..d57025b52 100644 --- a/docs/operations/scanner-benchmark-runbook.md +++ b/docs/operations/scanner-benchmark-runbook.md @@ -292,8 +292,14 @@ Compare these fields between baseline and tuned runs: | `metrics.source_work` | Shows cumulative work found, queued, skipped, missed, executed, and failed by source. | | `metrics.current_cycle_source_work` | Shows which source is consuming the current scan cycle. | | `metrics.last_cycle_source_work` | Shows which source consumed the previous scan cycle. | +| `metrics.lifecycle_expiry.current_queued` | Shows scanner-driven expiry/delete work waiting in the expiry worker queue. | +| `metrics.lifecycle_expiry.current_active` | Shows scanner-driven expiry/delete work currently running in expiry workers. | +| `metrics.lifecycle_expiry.queue_missed` | Shows expiry/delete queue admission failures outside the scanner walk itself. | +| `metrics.lifecycle_expiry.scanner_missed` | Shows scanner-discovered expiry/delete work that could not be queued. | | `metrics.lifecycle_transition.scanner_missed` | Shows scanner-discovered transition work that could not be queued. | | `metrics.lifecycle_transition.queue_full` | Shows transition queue pressure outside the scanner walk itself. | +| `metrics.lifecycle_transition.compensation_pending` | Shows transition compensation still pending or running after queue pressure. | +| `metrics.lifecycle_transition.failed` | Shows transition worker failures, which should also surface as lifecycle source failure. | | `metrics.scan_checkpoint` | Confirms partial cycles preserve resume context. | | `metrics.oldest_active_path_age_seconds` | Helps identify scanner paths that may be stuck. | @@ -318,8 +324,12 @@ Treat these as failure signals: - CPU drops only because the scanner stops making progress; - `primary_pressure` stays at `queued_scans` while queues grow; - `last_cycle_partial_reason` repeats forever with no checkpoint movement; -- lifecycle transition `scanner_missed` or `queue_full` grows during a run that - was expected to reduce backlog; +- lifecycle expiry `queue_missed`, `scanner_missed`, `current_queued`, or + `current_active` grows during a run that was expected to reduce expiry + backlog; +- lifecycle transition `scanner_missed`, `queue_full`, + `compensation_pending`, or `failed` grows during a run that was expected to + reduce backlog; - heal or bitrot work moves from `queued` to `missed` after a scanner pacing change. diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md index bbb3d01c5..72b0d08a9 100644 --- a/docs/operations/scanner-runtime-controls.md +++ b/docs/operations/scanner-runtime-controls.md @@ -173,7 +173,7 @@ Each `metrics.maintenance_control.sources[]` entry has: |---|---| | `source` | Scanner source such as `usage`, `lifecycle`, `bucket_replication`, `site_replication`, `heal`, `bitrot`, or `alerts`. | | `state` | `idle`, `active`, `deferred`, or `blocked`. | -| `reason` | Derived reason such as `active_work`, `queued_work`, `partial_cycle`, `missed_work`, `transition_queue_backlog`, or `transition_queue_full`. | +| `reason` | Derived reason such as `active_work`, `queued_work`, `partial_cycle`, `missed_work`, `expiry_queue_backlog`, `transition_failed`, `transition_compensation_backlog`, `transition_queue_backlog`, or `transition_queue_full`. | | `backlog` | Current source-level backlog estimate from queued or missed work. | | `current_checked` | Current-cycle checked work for this source, or the last completed cycle when no scan cycle is active. | | `current_queued` | Current-cycle queued work for this source, or the last completed cycle when no scan cycle is active. | @@ -184,7 +184,25 @@ Each `metrics.maintenance_control.sources[]` entry has: Use this snapshot before changing scanner controls. For example, `blocked_source` with `lifecycle/missed_work` points at downstream lifecycle admission, while `deferred_source` with `usage/partial_cycle` points at scanner -cycle budgets. +cycle budgets. `lifecycle/expiry_queue_backlog` means scanner-driven expiry or +delete work is still queued or active in the expiry worker pool. +`lifecycle/transition_failed` means transition worker execution failed during +the current or last completed scan cycle, while +`lifecycle/transition_compensation_backlog` means transition compensation is +still pending or running after queue backpressure. + +`metrics.lifecycle_expiry` exposes the expiry/delete worker queue observed by +scanner-driven lifecycle work: + +| Field | Meaning | +|---|---| +| `current_queue_capacity` | Effective expiry worker queue capacity for this node. | +| `current_queued` | Expiry/delete tasks currently waiting in the worker queue. | +| `current_active` | Expiry/delete tasks currently running in a worker. | +| `current_workers` | Configured expiry worker count. | +| `queue_missed` | Expiry/delete tasks that could not be queued because no worker channel was available or the queue was closed. | +| `scanner_queued` | Scanner-discovered expiry/delete object versions admitted to the expiry queue. | +| `scanner_missed` | Scanner-discovered expiry/delete object versions that could not be admitted. | ## Reading Distributed Metrics @@ -209,9 +227,10 @@ done ``` The `aggregated.scanner` payload preserves the same scanner progress, -checkpoint, pacing, source work, maintenance control, and lifecycle transition -fields used by the local scanner status, but only for the node that returned -the response. The `by_host.*.scanner` payload keeps that node's host view. +checkpoint, pacing, source work, maintenance control, lifecycle expiry, and +lifecycle transition fields used by the local scanner status, but only for the +node that returned the response. The `by_host.*.scanner` payload keeps that +node's host view. Compare the per-node artifacts externally to find old active paths, partial checkpoints, pacing pressure, source-level control pressure, or downstream queue admission problems across the deployment. @@ -230,6 +249,7 @@ work: | `queue_full` | Queue-full observations in the transition state. | | `queue_send_timeout` | Send timeouts for transition queue admission. | | `compensation_scheduled` | Buckets scheduled for transition compensation. | +| `compensation_pending` | Buckets with transition compensation still pending or running. | | `compensation_running` | Transition compensation tasks currently running. | | `scanner_queued` | Scanner transition tasks admitted to the queue. | | `scanner_missed` | Scanner transition tasks that could not be admitted. |