diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index f6b7f0e64..59b5326fb 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -124,6 +124,7 @@ pub enum Metric { Ilm, CheckReplication, Yield, + ThrottleSleep, CleanAbandoned, ApplyNonCurrent, HealAbandonedVersion, @@ -167,6 +168,7 @@ impl Metric { Self::Ilm => "ilm", Self::CheckReplication => "check_replication", Self::Yield => "yield", + Self::ThrottleSleep => "throttle_sleep", Self::CleanAbandoned => "clean_abandoned", Self::ApplyNonCurrent => "apply_non_current", Self::HealAbandonedVersion => "heal_abandoned_version", @@ -203,23 +205,24 @@ impl Metric { 7 => Some(Self::Ilm), 8 => Some(Self::CheckReplication), 9 => Some(Self::Yield), - 10 => Some(Self::CleanAbandoned), - 11 => Some(Self::ApplyNonCurrent), - 12 => Some(Self::HealAbandonedVersion), - 13 => Some(Self::QuotaCheck), - 14 => Some(Self::QuotaViolation), - 15 => Some(Self::QuotaSync), - 16 => Some(Self::StartTrace), - 17 => Some(Self::ScanObject), - 18 => Some(Self::HealAbandonedObject), - 19 => Some(Self::LastRealtime), - 20 => Some(Self::ScanFolder), - 21 => Some(Self::ScanCycle), - 22 => Some(Self::ScanBucketDrive), - 23 => Some(Self::CompactFolder), - 24 => Some(Self::ScanBucketDriveStart), - 25 => Some(Self::ScanBucketDriveFailure), - 26 => Some(Self::Last), + 10 => Some(Self::ThrottleSleep), + 11 => Some(Self::CleanAbandoned), + 12 => Some(Self::ApplyNonCurrent), + 13 => Some(Self::HealAbandonedVersion), + 14 => Some(Self::QuotaCheck), + 15 => Some(Self::QuotaViolation), + 16 => Some(Self::QuotaSync), + 17 => Some(Self::StartTrace), + 18 => Some(Self::ScanObject), + 19 => Some(Self::HealAbandonedObject), + 20 => Some(Self::LastRealtime), + 21 => Some(Self::ScanFolder), + 22 => Some(Self::ScanCycle), + 23 => Some(Self::ScanBucketDrive), + 24 => Some(Self::CompactFolder), + 25 => Some(Self::ScanBucketDriveStart), + 26 => Some(Self::ScanBucketDriveFailure), + 27 => Some(Self::Last), _ => None, } } @@ -355,6 +358,8 @@ pub struct Metrics { current_scan_cycle_bucket_drive_failures_start: AtomicU64, current_scan_cycle_yield_events_start: AtomicU64, current_scan_cycle_yield_duration_millis_start: AtomicU64, + 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_heal_objects_start: AtomicU64, current_scan_cycle_replication_checks_start: AtomicU64, @@ -367,28 +372,35 @@ pub struct Metrics { last_scan_cycle_bucket_drive_failures: AtomicU64, last_scan_cycle_yield_events: AtomicU64, last_scan_cycle_yield_duration_millis: AtomicU64, + last_scan_cycle_throttle_sleep_events: AtomicU64, + last_scan_cycle_throttle_sleep_duration_millis: AtomicU64, last_scan_cycle_ilm_actions: AtomicU64, last_scan_cycle_heal_objects: AtomicU64, last_scan_cycle_replication_checks: AtomicU64, last_scan_cycle_usage_saves: AtomicU64, failed_scan_cycles: AtomicU64, scanner_yield_duration_millis: AtomicU64, + scanner_throttle_sleep_duration_millis: AtomicU64, scanner_ilm_actions: AtomicU64, scanner_throttle_idle_mode_enabled: AtomicBool, scanner_throttle_sleep_factor_micros: AtomicU64, scanner_throttle_max_sleep_millis: AtomicU64, scanner_yield_every_n_objects: AtomicU64, scanner_cycle_interval_millis: AtomicU64, + scanner_cycle_max_duration_millis: AtomicU64, scanner_bitrot_cycle_enabled: AtomicBool, scanner_bitrot_cycle_millis: AtomicU64, + partial_scan_cycles: AtomicU64, } const SCAN_CYCLE_RESULT_UNKNOWN: u8 = 0; const SCAN_CYCLE_RESULT_SUCCESS: u8 = 1; const SCAN_CYCLE_RESULT_ERROR: u8 = 2; +const SCAN_CYCLE_RESULT_PARTIAL: u8 = 3; const SCAN_CYCLE_RESULT_UNKNOWN_LABEL: &str = "unknown"; const SCAN_CYCLE_RESULT_SUCCESS_LABEL: &str = "success"; const SCAN_CYCLE_RESULT_ERROR_LABEL: &str = "error"; +const SCAN_CYCLE_RESULT_PARTIAL_LABEL: &str = "partial"; #[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct CurrentCycle { @@ -406,6 +418,8 @@ pub struct ScanCycleWorkSnapshot { bucket_drive_failures: u64, yield_events: u64, yield_duration_millis: u64, + throttle_sleep_events: u64, + throttle_sleep_duration_millis: u64, ilm_actions: u64, heal_objects: u64, replication_checks: u64, @@ -452,6 +466,10 @@ pub struct ScannerMetricsReport { #[serde(default)] pub current_cycle_yield_duration_seconds: f64, #[serde(default)] + pub current_cycle_throttle_sleep_events: u64, + #[serde(default)] + pub current_cycle_throttle_sleep_duration_seconds: f64, + #[serde(default)] pub current_cycle_ilm_actions: u64, #[serde(default)] pub current_cycle_heal_objects: u64, @@ -475,6 +493,10 @@ pub struct ScannerMetricsReport { #[serde(default)] pub last_cycle_yield_duration_seconds: f64, #[serde(default)] + pub last_cycle_throttle_sleep_events: u64, + #[serde(default)] + pub last_cycle_throttle_sleep_duration_seconds: f64, + #[serde(default)] pub last_cycle_ilm_actions: u64, #[serde(default)] pub last_cycle_heal_objects: u64, @@ -494,9 +516,13 @@ pub struct ScannerMetricsReport { #[serde(default)] pub cycle_interval_seconds: f64, #[serde(default)] + pub cycle_max_duration_seconds: f64, + #[serde(default)] pub bitrot_cycle_enabled: bool, #[serde(default)] pub bitrot_cycle_seconds: f64, + #[serde(default)] + pub partial_cycles: u64, } impl CurrentCycle { @@ -534,6 +560,7 @@ fn scan_cycle_result_label(result: u8) -> &'static str { match result { SCAN_CYCLE_RESULT_SUCCESS => SCAN_CYCLE_RESULT_SUCCESS_LABEL, SCAN_CYCLE_RESULT_ERROR => SCAN_CYCLE_RESULT_ERROR_LABEL, + SCAN_CYCLE_RESULT_PARTIAL => SCAN_CYCLE_RESULT_PARTIAL_LABEL, _ => SCAN_CYCLE_RESULT_UNKNOWN_LABEL, } } @@ -563,6 +590,11 @@ pub fn emit_scan_cycle_complete(success: bool, duration: Duration) { } } +pub fn emit_scan_cycle_partial(duration: Duration) { + global_metrics().record_scan_cycle_partial(duration); + metrics::counter!(OTEL_SCANNER_CYCLES, "result" => SCAN_CYCLE_RESULT_PARTIAL_LABEL).increment(1); +} + pub fn emit_scan_bucket_drive_complete(success: bool, bucket: &str, disk: &str, duration: Duration) { let result = if success { "success" } else { "error" }; metrics::counter!( @@ -580,6 +612,22 @@ pub fn emit_scan_bucket_drive_complete(success: bool, bucket: &str, disk: &str, .record(duration.as_secs_f64()); } +pub fn emit_scan_bucket_drive_partial(bucket: &str, disk: &str, duration: Duration) { + metrics::counter!( + OTEL_SCANNER_BUCKETS_SCANNED, + "result" => SCAN_CYCLE_RESULT_PARTIAL_LABEL, + "bucket" => bucket.to_owned(), + "disk" => disk.to_owned() + ) + .increment(1); + metrics::histogram!( + OTEL_SCANNER_BUCKET_DRIVE_DURATION_SECONDS, + "bucket" => bucket.to_owned(), + "disk" => disk.to_owned() + ) + .record(duration.as_secs_f64()); +} + impl Metrics { pub fn new() -> Self { Self { @@ -603,6 +651,8 @@ impl Metrics { current_scan_cycle_bucket_drive_failures_start: AtomicU64::new(0), current_scan_cycle_yield_events_start: AtomicU64::new(0), current_scan_cycle_yield_duration_millis_start: AtomicU64::new(0), + 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_heal_objects_start: AtomicU64::new(0), current_scan_cycle_replication_checks_start: AtomicU64::new(0), @@ -615,20 +665,25 @@ impl Metrics { last_scan_cycle_bucket_drive_failures: AtomicU64::new(0), last_scan_cycle_yield_events: AtomicU64::new(0), last_scan_cycle_yield_duration_millis: AtomicU64::new(0), + 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_heal_objects: AtomicU64::new(0), last_scan_cycle_replication_checks: AtomicU64::new(0), last_scan_cycle_usage_saves: AtomicU64::new(0), failed_scan_cycles: AtomicU64::new(0), scanner_yield_duration_millis: AtomicU64::new(0), + scanner_throttle_sleep_duration_millis: AtomicU64::new(0), scanner_ilm_actions: 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), scanner_yield_every_n_objects: AtomicU64::new(0), scanner_cycle_interval_millis: AtomicU64::new(0), + scanner_cycle_max_duration_millis: AtomicU64::new(0), scanner_bitrot_cycle_enabled: AtomicBool::new(false), scanner_bitrot_cycle_millis: AtomicU64::new(0), + partial_scan_cycles: AtomicU64::new(0), } } @@ -713,6 +768,9 @@ impl Metrics { Box::new(move |versions: u64| { Box::new(move || { let duration = SystemTime::now().duration_since(start).unwrap_or_default(); + let metric_idx = Metric::Ilm as usize; + global_metrics().operations[metric_idx].fetch_add(versions, Ordering::Relaxed); + emit_otel_counter(metric_idx, versions); global_metrics().actions[a_idx].fetch_add(versions, Ordering::Relaxed); global_metrics().actions_latency[a_idx].add(duration); }) @@ -744,6 +802,20 @@ impl Metrics { }); } + pub fn record_scanner_throttle_sleep(&self, duration: Duration) { + let metric_idx = Metric::ThrottleSleep as usize; + self.operations[metric_idx].fetch_add(1, Ordering::Relaxed); + if metric_idx < Metric::LastRealtime as usize { + self.latency[metric_idx].add(duration); + } + let duration_millis = duration_millis_saturated(duration); + let _ = self + .scanner_throttle_sleep_duration_millis + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { + Some(current.saturating_add(duration_millis)) + }); + } + pub fn record_scanner_ilm_action(&self, count: u64) { self.scanner_ilm_actions.fetch_add(count, Ordering::Relaxed); } @@ -773,9 +845,16 @@ impl Metrics { .store(yield_every_n_objects, Ordering::Relaxed); } - pub fn record_scanner_cycle_config(&self, cycle_interval: Duration, bitrot_cycle: Option) { + pub fn record_scanner_cycle_config( + &self, + cycle_interval: Duration, + bitrot_cycle: Option, + cycle_max_duration: Option, + ) { self.scanner_cycle_interval_millis .store(duration_millis_saturated(cycle_interval), Ordering::Relaxed); + self.scanner_cycle_max_duration_millis + .store(cycle_max_duration.map(duration_millis_saturated).unwrap_or_default(), Ordering::Relaxed); self.scanner_bitrot_cycle_enabled .store(bitrot_cycle.is_some(), Ordering::Relaxed); self.scanner_bitrot_cycle_millis @@ -839,6 +918,14 @@ impl Metrics { .store(duration_millis_saturated(duration), Ordering::Relaxed); } + pub fn record_scan_cycle_partial(&self, duration: Duration) { + self.partial_scan_cycles.fetch_add(1, Ordering::Relaxed); + self.last_scan_cycle_result + .store(SCAN_CYCLE_RESULT_PARTIAL, Ordering::Relaxed); + self.last_scan_cycle_duration_millis + .store(duration_millis_saturated(duration), Ordering::Relaxed); + } + pub fn start_scan_cycle_work(&self) -> ScanCycleWorkSnapshot { let snapshot = self.scan_cycle_work_snapshot(); self.current_scan_cycle_objects_start @@ -853,6 +940,10 @@ impl Metrics { .store(snapshot.yield_events, Ordering::Relaxed); self.current_scan_cycle_yield_duration_millis_start .store(snapshot.yield_duration_millis, Ordering::Relaxed); + self.current_scan_cycle_throttle_sleep_events_start + .store(snapshot.throttle_sleep_events, Ordering::Relaxed); + self.current_scan_cycle_throttle_sleep_duration_millis_start + .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_heal_objects_start @@ -879,6 +970,8 @@ impl Metrics { bucket_drive_failures: self.lifetime(Metric::ScanBucketDriveFailure), yield_events: self.lifetime(Metric::Yield), yield_duration_millis: self.scanner_yield_duration_millis.load(Ordering::Relaxed), + 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), heal_objects: self.lifetime(Metric::HealAbandonedObject), replication_checks: self.lifetime(Metric::CheckReplication), @@ -894,6 +987,10 @@ impl Metrics { bucket_drive_failures: self.current_scan_cycle_bucket_drive_failures_start.load(Ordering::Relaxed), yield_events: self.current_scan_cycle_yield_events_start.load(Ordering::Relaxed), yield_duration_millis: self.current_scan_cycle_yield_duration_millis_start.load(Ordering::Relaxed), + throttle_sleep_events: self.current_scan_cycle_throttle_sleep_events_start.load(Ordering::Relaxed), + throttle_sleep_duration_millis: self + .current_scan_cycle_throttle_sleep_duration_millis_start + .load(Ordering::Relaxed), ilm_actions: self.current_scan_cycle_ilm_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), @@ -910,6 +1007,10 @@ impl Metrics { bucket_drive_failures: current.bucket_drive_failures.saturating_sub(start.bucket_drive_failures), yield_events: current.yield_events.saturating_sub(start.yield_events), yield_duration_millis: current.yield_duration_millis.saturating_sub(start.yield_duration_millis), + throttle_sleep_events: current.throttle_sleep_events.saturating_sub(start.throttle_sleep_events), + throttle_sleep_duration_millis: current + .throttle_sleep_duration_millis + .saturating_sub(start.throttle_sleep_duration_millis), ilm_actions: current.ilm_actions.saturating_sub(start.ilm_actions), heal_objects: current.heal_objects.saturating_sub(start.heal_objects), replication_checks: current.replication_checks.saturating_sub(start.replication_checks), @@ -931,6 +1032,10 @@ impl Metrics { self.last_scan_cycle_yield_events.store(work.yield_events, Ordering::Relaxed); self.last_scan_cycle_yield_duration_millis .store(work.yield_duration_millis, Ordering::Relaxed); + self.last_scan_cycle_throttle_sleep_events + .store(work.throttle_sleep_events, Ordering::Relaxed); + 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_heal_objects.store(work.heal_objects, Ordering::Relaxed); self.last_scan_cycle_replication_checks @@ -982,6 +1087,8 @@ impl Metrics { m.current_cycle_bucket_drive_failures = current_work.bucket_drive_failures; m.current_cycle_yield_events = current_work.yield_events; m.current_cycle_yield_duration_seconds = current_work.yield_duration_millis as f64 / 1000.0; + 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_heal_objects = current_work.heal_objects; m.current_cycle_replication_checks = current_work.replication_checks; @@ -997,6 +1104,9 @@ impl Metrics { m.last_cycle_bucket_drive_failures = self.last_scan_cycle_bucket_drive_failures.load(Ordering::Relaxed); m.last_cycle_yield_events = self.last_scan_cycle_yield_events.load(Ordering::Relaxed); m.last_cycle_yield_duration_seconds = self.last_scan_cycle_yield_duration_millis.load(Ordering::Relaxed) as f64 / 1000.0; + m.last_cycle_throttle_sleep_events = self.last_scan_cycle_throttle_sleep_events.load(Ordering::Relaxed); + 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_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); @@ -1007,8 +1117,10 @@ impl Metrics { m.throttle_max_sleep_seconds = self.scanner_throttle_max_sleep_millis.load(Ordering::Relaxed) as f64 / 1000.0; m.yield_every_n_objects = self.scanner_yield_every_n_objects.load(Ordering::Relaxed); m.cycle_interval_seconds = self.scanner_cycle_interval_millis.load(Ordering::Relaxed) as f64 / 1000.0; + m.cycle_max_duration_seconds = self.scanner_cycle_max_duration_millis.load(Ordering::Relaxed) as f64 / 1000.0; m.bitrot_cycle_enabled = self.scanner_bitrot_cycle_enabled.load(Ordering::Relaxed); m.bitrot_cycle_seconds = self.scanner_bitrot_cycle_millis.load(Ordering::Relaxed) as f64 / 1000.0; + m.partial_cycles = self.partial_scan_cycles.load(Ordering::Relaxed); // Lifetime operation counts for i in 0..Metric::Last as usize { @@ -1223,6 +1335,20 @@ mod tests { assert_eq!(report.failed_cycles, 0); } + #[tokio::test] + async fn report_tracks_partial_scan_cycle_without_failed_increment() { + let metrics = Metrics::new(); + metrics.record_scan_cycle_partial(Duration::from_millis(2500)); + + let report = metrics.report().await; + + assert_eq!(report.last_cycle_result, SCAN_CYCLE_RESULT_PARTIAL_LABEL); + assert_eq!(report.last_cycle_result_code, SCAN_CYCLE_RESULT_PARTIAL as u64); + assert_eq!(report.last_cycle_duration_seconds, 2.5); + assert_eq!(report.failed_cycles, 0); + assert_eq!(report.partial_cycles, 1); + } + #[tokio::test] async fn report_includes_last_scan_cycle_work() { let metrics = Metrics::new(); @@ -1233,6 +1359,8 @@ mod tests { bucket_drive_failures: 2, yield_events: 5, yield_duration_millis: 250, + throttle_sleep_events: 7, + throttle_sleep_duration_millis: 500, ilm_actions: 13, heal_objects: 2, replication_checks: 4, @@ -1247,6 +1375,8 @@ mod tests { assert_eq!(report.last_cycle_bucket_drive_failures, 2); assert_eq!(report.last_cycle_yield_events, 5); assert_eq!(report.last_cycle_yield_duration_seconds, 0.25); + assert_eq!(report.last_cycle_throttle_sleep_events, 7); + assert_eq!(report.last_cycle_throttle_sleep_duration_seconds, 0.5); assert_eq!(report.last_cycle_ilm_actions, 13); assert_eq!(report.last_cycle_heal_objects, 2); assert_eq!(report.last_cycle_replication_checks, 4); @@ -1273,24 +1403,27 @@ mod tests { metrics.operations[Metric::ScanBucketDrive as usize].store(1, Ordering::Relaxed); metrics.operations[Metric::ScanBucketDriveFailure as usize].store(1, Ordering::Relaxed); metrics.operations[Metric::Yield as usize].store(2, Ordering::Relaxed); + metrics.operations[Metric::ThrottleSleep as usize].store(3, Ordering::Relaxed); + metrics.operations[Metric::Ilm as usize].store(6, Ordering::Relaxed); + metrics.record_scanner_ilm_action(6); metrics.operations[Metric::HealAbandonedObject as usize].store(4, Ordering::Relaxed); metrics.operations[Metric::CheckReplication as usize].store(5, Ordering::Relaxed); metrics.operations[Metric::SaveUsage as usize].store(3, Ordering::Relaxed); - metrics.actions[IlmAction::DeleteAction as usize].store(100, Ordering::Relaxed); - metrics.record_scanner_ilm_action(6); metrics.scanner_yield_duration_millis.store(100, Ordering::Relaxed); + metrics.scanner_throttle_sleep_duration_millis.store(200, Ordering::Relaxed); let start = metrics.start_scan_cycle_work(); metrics.operations[Metric::ScanObject as usize].store(17, Ordering::Relaxed); metrics.operations[Metric::ScanFolder as usize].store(8, Ordering::Relaxed); metrics.operations[Metric::ScanBucketDrive as usize].store(3, Ordering::Relaxed); metrics.operations[Metric::ScanBucketDriveFailure as usize].store(4, Ordering::Relaxed); + metrics.operations[Metric::Ilm as usize].store(15, Ordering::Relaxed); + metrics.record_scanner_ilm_action(9); metrics.operations[Metric::HealAbandonedObject as usize].store(7, Ordering::Relaxed); metrics.operations[Metric::CheckReplication as usize].store(11, Ordering::Relaxed); metrics.operations[Metric::SaveUsage as usize].store(5, Ordering::Relaxed); - metrics.actions[IlmAction::DeleteAction as usize].store(150, Ordering::Relaxed); - metrics.record_scanner_ilm_action(9); metrics.record_scanner_yield(Duration::from_millis(150)); + metrics.record_scanner_throttle_sleep(Duration::from_millis(250)); let report = metrics.report().await; @@ -1300,6 +1433,8 @@ mod tests { assert_eq!(report.current_cycle_bucket_drive_failures, 3); assert_eq!(report.current_cycle_yield_events, 1); assert_eq!(report.current_cycle_yield_duration_seconds, 0.15); + assert_eq!(report.current_cycle_throttle_sleep_events, 1); + assert_eq!(report.current_cycle_throttle_sleep_duration_seconds, 0.25); assert_eq!(report.current_cycle_ilm_actions, 9); assert_eq!(report.current_cycle_heal_objects, 3); assert_eq!(report.current_cycle_replication_checks, 6); @@ -1314,6 +1449,8 @@ mod tests { assert_eq!(report.current_cycle_bucket_drive_failures, 0); assert_eq!(report.current_cycle_yield_events, 0); assert_eq!(report.current_cycle_yield_duration_seconds, 0.0); + assert_eq!(report.current_cycle_throttle_sleep_events, 0); + assert_eq!(report.current_cycle_throttle_sleep_duration_seconds, 0.0); assert_eq!(report.current_cycle_ilm_actions, 0); assert_eq!(report.current_cycle_heal_objects, 0); assert_eq!(report.current_cycle_replication_checks, 0); @@ -1324,12 +1461,37 @@ mod tests { assert_eq!(report.last_cycle_bucket_drive_failures, 3); assert_eq!(report.last_cycle_yield_events, 1); assert_eq!(report.last_cycle_yield_duration_seconds, 0.15); + assert_eq!(report.last_cycle_throttle_sleep_events, 1); + assert_eq!(report.last_cycle_throttle_sleep_duration_seconds, 0.25); assert_eq!(report.last_cycle_ilm_actions, 9); assert_eq!(report.last_cycle_heal_objects, 3); assert_eq!(report.last_cycle_replication_checks, 6); assert_eq!(report.last_cycle_usage_saves, 2); } + #[tokio::test] + async fn scanner_cycle_ilm_actions_ignore_global_ilm_work() { + let metrics = Metrics::new(); + metrics.operations[Metric::Ilm as usize].store(4, Ordering::Relaxed); + + let start = metrics.start_scan_cycle_work(); + metrics.operations[Metric::Ilm as usize].store(11, Ordering::Relaxed); + + let report = metrics.report().await; + + assert_eq!(report.current_cycle_ilm_actions, 0); + + metrics.record_scanner_ilm_action(3); + let report = metrics.report().await; + + assert_eq!(report.current_cycle_ilm_actions, 3); + + metrics.finish_scan_cycle_work(start); + let report = metrics.report().await; + + assert_eq!(report.last_cycle_ilm_actions, 3); + } + #[test] fn record_scanner_yield_tracks_count_and_duration() { let metrics = Metrics::new(); @@ -1339,6 +1501,19 @@ mod tests { assert_eq!(metrics.lifetime(Metric::Yield), 1); assert_eq!(metrics.scanner_yield_duration_millis.load(Ordering::Relaxed), 42); assert_eq!(metrics.last_minute(Metric::Yield).n, 1); + assert_eq!(metrics.lifetime(Metric::ThrottleSleep), 0); + } + + #[test] + fn record_scanner_throttle_sleep_tracks_count_and_duration() { + let metrics = Metrics::new(); + + metrics.record_scanner_throttle_sleep(Duration::from_millis(42)); + + assert_eq!(metrics.lifetime(Metric::ThrottleSleep), 1); + assert_eq!(metrics.scanner_throttle_sleep_duration_millis.load(Ordering::Relaxed), 42); + assert_eq!(metrics.last_minute(Metric::ThrottleSleep).n, 1); + assert_eq!(metrics.lifetime(Metric::Yield), 0); } #[tokio::test] @@ -1371,17 +1546,23 @@ mod tests { async fn report_includes_scanner_cycle_config() { let metrics = Metrics::new(); - metrics.record_scanner_cycle_config(Duration::from_secs(3600), Some(Duration::from_secs(86400))); + metrics.record_scanner_cycle_config( + Duration::from_secs(3600), + Some(Duration::from_secs(86400)), + Some(Duration::from_secs(1800)), + ); let report = metrics.report().await; assert_eq!(report.cycle_interval_seconds, 3600.0); + assert_eq!(report.cycle_max_duration_seconds, 1800.0); assert!(report.bitrot_cycle_enabled); assert_eq!(report.bitrot_cycle_seconds, 86400.0); - metrics.record_scanner_cycle_config(Duration::from_secs(60), None); + metrics.record_scanner_cycle_config(Duration::from_secs(60), None, None); let report = metrics.report().await; assert_eq!(report.cycle_interval_seconds, 60.0); + assert_eq!(report.cycle_max_duration_seconds, 0.0); assert!(!report.bitrot_cycle_enabled); assert_eq!(report.bitrot_cycle_seconds, 0.0); } diff --git a/crates/config/src/constants/scanner.rs b/crates/config/src/constants/scanner.rs index 4a15c9eae..b0b81b4fb 100644 --- a/crates/config/src/constants/scanner.rs +++ b/crates/config/src/constants/scanner.rs @@ -31,6 +31,12 @@ pub const ENV_DATA_SCANNER_START_DELAY_SECS: &str = "RUSTFS_DATA_SCANNER_START_D /// - Example: `export RUSTFS_SCANNER_CYCLE=3600` (1 hour) pub const ENV_SCANNER_CYCLE: &str = "RUSTFS_SCANNER_CYCLE"; +/// Environment variable that caps one scanner cycle's runtime in seconds. +/// A value of `0` disables the cycle runtime budget. +/// - Unit: seconds (u64). +/// - Example: `export RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS=1800` +pub const ENV_SCANNER_CYCLE_MAX_DURATION_SECS: &str = "RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS"; + /// Environment variable that selects the scanner speed preset. /// Valid values: `fastest`, `fast`, `default`, `slow`, `slowest`. /// Controls the sleep factor, maximum sleep duration, and cycle interval. @@ -40,6 +46,10 @@ pub const ENV_SCANNER_SPEED: &str = "RUSTFS_SCANNER_SPEED"; /// Default scanner speed preset. pub const DEFAULT_SCANNER_SPEED: &str = "default"; +/// Default scanner cycle runtime budget. +/// `0` keeps the existing unbounded per-cycle behavior. +pub const DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS: u64 = 0; + /// Environment variable that specifies the periodic bitrot scan cycle in seconds. /// When set to `0`, `true`, `on`, or `yes`, every scanner cycle runs in deep mode. /// When set to `false`, `off`, `no`, or `disabled`, periodic deep scans are disabled. diff --git a/crates/obs/src/metrics/collectors/scanner.rs b/crates/obs/src/metrics/collectors/scanner.rs index b11c4ff80..d3a93c9b8 100644 --- a/crates/obs/src/metrics/collectors/scanner.rs +++ b/crates/obs/src/metrics/collectors/scanner.rs @@ -28,15 +28,18 @@ use crate::metrics::schema::scanner::{ SCANNER_CURRENT_CYCLE_DIRECTORIES_PER_SECOND_MD, SCANNER_CURRENT_CYCLE_DIRECTORIES_SCANNED_MD, SCANNER_CURRENT_CYCLE_HEAL_OBJECTS_MD, SCANNER_CURRENT_CYCLE_ILM_ACTIONS_MD, SCANNER_CURRENT_CYCLE_MD, SCANNER_CURRENT_CYCLE_OBJECTS_PER_SECOND_MD, SCANNER_CURRENT_CYCLE_OBJECTS_SCANNED_MD, - SCANNER_CURRENT_CYCLE_REPLICATION_CHECKS_MD, SCANNER_CURRENT_CYCLE_USAGE_SAVES_MD, + SCANNER_CURRENT_CYCLE_REPLICATION_CHECKS_MD, SCANNER_CURRENT_CYCLE_THROTTLE_SLEEP_DURATION_SECONDS_MD, + SCANNER_CURRENT_CYCLE_THROTTLE_SLEEP_EVENTS_MD, SCANNER_CURRENT_CYCLE_USAGE_SAVES_MD, SCANNER_CURRENT_CYCLE_YIELD_DURATION_SECONDS_MD, SCANNER_CURRENT_CYCLE_YIELD_EVENTS_MD, SCANNER_CURRENT_SCAN_MODE_MD, - SCANNER_CYCLE_INTERVAL_SECONDS_MD, SCANNER_DIRECTORIES_SCANNED_MD, SCANNER_FAILED_CYCLES_MD, - SCANNER_LAST_ACTIVITY_SECONDS_MD, SCANNER_LAST_CYCLE_BUCKET_DRIVE_FAILURES_MD, SCANNER_LAST_CYCLE_BUCKET_DRIVE_SCANS_MD, - SCANNER_LAST_CYCLE_BUCKET_DRIVE_SCANS_PER_SECOND_MD, SCANNER_LAST_CYCLE_DIRECTORIES_PER_SECOND_MD, - SCANNER_LAST_CYCLE_DIRECTORIES_SCANNED_MD, SCANNER_LAST_CYCLE_DURATION_SECONDS_MD, SCANNER_LAST_CYCLE_HEAL_OBJECTS_MD, - SCANNER_LAST_CYCLE_ILM_ACTIONS_MD, SCANNER_LAST_CYCLE_OBJECTS_PER_SECOND_MD, SCANNER_LAST_CYCLE_OBJECTS_SCANNED_MD, - SCANNER_LAST_CYCLE_REPLICATION_CHECKS_MD, SCANNER_LAST_CYCLE_RESULT_MD, SCANNER_LAST_CYCLE_USAGE_SAVES_MD, - SCANNER_LAST_CYCLE_YIELD_DURATION_SECONDS_MD, SCANNER_LAST_CYCLE_YIELD_EVENTS_MD, SCANNER_OBJECTS_SCANNED_MD, + SCANNER_CYCLE_INTERVAL_SECONDS_MD, SCANNER_CYCLE_MAX_DURATION_SECONDS_MD, SCANNER_DIRECTORIES_SCANNED_MD, + SCANNER_FAILED_CYCLES_MD, SCANNER_LAST_ACTIVITY_SECONDS_MD, SCANNER_LAST_CYCLE_BUCKET_DRIVE_FAILURES_MD, + SCANNER_LAST_CYCLE_BUCKET_DRIVE_SCANS_MD, SCANNER_LAST_CYCLE_BUCKET_DRIVE_SCANS_PER_SECOND_MD, + SCANNER_LAST_CYCLE_DIRECTORIES_PER_SECOND_MD, SCANNER_LAST_CYCLE_DIRECTORIES_SCANNED_MD, + SCANNER_LAST_CYCLE_DURATION_SECONDS_MD, SCANNER_LAST_CYCLE_HEAL_OBJECTS_MD, SCANNER_LAST_CYCLE_ILM_ACTIONS_MD, + SCANNER_LAST_CYCLE_OBJECTS_PER_SECOND_MD, SCANNER_LAST_CYCLE_OBJECTS_SCANNED_MD, SCANNER_LAST_CYCLE_REPLICATION_CHECKS_MD, + SCANNER_LAST_CYCLE_RESULT_MD, SCANNER_LAST_CYCLE_THROTTLE_SLEEP_DURATION_SECONDS_MD, + SCANNER_LAST_CYCLE_THROTTLE_SLEEP_EVENTS_MD, SCANNER_LAST_CYCLE_USAGE_SAVES_MD, SCANNER_LAST_CYCLE_YIELD_DURATION_SECONDS_MD, + SCANNER_LAST_CYCLE_YIELD_EVENTS_MD, SCANNER_OBJECTS_SCANNED_MD, SCANNER_PARTIAL_CYCLES_MD, SCANNER_THROTTLE_IDLE_MODE_ENABLED_MD, SCANNER_THROTTLE_MAX_SLEEP_SECONDS_MD, SCANNER_THROTTLE_SLEEP_FACTOR_MD, SCANNER_VERSIONS_SCANNED_MD, SCANNER_YIELD_EVERY_N_OBJECTS_MD, }; @@ -61,7 +64,7 @@ pub struct ScannerStats { /// Number of scanner paths currently being processed pub active_paths: u64, /// Whether scanner idle-mode self-throttling is enabled - pub throttle_idle_mode_enabled: u64, + pub throttle_idle_mode_enabled: bool, /// Effective scanner sleep factor pub throttle_sleep_factor: f64, /// Effective scanner maximum self-throttle sleep duration in seconds @@ -70,8 +73,10 @@ pub struct ScannerStats { pub yield_every_n_objects: u64, /// Effective scanner cycle interval in seconds pub cycle_interval_seconds: f64, + /// Effective maximum scanner cycle runtime in seconds + pub cycle_max_duration_seconds: f64, /// Whether periodic scanner bitrot deep scans are enabled - pub bitrot_cycle_enabled: u64, + pub bitrot_cycle_enabled: bool, /// Effective scanner bitrot deep-scan interval in seconds pub bitrot_cycle_seconds: f64, /// Current scanner cycle number, or zero when idle @@ -94,10 +99,14 @@ pub struct ScannerStats { pub current_cycle_directories_per_second: f64, /// Bucket-drive scan rate for the currently running scanner cycle pub current_cycle_bucket_drive_scans_per_second: f64, - /// Number of scanner self-throttle yield events in the current scanner cycle + /// Number of scanner cooperative yield events in the current scanner cycle pub current_cycle_yield_events: u64, - /// Total scanner self-throttle yield duration in seconds for the current scanner cycle + /// Total scanner cooperative yield duration in seconds for the current scanner cycle pub current_cycle_yield_duration_seconds: f64, + /// Number of scanner self-throttle sleep events in the current scanner cycle + pub current_cycle_throttle_sleep_events: u64, + /// Total scanner self-throttle sleep duration in seconds for the current scanner cycle + pub current_cycle_throttle_sleep_duration_seconds: f64, /// Number of lifecycle actions applied by the current scanner cycle pub current_cycle_ilm_actions: u64, /// Number of object heal candidates enqueued by the current scanner cycle @@ -108,7 +117,7 @@ pub struct ScannerStats { pub current_cycle_usage_saves: u64, /// Current scanner mode: 0 unknown or idle, 1 normal, 2 deep bitrot scan pub current_scan_mode: u64, - /// Last scanner cycle result: 0 unknown, 1 success, 2 error + /// Last scanner cycle result: 0 unknown, 1 success, 2 error, 3 partial pub last_cycle_result: u64, /// Duration in seconds of the last finished scanner cycle pub last_cycle_duration_seconds: f64, @@ -126,10 +135,14 @@ pub struct ScannerStats { pub last_cycle_directories_per_second: f64, /// Bucket-drive scan rate for the last finished scanner cycle pub last_cycle_bucket_drive_scans_per_second: f64, - /// Number of scanner self-throttle yield events in the last finished scanner cycle + /// Number of scanner cooperative yield events in the last finished scanner cycle pub last_cycle_yield_events: u64, - /// Total scanner self-throttle yield duration in seconds for the last finished scanner cycle + /// Total scanner cooperative yield duration in seconds for the last finished scanner cycle pub last_cycle_yield_duration_seconds: f64, + /// Number of scanner self-throttle sleep events in the last finished scanner cycle + pub last_cycle_throttle_sleep_events: u64, + /// Total scanner self-throttle sleep duration in seconds for the last finished scanner cycle + pub last_cycle_throttle_sleep_duration_seconds: f64, /// Number of lifecycle actions applied by the last finished scanner cycle pub last_cycle_ilm_actions: u64, /// Number of object heal candidates enqueued by the last finished scanner cycle @@ -140,6 +153,8 @@ pub struct ScannerStats { pub last_cycle_usage_saves: u64, /// Number of scanner cycles that failed since server start pub failed_cycles: u64, + /// Number of scanner cycles stopped by runtime budget since server start + pub partial_cycles: u64, } /// Collects scanner metrics from the given stats. @@ -156,12 +171,16 @@ pub fn collect_scanner_metrics(stats: &ScannerStats) -> Vec { PrometheusMetric::from_descriptor(&SCANNER_VERSIONS_SCANNED_MD, stats.versions_scanned as f64), PrometheusMetric::from_descriptor(&SCANNER_LAST_ACTIVITY_SECONDS_MD, stats.last_activity_seconds as f64), PrometheusMetric::from_descriptor(&SCANNER_ACTIVE_PATHS_MD, stats.active_paths as f64), - PrometheusMetric::from_descriptor(&SCANNER_THROTTLE_IDLE_MODE_ENABLED_MD, stats.throttle_idle_mode_enabled as f64), + PrometheusMetric::from_descriptor( + &SCANNER_THROTTLE_IDLE_MODE_ENABLED_MD, + bool_metric_value(stats.throttle_idle_mode_enabled), + ), PrometheusMetric::from_descriptor(&SCANNER_THROTTLE_SLEEP_FACTOR_MD, stats.throttle_sleep_factor), PrometheusMetric::from_descriptor(&SCANNER_THROTTLE_MAX_SLEEP_SECONDS_MD, stats.throttle_max_sleep_seconds), PrometheusMetric::from_descriptor(&SCANNER_YIELD_EVERY_N_OBJECTS_MD, stats.yield_every_n_objects as f64), PrometheusMetric::from_descriptor(&SCANNER_CYCLE_INTERVAL_SECONDS_MD, stats.cycle_interval_seconds), - PrometheusMetric::from_descriptor(&SCANNER_BITROT_CYCLE_ENABLED_MD, stats.bitrot_cycle_enabled as f64), + PrometheusMetric::from_descriptor(&SCANNER_CYCLE_MAX_DURATION_SECONDS_MD, stats.cycle_max_duration_seconds), + PrometheusMetric::from_descriptor(&SCANNER_BITROT_CYCLE_ENABLED_MD, bool_metric_value(stats.bitrot_cycle_enabled)), PrometheusMetric::from_descriptor(&SCANNER_BITROT_CYCLE_SECONDS_MD, stats.bitrot_cycle_seconds), PrometheusMetric::from_descriptor(&SCANNER_CURRENT_CYCLE_MD, stats.current_cycle as f64), PrometheusMetric::from_descriptor(&SCANNER_COMPLETED_CYCLES_MD, stats.completed_cycles as f64), @@ -193,6 +212,14 @@ pub fn collect_scanner_metrics(stats: &ScannerStats) -> Vec { &SCANNER_CURRENT_CYCLE_YIELD_DURATION_SECONDS_MD, stats.current_cycle_yield_duration_seconds, ), + PrometheusMetric::from_descriptor( + &SCANNER_CURRENT_CYCLE_THROTTLE_SLEEP_EVENTS_MD, + stats.current_cycle_throttle_sleep_events as f64, + ), + PrometheusMetric::from_descriptor( + &SCANNER_CURRENT_CYCLE_THROTTLE_SLEEP_DURATION_SECONDS_MD, + stats.current_cycle_throttle_sleep_duration_seconds, + ), PrometheusMetric::from_descriptor(&SCANNER_CURRENT_CYCLE_ILM_ACTIONS_MD, stats.current_cycle_ilm_actions as f64), PrometheusMetric::from_descriptor(&SCANNER_CURRENT_CYCLE_HEAL_OBJECTS_MD, stats.current_cycle_heal_objects as f64), PrometheusMetric::from_descriptor( @@ -221,14 +248,27 @@ pub fn collect_scanner_metrics(stats: &ScannerStats) -> Vec { ), PrometheusMetric::from_descriptor(&SCANNER_LAST_CYCLE_YIELD_EVENTS_MD, stats.last_cycle_yield_events as f64), PrometheusMetric::from_descriptor(&SCANNER_LAST_CYCLE_YIELD_DURATION_SECONDS_MD, stats.last_cycle_yield_duration_seconds), + PrometheusMetric::from_descriptor( + &SCANNER_LAST_CYCLE_THROTTLE_SLEEP_EVENTS_MD, + stats.last_cycle_throttle_sleep_events as f64, + ), + PrometheusMetric::from_descriptor( + &SCANNER_LAST_CYCLE_THROTTLE_SLEEP_DURATION_SECONDS_MD, + stats.last_cycle_throttle_sleep_duration_seconds, + ), PrometheusMetric::from_descriptor(&SCANNER_LAST_CYCLE_ILM_ACTIONS_MD, stats.last_cycle_ilm_actions as f64), PrometheusMetric::from_descriptor(&SCANNER_LAST_CYCLE_HEAL_OBJECTS_MD, stats.last_cycle_heal_objects as f64), PrometheusMetric::from_descriptor(&SCANNER_LAST_CYCLE_REPLICATION_CHECKS_MD, stats.last_cycle_replication_checks as f64), PrometheusMetric::from_descriptor(&SCANNER_LAST_CYCLE_USAGE_SAVES_MD, stats.last_cycle_usage_saves as f64), PrometheusMetric::from_descriptor(&SCANNER_FAILED_CYCLES_MD, stats.failed_cycles as f64), + PrometheusMetric::from_descriptor(&SCANNER_PARTIAL_CYCLES_MD, stats.partial_cycles as f64), ] } +fn bool_metric_value(enabled: bool) -> f64 { + if enabled { 1.0 } else { 0.0 } +} + #[cfg(test)] mod tests { use super::*; @@ -245,12 +285,13 @@ mod tests { versions_scanned: 1500000, last_activity_seconds: 30, active_paths: 4, - throttle_idle_mode_enabled: 1, + throttle_idle_mode_enabled: true, throttle_sleep_factor: 10.0, throttle_max_sleep_seconds: 15.0, yield_every_n_objects: 128, cycle_interval_seconds: 3600.0, - bitrot_cycle_enabled: 1, + cycle_max_duration_seconds: 1800.0, + bitrot_cycle_enabled: true, bitrot_cycle_seconds: 86400.0, current_cycle: 12, completed_cycles: 11, @@ -264,6 +305,8 @@ mod tests { current_cycle_bucket_drive_scans_per_second: 0.1, current_cycle_yield_events: 8, current_cycle_yield_duration_seconds: 1.25, + current_cycle_throttle_sleep_events: 4, + current_cycle_throttle_sleep_duration_seconds: 2.5, current_cycle_ilm_actions: 6, current_cycle_heal_objects: 2, current_cycle_replication_checks: 5, @@ -280,17 +323,20 @@ mod tests { last_cycle_bucket_drive_scans_per_second: 0.12, last_cycle_yield_events: 30, last_cycle_yield_duration_seconds: 9.5, + last_cycle_throttle_sleep_events: 12, + last_cycle_throttle_sleep_duration_seconds: 6.75, last_cycle_ilm_actions: 44, last_cycle_heal_objects: 7, last_cycle_replication_checks: 12, last_cycle_usage_saves: 9, failed_cycles: 3, + partial_cycles: 2, }; let metrics = collect_scanner_metrics(&stats); report_metrics(&metrics); - assert_eq!(metrics.len(), 48); + assert_eq!(metrics.len(), 54); let objects = metrics.iter().find(|m| m.value == 1000000.0); assert!(objects.is_some()); @@ -333,6 +379,11 @@ mod tests { .find(|m| m.name == SCANNER_CYCLE_INTERVAL_SECONDS_MD.get_full_metric_name()); assert_eq!(cycle_interval_seconds.map(|m| m.value), Some(3600.0)); + let cycle_max_duration_seconds = metrics + .iter() + .find(|m| m.name == SCANNER_CYCLE_MAX_DURATION_SECONDS_MD.get_full_metric_name()); + assert_eq!(cycle_max_duration_seconds.map(|m| m.value), Some(1800.0)); + let bitrot_cycle_enabled = metrics .iter() .find(|m| m.name == SCANNER_BITROT_CYCLE_ENABLED_MD.get_full_metric_name()); @@ -403,6 +454,16 @@ mod tests { .find(|m| m.name == SCANNER_CURRENT_CYCLE_YIELD_DURATION_SECONDS_MD.get_full_metric_name()); assert_eq!(current_cycle_yield_duration.map(|m| m.value), Some(1.25)); + let current_cycle_throttle_sleep_events = metrics + .iter() + .find(|m| m.name == SCANNER_CURRENT_CYCLE_THROTTLE_SLEEP_EVENTS_MD.get_full_metric_name()); + assert_eq!(current_cycle_throttle_sleep_events.map(|m| m.value), Some(4.0)); + + let current_cycle_throttle_sleep_duration = metrics + .iter() + .find(|m| m.name == SCANNER_CURRENT_CYCLE_THROTTLE_SLEEP_DURATION_SECONDS_MD.get_full_metric_name()); + assert_eq!(current_cycle_throttle_sleep_duration.map(|m| m.value), Some(2.5)); + let current_cycle_ilm_actions = metrics .iter() .find(|m| m.name == SCANNER_CURRENT_CYCLE_ILM_ACTIONS_MD.get_full_metric_name()); @@ -483,6 +544,16 @@ mod tests { .find(|m| m.name == SCANNER_LAST_CYCLE_YIELD_DURATION_SECONDS_MD.get_full_metric_name()); assert_eq!(last_cycle_yield_duration.map(|m| m.value), Some(9.5)); + let last_cycle_throttle_sleep_events = metrics + .iter() + .find(|m| m.name == SCANNER_LAST_CYCLE_THROTTLE_SLEEP_EVENTS_MD.get_full_metric_name()); + assert_eq!(last_cycle_throttle_sleep_events.map(|m| m.value), Some(12.0)); + + let last_cycle_throttle_sleep_duration = metrics + .iter() + .find(|m| m.name == SCANNER_LAST_CYCLE_THROTTLE_SLEEP_DURATION_SECONDS_MD.get_full_metric_name()); + assert_eq!(last_cycle_throttle_sleep_duration.map(|m| m.value), Some(6.75)); + let last_cycle_ilm_actions = metrics .iter() .find(|m| m.name == SCANNER_LAST_CYCLE_ILM_ACTIONS_MD.get_full_metric_name()); @@ -507,6 +578,11 @@ mod tests { .iter() .find(|m| m.name == SCANNER_FAILED_CYCLES_MD.get_full_metric_name()); assert_eq!(failed_cycles.map(|m| m.value), Some(3.0)); + + let partial_cycles = metrics + .iter() + .find(|m| m.name == SCANNER_PARTIAL_CYCLES_MD.get_full_metric_name()); + assert_eq!(partial_cycles.map(|m| m.value), Some(2.0)); } #[test] @@ -514,7 +590,7 @@ mod tests { let stats = ScannerStats::default(); let metrics = collect_scanner_metrics(&stats); - assert_eq!(metrics.len(), 48); + assert_eq!(metrics.len(), 54); for metric in &metrics { assert_eq!(metric.value, 0.0); assert!(metric.labels.is_empty()); diff --git a/crates/obs/src/metrics/schema/entry/metric_name.rs b/crates/obs/src/metrics/schema/entry/metric_name.rs index ec70d5ba1..6704e8555 100644 --- a/crates/obs/src/metrics/schema/entry/metric_name.rs +++ b/crates/obs/src/metrics/schema/entry/metric_name.rs @@ -285,6 +285,7 @@ pub enum MetricName { ScannerThrottleMaxSleepSeconds, ScannerYieldEveryNObjects, ScannerCycleIntervalSeconds, + ScannerCycleMaxDurationSeconds, ScannerBitrotCycleEnabled, ScannerBitrotCycleSeconds, ScannerCurrentCycle, @@ -299,6 +300,8 @@ pub enum MetricName { ScannerCurrentCycleBucketDriveScansPerSecond, ScannerCurrentCycleYieldEvents, ScannerCurrentCycleYieldDurationSeconds, + ScannerCurrentCycleThrottleSleepEvents, + ScannerCurrentCycleThrottleSleepDurationSeconds, ScannerCurrentCycleIlmActions, ScannerCurrentCycleHealObjects, ScannerCurrentCycleReplicationChecks, @@ -315,11 +318,14 @@ pub enum MetricName { ScannerLastCycleBucketDriveScansPerSecond, ScannerLastCycleYieldEvents, ScannerLastCycleYieldDurationSeconds, + ScannerLastCycleThrottleSleepEvents, + ScannerLastCycleThrottleSleepDurationSeconds, ScannerLastCycleIlmActions, ScannerLastCycleHealObjects, ScannerLastCycleReplicationChecks, ScannerLastCycleUsageSaves, ScannerFailedCycles, + ScannerPartialCycles, // CPU system-related metrics SysCPUAvgIdle, @@ -667,6 +673,7 @@ impl MetricName { Self::ScannerThrottleMaxSleepSeconds => "throttle_max_sleep_seconds".to_string(), Self::ScannerYieldEveryNObjects => "yield_every_n_objects".to_string(), Self::ScannerCycleIntervalSeconds => "cycle_interval_seconds".to_string(), + Self::ScannerCycleMaxDurationSeconds => "cycle_max_duration_seconds".to_string(), Self::ScannerBitrotCycleEnabled => "bitrot_cycle_enabled".to_string(), Self::ScannerBitrotCycleSeconds => "bitrot_cycle_seconds".to_string(), Self::ScannerCurrentCycle => "current_cycle".to_string(), @@ -681,6 +688,8 @@ impl MetricName { Self::ScannerCurrentCycleBucketDriveScansPerSecond => "current_cycle_bucket_drive_scans_per_second".to_string(), Self::ScannerCurrentCycleYieldEvents => "current_cycle_yield_events".to_string(), Self::ScannerCurrentCycleYieldDurationSeconds => "current_cycle_yield_duration_seconds".to_string(), + Self::ScannerCurrentCycleThrottleSleepEvents => "current_cycle_throttle_sleep_events".to_string(), + Self::ScannerCurrentCycleThrottleSleepDurationSeconds => "current_cycle_throttle_sleep_duration_seconds".to_string(), Self::ScannerCurrentCycleIlmActions => "current_cycle_ilm_actions".to_string(), Self::ScannerCurrentCycleHealObjects => "current_cycle_heal_objects".to_string(), Self::ScannerCurrentCycleReplicationChecks => "current_cycle_replication_checks".to_string(), @@ -697,11 +706,14 @@ impl MetricName { Self::ScannerLastCycleBucketDriveScansPerSecond => "last_cycle_bucket_drive_scans_per_second".to_string(), Self::ScannerLastCycleYieldEvents => "last_cycle_yield_events".to_string(), Self::ScannerLastCycleYieldDurationSeconds => "last_cycle_yield_duration_seconds".to_string(), + Self::ScannerLastCycleThrottleSleepEvents => "last_cycle_throttle_sleep_events".to_string(), + Self::ScannerLastCycleThrottleSleepDurationSeconds => "last_cycle_throttle_sleep_duration_seconds".to_string(), Self::ScannerLastCycleIlmActions => "last_cycle_ilm_actions".to_string(), Self::ScannerLastCycleHealObjects => "last_cycle_heal_objects".to_string(), Self::ScannerLastCycleReplicationChecks => "last_cycle_replication_checks".to_string(), Self::ScannerLastCycleUsageSaves => "last_cycle_usage_saves".to_string(), Self::ScannerFailedCycles => "failed_cycles".to_string(), + Self::ScannerPartialCycles => "partial_cycles".to_string(), // CPU system-related metrics Self::SysCPUAvgIdle => "avg_idle".to_string(), diff --git a/crates/obs/src/metrics/schema/scanner.rs b/crates/obs/src/metrics/schema/scanner.rs index ac1c16516..cd043a31a 100644 --- a/crates/obs/src/metrics/schema/scanner.rs +++ b/crates/obs/src/metrics/schema/scanner.rs @@ -134,6 +134,15 @@ pub static SCANNER_CYCLE_INTERVAL_SECONDS_MD: LazyLock = LazyL ) }); +pub static SCANNER_CYCLE_MAX_DURATION_SECONDS_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCycleMaxDurationSeconds, + "Effective maximum scanner cycle runtime in seconds; zero means unlimited.", + &[], + subsystems::SCANNER, + ) +}); + pub static SCANNER_BITROT_CYCLE_ENABLED_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::ScannerBitrotCycleEnabled, @@ -245,7 +254,7 @@ pub static SCANNER_CURRENT_CYCLE_BUCKET_DRIVE_SCANS_PER_SECOND_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::ScannerCurrentCycleYieldEvents, - "Number of scanner self-throttle yield events in the currently running scanner cycle.", + "Number of scanner cooperative yield events in the currently running scanner cycle.", &[], subsystems::SCANNER, ) @@ -254,7 +263,25 @@ pub static SCANNER_CURRENT_CYCLE_YIELD_EVENTS_MD: LazyLock = L pub static SCANNER_CURRENT_CYCLE_YIELD_DURATION_SECONDS_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::ScannerCurrentCycleYieldDurationSeconds, - "Total scanner self-throttle yield duration in seconds for the currently running scanner cycle.", + "Total scanner cooperative yield duration in seconds for the currently running scanner cycle.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_CURRENT_CYCLE_THROTTLE_SLEEP_EVENTS_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCurrentCycleThrottleSleepEvents, + "Number of scanner self-throttle sleep events in the currently running scanner cycle.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_CURRENT_CYCLE_THROTTLE_SLEEP_DURATION_SECONDS_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCurrentCycleThrottleSleepDurationSeconds, + "Total scanner self-throttle sleep duration in seconds for the currently running scanner cycle.", &[], subsystems::SCANNER, ) @@ -308,7 +335,7 @@ pub static SCANNER_CURRENT_SCAN_MODE_MD: LazyLock = LazyLock:: pub static SCANNER_LAST_CYCLE_RESULT_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::ScannerLastCycleResult, - "Last scanner cycle result: 0 unknown, 1 success, 2 error.", + "Last scanner cycle result: 0 unknown, 1 success, 2 error, 3 partial.", &[], subsystems::SCANNER, ) @@ -389,7 +416,7 @@ pub static SCANNER_LAST_CYCLE_BUCKET_DRIVE_SCANS_PER_SECOND_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::ScannerLastCycleYieldEvents, - "Number of scanner self-throttle yield events in the last finished scanner cycle.", + "Number of scanner cooperative yield events in the last finished scanner cycle.", &[], subsystems::SCANNER, ) @@ -398,7 +425,25 @@ pub static SCANNER_LAST_CYCLE_YIELD_EVENTS_MD: LazyLock = Lazy pub static SCANNER_LAST_CYCLE_YIELD_DURATION_SECONDS_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::ScannerLastCycleYieldDurationSeconds, - "Total scanner self-throttle yield duration in seconds for the last finished scanner cycle.", + "Total scanner cooperative yield duration in seconds for the last finished scanner cycle.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_LAST_CYCLE_THROTTLE_SLEEP_EVENTS_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerLastCycleThrottleSleepEvents, + "Number of scanner self-throttle sleep events in the last finished scanner cycle.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_LAST_CYCLE_THROTTLE_SLEEP_DURATION_SECONDS_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerLastCycleThrottleSleepDurationSeconds, + "Total scanner self-throttle sleep duration in seconds for the last finished scanner cycle.", &[], subsystems::SCANNER, ) @@ -448,3 +493,12 @@ pub static SCANNER_FAILED_CYCLES_MD: LazyLock = LazyLock::new( subsystems::SCANNER, ) }); + +pub static SCANNER_PARTIAL_CYCLES_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::ScannerPartialCycles, + "Total number of scanner cycles stopped before completion by runtime budget.", + &[], + subsystems::SCANNER, + ) +}); diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 797c24488..390c7c310 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -934,12 +934,13 @@ pub async fn collect_scanner_metric_stats() -> Option { versions_scanned, last_activity_seconds, active_paths, - throttle_idle_mode_enabled: u64::from(metrics.throttle_idle_mode_enabled), + throttle_idle_mode_enabled: metrics.throttle_idle_mode_enabled, throttle_sleep_factor: metrics.throttle_sleep_factor, throttle_max_sleep_seconds: metrics.throttle_max_sleep_seconds, yield_every_n_objects: metrics.yield_every_n_objects, cycle_interval_seconds: metrics.cycle_interval_seconds, - bitrot_cycle_enabled: u64::from(metrics.bitrot_cycle_enabled), + cycle_max_duration_seconds: metrics.cycle_max_duration_seconds, + bitrot_cycle_enabled: metrics.bitrot_cycle_enabled, bitrot_cycle_seconds: metrics.bitrot_cycle_seconds, current_cycle: metrics.current_cycle, completed_cycles, @@ -959,6 +960,8 @@ pub async fn collect_scanner_metric_stats() -> Option { ), current_cycle_yield_events: metrics.current_cycle_yield_events, current_cycle_yield_duration_seconds: metrics.current_cycle_yield_duration_seconds, + current_cycle_throttle_sleep_events: metrics.current_cycle_throttle_sleep_events, + current_cycle_throttle_sleep_duration_seconds: metrics.current_cycle_throttle_sleep_duration_seconds, current_cycle_ilm_actions: metrics.current_cycle_ilm_actions, current_cycle_heal_objects: metrics.current_cycle_heal_objects, current_cycle_replication_checks: metrics.current_cycle_replication_checks, @@ -981,11 +984,14 @@ pub async fn collect_scanner_metric_stats() -> Option { ), last_cycle_yield_events: metrics.last_cycle_yield_events, last_cycle_yield_duration_seconds: metrics.last_cycle_yield_duration_seconds, + last_cycle_throttle_sleep_events: metrics.last_cycle_throttle_sleep_events, + last_cycle_throttle_sleep_duration_seconds: metrics.last_cycle_throttle_sleep_duration_seconds, last_cycle_ilm_actions: metrics.last_cycle_ilm_actions, last_cycle_heal_objects: metrics.last_cycle_heal_objects, last_cycle_replication_checks: metrics.last_cycle_replication_checks, last_cycle_usage_saves: metrics.last_cycle_usage_saves, failed_cycles: metrics.failed_cycles, + partial_cycles: metrics.partial_cycles, }) } diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 1cb3c116e..2dc849c83 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -1377,8 +1377,7 @@ mod tests { let attempts = attempts_clone.clone(); async move { attempts.fetch_add(1, Ordering::SeqCst); - tokio::time::sleep(Duration::from_millis(50)).await; - Ok(()) + std::future::pending::>().await } }) .await; diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 229a9cee0..17eb8c914 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -14,7 +14,7 @@ use std::sync::{ Arc, - atomic::{AtomicU64, Ordering}, + atomic::{AtomicBool, AtomicU64, Ordering}, }; use crate::data_usage_define::{BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH}; @@ -24,11 +24,11 @@ use crate::sleeper::{SCANNER_SLEEPER, scanner_speed_from_env_or_default, set_sca use crate::{DataUsageInfo, ScannerActivityGuard, ScannerError}; use chrono::{DateTime, Utc}; use rustfs_common::heal_channel::HealScanMode; -use rustfs_common::metrics::{CurrentCycle, Metric, Metrics, emit_scan_cycle_complete, global_metrics}; +use rustfs_common::metrics::{CurrentCycle, Metric, Metrics, emit_scan_cycle_complete, emit_scan_cycle_partial, global_metrics}; use rustfs_config::ScannerSpeed; use rustfs_config::{ - DEFAULT_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, - ENV_SCANNER_START_DELAY_SECS, + DEFAULT_SCANNER_BITROT_CYCLE_SECS, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CYCLE, + ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS, }; use rustfs_ecstore::StorageAPI as _; use rustfs_ecstore::bucket::lifecycle::lifecycle::Lifecycle as _; @@ -87,6 +87,13 @@ fn scanner_start_delay_secs() -> Option { rustfs_utils::get_env_opt_u64_with_aliases(ENV_SCANNER_START_DELAY_SECS, &deprecated) } +fn scanner_cycle_max_duration() -> Option { + match rustfs_utils::get_env_u64(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS) { + 0 => None, + secs => Some(Duration::from_secs(secs)), + } +} + /// Compute a randomized inter-cycle sleep. // Delay is scan interval +- 10%, with a floor of 1 second. fn randomized_cycle_delay() -> Duration { @@ -432,13 +439,78 @@ fn get_lock_acquire_timeout() -> Duration { Duration::from_secs(rustfs_utils::get_env_u64("RUSTFS_LOCK_ACQUIRE_TIMEOUT", 5)) } +struct ScannerCycleBudget { + token: CancellationToken, + elapsed: Arc, + max_duration: Option, +} + +impl ScannerCycleBudget { + fn new(parent: &CancellationToken, max_duration: Option) -> Self { + let token = parent.child_token(); + let elapsed = Arc::new(AtomicBool::new(false)); + + if let Some(duration) = max_duration { + let parent = parent.clone(); + let token_wait = token.clone(); + let token_cancel = token.clone(); + let elapsed = elapsed.clone(); + tokio::spawn(async move { + tokio::select! { + _ = parent.cancelled() => {} + _ = token_wait.cancelled() => {} + _ = tokio::time::sleep(duration) => { + elapsed.store(true, Ordering::Relaxed); + token_cancel.cancel(); + } + } + }); + } + + Self { + token, + elapsed, + max_duration, + } + } + + fn token(&self) -> CancellationToken { + self.token.clone() + } + + fn budget_elapsed(&self) -> bool { + self.elapsed.load(Ordering::Relaxed) + } + + fn max_duration(&self) -> Option { + self.max_duration + } +} + +impl Drop for ScannerCycleBudget { + fn drop(&mut self) { + self.token.cancel(); + } +} + +async fn mark_scan_cycle_idle(cycle_info: &mut CurrentCycle) { + cycle_info.current = 0; + global_metrics().clear_current_scan_mode(); + global_metrics().set_cycle(Some(cycle_info.clone())).await; +} + #[instrument(skip_all)] async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc, cycle_info: &mut CurrentCycle) { let _activity_guard = ScannerActivityGuard::new(); SCANNER_SLEEPER.refresh_from_env(); let configured_cycle_interval = cycle_interval(); let configured_bitrot_cycle = bitrot_scan_cycle(); - global_metrics().record_scanner_cycle_config(configured_cycle_interval, configured_bitrot_cycle); + let configured_cycle_max_duration = scanner_cycle_max_duration(); + global_metrics().record_scanner_cycle_config( + configured_cycle_interval, + configured_bitrot_cycle, + configured_cycle_max_duration, + ); info!("Start run data scanner cycle"); cycle_info.current = cycle_info.next; let now = Instant::now(); @@ -476,19 +548,42 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc let done_cycle = Metrics::time(Metric::ScanCycle); let cycle_start = std::time::Instant::now(); let cycle_work_start = global_metrics().start_scan_cycle_work(); + let cycle_budget = ScannerCycleBudget::new(ctx, configured_cycle_max_duration); if let Err(e) = storeapi .clone() - .nsscanner(ctx.clone(), sender, cycle_info.current, scan_mode) + .nsscanner(cycle_budget.token(), sender, cycle_info.current, scan_mode) .await { - error!(duration = ?now.elapsed(), "Fail run data scanner cycle: {e}"); + let budget_elapsed = cycle_budget.budget_elapsed() && !ctx.is_cancelled(); global_metrics().finish_scan_cycle_work(cycle_work_start); + if budget_elapsed { + warn!( + duration = ?now.elapsed(), + max_duration = ?cycle_budget.max_duration(), + "Data scanner cycle stopped after reaching its runtime budget" + ); + emit_scan_cycle_partial(cycle_start.elapsed()); + mark_scan_cycle_idle(cycle_info).await; + return; + } + error!(duration = ?now.elapsed(), "Fail run data scanner cycle: {e}"); emit_scan_cycle_complete(false, cycle_start.elapsed()); if let Some(new_heal_info) = background_heal_info_for_scan_complete(background_heal_info.clone(), scan_mode) { save_background_heal_info(storeapi.clone(), new_heal_info).await; } return; } + if cycle_budget.budget_elapsed() && !ctx.is_cancelled() { + warn!( + duration = ?now.elapsed(), + max_duration = ?cycle_budget.max_duration(), + "Data scanner cycle stopped after reaching its runtime budget" + ); + global_metrics().finish_scan_cycle_work(cycle_work_start); + emit_scan_cycle_partial(cycle_start.elapsed()); + mark_scan_cycle_idle(cycle_info).await; + return; + } done_cycle(); global_metrics().finish_scan_cycle_work(cycle_work_start); emit_scan_cycle_complete(true, cycle_start.elapsed()); @@ -504,7 +599,6 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc info!(duration = ?now.elapsed(), cycles_total=cycle_info.cycle_completed.len(), "Success run data scanner cycle"); retain_recent_cycle_completions(&mut cycle_info.cycle_completed); - global_metrics().set_cycle(Some(cycle_info.clone())).await; let cycle_info_buf = cycle_info.marshal().unwrap_or_default(); @@ -720,6 +814,75 @@ mod tests { }); } + #[test] + #[serial] + fn test_scanner_cycle_max_duration_uses_env() { + with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("42"), || { + assert_eq!(scanner_cycle_max_duration(), Some(Duration::from_secs(42))); + }); + } + + #[test] + #[serial] + fn test_scanner_cycle_max_duration_default_is_disabled() { + with_var_unset(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, || { + assert_eq!(scanner_cycle_max_duration(), None); + }); + } + + #[tokio::test] + async fn test_scanner_cycle_budget_cancels_after_duration() { + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new(&parent, Some(Duration::from_millis(1))); + + tokio::time::timeout(Duration::from_secs(5), budget.token().cancelled()) + .await + .expect("scanner cycle budget should cancel after max duration"); + + assert!(budget.budget_elapsed()); + assert!(budget.token().is_cancelled()); + } + + #[tokio::test] + async fn test_scanner_cycle_budget_drop_cancels_child_without_elapsed() { + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new(&parent, Some(Duration::from_secs(60))); + let token = budget.token(); + + drop(budget); + + assert!(token.is_cancelled()); + } + + #[tokio::test] + #[serial] + async fn test_mark_scan_cycle_idle_clears_published_cycle_state() { + let mut cycle_info = CurrentCycle { + current: 12, + next: 13, + cycle_completed: vec![Utc::now()], + started: Utc::now(), + }; + + global_metrics().set_current_scan_mode(HealScanMode::Deep); + global_metrics().set_cycle(Some(cycle_info.clone())).await; + + mark_scan_cycle_idle(&mut cycle_info).await; + + let published = global_metrics() + .get_cycle() + .await + .expect("scanner cycle state should remain published"); + + assert_eq!(cycle_info.current, 0); + assert_eq!(cycle_info.next, 13); + assert_eq!(published.current, 0); + assert_eq!(published.next, 13); + assert_eq!(global_metrics().current_scan_mode(), HealScanMode::Unknown); + + global_metrics().set_cycle(None).await; + } + #[test] #[serial] fn test_cycle_interval_prefers_explicit_cycle_override() { diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 2e6af1e06..4c070b880 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -22,7 +22,7 @@ use futures::future::join_all; use metrics::counter; use rand::seq::SliceRandom as _; use rustfs_common::heal_channel::HealScanMode; -use rustfs_common::metrics::{Metric, Metrics, emit_scan_bucket_drive_complete, global_metrics}; +use rustfs_common::metrics::{Metric, Metrics, emit_scan_bucket_drive_complete, emit_scan_bucket_drive_partial, global_metrics}; use rustfs_config::{ DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS, ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, @@ -125,7 +125,7 @@ impl BucketDriveFailureGuard { Self { failed: true } } - fn mark_success(&mut self) { + fn mark_not_failed(&mut self) { self.failed = false; } } @@ -395,6 +395,16 @@ impl ScannerIO for ECStore { .nsscanner_cache(child_token_clone.clone(), all_buckets_clone, tx, want_cycle_clone, scan_mode_clone) .await { + if child_token_clone.is_cancelled() { + debug!( + pool = %pool_label, + set = %set_label, + error = %e, + "Scanner set scan stopped after cancellation" + ); + return; + } + counter!( "rustfs_scanner_set_failure_total", "pool" => pool_label.clone(), @@ -752,7 +762,11 @@ impl ScannerIOCache for SetDisks { { Ok(cache) => cache, Err(e) => { - error!("Failed to scan disk: {}", e); + if ctx_clone.is_cancelled() { + debug!("Scanner disk scan stopped after cancellation: {}", e); + } else { + error!("Failed to scan disk: {}", e); + } if let (Some(last_update), Some(before_update)) = (cache.info.last_update, before) && last_update > before_update @@ -990,19 +1004,24 @@ impl ScannerIODisk for Disk { let disks = disks_result.into_iter().flatten().collect::>>(); - let result = scan_data_folder(ctx, disks, local_disk, cache, updates, scan_mode, SCANNER_SLEEPER.clone()).await; + let result = scan_data_folder(ctx.clone(), disks, local_disk, cache, updates, scan_mode, SCANNER_SLEEPER.clone()).await; match result { Ok(mut data_usage_info) => { done_drive(); emit_scan_bucket_drive_complete(true, &bucket, &disk_path, drive_start.elapsed()); data_usage_info.info.last_update = Some(SystemTime::now()); - failure_guard.mark_success(); + failure_guard.mark_not_failed(); Ok(data_usage_info) } Err(e) => { - done_drive(); - emit_scan_bucket_drive_complete(false, &bucket, &disk_path, drive_start.elapsed()); + if ctx.is_cancelled() { + emit_scan_bucket_drive_partial(&bucket, &disk_path, drive_start.elapsed()); + failure_guard.mark_not_failed(); + } else { + done_drive(); + emit_scan_bucket_drive_complete(false, &bucket, &disk_path, drive_start.elapsed()); + } Err(StorageError::other(format!("Failed to scan data folder: {e}"))) } } diff --git a/crates/scanner/src/sleeper.rs b/crates/scanner/src/sleeper.rs index fbdf9a944..51170fd40 100644 --- a/crates/scanner/src/sleeper.rs +++ b/crates/scanner/src/sleeper.rs @@ -134,7 +134,7 @@ impl DynamicSleeper { let sleep_dur = Duration::from_secs_f64(MIN_SLEEP.as_secs_f64() * factor).min(max_sleep); if !sleep_dur.is_zero() { tokio::time::sleep(sleep_dur).await; - global_metrics().record_scanner_yield(sleep_dur); + global_metrics().record_scanner_throttle_sleep(sleep_dur); } } @@ -199,7 +199,7 @@ impl SleepTimer { .min(max_sleep); if !sleep_dur.is_zero() { tokio::time::sleep(sleep_dur).await; - global_metrics().record_scanner_yield(sleep_dur); + global_metrics().record_scanner_throttle_sleep(sleep_dur); } } }