diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index f5357f074..32c7c724d 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -600,6 +600,8 @@ pub struct Metrics { current_scan_cycle_throttle_sleep_events_start: AtomicU64, current_scan_cycle_throttle_sleep_duration_millis_start: AtomicU64, current_scan_cycle_ilm_actions_start: AtomicU64, + current_scan_cycle_lifecycle_expiry_actions_start: AtomicU64, + current_scan_cycle_lifecycle_transition_actions_start: AtomicU64, current_scan_cycle_heal_objects_start: AtomicU64, current_scan_cycle_replication_checks_start: AtomicU64, current_scan_cycle_usage_saves_start: AtomicU64, @@ -620,6 +622,8 @@ pub struct Metrics { last_scan_cycle_throttle_sleep_events: AtomicU64, last_scan_cycle_throttle_sleep_duration_millis: AtomicU64, last_scan_cycle_ilm_actions: AtomicU64, + last_scan_cycle_lifecycle_expiry_actions: AtomicU64, + last_scan_cycle_lifecycle_transition_actions: AtomicU64, last_scan_cycle_heal_objects: AtomicU64, last_scan_cycle_replication_checks: AtomicU64, last_scan_cycle_usage_saves: AtomicU64, @@ -632,6 +636,20 @@ pub struct Metrics { scanner_yield_duration_millis: AtomicU64, scanner_throttle_sleep_duration_millis: AtomicU64, scanner_ilm_actions: AtomicU64, + scanner_lifecycle_expiry_actions: AtomicU64, + scanner_lifecycle_transition_actions: AtomicU64, + scanner_transition_queue_capacity: AtomicU64, + scanner_transition_queued: AtomicU64, + scanner_transition_active: AtomicU64, + scanner_transition_workers: AtomicU64, + scanner_transition_queue_full: AtomicU64, + scanner_transition_queue_send_timeout: AtomicU64, + scanner_transition_compensation_scheduled: AtomicU64, + scanner_transition_compensation_running: AtomicU64, + scanner_transition_queued_total: AtomicU64, + scanner_transition_missed_total: AtomicU64, + scanner_transition_completed: AtomicU64, + scanner_transition_failed: AtomicU64, scanner_throttle_idle_mode_enabled: AtomicBool, scanner_throttle_sleep_factor_micros: AtomicU64, scanner_throttle_max_sleep_millis: AtomicU64, @@ -710,6 +728,8 @@ pub struct ScanCycleWorkSnapshot { throttle_sleep_events: u64, throttle_sleep_duration_millis: u64, ilm_actions: u64, + lifecycle_expiry_actions: u64, + lifecycle_transition_actions: u64, heal_objects: u64, replication_checks: u64, usage_saves: u64, @@ -775,6 +795,34 @@ impl Default for ScannerPacingPressureSnapshot { } } +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ScannerLifecycleTransitionStateUpdate { + pub queue_capacity: u64, + pub queued: u64, + pub active: u64, + pub workers: u64, + pub queue_full: u64, + pub queue_send_timeout: u64, + pub compensation_scheduled: u64, + pub compensation_running: u64, +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct ScannerLifecycleTransitionSnapshot { + pub current_queue_capacity: u64, + pub current_queued: u64, + pub current_active: u64, + pub current_workers: u64, + pub queue_full: u64, + pub queue_send_timeout: u64, + pub compensation_scheduled: u64, + pub compensation_running: u64, + pub scanner_queued: u64, + pub scanner_missed: u64, + pub completed: u64, + pub failed: u64, +} + #[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct ScannerLastMinute { pub actions: HashMap, @@ -828,6 +876,10 @@ pub struct ScannerMetricsReport { #[serde(default)] pub current_cycle_ilm_actions: u64, #[serde(default)] + pub current_cycle_lifecycle_expiry_actions: u64, + #[serde(default)] + pub current_cycle_lifecycle_transition_actions: u64, + #[serde(default)] pub current_cycle_heal_objects: u64, #[serde(default)] pub current_cycle_replication_checks: u64, @@ -863,6 +915,10 @@ pub struct ScannerMetricsReport { #[serde(default)] pub last_cycle_ilm_actions: u64, #[serde(default)] + pub last_cycle_lifecycle_expiry_actions: u64, + #[serde(default)] + pub last_cycle_lifecycle_transition_actions: u64, + #[serde(default)] pub last_cycle_heal_objects: u64, #[serde(default)] pub last_cycle_replication_checks: u64, @@ -882,6 +938,8 @@ pub struct ScannerMetricsReport { #[serde(default)] pub pacing_pressure: ScannerPacingPressureSnapshot, #[serde(default)] + pub lifecycle_transition: ScannerLifecycleTransitionSnapshot, + #[serde(default)] pub throttle_idle_mode_enabled: bool, #[serde(default)] pub throttle_sleep_factor: f64, @@ -1134,6 +1192,8 @@ impl Metrics { current_scan_cycle_throttle_sleep_events_start: AtomicU64::new(0), current_scan_cycle_throttle_sleep_duration_millis_start: AtomicU64::new(0), current_scan_cycle_ilm_actions_start: AtomicU64::new(0), + current_scan_cycle_lifecycle_expiry_actions_start: AtomicU64::new(0), + current_scan_cycle_lifecycle_transition_actions_start: AtomicU64::new(0), current_scan_cycle_heal_objects_start: AtomicU64::new(0), current_scan_cycle_replication_checks_start: AtomicU64::new(0), current_scan_cycle_usage_saves_start: AtomicU64::new(0), @@ -1154,6 +1214,8 @@ impl Metrics { last_scan_cycle_throttle_sleep_events: AtomicU64::new(0), last_scan_cycle_throttle_sleep_duration_millis: AtomicU64::new(0), last_scan_cycle_ilm_actions: AtomicU64::new(0), + last_scan_cycle_lifecycle_expiry_actions: AtomicU64::new(0), + last_scan_cycle_lifecycle_transition_actions: AtomicU64::new(0), last_scan_cycle_heal_objects: AtomicU64::new(0), last_scan_cycle_replication_checks: AtomicU64::new(0), last_scan_cycle_usage_saves: AtomicU64::new(0), @@ -1166,6 +1228,20 @@ impl Metrics { scanner_yield_duration_millis: AtomicU64::new(0), scanner_throttle_sleep_duration_millis: AtomicU64::new(0), scanner_ilm_actions: AtomicU64::new(0), + scanner_lifecycle_expiry_actions: AtomicU64::new(0), + scanner_lifecycle_transition_actions: AtomicU64::new(0), + scanner_transition_queue_capacity: AtomicU64::new(0), + scanner_transition_queued: AtomicU64::new(0), + scanner_transition_active: AtomicU64::new(0), + scanner_transition_workers: AtomicU64::new(0), + 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_running: AtomicU64::new(0), + scanner_transition_queued_total: AtomicU64::new(0), + scanner_transition_missed_total: AtomicU64::new(0), + scanner_transition_completed: AtomicU64::new(0), + scanner_transition_failed: AtomicU64::new(0), scanner_throttle_idle_mode_enabled: AtomicBool::new(false), scanner_throttle_sleep_factor_micros: AtomicU64::new(0), scanner_throttle_max_sleep_millis: AtomicU64::new(0), @@ -1337,6 +1413,19 @@ impl Metrics { self.record_scanner_source_executed(ScannerWorkSource::Lifecycle, count); } + pub fn record_scanner_lifecycle_action(&self, action: IlmAction, count: u64) { + self.record_scanner_ilm_action(count); + match action { + IlmAction::TransitionAction | IlmAction::TransitionVersionAction => { + self.scanner_lifecycle_transition_actions.fetch_add(count, Ordering::Relaxed); + } + _ if action.delete() => { + self.scanner_lifecycle_expiry_actions.fetch_add(count, Ordering::Relaxed); + } + _ => {} + } + } + pub fn record_scanner_ilm_enqueue_result(&self, count: u64, queued: bool) { if queued { self.record_scanner_source_queued(ScannerWorkSource::Lifecycle, count); @@ -1345,6 +1434,38 @@ impl Metrics { } } + pub fn record_scanner_transition_enqueue_result(&self, count: u64, queued: bool) { + self.record_scanner_ilm_enqueue_result(count, queued); + if queued { + self.scanner_transition_queued_total.fetch_add(count, Ordering::Relaxed); + } else { + self.scanner_transition_missed_total.fetch_add(count, Ordering::Relaxed); + } + } + + pub fn record_scanner_lifecycle_transition_state(&self, state: ScannerLifecycleTransitionStateUpdate) { + self.scanner_transition_queue_capacity + .store(state.queue_capacity, Ordering::Relaxed); + self.scanner_transition_queued.store(state.queued, Ordering::Relaxed); + self.scanner_transition_active.store(state.active, Ordering::Relaxed); + self.scanner_transition_workers.store(state.workers, Ordering::Relaxed); + self.scanner_transition_queue_full.store(state.queue_full, Ordering::Relaxed); + self.scanner_transition_queue_send_timeout + .store(state.queue_send_timeout, Ordering::Relaxed); + self.scanner_transition_compensation_scheduled + .store(state.compensation_scheduled, Ordering::Relaxed); + self.scanner_transition_compensation_running + .store(state.compensation_running, Ordering::Relaxed); + } + + pub fn record_scanner_transition_completed(&self, count: u64) { + self.scanner_transition_completed.fetch_add(count, Ordering::Relaxed); + } + + pub fn record_scanner_transition_failed(&self, count: u64) { + self.scanner_transition_failed.fetch_add(count, Ordering::Relaxed); + } + pub fn record_scanner_checkpoint_set(&self, version: u16, resume_after: impl Into, reason: impl Into) { let checkpoint = ScannerCheckpointReport { version, @@ -1643,6 +1764,10 @@ impl Metrics { .store(snapshot.throttle_sleep_duration_millis, Ordering::Relaxed); self.current_scan_cycle_ilm_actions_start .store(snapshot.ilm_actions, Ordering::Relaxed); + self.current_scan_cycle_lifecycle_expiry_actions_start + .store(snapshot.lifecycle_expiry_actions, Ordering::Relaxed); + self.current_scan_cycle_lifecycle_transition_actions_start + .store(snapshot.lifecycle_transition_actions, Ordering::Relaxed); self.current_scan_cycle_heal_objects_start .store(snapshot.heal_objects, Ordering::Relaxed); self.current_scan_cycle_replication_checks_start @@ -1673,6 +1798,8 @@ impl Metrics { throttle_sleep_events: self.lifetime(Metric::ThrottleSleep), throttle_sleep_duration_millis: self.scanner_throttle_sleep_duration_millis.load(Ordering::Relaxed), ilm_actions: self.scanner_ilm_actions.load(Ordering::Relaxed), + lifecycle_expiry_actions: self.scanner_lifecycle_expiry_actions.load(Ordering::Relaxed), + lifecycle_transition_actions: self.scanner_lifecycle_transition_actions.load(Ordering::Relaxed), heal_objects: self.lifetime(Metric::HealAbandonedObject), replication_checks: self.lifetime(Metric::CheckReplication), usage_saves: self.lifetime(Metric::SaveUsage), @@ -1692,6 +1819,10 @@ impl Metrics { .current_scan_cycle_throttle_sleep_duration_millis_start .load(Ordering::Relaxed), ilm_actions: self.current_scan_cycle_ilm_actions_start.load(Ordering::Relaxed), + lifecycle_expiry_actions: self.current_scan_cycle_lifecycle_expiry_actions_start.load(Ordering::Relaxed), + lifecycle_transition_actions: self + .current_scan_cycle_lifecycle_transition_actions_start + .load(Ordering::Relaxed), heal_objects: self.current_scan_cycle_heal_objects_start.load(Ordering::Relaxed), replication_checks: self.current_scan_cycle_replication_checks_start.load(Ordering::Relaxed), usage_saves: self.current_scan_cycle_usage_saves_start.load(Ordering::Relaxed), @@ -1712,6 +1843,12 @@ impl Metrics { .throttle_sleep_duration_millis .saturating_sub(start.throttle_sleep_duration_millis), ilm_actions: current.ilm_actions.saturating_sub(start.ilm_actions), + lifecycle_expiry_actions: current + .lifecycle_expiry_actions + .saturating_sub(start.lifecycle_expiry_actions), + lifecycle_transition_actions: current + .lifecycle_transition_actions + .saturating_sub(start.lifecycle_transition_actions), heal_objects: current.heal_objects.saturating_sub(start.heal_objects), replication_checks: current.replication_checks.saturating_sub(start.replication_checks), usage_saves: current.usage_saves.saturating_sub(start.usage_saves), @@ -1789,6 +1926,10 @@ impl Metrics { self.last_scan_cycle_throttle_sleep_duration_millis .store(work.throttle_sleep_duration_millis, Ordering::Relaxed); self.last_scan_cycle_ilm_actions.store(work.ilm_actions, Ordering::Relaxed); + self.last_scan_cycle_lifecycle_expiry_actions + .store(work.lifecycle_expiry_actions, Ordering::Relaxed); + self.last_scan_cycle_lifecycle_transition_actions + .store(work.lifecycle_transition_actions, Ordering::Relaxed); self.last_scan_cycle_heal_objects.store(work.heal_objects, Ordering::Relaxed); self.last_scan_cycle_replication_checks .store(work.replication_checks, Ordering::Relaxed); @@ -1880,6 +2021,8 @@ impl Metrics { m.current_cycle_throttle_sleep_events = current_work.throttle_sleep_events; m.current_cycle_throttle_sleep_duration_seconds = current_work.throttle_sleep_duration_millis as f64 / 1000.0; m.current_cycle_ilm_actions = current_work.ilm_actions; + m.current_cycle_lifecycle_expiry_actions = current_work.lifecycle_expiry_actions; + m.current_cycle_lifecycle_transition_actions = current_work.lifecycle_transition_actions; m.current_cycle_heal_objects = current_work.heal_objects; m.current_cycle_replication_checks = current_work.replication_checks; m.current_cycle_usage_saves = current_work.usage_saves; @@ -1908,6 +2051,8 @@ impl Metrics { m.last_cycle_throttle_sleep_duration_seconds = self.last_scan_cycle_throttle_sleep_duration_millis.load(Ordering::Relaxed) as f64 / 1000.0; m.last_cycle_ilm_actions = self.last_scan_cycle_ilm_actions.load(Ordering::Relaxed); + m.last_cycle_lifecycle_expiry_actions = self.last_scan_cycle_lifecycle_expiry_actions.load(Ordering::Relaxed); + m.last_cycle_lifecycle_transition_actions = self.last_scan_cycle_lifecycle_transition_actions.load(Ordering::Relaxed); m.last_cycle_heal_objects = self.last_scan_cycle_heal_objects.load(Ordering::Relaxed); m.last_cycle_replication_checks = self.last_scan_cycle_replication_checks.load(Ordering::Relaxed); m.last_cycle_usage_saves = self.last_scan_cycle_usage_saves.load(Ordering::Relaxed); @@ -1928,6 +2073,20 @@ impl Metrics { }) }) .collect(); + m.lifecycle_transition = ScannerLifecycleTransitionSnapshot { + current_queue_capacity: self.scanner_transition_queue_capacity.load(Ordering::Relaxed), + current_queued: self.scanner_transition_queued.load(Ordering::Relaxed), + current_active: self.scanner_transition_active.load(Ordering::Relaxed), + current_workers: self.scanner_transition_workers.load(Ordering::Relaxed), + 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_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), + completed: self.scanner_transition_completed.load(Ordering::Relaxed), + failed: self.scanner_transition_failed.load(Ordering::Relaxed), + }; m.throttle_idle_mode_enabled = self.scanner_throttle_idle_mode_enabled.load(Ordering::Relaxed); m.throttle_sleep_factor = self.scanner_throttle_sleep_factor_micros.load(Ordering::Relaxed) as f64 / 1_000_000.0; m.throttle_max_sleep_seconds = self.scanner_throttle_max_sleep_millis.load(Ordering::Relaxed) as f64 / 1000.0; @@ -2262,6 +2421,64 @@ mod tests { assert_eq!(lifecycle.missed, 8); } + #[tokio::test] + async fn report_splits_scanner_lifecycle_actions_by_expiry_and_transition() { + let metrics = Metrics::new(); + let start = metrics.start_scan_cycle_work(); + + metrics.record_scanner_lifecycle_action(IlmAction::DeleteAction, 2); + metrics.record_scanner_lifecycle_action(IlmAction::DeleteRestoredVersionAction, 3); + metrics.record_scanner_lifecycle_action(IlmAction::TransitionAction, 5); + metrics.record_scanner_lifecycle_action(IlmAction::TransitionVersionAction, 7); + + let report = metrics.report().await; + + assert_eq!(report.current_cycle_ilm_actions, 17); + assert_eq!(report.current_cycle_lifecycle_expiry_actions, 5); + assert_eq!(report.current_cycle_lifecycle_transition_actions, 12); + + metrics.finish_scan_cycle_work(start); + let report = metrics.report().await; + + assert_eq!(report.last_cycle_ilm_actions, 17); + assert_eq!(report.last_cycle_lifecycle_expiry_actions, 5); + assert_eq!(report.last_cycle_lifecycle_transition_actions, 12); + } + + #[tokio::test] + async fn report_includes_scanner_lifecycle_transition_status() { + let metrics = Metrics::new(); + metrics.record_scanner_lifecycle_transition_state(ScannerLifecycleTransitionStateUpdate { + queue_capacity: 16, + queued: 5, + active: 2, + workers: 4, + queue_full: 3, + queue_send_timeout: 1, + compensation_scheduled: 2, + compensation_running: 1, + }); + metrics.record_scanner_transition_enqueue_result(6, true); + metrics.record_scanner_transition_enqueue_result(2, false); + metrics.record_scanner_transition_completed(4); + metrics.record_scanner_transition_failed(1); + + let report = metrics.report().await; + + assert_eq!(report.lifecycle_transition.current_queue_capacity, 16); + assert_eq!(report.lifecycle_transition.current_queued, 5); + assert_eq!(report.lifecycle_transition.current_active, 2); + assert_eq!(report.lifecycle_transition.current_workers, 4); + 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_running, 1); + assert_eq!(report.lifecycle_transition.scanner_queued, 6); + assert_eq!(report.lifecycle_transition.scanner_missed, 2); + assert_eq!(report.lifecycle_transition.completed, 4); + assert_eq!(report.lifecycle_transition.failed, 1); + } + #[tokio::test] async fn scanner_metrics_update_source_work() { let metrics = Metrics::new(); @@ -2485,6 +2702,7 @@ mod tests { heal_objects: 2, replication_checks: 4, usage_saves: 6, + ..Default::default() }); let report = metrics.report().await; diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 8e7d44aeb..f8f1becde 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, global_metrics}; +use rustfs_common::metrics::{IlmAction, Metrics, 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, @@ -151,6 +151,20 @@ fn record_scanner_lifecycle_enqueue_result(src: &LcEventSrc, count: u64, queued: } } +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); + } +} + +fn nonnegative_i64_to_u64(value: i64) -> u64 { + u64::try_from(value).unwrap_or_default() +} + +fn usize_to_u64_saturated(value: usize) -> u64 { + u64::try_from(value).unwrap_or(u64::MAX) +} + #[cfg(any(test, debug_assertions))] fn should_force_immediate_transition_enqueue_timeout() -> bool { env::var(rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT) @@ -664,14 +678,17 @@ impl TransitionState { return false; } Self::inc_counter(&self.compensation_scheduled_tasks); + self.record_scanner_transition_state(); let bucket = bucket.to_string(); let scheduled = Arc::clone(&self.compensation_buckets); let state = Arc::clone(self); tokio::spawn(async move { Self::inc_counter(&state.compensation_running_tasks); + state.record_scanner_transition_state(); let Some(api) = crate::new_object_layer_fn() else { scheduled.lock().unwrap().remove(&bucket); Self::add_counter(&state.compensation_running_tasks, -1); + state.record_scanner_transition_state(); warn!(bucket = %bucket, "transition compensation skipped because object layer is unavailable"); return; }; @@ -684,6 +701,7 @@ impl TransitionState { scheduled.lock().unwrap().remove(&bucket); Self::add_counter(&state.compensation_running_tasks, -1); + state.record_scanner_transition_state(); }); true } @@ -703,6 +721,23 @@ impl TransitionState { counter.load(Ordering::Relaxed) } + fn scanner_transition_state_update(&self) -> ScannerLifecycleTransitionStateUpdate { + ScannerLifecycleTransitionStateUpdate { + queue_capacity: usize_to_u64_saturated(self.transition_queue_capacity), + queued: usize_to_u64_saturated(self.transition_rx.len()), + active: nonnegative_i64_to_u64(Self::counter_value(&self.active_tasks)), + workers: nonnegative_i64_to_u64(Self::counter_value(&self.num_workers)), + 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_running: nonnegative_i64_to_u64(Self::counter_value(&self.compensation_running_tasks)), + } + } + + fn record_scanner_transition_state(&self) { + global_metrics().record_scanner_lifecycle_transition_state(self.scanner_transition_state_update()); + } + fn handle_immediate_enqueue_failure(self: &Arc, oi: &ObjectInfo, src: &LcEventSrc, failure: ImmediateEnqueueFailure) { Self::inc_counter(&self.missed_immediate_tasks); let scheduled = self.schedule_bucket_compensation(&oi.bucket); @@ -755,7 +790,8 @@ impl TransitionState { 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); - record_scanner_lifecycle_enqueue_result(src, 1, false); + record_scanner_transition_enqueue_result(src, 1, false); + self.record_scanner_transition_state(); return false; } @@ -797,13 +833,15 @@ impl TransitionState { self.handle_immediate_enqueue_failure(oi, src, ImmediateEnqueueFailure::QueueClosed { timeout_ms: None }); } } - record_scanner_lifecycle_enqueue_result(src, 1, queued); + record_scanner_transition_enqueue_result(src, 1, queued); + self.record_scanner_transition_state(); return queued; } match self.transition_tx.try_send(Some(task)) { Ok(()) => queued = true, Err(async_channel::TrySendError::Full(_)) => { + Self::inc_counter(&self.queue_full_tasks); debug!( bucket = %oi.bucket, object = %oi.name, @@ -820,7 +858,8 @@ impl TransitionState { ); } } - record_scanner_lifecycle_enqueue_result(src, 1, queued); + record_scanner_transition_enqueue_result(src, 1, queued); + self.record_scanner_transition_state(); queued } @@ -893,6 +932,7 @@ impl TransitionState { let task = task.as_any().downcast_ref::().expect("TransitionTask downcast failed"); TransitionState::inc_counter(&GLOBAL_TransitionState.active_tasks); + GLOBAL_TransitionState.record_scanner_transition_state(); let obj_info_for_event = ObjectInfo { bucket: task.obj_info.bucket.clone(), @@ -903,6 +943,7 @@ impl TransitionState { }; if let Err(err) = transition_object(api.clone(), &task.obj_info, LcAuditEvent::new(task.event.clone(), task.src.clone())).await { + global_metrics().record_scanner_transition_failed(1); if !is_err_version_not_found(&err) && !is_err_object_not_found(&err) && !is_network_or_host_down(&err.to_string(), false) && !err.to_string().contains("use of closed network connection") { error!("Transition to {} failed for {}/{} version:{} with {}", task.event.storage_class, task.obj_info.bucket, task.obj_info.name, task.obj_info.version_id.map(|v| v.to_string()).unwrap_or_default(), err.to_string()); @@ -917,6 +958,7 @@ impl TransitionState { ..Default::default() }); } else { + global_metrics().record_scanner_transition_completed(1); let mut ts = TierStats { total_size: task.obj_info.size as u64, num_versions: 1, @@ -938,6 +980,7 @@ impl TransitionState { }); } TransitionState::add_counter(&GLOBAL_TransitionState.active_tasks, -1); + GLOBAL_TransitionState.record_scanner_transition_state(); } } else => () @@ -1006,6 +1049,7 @@ impl TransitionState { let current_workers = workers.len() as i64; GLOBAL_TransitionState.num_workers.store(current_workers, Ordering::SeqCst); + GLOBAL_TransitionState.record_scanner_transition_state(); info!( requested_transition_workers = requested, @@ -2333,7 +2377,7 @@ mod tests { BucketOperations, BucketOptions, MakeBucketOptions, MultipartOperations, ObjectInfo, ObjectOptions, PutObjReader, }; use futures::FutureExt; - use rustfs_common::metrics::IlmAction; + use rustfs_common::metrics::{IlmAction, global_metrics}; use rustfs_config::ENV_TRANSITION_WORKERS_ABSOLUTE_MAX; use rustfs_filemeta::{ReplicateDecision, VersionPurgeStatusType}; use s3s::dto::{BucketLifecycleConfiguration, ExpirationStatus, LifecycleExpiration, LifecycleRule, Timestamp}; @@ -2372,6 +2416,7 @@ mod tests { } #[tokio::test] + #[serial] async fn scanner_transition_enqueue_reports_full_queue() { let state = TransitionState::new_with_capacity(1); let object = ObjectInfo { @@ -2392,6 +2437,35 @@ mod tests { assert_eq!(state.transition_rx.len(), 1); } + #[tokio::test] + #[serial] + async fn scanner_transition_enqueue_updates_transition_status() { + let before = global_metrics().report().await.lifecycle_transition; + 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); + let after = global_metrics().report().await.lifecycle_transition; + assert_eq!(after.scanner_queued.saturating_sub(before.scanner_queued), 1); + assert_eq!(after.scanner_missed.saturating_sub(before.scanner_missed), 1); + assert_eq!(after.queue_full, 1); + assert_eq!(after.current_queue_capacity, 1); + assert_eq!(after.current_queued, 1); + assert_eq!(after.current_active, 0); + } + #[test] fn mark_delete_opts_skip_decommissioned_on_remote_success_sets_flag_on_success() { let mut opts = ObjectOptions::default(); diff --git a/crates/ecstore/src/metrics_realtime.rs b/crates/ecstore/src/metrics_realtime.rs index dd5c8a6d1..d2e6b7313 100644 --- a/crates/ecstore/src/metrics_realtime.rs +++ b/crates/ecstore/src/metrics_realtime.rs @@ -18,7 +18,8 @@ use rustfs_common::{GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_ADDR, heal_channel::Dr use rustfs_io_metrics::internode_metrics::global_internode_metrics; use rustfs_madmin::metrics::{ DiskIOStats, DiskMetric, LastMinute as MadminLastMinute, NetDevLine, NetMetrics, RPCMetrics, RealtimeMetrics, - ScannerMetrics as MadminScannerMetrics, ScannerPacingPressureSnapshot as MadminScannerPacingPressureSnapshot, + ScannerLifecycleTransitionSnapshot as MadminScannerLifecycleTransitionSnapshot, ScannerMetrics as MadminScannerMetrics, + ScannerPacingPressureSnapshot as MadminScannerPacingPressureSnapshot, ScannerSourceCycleSnapshot as MadminScannerSourceCycleSnapshot, TimedAction as MadminTimedAction, }; use rustfs_utils::os::get_drive_stats; @@ -100,6 +101,10 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo .collect(), }, active_paths: metrics.active_paths, + current_cycle_lifecycle_expiry_actions: metrics.current_cycle_lifecycle_expiry_actions, + current_cycle_lifecycle_transition_actions: metrics.current_cycle_lifecycle_transition_actions, + last_cycle_lifecycle_expiry_actions: metrics.last_cycle_lifecycle_expiry_actions, + last_cycle_lifecycle_transition_actions: metrics.last_cycle_lifecycle_transition_actions, last_cycle_partial_source: metrics.last_cycle_partial_source, last_cycle_partial_source_code: metrics.last_cycle_partial_source_code, pacing_pressure: MadminScannerPacingPressureSnapshot { @@ -112,6 +117,20 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo last_cycle_yield_ratio: metrics.pacing_pressure.last_cycle_yield_ratio, last_cycle_total_pause_ratio: metrics.pacing_pressure.last_cycle_total_pause_ratio, }, + lifecycle_transition: MadminScannerLifecycleTransitionSnapshot { + current_queue_capacity: metrics.lifecycle_transition.current_queue_capacity, + current_queued: metrics.lifecycle_transition.current_queued, + current_active: metrics.lifecycle_transition.current_active, + current_workers: metrics.lifecycle_transition.current_workers, + queue_full: metrics.lifecycle_transition.queue_full, + queue_send_timeout: metrics.lifecycle_transition.queue_send_timeout, + compensation_scheduled: metrics.lifecycle_transition.compensation_scheduled, + 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, + }, partial_cycles_by_source: metrics .partial_cycles_by_source .into_iter() @@ -412,4 +431,46 @@ mod test { assert_eq!(scanner.pacing_pressure.last_cycle_yield_ratio, 0.05); assert_eq!(scanner.pacing_pressure.last_cycle_total_pause_ratio, 0.3); } + + #[test] + fn scanner_metrics_mapping_preserves_lifecycle_transition_status() { + let scanner = to_madmin_scanner_metrics(rustfs_common::metrics::ScannerMetricsReport { + current_cycle_lifecycle_expiry_actions: 2, + current_cycle_lifecycle_transition_actions: 3, + last_cycle_lifecycle_expiry_actions: 5, + last_cycle_lifecycle_transition_actions: 7, + lifecycle_transition: rustfs_common::metrics::ScannerLifecycleTransitionSnapshot { + current_queue_capacity: 16, + current_queued: 5, + current_active: 2, + current_workers: 4, + queue_full: 3, + queue_send_timeout: 1, + compensation_scheduled: 2, + compensation_running: 1, + scanner_queued: 6, + scanner_missed: 2, + completed: 4, + failed: 1, + }, + ..Default::default() + }); + + 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); + assert_eq!(scanner.lifecycle_transition.current_workers, 4); + 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_running, 1); + assert_eq!(scanner.lifecycle_transition.scanner_queued, 6); + assert_eq!(scanner.lifecycle_transition.scanner_missed, 2); + assert_eq!(scanner.lifecycle_transition.completed, 4); + assert_eq!(scanner.lifecycle_transition.failed, 1); + assert_eq!(scanner.current_cycle_lifecycle_expiry_actions, 2); + assert_eq!(scanner.current_cycle_lifecycle_transition_actions, 3); + assert_eq!(scanner.last_cycle_lifecycle_expiry_actions, 5); + assert_eq!(scanner.last_cycle_lifecycle_transition_actions, 7); + } } diff --git a/crates/madmin/src/metrics.rs b/crates/madmin/src/metrics.rs index 41be04e31..de329e357 100644 --- a/crates/madmin/src/metrics.rs +++ b/crates/madmin/src/metrics.rs @@ -203,6 +203,51 @@ impl ScannerPacingPressureSnapshot { } } +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct ScannerLifecycleTransitionSnapshot { + #[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_full", default)] + pub queue_full: u64, + #[serde(rename = "queue_send_timeout", default)] + pub queue_send_timeout: u64, + #[serde(rename = "compensation_scheduled", default)] + pub compensation_scheduled: u64, + #[serde(rename = "compensation_running", default)] + pub compensation_running: u64, + #[serde(rename = "scanner_queued", default)] + pub scanner_queued: u64, + #[serde(rename = "scanner_missed", default)] + pub scanner_missed: u64, + #[serde(rename = "completed", default)] + pub completed: u64, + #[serde(rename = "failed", default)] + pub failed: u64, +} + +impl ScannerLifecycleTransitionSnapshot { + 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_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_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); + self.completed = self.completed.saturating_add(other.completed); + self.failed = self.failed.saturating_add(other.failed); + } +} + #[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct ScannerMetrics { #[serde(rename = "collected")] @@ -223,6 +268,14 @@ pub struct ScannerMetrics { pub last_minute: LastMinute, #[serde(rename = "active")] pub active_paths: Vec, + #[serde(rename = "current_cycle_lifecycle_expiry_actions", default)] + pub current_cycle_lifecycle_expiry_actions: u64, + #[serde(rename = "current_cycle_lifecycle_transition_actions", default)] + pub current_cycle_lifecycle_transition_actions: u64, + #[serde(rename = "last_cycle_lifecycle_expiry_actions", default)] + pub last_cycle_lifecycle_expiry_actions: u64, + #[serde(rename = "last_cycle_lifecycle_transition_actions", default)] + pub last_cycle_lifecycle_transition_actions: u64, #[serde(rename = "last_cycle_partial_source", default)] pub last_cycle_partial_source: String, #[serde(rename = "last_cycle_partial_source_code", default)] @@ -231,6 +284,8 @@ pub struct ScannerMetrics { pub partial_cycles_by_source: Vec, #[serde(rename = "pacing_pressure", default)] pub pacing_pressure: ScannerPacingPressureSnapshot, + #[serde(rename = "lifecycle_transition", default)] + pub lifecycle_transition: ScannerLifecycleTransitionSnapshot, } impl ScannerMetrics { @@ -243,6 +298,19 @@ impl ScannerMetrics { } self.pacing_pressure.merge(&other.pacing_pressure); + self.lifecycle_transition.merge(&other.lifecycle_transition); + self.current_cycle_lifecycle_expiry_actions = self + .current_cycle_lifecycle_expiry_actions + .saturating_add(other.current_cycle_lifecycle_expiry_actions); + self.current_cycle_lifecycle_transition_actions = self + .current_cycle_lifecycle_transition_actions + .saturating_add(other.current_cycle_lifecycle_transition_actions); + self.last_cycle_lifecycle_expiry_actions = self + .last_cycle_lifecycle_expiry_actions + .saturating_add(other.last_cycle_lifecycle_expiry_actions); + self.last_cycle_lifecycle_transition_actions = self + .last_cycle_lifecycle_transition_actions + .saturating_add(other.last_cycle_lifecycle_transition_actions); if self.ongoing_buckets < other.ongoing_buckets { self.ongoing_buckets = other.ongoing_buckets; @@ -893,4 +961,71 @@ mod tests { assert_eq!(pacing_pressure.primary_pressure, "none"); assert_eq!(pacing_pressure.current_active_scans, 1); } + + #[test] + fn scanner_metrics_merge_aggregates_lifecycle_transition_status() { + let collected_at = Utc::now(); + let mut scanner = ScannerMetrics { + collected_at, + current_cycle_lifecycle_expiry_actions: 2, + current_cycle_lifecycle_transition_actions: 3, + last_cycle_lifecycle_expiry_actions: 5, + last_cycle_lifecycle_transition_actions: 7, + lifecycle_transition: ScannerLifecycleTransitionSnapshot { + current_queue_capacity: 8, + current_queued: 2, + current_active: 1, + current_workers: 2, + queue_full: 3, + queue_send_timeout: 1, + compensation_scheduled: 1, + compensation_running: 1, + scanner_queued: 5, + scanner_missed: 2, + completed: 7, + failed: 1, + }, + ..Default::default() + }; + + scanner.merge(&ScannerMetrics { + collected_at: collected_at + chrono::Duration::seconds(1), + current_cycle_lifecycle_expiry_actions: 11, + current_cycle_lifecycle_transition_actions: 13, + last_cycle_lifecycle_expiry_actions: 17, + last_cycle_lifecycle_transition_actions: 19, + lifecycle_transition: ScannerLifecycleTransitionSnapshot { + current_queue_capacity: 4, + current_queued: 3, + current_active: 2, + current_workers: 1, + queue_full: 2, + queue_send_timeout: 3, + compensation_scheduled: 4, + compensation_running: 0, + scanner_queued: 6, + scanner_missed: 4, + completed: 8, + failed: 2, + }, + ..Default::default() + }); + + 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); + assert_eq!(scanner.lifecycle_transition.current_workers, 3); + 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_running, 1); + assert_eq!(scanner.lifecycle_transition.scanner_queued, 11); + assert_eq!(scanner.lifecycle_transition.scanner_missed, 6); + assert_eq!(scanner.lifecycle_transition.completed, 15); + assert_eq!(scanner.lifecycle_transition.failed, 3); + assert_eq!(scanner.current_cycle_lifecycle_expiry_actions, 13); + assert_eq!(scanner.current_cycle_lifecycle_transition_actions, 16); + assert_eq!(scanner.last_cycle_lifecycle_expiry_actions, 22); + assert_eq!(scanner.last_cycle_lifecycle_transition_actions, 26); + } } diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 0a6f41741..7dd5ba703 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -188,9 +188,9 @@ fn should_log_failed_object(failed_objects: usize) -> bool { failed_objects <= SCANNER_FAILED_OBJECT_LOG_INITIAL_LIMIT || failed_objects.is_multiple_of(SCANNER_FAILED_OBJECT_LOG_EVERY) } -fn record_scanner_ilm_action_if_queued(metrics: &Metrics, count: u64, queued: bool) -> bool { +fn record_scanner_ilm_action_if_queued(metrics: &Metrics, action: IlmAction, count: u64, queued: bool) -> bool { if queued { - metrics.record_scanner_ilm_action(count); + metrics.record_scanner_lifecycle_action(action, count); } queued } @@ -679,7 +679,7 @@ impl ScannerItem { debug!("apply_actions: applying expiry rule for object: {} {}", oi.name, event.action); 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) { + if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) { done_ilm(1)(); remaining_versions = 0; } else { @@ -712,7 +712,7 @@ impl ScannerItem { debug!("apply_actions: applying expiry rule for object: {} {}", oi.name, event.action); 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) { + if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) { done_ilm(1)(); if !versioning_config.prefix_enabled(&self.object_path()) && event.action == IlmAction::DeleteAction { remaining_versions -= 1; @@ -740,7 +740,7 @@ impl ScannerItem { debug!("apply_actions: applying transition rule for object: {} {}", oi.name, event.action); 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) { + if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) { done_ilm(1)(); } } @@ -768,7 +768,7 @@ impl ScannerItem { .await .enqueue_by_newer_noncurrent(&self.bucket, to_delete_objs, event, &LcEventSrc::Scanner) .await; - if record_scanner_ilm_action_if_queued(global_metrics(), count, queued) { + if record_scanner_ilm_action_if_queued(global_metrics(), action, count, queued) { done_ilm(count)(); remaining_versions = remaining_versions.saturating_sub(noncurrent_accounting.len()); } @@ -2169,8 +2169,9 @@ mod tests { #[tokio::test] async fn test_scanner_ilm_action_accounting_requires_enqueue_success() { let metrics = Metrics::new(); + let _start = metrics.start_scan_cycle_work(); - record_scanner_ilm_action_if_queued(&metrics, 2, false); + record_scanner_ilm_action_if_queued(&metrics, IlmAction::DeleteAction, 2, false); let report = metrics.report().await; let lifecycle = report .source_work @@ -2178,8 +2179,9 @@ mod tests { .find(|work| work.source == ScannerWorkSource::Lifecycle.as_str()) .expect("lifecycle source work should be visible"); assert_eq!(lifecycle.executed, 0); + assert_eq!(report.current_cycle_lifecycle_expiry_actions, 0); - record_scanner_ilm_action_if_queued(&metrics, 3, true); + record_scanner_ilm_action_if_queued(&metrics, IlmAction::DeleteAction, 3, true); let report = metrics.report().await; let lifecycle = report .source_work @@ -2187,6 +2189,19 @@ mod tests { .find(|work| work.source == ScannerWorkSource::Lifecycle.as_str()) .expect("lifecycle source work should be visible"); assert_eq!(lifecycle.executed, 3); + assert_eq!(report.current_cycle_lifecycle_expiry_actions, 3); + assert_eq!(report.current_cycle_lifecycle_transition_actions, 0); + + record_scanner_ilm_action_if_queued(&metrics, IlmAction::TransitionAction, 4, 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, 7); + assert_eq!(report.current_cycle_lifecycle_expiry_actions, 3); + assert_eq!(report.current_cycle_lifecycle_transition_actions, 4); } #[test]