From f49df41db933324577022e2fce1921458711a62a Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Mon, 8 Jun 2026 00:18:34 +0800 Subject: [PATCH] fix(lifecycle): harden scanner ILM expiry accounting (#3257) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(lifecycle): harden scanner ILM expiry accounting * fix(scanner): gate ILM action accounting on enqueue * fix(metrics): avoid scanner source work argument list * fix(scanner): gate local ILM accounting on enqueue --------- Co-authored-by: Henry Guo Co-authored-by: 安正超 Co-authored-by: houseme --- crates/common/src/metrics.rs | 180 +++++++++++++++--- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 153 ++++++++++----- crates/scanner/src/scanner_folder.rs | 170 ++++++++++++++--- 3 files changed, 400 insertions(+), 103 deletions(-) diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index 546a7fcc7..fbfd5a50e 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -316,15 +316,73 @@ struct ScannerSourceWorkCounters { executed: AtomicU64, failed: AtomicU64, skipped: AtomicU64, + missed: AtomicU64, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ScannerSourceWorkUpdate { + pub checked: u64, + pub queued: u64, + pub executed: u64, + pub failed: u64, + pub skipped: u64, + pub missed: u64, +} + +impl ScannerSourceWorkUpdate { + pub const fn checked(count: u64) -> Self { + Self { + checked: count, + queued: 0, + executed: 0, + failed: 0, + skipped: 0, + missed: 0, + } + } + + pub const fn queued(count: u64) -> Self { + Self { + checked: 0, + queued: count, + executed: 0, + failed: 0, + skipped: 0, + missed: 0, + } + } + + pub const fn executed(count: u64) -> Self { + Self { + checked: 0, + queued: 0, + executed: count, + failed: 0, + skipped: 0, + missed: 0, + } + } + + pub const fn missed(count: u64) -> Self { + Self { + checked: 0, + queued: 0, + executed: 0, + failed: 0, + skipped: 0, + missed: count, + } + } } impl ScannerSourceWorkCounters { - fn add(&self, checked: u64, queued: u64, executed: u64, failed: u64, skipped: u64) { - self.checked.fetch_add(checked, Ordering::Relaxed); - self.queued.fetch_add(queued, Ordering::Relaxed); - self.executed.fetch_add(executed, Ordering::Relaxed); - self.failed.fetch_add(failed, Ordering::Relaxed); - self.skipped.fetch_add(skipped, Ordering::Relaxed); + fn add(&self, work: ScannerSourceWorkUpdate) { + self.checked.fetch_add(work.checked, Ordering::Relaxed); + self.queued.fetch_add(work.queued, Ordering::Relaxed); + self.executed.fetch_add(work.executed, Ordering::Relaxed); + self.failed.fetch_add(work.failed, Ordering::Relaxed); + self.skipped.fetch_add(work.skipped, Ordering::Relaxed); + self.missed.fetch_add(work.missed, Ordering::Relaxed); } fn store(&self, values: ScannerSourceWorkValues) { @@ -333,6 +391,7 @@ impl ScannerSourceWorkCounters { self.executed.store(values.executed, Ordering::Relaxed); self.failed.store(values.failed, Ordering::Relaxed); self.skipped.store(values.skipped, Ordering::Relaxed); + self.missed.store(values.missed, Ordering::Relaxed); } fn values(&self) -> ScannerSourceWorkValues { @@ -342,6 +401,7 @@ impl ScannerSourceWorkCounters { executed: self.executed.load(Ordering::Relaxed), failed: self.failed.load(Ordering::Relaxed), skipped: self.skipped.load(Ordering::Relaxed), + missed: self.missed.load(Ordering::Relaxed), } } @@ -357,6 +417,7 @@ struct ScannerSourceWorkValues { executed: u64, failed: u64, skipped: u64, + missed: u64, } impl ScannerSourceWorkValues { @@ -367,6 +428,7 @@ impl ScannerSourceWorkValues { executed: self.executed.saturating_sub(start.executed), failed: self.failed.saturating_sub(start.failed), skipped: self.skipped.saturating_sub(start.skipped), + missed: self.missed.saturating_sub(start.missed), } } @@ -378,6 +440,7 @@ impl ScannerSourceWorkValues { executed: self.executed, failed: self.failed, skipped: self.skipped, + missed: self.missed, } } } @@ -675,6 +738,8 @@ pub struct ScannerSourceWorkSnapshot { pub executed: u64, pub failed: u64, pub skipped: u64, + #[serde(default)] + pub missed: u64, } #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] @@ -1170,7 +1235,15 @@ impl Metrics { pub fn record_scanner_ilm_action(&self, count: u64) { self.scanner_ilm_actions.fetch_add(count, Ordering::Relaxed); - self.record_scanner_source_work(ScannerWorkSource::Lifecycle, 0, 0, count, 0, 0); + self.record_scanner_source_executed(ScannerWorkSource::Lifecycle, count); + } + + pub fn record_scanner_ilm_enqueue_result(&self, count: u64, queued: bool) { + if queued { + self.record_scanner_source_queued(ScannerWorkSource::Lifecycle, count); + } else { + self.record_scanner_source_missed(ScannerWorkSource::Lifecycle, count); + } } pub fn record_scanner_checkpoint_set(&self, version: u16, resume_after: impl Into, reason: impl Into) { @@ -1209,20 +1282,28 @@ impl Metrics { self.update_scanner_checkpoint_event(SCANNER_CHECKPOINT_EVENT_STALE); } - pub fn record_scanner_source_work( - &self, - source: ScannerWorkSource, - checked: u64, - queued: u64, - executed: u64, - failed: u64, - skipped: u64, - ) { + pub fn record_scanner_source_work(&self, source: ScannerWorkSource, work: ScannerSourceWorkUpdate) { if let Some(counters) = self.scanner_source_work.get(source.index()) { - counters.add(checked, queued, executed, failed, skipped); + counters.add(work); } } + pub fn record_scanner_source_checked(&self, source: ScannerWorkSource, count: u64) { + self.record_scanner_source_work(source, ScannerSourceWorkUpdate::checked(count)); + } + + pub fn record_scanner_source_queued(&self, source: ScannerWorkSource, count: u64) { + self.record_scanner_source_work(source, ScannerSourceWorkUpdate::queued(count)); + } + + pub fn record_scanner_source_executed(&self, source: ScannerWorkSource, count: u64) { + self.record_scanner_source_work(source, ScannerSourceWorkUpdate::executed(count)); + } + + pub fn record_scanner_source_missed(&self, source: ScannerWorkSource, count: u64) { + self.record_scanner_source_work(source, ScannerSourceWorkUpdate::missed(count)); + } + fn update_scanner_checkpoint_event(&self, event: &str) { match self.scanner_checkpoint.lock() { Ok(mut current) => { @@ -1242,19 +1323,19 @@ impl Metrics { fn record_source_work_for_metric(&self, metric: Metric, count: u64) { match metric { Metric::ScanObject | Metric::ScanFolder => { - self.record_scanner_source_work(ScannerWorkSource::Usage, count, 0, 0, 0, 0); + self.record_scanner_source_checked(ScannerWorkSource::Usage, count); } Metric::SaveUsage => { - self.record_scanner_source_work(ScannerWorkSource::Usage, 0, 0, count, 0, 0); + self.record_scanner_source_executed(ScannerWorkSource::Usage, count); } Metric::CheckReplication => { - self.record_scanner_source_work(ScannerWorkSource::BucketReplication, count, 0, 0, 0, 0); + self.record_scanner_source_checked(ScannerWorkSource::BucketReplication, count); } Metric::HealCheck => { - self.record_scanner_source_work(ScannerWorkSource::Heal, count, 0, 0, 0, 0); + self.record_scanner_source_checked(ScannerWorkSource::Heal, count); } Metric::HealAbandonedObject => { - self.record_scanner_source_work(ScannerWorkSource::Heal, 0, 0, count, 0, 0); + self.record_scanner_source_executed(ScannerWorkSource::Heal, count); } _ => {} } @@ -2011,8 +2092,26 @@ mod tests { #[tokio::test] async fn report_includes_scanner_source_work() { let metrics = Metrics::new(); - metrics.record_scanner_source_work(ScannerWorkSource::Usage, 3, 0, 1, 0, 0); - metrics.record_scanner_source_work(ScannerWorkSource::Lifecycle, 0, 2, 1, 1, 0); + metrics.record_scanner_source_work( + ScannerWorkSource::Usage, + ScannerSourceWorkUpdate { + checked: 3, + executed: 1, + ..Default::default() + }, + ); + metrics.record_scanner_source_work( + ScannerWorkSource::Lifecycle, + ScannerSourceWorkUpdate { + queued: 2, + executed: 1, + failed: 1, + missed: 3, + ..Default::default() + }, + ); + metrics.record_scanner_ilm_enqueue_result(4, true); + metrics.record_scanner_ilm_enqueue_result(5, false); let report = metrics.report().await; @@ -2029,9 +2128,10 @@ mod tests { .iter() .find(|work| work.source == ScannerWorkSource::Lifecycle.as_str()) .expect("lifecycle source work should be visible"); - assert_eq!(lifecycle.queued, 2); + assert_eq!(lifecycle.queued, 6); assert_eq!(lifecycle.executed, 1); assert_eq!(lifecycle.failed, 1); + assert_eq!(lifecycle.missed, 8); } #[tokio::test] @@ -2073,11 +2173,33 @@ mod tests { #[tokio::test] async fn report_includes_scan_cycle_source_work() { let metrics = Metrics::new(); - metrics.record_scanner_source_work(ScannerWorkSource::Usage, 10, 0, 1, 0, 0); + metrics.record_scanner_source_work( + ScannerWorkSource::Usage, + ScannerSourceWorkUpdate { + checked: 10, + executed: 1, + ..Default::default() + }, + ); let start = metrics.start_scan_cycle_work(); - metrics.record_scanner_source_work(ScannerWorkSource::Usage, 3, 0, 2, 0, 0); - metrics.record_scanner_source_work(ScannerWorkSource::Lifecycle, 0, 1, 1, 0, 0); + metrics.record_scanner_source_work( + ScannerWorkSource::Usage, + ScannerSourceWorkUpdate { + checked: 3, + executed: 2, + ..Default::default() + }, + ); + metrics.record_scanner_source_work( + ScannerWorkSource::Lifecycle, + ScannerSourceWorkUpdate { + queued: 1, + executed: 1, + missed: 2, + ..Default::default() + }, + ); let report = metrics.report().await; let usage = report @@ -2095,6 +2217,7 @@ mod tests { .expect("current cycle lifecycle source work should be visible"); assert_eq!(lifecycle.queued, 1); assert_eq!(lifecycle.executed, 1); + assert_eq!(lifecycle.missed, 2); metrics.finish_scan_cycle_work(start); let report = metrics.report().await; @@ -2115,6 +2238,7 @@ mod tests { .expect("last cycle lifecycle source work should be visible"); assert_eq!(lifecycle.queued, 1); assert_eq!(lifecycle.executed, 1); + assert_eq!(lifecycle.missed, 2); } #[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 9f397f15b..8e7d44aeb 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -45,7 +45,7 @@ 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}; +use rustfs_common::metrics::{IlmAction, Metrics, 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, @@ -145,6 +145,12 @@ 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); + } +} + #[cfg(any(test, debug_assertions))] fn should_force_immediate_transition_enqueue_timeout() -> bool { env::var(rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT) @@ -357,7 +363,7 @@ impl ExpiryState { } } - pub async fn enqueue_by_days(&mut self, oi: &ObjectInfo, event: &lifecycle::Event, src: &LcEventSrc) { + pub async fn enqueue_by_days(&mut self, oi: &ObjectInfo, event: &lifecycle::Event, src: &LcEventSrc) -> bool { let task = ExpiryTask { obj_info: oi.clone(), event: event.clone(), @@ -366,22 +372,29 @@ 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; - return; + record_scanner_lifecycle_enqueue_result(src, 1, false); + return false; } 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_expiry_tasks.get_mut() += 1; - } + let queued = wrkr.send(Some(Box::new(task))).await.is_ok(); + if !queued { + *self.stats.as_mut().expect("stats lock").missed_expiry_tasks.get_mut() += 1; } + record_scanner_lifecycle_enqueue_result(src, 1, queued); + queued } - pub async fn enqueue_by_newer_noncurrent(&mut self, bucket: &str, versions: Vec, lc_event: lifecycle::Event) { + pub async fn enqueue_by_newer_noncurrent( + &mut self, + bucket: &str, + versions: Vec, + lc_event: lifecycle::Event, + src: &LcEventSrc, + ) -> bool { if versions.is_empty() { - return; + return true; } + let version_count = u64::try_from(versions.len()).unwrap_or(u64::MAX); let task = NewerNoncurrentTask { bucket: String::from(bucket), @@ -391,16 +404,16 @@ 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; - return; + record_scanner_lifecycle_enqueue_result(src, version_count, false); + return false; } 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_expiry_tasks.get_mut() += 1; - } + let queued = wrkr.send(Some(Box::new(task))).await.is_ok(); + if !queued { + *self.stats.as_mut().expect("stats lock").missed_expiry_tasks.get_mut() += 1; } + record_scanner_lifecycle_enqueue_result(src, version_count, queued); + queued } pub fn get_worker_ch(&self, h: u64) -> Option>> { @@ -739,10 +752,11 @@ impl TransitionState { } } - pub async fn queue_transition_task(self: &Arc, oi: &ObjectInfo, event: &lifecycle::Event, src: &LcEventSrc) { + pub async fn queue_transition_task(self: &Arc, oi: &ObjectInfo, event: &lifecycle::Event, src: &LcEventSrc) -> bool { if is_immediate_transition_source(src) && should_force_immediate_transition_enqueue_timeout() { self.handle_immediate_enqueue_failure(oi, src, ImmediateEnqueueFailure::ForcedTimeout); - return; + record_scanner_lifecycle_enqueue_result(src, 1, false); + return false; } let task = TransitionTask { @@ -750,14 +764,15 @@ impl TransitionState { src: src.clone(), event: event.clone(), }; + let mut queued = false; if is_immediate_transition_source(src) { match self.transition_tx.try_send(Some(task)) { - Ok(()) => {} + Ok(()) => queued = true, Err(async_channel::TrySendError::Full(task)) => { Self::inc_counter(&self.queue_full_tasks); let send_timeout = self.transition_queue_send_timeout; match tokio::time::timeout(send_timeout, self.transition_tx.send(task)).await { - Ok(Ok(())) => {} + Ok(Ok(())) => queued = true, Ok(Err(_)) => { self.handle_immediate_enqueue_failure( oi, @@ -782,29 +797,31 @@ impl TransitionState { self.handle_immediate_enqueue_failure(oi, src, ImmediateEnqueueFailure::QueueClosed { timeout_ms: None }); } } - return; + record_scanner_lifecycle_enqueue_result(src, 1, queued); + return queued; } - if let Err(err) = self.transition_tx.try_send(Some(task)) { - match err { - async_channel::TrySendError::Full(_) => { - debug!( - bucket = %oi.bucket, - object = %oi.name, - source = ?src, - "transition queue is full; deferring to scanner/backfill" - ); - } - async_channel::TrySendError::Closed(_) => { - warn!( - bucket = %oi.bucket, - object = %oi.name, - source = ?src, - "transition enqueue failed because the queue is closed" - ); - } + match self.transition_tx.try_send(Some(task)) { + Ok(()) => queued = true, + Err(async_channel::TrySendError::Full(_)) => { + debug!( + bucket = %oi.bucket, + object = %oi.name, + source = ?src, + "transition queue is full; deferring to scanner/backfill" + ); + } + Err(async_channel::TrySendError::Closed(_)) => { + warn!( + bucket = %oi.bucket, + object = %oi.name, + source = ?src, + "transition enqueue failed because the queue is closed" + ); } } + record_scanner_lifecycle_enqueue_result(src, 1, queued); + queued } pub async fn init(api: Arc) { @@ -1482,7 +1499,7 @@ pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) { GLOBAL_ExpiryState .write() .await - .enqueue_by_newer_noncurrent(&oi.bucket, to_delete_objs, event) + .enqueue_by_newer_noncurrent(&oi.bucket, to_delete_objs, event, &src) .await; } } @@ -2045,8 +2062,7 @@ pub async fn apply_transition_rule(event: &lifecycle::Event, src: &LcEventSrc, o if oi.delete_marker || oi.is_dir { return false; } - GLOBAL_TransitionState.queue_transition_task(oi, event, src).await; - true + GLOBAL_TransitionState.queue_transition_task(oi, event, src).await } pub async fn apply_expiry_on_transitioned_object( @@ -2135,8 +2151,7 @@ pub async fn apply_expiry_on_non_transitioned_objects( pub async fn apply_expiry_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool { let mut expiry_state = GLOBAL_ExpiryState.write().await; - expiry_state.enqueue_by_days(oi, event, src).await; - true + expiry_state.enqueue_by_days(oi, event, src).await } fn lifecycle_deleted_object(oi: &ObjectInfo, dobj: &ObjectInfo) -> crate::store_api::DeletedObject { @@ -2297,7 +2312,7 @@ pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, mod tests { use super::{ DATE_EXPIRY_EXISTING_OBJECTS_GRACE_SECS, DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX, - DEFAULT_TRANSITION_WORKERS_CAP, GLOBAL_TransitionState, StaleMultipartUploadCandidate, TransitionState, + DEFAULT_TRANSITION_WORKERS_CAP, ExpiryState, GLOBAL_TransitionState, StaleMultipartUploadCandidate, TransitionState, cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at, lifecycle_deleted_object, lifecycle_rule_has_date_expiration, lifecycle_version_purge_state_from_completed_targets, mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate, replication_state_for_delete, @@ -2305,6 +2320,7 @@ mod tests { resolve_transition_workers_absolute_max, should_defer_date_expiry_for_recent_config_update, should_reuse_lifecycle_delete_replication_state, }; + use crate::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc; use crate::bucket::metadata::BUCKET_LIFECYCLE_CONFIG; use crate::bucket::metadata_sys; use crate::disk::RUSTFS_META_MULTIPART_BUCKET; @@ -2317,6 +2333,7 @@ mod tests { BucketOperations, BucketOptions, MakeBucketOptions, MultipartOperations, ObjectInfo, ObjectOptions, PutObjReader, }; use futures::FutureExt; + use rustfs_common::metrics::IlmAction; use rustfs_config::ENV_TRANSITION_WORKERS_ABSOLUTE_MAX; use rustfs_filemeta::{ReplicateDecision, VersionPurgeStatusType}; use s3s::dto::{BucketLifecycleConfiguration, ExpirationStatus, LifecycleExpiration, LifecycleRule, Timestamp}; @@ -2333,6 +2350,48 @@ mod tests { use tokio_util::sync::CancellationToken; use uuid::Uuid; + #[tokio::test] + async fn expiry_enqueue_reports_missed_without_worker_channel() { + let state = ExpiryState::new(); + let mut state = state.write().await; + let object = ObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + ..Default::default() + }; + let event = crate::bucket::lifecycle::lifecycle::Event { + action: IlmAction::DeleteAction, + ..Default::default() + }; + + 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); + } + + #[tokio::test] + async fn scanner_transition_enqueue_reports_full_queue() { + let state = TransitionState::new_with_capacity(1); + let object = ObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + ..Default::default() + }; + let event = crate::bucket::lifecycle::lifecycle::Event { + action: IlmAction::TransitionAction, + ..Default::default() + }; + + let first = state.queue_transition_task(&object, &event, &LcEventSrc::Scanner).await; + let second = state.queue_transition_task(&object, &event, &LcEventSrc::Scanner).await; + + assert!(first); + assert!(!second); + assert_eq!(state.transition_rx.len(), 1); + } + #[test] fn mark_delete_opts_skip_decommissioned_on_remote_success_sets_flag_on_success() { let mut opts = ObjectOptions::default(); diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index c7ecc0932..d72c8eb88 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -147,6 +147,28 @@ fn should_yield_after_object(object_count: u64, yield_every: u64) -> bool { yield_every > 0 && object_count.is_multiple_of(yield_every) } +fn record_scanner_ilm_action_if_queued(metrics: &Metrics, count: u64, queued: bool) -> bool { + if queued { + metrics.record_scanner_ilm_action(count); + } + queued +} + +#[derive(Clone, Copy)] +struct PendingScannerAccounting<'a> { + object: &'a ObjectInfo, + retained_size: i64, + expired_size: i64, +} + +impl PendingScannerAccounting<'_> { + fn apply(self, size_summary: &mut SizeSummary, cumulative_size: &mut i64, queued: bool) { + let size = if queued { self.expired_size } else { self.retained_size }; + size_summary.actions_accounting(self.object, size, self.retained_size); + *cumulative_size += size; + } +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum FolderResumeMatch { Exact, @@ -559,6 +581,7 @@ impl ScannerItem { }; let mut to_delete_objs: Vec = Vec::new(); let mut noncurrent_events: Vec = Vec::new(); + let mut noncurrent_accounting: Vec> = Vec::new(); let mut cumulative_size = 0; let mut remaining_versions = object_infos.len(); 'eventLoop: { @@ -573,48 +596,77 @@ impl ScannerItem { }; let mut size = actual_size; + let mut account_now = true; - let done_ilm = Metrics::time_ilm(event.action); match event.action { IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction => { - remaining_versions = 0; debug!("apply_actions: applying expiry rule for object: {} {}", oi.name, event.action); - apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await; - done_ilm(1)(); - global_metrics().record_scanner_ilm_action(1); + let done_ilm = Metrics::time_ilm(event.action); + let queued = apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await; + if record_scanner_ilm_action_if_queued(global_metrics(), 1, queued) { + done_ilm(1)(); + remaining_versions = 0; + } else { + PendingScannerAccounting { + object: oi, + retained_size: actual_size, + expired_size: 0, + } + .apply(size_summary, &mut cumulative_size, false); + for retained in object_infos.iter().skip(i + 1) { + let retained_size = match retained.get_actual_size() { + Ok(size) => size, + Err(_) => { + warn!("apply_actions: Failed to get actual size for object {}", retained.name); + 0 + } + }; + PendingScannerAccounting { + object: retained, + retained_size, + expired_size: 0, + } + .apply(size_summary, &mut cumulative_size, false); + } + } break 'eventLoop; } IlmAction::DeleteAction | IlmAction::DeleteRestoredAction | IlmAction::DeleteRestoredVersionAction => { - if !versioning_config.prefix_enabled(&self.object_path()) && event.action == IlmAction::DeleteAction { - remaining_versions -= 1; - size = 0; - } - debug!("apply_actions: applying expiry rule for object: {} {}", oi.name, event.action); - apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await; - done_ilm(1)(); - global_metrics().record_scanner_ilm_action(1); + let done_ilm = Metrics::time_ilm(event.action); + let queued = apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await; + if record_scanner_ilm_action_if_queued(global_metrics(), 1, queued) { + done_ilm(1)(); + if !versioning_config.prefix_enabled(&self.object_path()) && event.action == IlmAction::DeleteAction { + remaining_versions -= 1; + size = 0; + } + } } IlmAction::DeleteVersionAction => { - remaining_versions -= 1; - size = 0; if let Some(opt) = object_opts.get(i) { to_delete_objs.push(ObjectToDelete { object_name: opt.name.clone(), version_id: opt.version_id, ..Default::default() }); + noncurrent_accounting.push(PendingScannerAccounting { + object: oi, + retained_size: actual_size, + expired_size: 0, + }); + account_now = false; } noncurrent_events.push(event.clone()); - done_ilm(1)(); - global_metrics().record_scanner_ilm_action(1); } IlmAction::TransitionAction | IlmAction::TransitionVersionAction => { debug!("apply_actions: applying transition rule for object: {} {}", oi.name, event.action); - apply_transition_rule(event, &LcEventSrc::Scanner, oi).await; - done_ilm(1)(); - global_metrics().record_scanner_ilm_action(1); + let done_ilm = Metrics::time_ilm(event.action); + let queued = apply_transition_rule(event, &LcEventSrc::Scanner, oi).await; + if record_scanner_ilm_action_if_queued(global_metrics(), 1, queued) { + done_ilm(1)(); + } } IlmAction::NoneAction | IlmAction::ActionCount => { @@ -622,20 +674,31 @@ impl ScannerItem { } } - size_summary.actions_accounting(oi, size, actual_size); - - cumulative_size += size; + if account_now { + size_summary.actions_accounting(oi, size, actual_size); + cumulative_size += size; + } } } if !to_delete_objs.is_empty() && let Some(event) = noncurrent_events.first().cloned() { - GLOBAL_ExpiryState + let action = event.action; + let count = u64::try_from(to_delete_objs.len()).unwrap_or(u64::MAX); + let done_ilm = Metrics::time_ilm(action); + let queued = GLOBAL_ExpiryState .write() .await - .enqueue_by_newer_noncurrent(&self.bucket, to_delete_objs, event) + .enqueue_by_newer_noncurrent(&self.bucket, to_delete_objs, event, &LcEventSrc::Scanner) .await; + if record_scanner_ilm_action_if_queued(global_metrics(), count, queued) { + done_ilm(count)(); + remaining_versions = remaining_versions.saturating_sub(noncurrent_accounting.len()); + } + for pending in noncurrent_accounting { + pending.apply(size_summary, &mut cumulative_size, queued); + } } self.alert_excessive_versions(remaining_versions, cumulative_size); } @@ -756,7 +819,7 @@ impl ScannerItem { ensure_scanner_alert_metrics_registered(); let (too_many_versions, too_large_versions) = should_alert_excessive_versions(remaining_versions, cumulative_size); if too_many_versions { - global_metrics().record_scanner_source_work(ScannerWorkSource::Alerts, 0, 0, 1, 0, 0); + global_metrics().record_scanner_source_executed(ScannerWorkSource::Alerts, 1); counter!( METRIC_SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL, "bucket" => self.bucket.clone() @@ -771,7 +834,7 @@ impl ScannerItem { ); } if too_large_versions { - global_metrics().record_scanner_source_work(ScannerWorkSource::Alerts, 0, 0, 1, 0, 0); + global_metrics().record_scanner_source_executed(ScannerWorkSource::Alerts, 1); counter!( METRIC_SCANNER_EXCESS_OBJECT_VERSION_SIZE_TOTAL, "bucket" => self.bucket.clone() @@ -911,7 +974,7 @@ impl FolderScanner { } ensure_scanner_alert_metrics_registered(); - global_metrics().record_scanner_source_work(ScannerWorkSource::Alerts, 0, 0, 1, 0, 0); + global_metrics().record_scanner_source_executed(ScannerWorkSource::Alerts, 1); counter!( METRIC_SCANNER_EXCESS_FOLDERS_TOTAL, "root" => self.root.clone() @@ -1987,6 +2050,57 @@ mod tests { assert!(!ScannerItem::should_account_replication_stats(&purge_version)); } + #[tokio::test] + async fn test_scanner_ilm_action_accounting_requires_enqueue_success() { + let metrics = Metrics::new(); + + record_scanner_ilm_action_if_queued(&metrics, 2, false); + let report = metrics.report().await; + let lifecycle = report + .source_work + .iter() + .find(|work| work.source == ScannerWorkSource::Lifecycle.as_str()) + .expect("lifecycle source work should be visible"); + assert_eq!(lifecycle.executed, 0); + + record_scanner_ilm_action_if_queued(&metrics, 3, true); + let report = metrics.report().await; + let lifecycle = report + .source_work + .iter() + .find(|work| work.source == ScannerWorkSource::Lifecycle.as_str()) + .expect("lifecycle source work should be visible"); + assert_eq!(lifecycle.executed, 3); + } + + #[test] + fn test_pending_scanner_accounting_requires_enqueue_success() { + let object = ObjectInfo { + size: 10, + version_id: Some(uuid::Uuid::new_v4()), + ..Default::default() + }; + let pending = PendingScannerAccounting { + object: &object, + retained_size: 10, + expired_size: 0, + }; + + let mut failed_summary = SizeSummary::default(); + let mut failed_cumulative_size = 0; + pending.apply(&mut failed_summary, &mut failed_cumulative_size, false); + assert_eq!(failed_summary.versions, 1); + assert_eq!(failed_summary.total_size, 10); + assert_eq!(failed_cumulative_size, 10); + + let mut queued_summary = SizeSummary::default(); + let mut queued_cumulative_size = 0; + pending.apply(&mut queued_summary, &mut queued_cumulative_size, true); + assert_eq!(queued_summary.versions, 0); + assert_eq!(queued_summary.total_size, 0); + assert_eq!(queued_cumulative_size, 0); + } + #[test] #[serial] fn test_excessive_version_alert_thresholds_use_env() {