From ad1a489f750d16f37d3e1951debd3373ed054502 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Wed, 3 Jun 2026 22:37:58 +0800 Subject: [PATCH] feat(scanner): add scanner budget progress controls (#3185) Co-authored-by: Henry Guo --- crates/common/src/metrics.rs | 314 ++++++++++++++++- crates/config/README.md | 3 + crates/config/src/constants/scanner.rs | 18 + crates/obs/src/metrics/collectors/scanner.rs | 193 ++++++++++- .../src/metrics/schema/entry/metric_name.rs | 22 ++ crates/obs/src/metrics/schema/scanner.rs | 101 +++++- crates/obs/src/metrics/stats_collector.rs | 14 + crates/scanner/src/error.rs | 6 + crates/scanner/src/lib.rs | 1 + crates/scanner/src/scanner.rs | 171 ++++++---- crates/scanner/src/scanner_budget.rs | 231 +++++++++++++ crates/scanner/src/scanner_folder.rs | 144 +++++++- crates/scanner/src/scanner_io.rs | 317 +++++++++++++----- 13 files changed, 1346 insertions(+), 189 deletions(-) create mode 100644 crates/scanner/src/scanner_budget.rs diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index 59b5326fb..da392d363 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -319,26 +319,48 @@ impl LockedLastMinuteLatency { // lives inside async path-update callbacks. // --------------------------------------------------------------------------- +#[derive(Clone, Debug)] +struct CurrentPathState { + path: String, + updated_at: DateTime, +} + struct CurrentPathTracker { - current_path: Arc>, + state: Arc>, } impl CurrentPathTracker { fn new(initial_path: String) -> Self { + Self::new_at(initial_path, Utc::now()) + } + + fn new_at(initial_path: String, updated_at: DateTime) -> Self { Self { - current_path: Arc::new(RwLock::new(initial_path)), + state: Arc::new(RwLock::new(CurrentPathState { + path: initial_path, + updated_at, + })), } } async fn update_path(&self, path: String) { - *self.current_path.write().await = path; + let mut state = self.state.write().await; + state.path = path; + state.updated_at = Utc::now(); } - async fn get_path(&self) -> String { - self.current_path.read().await.clone() + async fn get_state(&self) -> CurrentPathState { + self.state.read().await.clone() } } +#[derive(Clone, Copy, Debug, Default)] +struct ScannerDiskBucketScanState { + concurrency_limit: u64, + queued: u64, + active: u64, +} + // --------------------------------------------------------------------------- // Metrics // --------------------------------------------------------------------------- @@ -364,7 +386,12 @@ pub struct Metrics { current_scan_cycle_heal_objects_start: AtomicU64, current_scan_cycle_replication_checks_start: AtomicU64, current_scan_cycle_usage_saves_start: AtomicU64, + scanner_set_scan_concurrency_limit: AtomicU64, + scanner_set_scans_queued: AtomicU64, + scanner_set_scans_active: AtomicU64, + scanner_disk_bucket_scan_states: Mutex>, last_scan_cycle_result: AtomicU8, + last_scan_cycle_partial_reason: AtomicU8, last_scan_cycle_duration_millis: AtomicU64, last_scan_cycle_objects_scanned: AtomicU64, last_scan_cycle_directories_scanned: AtomicU64, @@ -379,6 +406,10 @@ pub struct Metrics { last_scan_cycle_replication_checks: AtomicU64, last_scan_cycle_usage_saves: AtomicU64, failed_scan_cycles: AtomicU64, + partial_scan_cycles_unknown: AtomicU64, + partial_scan_cycles_runtime: AtomicU64, + partial_scan_cycles_objects: AtomicU64, + partial_scan_cycles_directories: AtomicU64, scanner_yield_duration_millis: AtomicU64, scanner_throttle_sleep_duration_millis: AtomicU64, scanner_ilm_actions: AtomicU64, @@ -388,6 +419,8 @@ pub struct Metrics { scanner_yield_every_n_objects: AtomicU64, scanner_cycle_interval_millis: AtomicU64, scanner_cycle_max_duration_millis: AtomicU64, + scanner_cycle_max_objects: AtomicU64, + scanner_cycle_max_directories: AtomicU64, scanner_bitrot_cycle_enabled: AtomicBool, scanner_bitrot_cycle_millis: AtomicU64, partial_scan_cycles: AtomicU64, @@ -402,6 +435,35 @@ 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, Copy, Debug, Default, PartialEq, Eq)] +pub enum ScanCyclePartialReason { + #[default] + Unknown = 0, + Runtime = 1, + Objects = 2, + Directories = 3, +} + +impl ScanCyclePartialReason { + pub fn as_str(self) -> &'static str { + match self { + Self::Unknown => SCAN_CYCLE_RESULT_UNKNOWN_LABEL, + Self::Runtime => "runtime", + Self::Objects => "objects", + Self::Directories => "directories", + } + } + + fn from_code(code: u8) -> Self { + match code { + 1 => Self::Runtime, + 2 => Self::Objects, + 3 => Self::Directories, + _ => Self::Unknown, + } + } +} + #[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct CurrentCycle { pub current: u64, @@ -448,12 +510,26 @@ pub struct ScannerMetricsReport { pub ongoing_buckets: usize, #[serde(default)] pub active_scan_paths: usize, + #[serde(default)] + pub oldest_active_path_age_seconds: u64, pub life_time_ops: HashMap, pub life_time_ilm: HashMap, pub last_minute: ScannerLastMinute, pub active_paths: Vec, pub current_scan_mode: String, #[serde(default)] + pub current_set_scan_concurrency_limit: u64, + #[serde(default)] + pub current_set_scans_queued: u64, + #[serde(default)] + pub current_set_scans_active: u64, + #[serde(default)] + pub current_disk_scan_concurrency_limit: u64, + #[serde(default)] + pub current_disk_bucket_scans_queued: u64, + #[serde(default)] + pub current_disk_bucket_scans_active: u64, + #[serde(default)] pub current_cycle_objects_scanned: u64, #[serde(default)] pub current_cycle_directories_scanned: u64, @@ -479,6 +555,10 @@ pub struct ScannerMetricsReport { pub current_cycle_usage_saves: u64, pub last_cycle_result: String, pub last_cycle_result_code: u64, + #[serde(default)] + pub last_cycle_partial_reason: String, + #[serde(default)] + pub last_cycle_partial_reason_code: u64, pub last_cycle_duration_seconds: f64, #[serde(default)] pub last_cycle_objects_scanned: u64, @@ -506,6 +586,14 @@ pub struct ScannerMetricsReport { pub last_cycle_usage_saves: u64, pub failed_cycles: u64, #[serde(default)] + pub partial_cycles_unknown: u64, + #[serde(default)] + pub partial_cycles_runtime: u64, + #[serde(default)] + pub partial_cycles_objects: u64, + #[serde(default)] + pub partial_cycles_directories: u64, + #[serde(default)] pub throttle_idle_mode_enabled: bool, #[serde(default)] pub throttle_sleep_factor: f64, @@ -518,6 +606,10 @@ pub struct ScannerMetricsReport { #[serde(default)] pub cycle_max_duration_seconds: f64, #[serde(default)] + pub cycle_max_objects: u64, + #[serde(default)] + pub cycle_max_directories: u64, + #[serde(default)] pub bitrot_cycle_enabled: bool, #[serde(default)] pub bitrot_cycle_seconds: f64, @@ -590,8 +682,8 @@ 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); +pub fn emit_scan_cycle_partial(duration: Duration, reason: ScanCyclePartialReason) { + global_metrics().record_scan_cycle_partial(duration, reason); metrics::counter!(OTEL_SCANNER_CYCLES, "result" => SCAN_CYCLE_RESULT_PARTIAL_LABEL).increment(1); } @@ -657,7 +749,12 @@ impl Metrics { 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), + scanner_set_scan_concurrency_limit: AtomicU64::new(0), + scanner_set_scans_queued: AtomicU64::new(0), + scanner_set_scans_active: AtomicU64::new(0), + scanner_disk_bucket_scan_states: Mutex::new(HashMap::new()), last_scan_cycle_result: AtomicU8::new(SCAN_CYCLE_RESULT_UNKNOWN), + last_scan_cycle_partial_reason: AtomicU8::new(ScanCyclePartialReason::Unknown as u8), last_scan_cycle_duration_millis: AtomicU64::new(0), last_scan_cycle_objects_scanned: AtomicU64::new(0), last_scan_cycle_directories_scanned: AtomicU64::new(0), @@ -672,6 +769,10 @@ impl Metrics { last_scan_cycle_replication_checks: AtomicU64::new(0), last_scan_cycle_usage_saves: AtomicU64::new(0), failed_scan_cycles: AtomicU64::new(0), + partial_scan_cycles_unknown: AtomicU64::new(0), + partial_scan_cycles_runtime: AtomicU64::new(0), + partial_scan_cycles_objects: AtomicU64::new(0), + partial_scan_cycles_directories: AtomicU64::new(0), scanner_yield_duration_millis: AtomicU64::new(0), scanner_throttle_sleep_duration_millis: AtomicU64::new(0), scanner_ilm_actions: AtomicU64::new(0), @@ -681,6 +782,8 @@ impl Metrics { scanner_yield_every_n_objects: AtomicU64::new(0), scanner_cycle_interval_millis: AtomicU64::new(0), scanner_cycle_max_duration_millis: AtomicU64::new(0), + scanner_cycle_max_objects: AtomicU64::new(0), + scanner_cycle_max_directories: AtomicU64::new(0), scanner_bitrot_cycle_enabled: AtomicBool::new(false), scanner_bitrot_cycle_millis: AtomicU64::new(0), partial_scan_cycles: AtomicU64::new(0), @@ -850,17 +953,69 @@ impl Metrics { cycle_interval: Duration, bitrot_cycle: Option, cycle_max_duration: Option, + cycle_max_objects: Option, + cycle_max_directories: 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_cycle_max_objects + .store(cycle_max_objects.unwrap_or_default(), Ordering::Relaxed); + self.scanner_cycle_max_directories + .store(cycle_max_directories.unwrap_or_default(), Ordering::Relaxed); self.scanner_bitrot_cycle_enabled .store(bitrot_cycle.is_some(), Ordering::Relaxed); self.scanner_bitrot_cycle_millis .store(bitrot_cycle.map(duration_millis_saturated).unwrap_or_default(), Ordering::Relaxed); } + pub fn record_scanner_set_scan_state(&self, concurrency_limit: Option, queued: Option, active: Option) { + if let Some(concurrency_limit) = concurrency_limit { + self.scanner_set_scan_concurrency_limit + .store(concurrency_limit as u64, Ordering::Relaxed); + } + if let Some(queued) = queued { + self.scanner_set_scans_queued.store(queued as u64, Ordering::Relaxed); + } + if let Some(active) = active { + self.scanner_set_scans_active.store(active as u64, Ordering::Relaxed); + } + } + + pub fn reset_scanner_set_scan_state(&self) { + self.record_scanner_set_scan_state(Some(0), Some(0), Some(0)); + match self.scanner_disk_bucket_scan_states.lock() { + Ok(mut states) => states.clear(), + Err(poisoned) => poisoned.into_inner().clear(), + } + } + + pub fn record_scanner_disk_bucket_scan_state( + &self, + pool: &str, + set: &str, + concurrency_limit: Option, + queued: Option, + active: Option, + ) { + let key = format!("{pool}/{set}"); + let mut states = match self.scanner_disk_bucket_scan_states.lock() { + Ok(states) => states, + Err(poisoned) => poisoned.into_inner(), + }; + let state = states.entry(key).or_default(); + if let Some(concurrency_limit) = concurrency_limit { + state.concurrency_limit = concurrency_limit as u64; + } + if let Some(queued) = queued { + state.queued = queued as u64; + } + if let Some(active) = active { + state.active = active as u64; + } + } + // ----------------------------------------------------------------------- // Read-side helpers // ----------------------------------------------------------------------- @@ -914,14 +1069,24 @@ impl Metrics { SCAN_CYCLE_RESULT_ERROR }; self.last_scan_cycle_result.store(result, Ordering::Relaxed); + self.last_scan_cycle_partial_reason + .store(ScanCyclePartialReason::Unknown as u8, Ordering::Relaxed); self.last_scan_cycle_duration_millis .store(duration_millis_saturated(duration), Ordering::Relaxed); } - pub fn record_scan_cycle_partial(&self, duration: Duration) { + pub fn record_scan_cycle_partial(&self, duration: Duration, reason: ScanCyclePartialReason) { self.partial_scan_cycles.fetch_add(1, Ordering::Relaxed); + match reason { + ScanCyclePartialReason::Unknown => &self.partial_scan_cycles_unknown, + ScanCyclePartialReason::Runtime => &self.partial_scan_cycles_runtime, + ScanCyclePartialReason::Objects => &self.partial_scan_cycles_objects, + ScanCyclePartialReason::Directories => &self.partial_scan_cycles_directories, + } + .fetch_add(1, Ordering::Relaxed); self.last_scan_cycle_result .store(SCAN_CYCLE_RESULT_PARTIAL, Ordering::Relaxed); + self.last_scan_cycle_partial_reason.store(reason as u8, Ordering::Relaxed); self.last_scan_cycle_duration_millis .store(duration_millis_saturated(duration), Ordering::Relaxed); } @@ -1045,11 +1210,20 @@ impl Metrics { /// Snapshot of every path currently being scanned. pub async fn get_current_paths(&self) -> Vec { + self.current_path_snapshots() + .await + .into_iter() + .map(|(disk, state)| format!("{disk}/{}", state.path)) + .collect() + } + + async fn current_path_snapshots(&self) -> Vec<(String, CurrentPathState)> { let paths = self.current_paths.read().await; let mut result = Vec::with_capacity(paths.len()); for (disk, tracker) in paths.iter() { - result.push(format!("{disk}/{}", tracker.get_path().await)); + result.push((disk.clone(), tracker.get_state().await)); } + result.sort_by(|(left, _), (right, _)| left.cmp(right)); result } @@ -1076,9 +1250,33 @@ impl Metrics { } m.collected_at = Utc::now(); - m.active_paths = self.get_current_paths().await; - m.active_scan_paths = m.active_paths.len(); + let current_path_snapshots = self.current_path_snapshots().await; + m.active_scan_paths = current_path_snapshots.len(); + m.oldest_active_path_age_seconds = current_path_snapshots + .iter() + .map(|(_, state)| m.collected_at.signed_duration_since(state.updated_at).num_seconds().max(0) as u64) + .max() + .unwrap_or_default(); + m.active_paths = current_path_snapshots + .into_iter() + .map(|(disk, state)| format!("{disk}/{}", state.path)) + .collect(); m.current_scan_mode = self.current_scan_mode().as_str().to_string(); + m.current_set_scan_concurrency_limit = self.scanner_set_scan_concurrency_limit.load(Ordering::Relaxed); + m.current_set_scans_queued = self.scanner_set_scans_queued.load(Ordering::Relaxed); + m.current_set_scans_active = self.scanner_set_scans_active.load(Ordering::Relaxed); + let (disk_scan_concurrency_limit, disk_bucket_scans_queued, disk_bucket_scans_active) = + match self.scanner_disk_bucket_scan_states.lock() { + Ok(states) => states.values().fold((0, 0, 0), |acc, state| { + (acc.0 + state.concurrency_limit, acc.1 + state.queued, acc.2 + state.active) + }), + Err(poisoned) => poisoned.into_inner().values().fold((0, 0, 0), |acc, state| { + (acc.0 + state.concurrency_limit, acc.1 + state.queued, acc.2 + state.active) + }), + }; + m.current_disk_scan_concurrency_limit = disk_scan_concurrency_limit; + m.current_disk_bucket_scans_queued = disk_bucket_scans_queued; + m.current_disk_bucket_scans_active = disk_bucket_scans_active; if self.current_scan_cycle_work_active.load(Ordering::Relaxed) { let current_work = self.scan_cycle_work_since(self.current_scan_cycle_work_start()); m.current_cycle_objects_scanned = current_work.objects_scanned; @@ -1097,6 +1295,9 @@ impl Metrics { let last_cycle_result = self.last_scan_cycle_result.load(Ordering::Relaxed); m.last_cycle_result = scan_cycle_result_label(last_cycle_result).to_string(); m.last_cycle_result_code = last_cycle_result as u64; + let partial_reason = ScanCyclePartialReason::from_code(self.last_scan_cycle_partial_reason.load(Ordering::Relaxed)); + m.last_cycle_partial_reason = partial_reason.as_str().to_string(); + m.last_cycle_partial_reason_code = partial_reason as u8 as u64; m.last_cycle_duration_seconds = self.last_scan_cycle_duration_millis.load(Ordering::Relaxed) as f64 / 1000.0; m.last_cycle_objects_scanned = self.last_scan_cycle_objects_scanned.load(Ordering::Relaxed); m.last_cycle_directories_scanned = self.last_scan_cycle_directories_scanned.load(Ordering::Relaxed); @@ -1112,12 +1313,18 @@ impl Metrics { 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); m.failed_cycles = self.failed_scan_cycles.load(Ordering::Relaxed); + m.partial_cycles_unknown = self.partial_scan_cycles_unknown.load(Ordering::Relaxed); + m.partial_cycles_runtime = self.partial_scan_cycles_runtime.load(Ordering::Relaxed); + m.partial_cycles_objects = self.partial_scan_cycles_objects.load(Ordering::Relaxed); + m.partial_cycles_directories = self.partial_scan_cycles_directories.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; 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.cycle_max_objects = self.scanner_cycle_max_objects.load(Ordering::Relaxed); + m.cycle_max_directories = self.scanner_cycle_max_directories.load(Ordering::Relaxed); 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); @@ -1264,19 +1471,74 @@ mod tests { #[tokio::test] async fn report_counts_active_scan_paths() { let metrics = Metrics::new(); - metrics - .current_paths - .write() - .await - .insert("disk-a".to_string(), Arc::new(CurrentPathTracker::new("bucket-a".to_string()))); + let updated_at = Utc::now() - chrono::Duration::seconds(12); + metrics.current_paths.write().await.insert( + "disk-a".to_string(), + Arc::new(CurrentPathTracker::new_at("bucket-a".to_string(), updated_at)), + ); let report = metrics.report().await; assert_eq!(report.active_scan_paths, 1); + assert!( + report.oldest_active_path_age_seconds >= 12, + "expected active path age to reflect last path update" + ); assert_eq!(report.ongoing_buckets, 0); assert_eq!(report.active_paths, vec!["disk-a/bucket-a".to_string()]); } + #[tokio::test] + async fn report_active_path_age_uses_last_path_update() { + let metrics = Metrics::new(); + let tracker = Arc::new(CurrentPathTracker::new_at( + "bucket-a".to_string(), + Utc::now() - chrono::Duration::hours(1), + )); + metrics + .current_paths + .write() + .await + .insert("disk-a".to_string(), Arc::clone(&tracker)); + + tracker.update_path("bucket-a/prefix".to_string()).await; + + let report = metrics.report().await; + + assert_eq!(report.active_paths, vec!["disk-a/bucket-a/prefix".to_string()]); + assert!( + report.oldest_active_path_age_seconds < 5, + "expected active path age to reset after path update" + ); + } + + #[tokio::test] + async fn report_includes_scanner_queue_state() { + let metrics = Metrics::new(); + metrics.record_scanner_set_scan_state(Some(3), Some(5), Some(2)); + metrics.record_scanner_disk_bucket_scan_state("0", "0", Some(2), Some(7), Some(1)); + metrics.record_scanner_disk_bucket_scan_state("0", "1", Some(4), Some(11), Some(3)); + + let report = metrics.report().await; + + assert_eq!(report.current_set_scan_concurrency_limit, 3); + assert_eq!(report.current_set_scans_queued, 5); + assert_eq!(report.current_set_scans_active, 2); + assert_eq!(report.current_disk_scan_concurrency_limit, 6); + assert_eq!(report.current_disk_bucket_scans_queued, 18); + assert_eq!(report.current_disk_bucket_scans_active, 4); + + metrics.reset_scanner_set_scan_state(); + let report = metrics.report().await; + + assert_eq!(report.current_set_scan_concurrency_limit, 0); + assert_eq!(report.current_set_scans_queued, 0); + assert_eq!(report.current_set_scans_active, 0); + assert_eq!(report.current_disk_scan_concurrency_limit, 0); + assert_eq!(report.current_disk_bucket_scans_queued, 0); + assert_eq!(report.current_disk_bucket_scans_active, 0); + } + #[tokio::test] async fn report_preserves_current_cycle_started_time() { let previous_init_time = *crate::globals::GLOBAL_INIT_TIME.read().await; @@ -1325,28 +1587,38 @@ mod tests { #[tokio::test] async fn report_tracks_successful_scan_cycle_without_failed_increment() { let metrics = Metrics::new(); + metrics.record_scan_cycle_partial(Duration::from_millis(500), ScanCyclePartialReason::Runtime); metrics.record_scan_cycle_complete(true, Duration::from_secs(2)); let report = metrics.report().await; assert_eq!(report.last_cycle_result, SCAN_CYCLE_RESULT_SUCCESS_LABEL); assert_eq!(report.last_cycle_result_code, SCAN_CYCLE_RESULT_SUCCESS as u64); + assert_eq!(report.last_cycle_partial_reason, SCAN_CYCLE_RESULT_UNKNOWN_LABEL); + assert_eq!(report.last_cycle_partial_reason_code, ScanCyclePartialReason::Unknown as u8 as u64); assert_eq!(report.last_cycle_duration_seconds, 2.0); assert_eq!(report.failed_cycles, 0); + assert_eq!(report.partial_cycles, 1); } #[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)); + metrics.record_scan_cycle_partial(Duration::from_millis(2500), ScanCyclePartialReason::Directories); 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_partial_reason, "directories"); + assert_eq!(report.last_cycle_partial_reason_code, ScanCyclePartialReason::Directories as u8 as u64); assert_eq!(report.last_cycle_duration_seconds, 2.5); assert_eq!(report.failed_cycles, 0); assert_eq!(report.partial_cycles, 1); + assert_eq!(report.partial_cycles_directories, 1); + assert_eq!(report.partial_cycles_runtime, 0); + assert_eq!(report.partial_cycles_objects, 0); + assert_eq!(report.partial_cycles_unknown, 0); } #[tokio::test] @@ -1550,19 +1822,25 @@ mod tests { Duration::from_secs(3600), Some(Duration::from_secs(86400)), Some(Duration::from_secs(1800)), + Some(1_000_000), + Some(100_000), ); let report = metrics.report().await; assert_eq!(report.cycle_interval_seconds, 3600.0); assert_eq!(report.cycle_max_duration_seconds, 1800.0); + assert_eq!(report.cycle_max_objects, 1_000_000); + assert_eq!(report.cycle_max_directories, 100_000); assert!(report.bitrot_cycle_enabled); assert_eq!(report.bitrot_cycle_seconds, 86400.0); - metrics.record_scanner_cycle_config(Duration::from_secs(60), None, None); + metrics.record_scanner_cycle_config(Duration::from_secs(60), None, None, 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_eq!(report.cycle_max_objects, 0); + assert_eq!(report.cycle_max_directories, 0); assert!(!report.bitrot_cycle_enabled); assert_eq!(report.bitrot_cycle_seconds, 0.0); } diff --git a/crates/config/README.md b/crates/config/README.md index bb921c2a5..c1e0c7829 100644 --- a/crates/config/README.md +++ b/crates/config/README.md @@ -70,6 +70,9 @@ Current guidance: - `RUSTFS_DATA_SCANNER_START_DELAY_SECS` (deprecated alias for compatibility) - `RUSTFS_SCANNER_IDLE_MODE` (canonical) - `RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS` (canonical) +- `RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS` (canonical) +- `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS` (canonical) +- `RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES` (canonical) ## Health compatibility switches diff --git a/crates/config/src/constants/scanner.rs b/crates/config/src/constants/scanner.rs index b0b81b4fb..f43f63ae4 100644 --- a/crates/config/src/constants/scanner.rs +++ b/crates/config/src/constants/scanner.rs @@ -37,6 +37,16 @@ pub const ENV_SCANNER_CYCLE: &str = "RUSTFS_SCANNER_CYCLE"; /// - 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 caps objects processed by one scanner cycle. +/// A value of `0` disables the per-cycle object budget. +/// - Example: `export RUSTFS_SCANNER_CYCLE_MAX_OBJECTS=1000000` +pub const ENV_SCANNER_CYCLE_MAX_OBJECTS: &str = "RUSTFS_SCANNER_CYCLE_MAX_OBJECTS"; + +/// Environment variable that caps directories entered by one scanner cycle. +/// A value of `0` disables the per-cycle directory budget. +/// - Example: `export RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES=100000` +pub const ENV_SCANNER_CYCLE_MAX_DIRECTORIES: &str = "RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES"; + /// 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. @@ -50,6 +60,14 @@ pub const DEFAULT_SCANNER_SPEED: &str = "default"; /// `0` keeps the existing unbounded per-cycle behavior. pub const DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS: u64 = 0; +/// Default scanner per-cycle object budget. +/// `0` keeps the existing unbounded per-cycle behavior. +pub const DEFAULT_SCANNER_CYCLE_MAX_OBJECTS: u64 = 0; + +/// Default scanner per-cycle directory budget. +/// `0` keeps the existing unbounded per-cycle behavior. +pub const DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES: 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 d3a93c9b8..f8e2abc55 100644 --- a/crates/obs/src/metrics/collectors/scanner.rs +++ b/crates/obs/src/metrics/collectors/scanner.rs @@ -30,18 +30,22 @@ use crate::metrics::schema::scanner::{ SCANNER_CURRENT_CYCLE_OBJECTS_PER_SECOND_MD, SCANNER_CURRENT_CYCLE_OBJECTS_SCANNED_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_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, + SCANNER_CURRENT_CYCLE_YIELD_DURATION_SECONDS_MD, SCANNER_CURRENT_CYCLE_YIELD_EVENTS_MD, + SCANNER_CURRENT_DISK_BUCKET_SCANS_ACTIVE_MD, SCANNER_CURRENT_DISK_BUCKET_SCANS_QUEUED_MD, + SCANNER_CURRENT_DISK_SCAN_CONCURRENCY_LIMIT_MD, SCANNER_CURRENT_SCAN_MODE_MD, SCANNER_CURRENT_SET_SCAN_CONCURRENCY_LIMIT_MD, + SCANNER_CURRENT_SET_SCANS_ACTIVE_MD, SCANNER_CURRENT_SET_SCANS_QUEUED_MD, SCANNER_CYCLE_INTERVAL_SECONDS_MD, + SCANNER_CYCLE_MAX_DIRECTORIES_MD, SCANNER_CYCLE_MAX_DURATION_SECONDS_MD, SCANNER_CYCLE_MAX_OBJECTS_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_PARTIAL_REASON_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_OLDEST_ACTIVE_PATH_AGE_SECONDS_MD, SCANNER_PARTIAL_CYCLES_BY_REASON_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, }; /// Scanner statistics. @@ -63,6 +67,20 @@ pub struct ScannerStats { pub last_activity_seconds: u64, /// Number of scanner paths currently being processed pub active_paths: u64, + /// Age in seconds of the oldest active scanner path update + pub oldest_active_path_age_seconds: u64, + /// Current aggregate set scan concurrency limit + pub current_set_scan_concurrency_limit: u64, + /// Current number of queued set scans + pub current_set_scans_queued: u64, + /// Current number of active set scans + pub current_set_scans_active: u64, + /// Current aggregate disk-bucket scan concurrency limit + pub current_disk_scan_concurrency_limit: u64, + /// Current number of queued disk-bucket scans + pub current_disk_bucket_scans_queued: u64, + /// Current number of active disk-bucket scans + pub current_disk_bucket_scans_active: u64, /// Whether scanner idle-mode self-throttling is enabled pub throttle_idle_mode_enabled: bool, /// Effective scanner sleep factor @@ -75,6 +93,10 @@ pub struct ScannerStats { pub cycle_interval_seconds: f64, /// Effective maximum scanner cycle runtime in seconds pub cycle_max_duration_seconds: f64, + /// Effective maximum objects processed by one scanner cycle + pub cycle_max_objects: u64, + /// Effective maximum directories entered by one scanner cycle + pub cycle_max_directories: u64, /// Whether periodic scanner bitrot deep scans are enabled pub bitrot_cycle_enabled: bool, /// Effective scanner bitrot deep-scan interval in seconds @@ -119,6 +141,8 @@ pub struct ScannerStats { pub current_scan_mode: u64, /// Last scanner cycle result: 0 unknown, 1 success, 2 error, 3 partial pub last_cycle_result: u64, + /// Last scanner partial cycle reason: 0 unknown, 1 runtime, 2 objects, 3 directories + pub last_cycle_partial_reason: u64, /// Duration in seconds of the last finished scanner cycle pub last_cycle_duration_seconds: f64, /// Number of objects scanned by the last finished scanner cycle @@ -155,6 +179,14 @@ pub struct ScannerStats { pub failed_cycles: u64, /// Number of scanner cycles stopped by runtime budget since server start pub partial_cycles: u64, + /// Number of scanner cycles stopped by an unknown budget reason + pub partial_cycles_unknown: u64, + /// Number of scanner cycles stopped by runtime budget + pub partial_cycles_runtime: u64, + /// Number of scanner cycles stopped by object budget + pub partial_cycles_objects: u64, + /// Number of scanner cycles stopped by directory budget + pub partial_cycles_directories: u64, } /// Collects scanner metrics from the given stats. @@ -171,6 +203,28 @@ 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_OLDEST_ACTIVE_PATH_AGE_SECONDS_MD, + stats.oldest_active_path_age_seconds as f64, + ), + PrometheusMetric::from_descriptor( + &SCANNER_CURRENT_SET_SCAN_CONCURRENCY_LIMIT_MD, + stats.current_set_scan_concurrency_limit as f64, + ), + PrometheusMetric::from_descriptor(&SCANNER_CURRENT_SET_SCANS_QUEUED_MD, stats.current_set_scans_queued as f64), + PrometheusMetric::from_descriptor(&SCANNER_CURRENT_SET_SCANS_ACTIVE_MD, stats.current_set_scans_active as f64), + PrometheusMetric::from_descriptor( + &SCANNER_CURRENT_DISK_SCAN_CONCURRENCY_LIMIT_MD, + stats.current_disk_scan_concurrency_limit as f64, + ), + PrometheusMetric::from_descriptor( + &SCANNER_CURRENT_DISK_BUCKET_SCANS_QUEUED_MD, + stats.current_disk_bucket_scans_queued as f64, + ), + PrometheusMetric::from_descriptor( + &SCANNER_CURRENT_DISK_BUCKET_SCANS_ACTIVE_MD, + stats.current_disk_bucket_scans_active as f64, + ), PrometheusMetric::from_descriptor( &SCANNER_THROTTLE_IDLE_MODE_ENABLED_MD, bool_metric_value(stats.throttle_idle_mode_enabled), @@ -180,6 +234,8 @@ pub fn collect_scanner_metrics(stats: &ScannerStats) -> Vec { 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_CYCLE_MAX_DURATION_SECONDS_MD, stats.cycle_max_duration_seconds), + PrometheusMetric::from_descriptor(&SCANNER_CYCLE_MAX_OBJECTS_MD, stats.cycle_max_objects as f64), + PrometheusMetric::from_descriptor(&SCANNER_CYCLE_MAX_DIRECTORIES_MD, stats.cycle_max_directories as f64), 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), @@ -229,6 +285,7 @@ pub fn collect_scanner_metrics(stats: &ScannerStats) -> Vec { PrometheusMetric::from_descriptor(&SCANNER_CURRENT_CYCLE_USAGE_SAVES_MD, stats.current_cycle_usage_saves as f64), PrometheusMetric::from_descriptor(&SCANNER_CURRENT_SCAN_MODE_MD, stats.current_scan_mode as f64), PrometheusMetric::from_descriptor(&SCANNER_LAST_CYCLE_RESULT_MD, stats.last_cycle_result as f64), + PrometheusMetric::from_descriptor(&SCANNER_LAST_CYCLE_PARTIAL_REASON_MD, stats.last_cycle_partial_reason as f64), PrometheusMetric::from_descriptor(&SCANNER_LAST_CYCLE_DURATION_SECONDS_MD, stats.last_cycle_duration_seconds), PrometheusMetric::from_descriptor(&SCANNER_LAST_CYCLE_OBJECTS_SCANNED_MD, stats.last_cycle_objects_scanned as f64), PrometheusMetric::from_descriptor( @@ -262,6 +319,14 @@ pub fn collect_scanner_metrics(stats: &ScannerStats) -> Vec { 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), + PrometheusMetric::from_descriptor(&SCANNER_PARTIAL_CYCLES_BY_REASON_MD, stats.partial_cycles_unknown as f64) + .with_label("reason", "unknown"), + PrometheusMetric::from_descriptor(&SCANNER_PARTIAL_CYCLES_BY_REASON_MD, stats.partial_cycles_runtime as f64) + .with_label("reason", "runtime"), + PrometheusMetric::from_descriptor(&SCANNER_PARTIAL_CYCLES_BY_REASON_MD, stats.partial_cycles_objects as f64) + .with_label("reason", "objects"), + PrometheusMetric::from_descriptor(&SCANNER_PARTIAL_CYCLES_BY_REASON_MD, stats.partial_cycles_directories as f64) + .with_label("reason", "directories"), ] } @@ -285,12 +350,21 @@ mod tests { versions_scanned: 1500000, last_activity_seconds: 30, active_paths: 4, + oldest_active_path_age_seconds: 17, + current_set_scan_concurrency_limit: 3, + current_set_scans_queued: 5, + current_set_scans_active: 2, + current_disk_scan_concurrency_limit: 6, + current_disk_bucket_scans_queued: 18, + current_disk_bucket_scans_active: 4, 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, cycle_max_duration_seconds: 1800.0, + cycle_max_objects: 1_000_000, + cycle_max_directories: 100_000, bitrot_cycle_enabled: true, bitrot_cycle_seconds: 86400.0, current_cycle: 12, @@ -313,6 +387,7 @@ mod tests { current_cycle_usage_saves: 3, current_scan_mode: 2, last_cycle_result: 1, + last_cycle_partial_reason: 3, last_cycle_duration_seconds: 42.5, last_cycle_objects_scanned: 900, last_cycle_directories_scanned: 80, @@ -330,13 +405,17 @@ mod tests { last_cycle_replication_checks: 12, last_cycle_usage_saves: 9, failed_cycles: 3, - partial_cycles: 2, + partial_cycles: 10, + partial_cycles_unknown: 1, + partial_cycles_runtime: 2, + partial_cycles_objects: 3, + partial_cycles_directories: 4, }; let metrics = collect_scanner_metrics(&stats); report_metrics(&metrics); - assert_eq!(metrics.len(), 54); + assert_eq!(metrics.len(), 68); let objects = metrics.iter().find(|m| m.value == 1000000.0); assert!(objects.is_some()); @@ -349,6 +428,41 @@ mod tests { .find(|m| m.name == SCANNER_ACTIVE_PATHS_MD.get_full_metric_name()); assert_eq!(active_paths.map(|m| m.value), Some(4.0)); + let oldest_active_path_age = metrics + .iter() + .find(|m| m.name == SCANNER_OLDEST_ACTIVE_PATH_AGE_SECONDS_MD.get_full_metric_name()); + assert_eq!(oldest_active_path_age.map(|m| m.value), Some(17.0)); + + let current_set_scan_concurrency_limit = metrics + .iter() + .find(|m| m.name == SCANNER_CURRENT_SET_SCAN_CONCURRENCY_LIMIT_MD.get_full_metric_name()); + assert_eq!(current_set_scan_concurrency_limit.map(|m| m.value), Some(3.0)); + + let current_set_scans_queued = metrics + .iter() + .find(|m| m.name == SCANNER_CURRENT_SET_SCANS_QUEUED_MD.get_full_metric_name()); + assert_eq!(current_set_scans_queued.map(|m| m.value), Some(5.0)); + + let current_set_scans_active = metrics + .iter() + .find(|m| m.name == SCANNER_CURRENT_SET_SCANS_ACTIVE_MD.get_full_metric_name()); + assert_eq!(current_set_scans_active.map(|m| m.value), Some(2.0)); + + let current_disk_scan_concurrency_limit = metrics + .iter() + .find(|m| m.name == SCANNER_CURRENT_DISK_SCAN_CONCURRENCY_LIMIT_MD.get_full_metric_name()); + assert_eq!(current_disk_scan_concurrency_limit.map(|m| m.value), Some(6.0)); + + let current_disk_bucket_scans_queued = metrics + .iter() + .find(|m| m.name == SCANNER_CURRENT_DISK_BUCKET_SCANS_QUEUED_MD.get_full_metric_name()); + assert_eq!(current_disk_bucket_scans_queued.map(|m| m.value), Some(18.0)); + + let current_disk_bucket_scans_active = metrics + .iter() + .find(|m| m.name == SCANNER_CURRENT_DISK_BUCKET_SCANS_ACTIVE_MD.get_full_metric_name()); + assert_eq!(current_disk_bucket_scans_active.map(|m| m.value), Some(4.0)); + let bucket_scans_failed = metrics .iter() .find(|m| m.name == SCANNER_BUCKET_SCANS_FAILED_MD.get_full_metric_name()); @@ -384,6 +498,16 @@ mod tests { .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 cycle_max_objects = metrics + .iter() + .find(|m| m.name == SCANNER_CYCLE_MAX_OBJECTS_MD.get_full_metric_name()); + assert_eq!(cycle_max_objects.map(|m| m.value), Some(1_000_000.0)); + + let cycle_max_directories = metrics + .iter() + .find(|m| m.name == SCANNER_CYCLE_MAX_DIRECTORIES_MD.get_full_metric_name()); + assert_eq!(cycle_max_directories.map(|m| m.value), Some(100_000.0)); + let bitrot_cycle_enabled = metrics .iter() .find(|m| m.name == SCANNER_BITROT_CYCLE_ENABLED_MD.get_full_metric_name()); @@ -494,6 +618,11 @@ mod tests { .find(|m| m.name == SCANNER_LAST_CYCLE_RESULT_MD.get_full_metric_name()); assert_eq!(last_cycle_result.map(|m| m.value), Some(1.0)); + let last_cycle_partial_reason = metrics + .iter() + .find(|m| m.name == SCANNER_LAST_CYCLE_PARTIAL_REASON_MD.get_full_metric_name()); + assert_eq!(last_cycle_partial_reason.map(|m| m.value), Some(3.0)); + let last_cycle_duration = metrics .iter() .find(|m| m.name == SCANNER_LAST_CYCLE_DURATION_SECONDS_MD.get_full_metric_name()); @@ -582,7 +711,31 @@ mod tests { 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)); + assert_eq!(partial_cycles.map(|m| m.value), Some(10.0)); + + let partial_cycles_runtime = metrics.iter().find(|m| { + m.name == SCANNER_PARTIAL_CYCLES_BY_REASON_MD.get_full_metric_name() + && m.labels + .iter() + .any(|(name, value)| *name == "reason" && value.as_ref() == "runtime") + }); + assert_eq!(partial_cycles_runtime.map(|m| m.value), Some(2.0)); + + let partial_cycles_objects = metrics.iter().find(|m| { + m.name == SCANNER_PARTIAL_CYCLES_BY_REASON_MD.get_full_metric_name() + && m.labels + .iter() + .any(|(name, value)| *name == "reason" && value.as_ref() == "objects") + }); + assert_eq!(partial_cycles_objects.map(|m| m.value), Some(3.0)); + + let partial_cycles_directories = metrics.iter().find(|m| { + m.name == SCANNER_PARTIAL_CYCLES_BY_REASON_MD.get_full_metric_name() + && m.labels + .iter() + .any(|(name, value)| *name == "reason" && value.as_ref() == "directories") + }); + assert_eq!(partial_cycles_directories.map(|m| m.value), Some(4.0)); } #[test] @@ -590,10 +743,16 @@ mod tests { let stats = ScannerStats::default(); let metrics = collect_scanner_metrics(&stats); - assert_eq!(metrics.len(), 54); + assert_eq!(metrics.len(), 68); for metric in &metrics { assert_eq!(metric.value, 0.0); - assert!(metric.labels.is_empty()); + if metric.name == SCANNER_PARTIAL_CYCLES_BY_REASON_MD.get_full_metric_name() { + assert_eq!(metric.labels.len(), 1); + assert_eq!(metric.labels[0].0, "reason"); + assert!(matches!(metric.labels[0].1.as_ref(), "unknown" | "runtime" | "objects" | "directories")); + } else { + 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 6704e8555..da02a9090 100644 --- a/crates/obs/src/metrics/schema/entry/metric_name.rs +++ b/crates/obs/src/metrics/schema/entry/metric_name.rs @@ -279,6 +279,13 @@ pub enum MetricName { ScannerVersionsScanned, ScannerLastActivitySeconds, ScannerActivePaths, + ScannerOldestActivePathAgeSeconds, + ScannerCurrentSetScanConcurrencyLimit, + ScannerCurrentSetScansQueued, + ScannerCurrentSetScansActive, + ScannerCurrentDiskScanConcurrencyLimit, + ScannerCurrentDiskBucketScansQueued, + ScannerCurrentDiskBucketScansActive, ScannerBucketScansFailed, ScannerThrottleIdleModeEnabled, ScannerThrottleSleepFactor, @@ -286,6 +293,8 @@ pub enum MetricName { ScannerYieldEveryNObjects, ScannerCycleIntervalSeconds, ScannerCycleMaxDurationSeconds, + ScannerCycleMaxObjects, + ScannerCycleMaxDirectories, ScannerBitrotCycleEnabled, ScannerBitrotCycleSeconds, ScannerCurrentCycle, @@ -308,6 +317,7 @@ pub enum MetricName { ScannerCurrentCycleUsageSaves, ScannerCurrentScanMode, ScannerLastCycleResult, + ScannerLastCyclePartialReason, ScannerLastCycleDurationSeconds, ScannerLastCycleObjectsScanned, ScannerLastCycleDirectoriesScanned, @@ -326,6 +336,7 @@ pub enum MetricName { ScannerLastCycleUsageSaves, ScannerFailedCycles, ScannerPartialCycles, + ScannerPartialCyclesByReason, // CPU system-related metrics SysCPUAvgIdle, @@ -667,6 +678,13 @@ impl MetricName { Self::ScannerVersionsScanned => "versions_scanned".to_string(), Self::ScannerLastActivitySeconds => "last_activity_seconds".to_string(), Self::ScannerActivePaths => "active_paths".to_string(), + Self::ScannerOldestActivePathAgeSeconds => "oldest_active_path_age_seconds".to_string(), + Self::ScannerCurrentSetScanConcurrencyLimit => "current_set_scan_concurrency_limit".to_string(), + Self::ScannerCurrentSetScansQueued => "current_set_scans_queued".to_string(), + Self::ScannerCurrentSetScansActive => "current_set_scans_active".to_string(), + Self::ScannerCurrentDiskScanConcurrencyLimit => "current_disk_scan_concurrency_limit".to_string(), + Self::ScannerCurrentDiskBucketScansQueued => "current_disk_bucket_scans_queued".to_string(), + Self::ScannerCurrentDiskBucketScansActive => "current_disk_bucket_scans_active".to_string(), Self::ScannerBucketScansFailed => "bucket_scans_failed".to_string(), Self::ScannerThrottleIdleModeEnabled => "throttle_idle_mode_enabled".to_string(), Self::ScannerThrottleSleepFactor => "throttle_sleep_factor".to_string(), @@ -674,6 +692,8 @@ impl MetricName { Self::ScannerYieldEveryNObjects => "yield_every_n_objects".to_string(), Self::ScannerCycleIntervalSeconds => "cycle_interval_seconds".to_string(), Self::ScannerCycleMaxDurationSeconds => "cycle_max_duration_seconds".to_string(), + Self::ScannerCycleMaxObjects => "cycle_max_objects".to_string(), + Self::ScannerCycleMaxDirectories => "cycle_max_directories".to_string(), Self::ScannerBitrotCycleEnabled => "bitrot_cycle_enabled".to_string(), Self::ScannerBitrotCycleSeconds => "bitrot_cycle_seconds".to_string(), Self::ScannerCurrentCycle => "current_cycle".to_string(), @@ -696,6 +716,7 @@ impl MetricName { Self::ScannerCurrentCycleUsageSaves => "current_cycle_usage_saves".to_string(), Self::ScannerCurrentScanMode => "current_scan_mode".to_string(), Self::ScannerLastCycleResult => "last_cycle_result".to_string(), + Self::ScannerLastCyclePartialReason => "last_cycle_partial_reason".to_string(), Self::ScannerLastCycleDurationSeconds => "last_cycle_duration_seconds".to_string(), Self::ScannerLastCycleObjectsScanned => "last_cycle_objects_scanned".to_string(), Self::ScannerLastCycleDirectoriesScanned => "last_cycle_directories_scanned".to_string(), @@ -714,6 +735,7 @@ impl MetricName { Self::ScannerLastCycleUsageSaves => "last_cycle_usage_saves".to_string(), Self::ScannerFailedCycles => "failed_cycles".to_string(), Self::ScannerPartialCycles => "partial_cycles".to_string(), + Self::ScannerPartialCyclesByReason => "partial_cycles_by_reason".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 cd043a31a..7f3ac768a 100644 --- a/crates/obs/src/metrics/schema/scanner.rs +++ b/crates/obs/src/metrics/schema/scanner.rs @@ -89,6 +89,69 @@ pub static SCANNER_ACTIVE_PATHS_MD: LazyLock = LazyLock::new(| ) }); +pub static SCANNER_OLDEST_ACTIVE_PATH_AGE_SECONDS_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerOldestActivePathAgeSeconds, + "Time elapsed (in seconds) since the oldest active scanner path was last updated, or zero when idle.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_CURRENT_SET_SCAN_CONCURRENCY_LIMIT_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCurrentSetScanConcurrencyLimit, + "Current aggregate scanner set scan concurrency limit across active scanner work.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_CURRENT_SET_SCANS_QUEUED_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCurrentSetScansQueued, + "Current number of queued scanner set scans across active scanner work.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_CURRENT_SET_SCANS_ACTIVE_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCurrentSetScansActive, + "Current number of active scanner set scans across active scanner work.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_CURRENT_DISK_SCAN_CONCURRENCY_LIMIT_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCurrentDiskScanConcurrencyLimit, + "Current aggregate scanner disk-bucket scan concurrency limit across active scanner work.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_CURRENT_DISK_BUCKET_SCANS_QUEUED_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCurrentDiskBucketScansQueued, + "Current number of queued scanner disk-bucket scans across active scanner work.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_CURRENT_DISK_BUCKET_SCANS_ACTIVE_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCurrentDiskBucketScansActive, + "Current number of active scanner disk-bucket scans across active scanner work.", + &[], + subsystems::SCANNER, + ) +}); + pub static SCANNER_THROTTLE_IDLE_MODE_ENABLED_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::ScannerThrottleIdleModeEnabled, @@ -143,6 +206,24 @@ pub static SCANNER_CYCLE_MAX_DURATION_SECONDS_MD: LazyLock = L ) }); +pub static SCANNER_CYCLE_MAX_OBJECTS_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCycleMaxObjects, + "Effective maximum objects processed by one scanner cycle; zero means unlimited.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_CYCLE_MAX_DIRECTORIES_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCycleMaxDirectories, + "Effective maximum directories entered by one scanner cycle; zero means unlimited.", + &[], + subsystems::SCANNER, + ) +}); + pub static SCANNER_BITROT_CYCLE_ENABLED_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::ScannerBitrotCycleEnabled, @@ -341,6 +422,15 @@ pub static SCANNER_LAST_CYCLE_RESULT_MD: LazyLock = LazyLock:: ) }); +pub static SCANNER_LAST_CYCLE_PARTIAL_REASON_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerLastCyclePartialReason, + "Last scanner partial cycle reason: 0 unknown, 1 runtime budget, 2 object budget, 3 directory budget.", + &[], + subsystems::SCANNER, + ) +}); + pub static SCANNER_LAST_CYCLE_DURATION_SECONDS_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::ScannerLastCycleDurationSeconds, @@ -497,8 +587,17 @@ pub static SCANNER_FAILED_CYCLES_MD: LazyLock = LazyLock::new( 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.", + "Total number of scanner cycles stopped before completion by cycle budget.", &[], subsystems::SCANNER, ) }); + +pub static SCANNER_PARTIAL_CYCLES_BY_REASON_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::ScannerPartialCyclesByReason, + "Total number of scanner cycles stopped before completion by cycle budget reason.", + &["reason"], + subsystems::SCANNER, + ) +}); diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 390c7c310..e94a5fa78 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -934,12 +934,21 @@ pub async fn collect_scanner_metric_stats() -> Option { versions_scanned, last_activity_seconds, active_paths, + oldest_active_path_age_seconds: metrics.oldest_active_path_age_seconds, + current_set_scan_concurrency_limit: metrics.current_set_scan_concurrency_limit, + current_set_scans_queued: metrics.current_set_scans_queued, + current_set_scans_active: metrics.current_set_scans_active, + current_disk_scan_concurrency_limit: metrics.current_disk_scan_concurrency_limit, + current_disk_bucket_scans_queued: metrics.current_disk_bucket_scans_queued, + current_disk_bucket_scans_active: metrics.current_disk_bucket_scans_active, 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, cycle_max_duration_seconds: metrics.cycle_max_duration_seconds, + cycle_max_objects: metrics.cycle_max_objects, + cycle_max_directories: metrics.cycle_max_directories, bitrot_cycle_enabled: metrics.bitrot_cycle_enabled, bitrot_cycle_seconds: metrics.bitrot_cycle_seconds, current_cycle: metrics.current_cycle, @@ -968,6 +977,7 @@ pub async fn collect_scanner_metric_stats() -> Option { current_cycle_usage_saves: metrics.current_cycle_usage_saves, current_scan_mode, last_cycle_result: metrics.last_cycle_result_code, + last_cycle_partial_reason: metrics.last_cycle_partial_reason_code, last_cycle_duration_seconds: metrics.last_cycle_duration_seconds, last_cycle_objects_scanned: metrics.last_cycle_objects_scanned, last_cycle_directories_scanned: metrics.last_cycle_directories_scanned, @@ -992,6 +1002,10 @@ pub async fn collect_scanner_metric_stats() -> Option { last_cycle_usage_saves: metrics.last_cycle_usage_saves, failed_cycles: metrics.failed_cycles, partial_cycles: metrics.partial_cycles, + partial_cycles_unknown: metrics.partial_cycles_unknown, + partial_cycles_runtime: metrics.partial_cycles_runtime, + partial_cycles_objects: metrics.partial_cycles_objects, + partial_cycles_directories: metrics.partial_cycles_directories, }) } diff --git a/crates/scanner/src/error.rs b/crates/scanner/src/error.rs index adf066391..cbd628c78 100644 --- a/crates/scanner/src/error.rs +++ b/crates/scanner/src/error.rs @@ -14,6 +14,8 @@ use thiserror::Error; +use crate::data_usage_define::DataUsageCache; + /// Scanner-related errors #[derive(Error, Debug)] #[non_exhaustive] @@ -33,4 +35,8 @@ pub enum ScannerError { /// Generic error #[error("Scanner error: {0}")] Other(String), + + /// Partial data usage cache produced before the scanner stopped. + #[error("Scanner stopped with partial data usage cache")] + PartialCache(Box), } diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 57bf92180..cd71b0a52 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -23,6 +23,7 @@ pub mod data_usage_define; pub mod error; pub mod scanner; +pub mod scanner_budget; pub mod scanner_folder; pub mod scanner_io; pub mod sleeper; diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 17eb8c914..99b99a11e 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -14,21 +14,25 @@ use std::sync::{ Arc, - atomic::{AtomicBool, AtomicU64, Ordering}, + atomic::{AtomicU64, Ordering}, }; use crate::data_usage_define::{BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH}; +use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetConfig, ScannerCycleBudgetReason}; use crate::scanner_folder::{data_usage_update_dir_cycles, heal_object_select_prob}; use crate::scanner_io::ScannerIO; use crate::sleeper::{SCANNER_SLEEPER, scanner_speed_from_env_or_default, set_scanner_default_speed}; 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, emit_scan_cycle_partial, global_metrics}; +use rustfs_common::metrics::{ + CurrentCycle, Metric, Metrics, ScanCyclePartialReason, emit_scan_cycle_complete, emit_scan_cycle_partial, global_metrics, +}; use rustfs_config::ScannerSpeed; use rustfs_config::{ - 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, + DEFAULT_SCANNER_BITROT_CYCLE_SECS, DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS, + DEFAULT_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DIRECTORIES, + ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS, }; use rustfs_ecstore::StorageAPI as _; use rustfs_ecstore::bucket::lifecycle::lifecycle::Lifecycle as _; @@ -94,6 +98,30 @@ fn scanner_cycle_max_duration() -> Option { } } +fn scanner_cycle_count_budget(env: &str, default: u64) -> Option { + match rustfs_utils::get_env_u64(env, default) { + 0 => None, + count => Some(count), + } +} + +fn scanner_cycle_budget_config() -> ScannerCycleBudgetConfig { + ScannerCycleBudgetConfig { + max_duration: scanner_cycle_max_duration(), + max_objects: scanner_cycle_count_budget(ENV_SCANNER_CYCLE_MAX_OBJECTS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS), + max_directories: scanner_cycle_count_budget(ENV_SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES), + } +} + +fn scan_cycle_partial_reason(reason: Option) -> ScanCyclePartialReason { + match reason { + Some(ScannerCycleBudgetReason::Runtime) => ScanCyclePartialReason::Runtime, + Some(ScannerCycleBudgetReason::Objects) => ScanCyclePartialReason::Objects, + Some(ScannerCycleBudgetReason::Directories) => ScanCyclePartialReason::Directories, + None => ScanCyclePartialReason::Unknown, + } +} + /// Compute a randomized inter-cycle sleep. // Delay is scan interval +- 10%, with a floor of 1 second. fn randomized_cycle_delay() -> Duration { @@ -439,60 +467,6 @@ 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(); @@ -505,11 +479,13 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc SCANNER_SLEEPER.refresh_from_env(); let configured_cycle_interval = cycle_interval(); let configured_bitrot_cycle = bitrot_scan_cycle(); - let configured_cycle_max_duration = scanner_cycle_max_duration(); + let cycle_budget_config = scanner_cycle_budget_config(); global_metrics().record_scanner_cycle_config( configured_cycle_interval, configured_bitrot_cycle, - configured_cycle_max_duration, + cycle_budget_config.max_duration, + cycle_budget_config.max_objects, + cycle_budget_config.max_directories, ); info!("Start run data scanner cycle"); cycle_info.current = cycle_info.next; @@ -548,10 +524,10 @@ 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); + let cycle_budget = ScannerCycleBudget::new(ctx, cycle_budget_config); if let Err(e) = storeapi .clone() - .nsscanner(cycle_budget.token(), sender, cycle_info.current, scan_mode) + .nsscanner(cycle_budget.token(), cycle_budget.clone(), sender, cycle_info.current, scan_mode) .await { let budget_elapsed = cycle_budget.budget_elapsed() && !ctx.is_cancelled(); @@ -559,10 +535,13 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc if budget_elapsed { warn!( duration = ?now.elapsed(), + reason = ?cycle_budget.reason(), max_duration = ?cycle_budget.max_duration(), - "Data scanner cycle stopped after reaching its runtime budget" + max_objects = ?cycle_budget.max_objects(), + max_directories = ?cycle_budget.max_directories(), + "Data scanner cycle stopped after reaching its cycle budget" ); - emit_scan_cycle_partial(cycle_start.elapsed()); + emit_scan_cycle_partial(cycle_start.elapsed(), scan_cycle_partial_reason(cycle_budget.reason())); mark_scan_cycle_idle(cycle_info).await; return; } @@ -576,11 +555,14 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc if cycle_budget.budget_elapsed() && !ctx.is_cancelled() { warn!( duration = ?now.elapsed(), + reason = ?cycle_budget.reason(), max_duration = ?cycle_budget.max_duration(), - "Data scanner cycle stopped after reaching its runtime budget" + max_objects = ?cycle_budget.max_objects(), + max_directories = ?cycle_budget.max_directories(), + "Data scanner cycle stopped after reaching its cycle budget" ); global_metrics().finish_scan_cycle_work(cycle_work_start); - emit_scan_cycle_partial(cycle_start.elapsed()); + emit_scan_cycle_partial(cycle_start.elapsed(), scan_cycle_partial_reason(cycle_budget.reason())); mark_scan_cycle_idle(cycle_info).await; return; } @@ -833,7 +815,13 @@ mod tests { #[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))); + let budget = ScannerCycleBudget::new( + &parent, + ScannerCycleBudgetConfig { + max_duration: Some(Duration::from_millis(1)), + ..Default::default() + }, + ); tokio::time::timeout(Duration::from_secs(5), budget.token().cancelled()) .await @@ -846,7 +834,13 @@ mod tests { #[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 budget = ScannerCycleBudget::new( + &parent, + ScannerCycleBudgetConfig { + max_duration: Some(Duration::from_secs(60)), + ..Default::default() + }, + ); let token = budget.token(); drop(budget); @@ -854,6 +848,47 @@ mod tests { assert!(token.is_cancelled()); } + #[test] + #[serial] + fn test_scanner_cycle_budget_config_uses_work_budget_env() { + with_var(ENV_SCANNER_CYCLE_MAX_OBJECTS, Some("100"), || { + with_var(ENV_SCANNER_CYCLE_MAX_DIRECTORIES, Some("25"), || { + let config = scanner_cycle_budget_config(); + assert_eq!(config.max_objects, Some(100)); + assert_eq!(config.max_directories, Some(25)); + }); + }); + } + + #[test] + #[serial] + fn test_scanner_cycle_budget_config_disables_zero_work_budgets() { + with_var(ENV_SCANNER_CYCLE_MAX_OBJECTS, Some("0"), || { + with_var(ENV_SCANNER_CYCLE_MAX_DIRECTORIES, Some("0"), || { + let config = scanner_cycle_budget_config(); + assert_eq!(config.max_objects, None); + assert_eq!(config.max_directories, None); + }); + }); + } + + #[test] + fn test_scan_cycle_partial_reason_maps_budget_reason() { + assert_eq!( + scan_cycle_partial_reason(Some(ScannerCycleBudgetReason::Runtime)), + ScanCyclePartialReason::Runtime + ); + assert_eq!( + scan_cycle_partial_reason(Some(ScannerCycleBudgetReason::Objects)), + ScanCyclePartialReason::Objects + ); + assert_eq!( + scan_cycle_partial_reason(Some(ScannerCycleBudgetReason::Directories)), + ScanCyclePartialReason::Directories + ); + assert_eq!(scan_cycle_partial_reason(None), ScanCyclePartialReason::Unknown); + } + #[tokio::test] #[serial] async fn test_mark_scan_cycle_idle_clears_published_cycle_state() { diff --git a/crates/scanner/src/scanner_budget.rs b/crates/scanner/src/scanner_budget.rs new file mode 100644 index 000000000..cf3eb3f3e --- /dev/null +++ b/crates/scanner/src/scanner_budget.rs @@ -0,0 +1,231 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::sync::{ + Arc, + atomic::{AtomicU8, AtomicU64, Ordering}, +}; + +use tokio::time::Duration; +use tokio_util::sync::CancellationToken; + +const BUDGET_REASON_NONE: u8 = 0; +const BUDGET_REASON_RUNTIME: u8 = 1; +const BUDGET_REASON_OBJECTS: u8 = 2; +const BUDGET_REASON_DIRECTORIES: u8 = 3; + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) struct ScannerCycleBudgetConfig { + pub max_duration: Option, + pub max_objects: Option, + pub max_directories: Option, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum ScannerCycleBudgetReason { + Runtime, + Objects, + Directories, +} + +impl ScannerCycleBudgetReason { + fn code(self) -> u8 { + match self { + Self::Runtime => BUDGET_REASON_RUNTIME, + Self::Objects => BUDGET_REASON_OBJECTS, + Self::Directories => BUDGET_REASON_DIRECTORIES, + } + } + + fn from_code(code: u8) -> Option { + match code { + BUDGET_REASON_RUNTIME => Some(Self::Runtime), + BUDGET_REASON_OBJECTS => Some(Self::Objects), + BUDGET_REASON_DIRECTORIES => Some(Self::Directories), + _ => None, + } + } +} + +pub struct ScannerCycleBudget { + token: CancellationToken, + reason: Arc, + max_duration: Option, + max_objects: Option, + max_directories: Option, + objects_scanned: AtomicU64, + directories_started: AtomicU64, +} + +impl ScannerCycleBudget { + pub(crate) fn new(parent: &CancellationToken, config: ScannerCycleBudgetConfig) -> Arc { + let token = parent.child_token(); + let reason = Arc::new(AtomicU8::new(BUDGET_REASON_NONE)); + + if let Some(duration) = config.max_duration { + let parent = parent.clone(); + let token_wait = token.clone(); + let token_cancel = token.clone(); + let reason = reason.clone(); + tokio::spawn(async move { + tokio::select! { + _ = parent.cancelled() => {} + _ = token_wait.cancelled() => {} + _ = tokio::time::sleep(duration) => { + Self::cancel_for_reason(&reason, &token_cancel, ScannerCycleBudgetReason::Runtime); + } + } + }); + } + + Arc::new(Self { + token, + reason, + max_duration: config.max_duration, + max_objects: config.max_objects, + max_directories: config.max_directories, + objects_scanned: AtomicU64::new(0), + directories_started: AtomicU64::new(0), + }) + } + + pub(crate) fn token(&self) -> CancellationToken { + self.token.clone() + } + + pub(crate) fn budget_elapsed(&self) -> bool { + self.reason.load(Ordering::Relaxed) != BUDGET_REASON_NONE + } + + pub(crate) fn reason(&self) -> Option { + ScannerCycleBudgetReason::from_code(self.reason.load(Ordering::Relaxed)) + } + + pub(crate) fn max_duration(&self) -> Option { + self.max_duration + } + + pub(crate) fn max_objects(&self) -> Option { + self.max_objects + } + + pub(crate) fn max_directories(&self) -> Option { + self.max_directories + } + + pub(crate) fn try_start_directory(&self) -> bool { + let Some(max_directories) = self.max_directories else { + return true; + }; + + let directories = self.directories_started.fetch_add(1, Ordering::Relaxed) + 1; + if directories <= max_directories { + return true; + } + + self.cancel_for(ScannerCycleBudgetReason::Directories); + false + } + + pub(crate) fn record_object_scanned(&self) { + let Some(max_objects) = self.max_objects else { + return; + }; + + let objects = self.objects_scanned.fetch_add(1, Ordering::Relaxed) + 1; + if objects >= max_objects { + self.cancel_for(ScannerCycleBudgetReason::Objects); + } + } + + fn cancel_for(&self, reason: ScannerCycleBudgetReason) { + Self::cancel_for_reason(&self.reason, &self.token, reason); + } + + fn cancel_for_reason(reason: &AtomicU8, token: &CancellationToken, budget_reason: ScannerCycleBudgetReason) { + if reason + .compare_exchange(BUDGET_REASON_NONE, budget_reason.code(), Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + token.cancel(); + } + } +} + +impl Drop for ScannerCycleBudget { + fn drop(&mut self) { + self.token.cancel(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn runtime_budget_cancels_child_token() { + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new( + &parent, + ScannerCycleBudgetConfig { + max_duration: Some(Duration::from_millis(1)), + ..Default::default() + }, + ); + + tokio::time::sleep(Duration::from_millis(10)).await; + + assert!(budget.budget_elapsed()); + assert_eq!(budget.reason(), Some(ScannerCycleBudgetReason::Runtime)); + assert!(budget.token().is_cancelled()); + } + + #[test] + fn object_budget_cancels_after_reaching_limit() { + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new( + &parent, + ScannerCycleBudgetConfig { + max_objects: Some(2), + ..Default::default() + }, + ); + + budget.record_object_scanned(); + assert!(!budget.budget_elapsed()); + + budget.record_object_scanned(); + assert!(budget.budget_elapsed()); + assert_eq!(budget.reason(), Some(ScannerCycleBudgetReason::Objects)); + assert!(budget.token().is_cancelled()); + } + + #[test] + fn directory_budget_rejects_directory_after_limit() { + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new( + &parent, + ScannerCycleBudgetConfig { + max_directories: Some(1), + ..Default::default() + }, + ); + + assert!(budget.try_start_directory()); + assert!(!budget.budget_elapsed()); + assert!(!budget.try_start_directory()); + assert_eq!(budget.reason(), Some(ScannerCycleBudgetReason::Directories)); + assert!(budget.token().is_cancelled()); + } +} diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 6b420f23e..cfabc9d0a 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -21,6 +21,7 @@ use std::time::{Duration, Instant, SystemTime}; use crate::ReplTargetSizeSummary; use crate::data_usage_define::{DataUsageCache, DataUsageEntry, DataUsageHash, DataUsageHashMap, SizeSummary, hash_path}; use crate::error::ScannerError; +use crate::scanner_budget::ScannerCycleBudget; use crate::scanner_io::ScannerIODisk as _; use crate::sleeper::{DynamicSleeper, scanner_yield_every_n_objects}; use metrics::{counter, describe_counter}; @@ -708,6 +709,7 @@ pub struct FolderScanner { update_current_path: UpdateCurrentPathFn, + budget: Arc, skip_heal: Arc, local_disk: Arc, } @@ -878,6 +880,9 @@ impl FolderScanner { if ctx.is_cancelled() { return Err(ScannerError::Other("Operation cancelled".to_string())); } + if !self.budget.try_start_directory() { + return Err(ScannerError::Other("Operation cancelled".to_string())); + } let this_hash = hash_path(&folder.name); // Store initial compaction state. @@ -1105,9 +1110,14 @@ impl FolderScanner { apply_scanner_size_summary(into, &sz); into.objects += 1; object_count += 1; + self.budget.record_object_scanned(); timer.sleep().await; + if ctx.is_cancelled() { + return Err(ScannerError::Other("Operation cancelled".to_string())); + } + if should_yield_after_object(object_count, yield_every_objects) { let yield_start = Instant::now(); tokio::task::yield_now().await; @@ -1115,6 +1125,10 @@ impl FolderScanner { } } + if ctx.is_cancelled() { + return Err(ScannerError::Other("Operation cancelled".to_string())); + } + if found_objects && is_erasure().await { // If we found an object in erasure mode, we skip subdirs (only datadirs)... info!("scan_folder: done for now found an object in erasure mode"); @@ -1194,6 +1208,9 @@ impl FolderScanner { // Use Box::pin for recursive async call let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst)); if let Err(e) = fut.await { + if ctx.is_cancelled() { + return Err(e); + } warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e); continue; } @@ -1247,6 +1264,9 @@ impl FolderScanner { // Use Box::pin for recursive async call let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst)); if let Err(e) = fut.await { + if ctx.is_cancelled() { + return Err(e); + } warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e); continue; } @@ -1483,6 +1503,9 @@ impl FolderScanner { // Use Box::pin for recursive async call let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst)); if let Err(e) = fut.await { + if ctx.is_cancelled() { + return Err(e); + } warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e); continue; } @@ -1570,6 +1593,7 @@ impl FolderScanner { #[allow(clippy::too_many_arguments)] pub async fn scan_data_folder( ctx: CancellationToken, + budget: Arc, disks: Vec>, local_disk: Arc, cache: DataUsageCache, @@ -1630,6 +1654,7 @@ pub async fn scan_data_folder( updates, last_update: SystemTime::UNIX_EPOCH, update_current_path, + budget, skip_heal, local_disk, }; @@ -1648,7 +1673,7 @@ pub async fn scan_data_folder( }; // Scan the folder - match scanner.scan_folder(ctx, folder, &mut root).await { + match scanner.scan_folder(ctx.clone(), folder, &mut root).await { Ok(()) => { // Get the new cache and finalize it let new_cache = scanner.as_mut_new_cache(); @@ -1660,6 +1685,26 @@ pub async fn scan_data_folder( Ok(new_cache.clone()) } Err(e) => { + if ctx.is_cancelled() { + let root_has_progress = !root.children.is_empty() + || root.size > 0 + || root.objects > 0 + || root.versions > 0 + || root.delete_markers > 0 + || root.failed_objects > 0 + || root.replication_stats.is_some(); + let new_cache = scanner.as_mut_new_cache(); + if root_has_progress { + new_cache.replace_hashed(&hash_path(&cache.info.name), &None, &root); + } + if new_cache.root().is_some() { + new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN); + new_cache.info.last_update = Some(SystemTime::now()); + new_cache.info.next_cycle = cache.info.next_cycle; + close_disk().await; + return Err(ScannerError::PartialCache(Box::new(new_cache.clone()))); + } + } close_disk().await; // No useful information, return original cache Err(e) @@ -1716,6 +1761,7 @@ mod tests { updates: None, last_update: SystemTime::UNIX_EPOCH, update_current_path, + budget: ScannerCycleBudget::new(&CancellationToken::new(), Default::default()), skip_heal: Arc::new(AtomicBool::new(false)), local_disk: disk, }; @@ -2138,6 +2184,102 @@ mod tests { .expect("scan_folder should finish successfully"); } + #[tokio::test] + #[serial] + async fn test_scan_folder_directory_budget_cancels_after_limit() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone()); + + let bucket_dir = temp_dir.join("bucket"); + tokio::fs::create_dir_all(bucket_dir.join("child")) + .await + .expect("failed to create child directory"); + + scanner.old_cache.info.name = "bucket".to_string(); + scanner.new_cache.info.name = "bucket".to_string(); + scanner.update_cache.info.name = "bucket".to_string(); + + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new( + &parent, + crate::scanner_budget::ScannerCycleBudgetConfig { + max_directories: Some(1), + ..Default::default() + }, + ); + let ctx = budget.token(); + scanner.budget = budget.clone(); + + let folder = CachedFolder { + name: "bucket".to_string(), + parent: None, + object_heal_prob_div: 1, + }; + + let mut into = DataUsageEntry::default(); + let result = scanner.scan_folder(ctx, folder, &mut into).await; + + assert!(result.is_err(), "directory budget cancellation should make the scan partial"); + assert!(budget.budget_elapsed()); + assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Directories)); + assert!(budget.token().is_cancelled()); + } + + #[tokio::test] + #[serial] + async fn test_scan_data_folder_returns_partial_cache_on_budget_cancel() { + let (mut scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone()); + + let bucket_dir = temp_dir.join("bucket"); + tokio::fs::create_dir_all(bucket_dir.join("child-a")) + .await + .expect("failed to create first child directory"); + tokio::fs::create_dir_all(bucket_dir.join("child-b")) + .await + .expect("failed to create second child directory"); + + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new( + &parent, + crate::scanner_budget::ScannerCycleBudgetConfig { + max_directories: Some(2), + ..Default::default() + }, + ); + let cache = DataUsageCache { + info: crate::data_usage_define::DataUsageCacheInfo { + name: "bucket".to_string(), + next_cycle: 7, + ..Default::default() + }, + ..Default::default() + }; + + let result = scan_data_folder( + budget.token(), + budget.clone(), + vec![scanner.local_disk.clone()], + scanner.local_disk.clone(), + cache, + None, + HealScanMode::Normal, + SCANNER_SLEEPER.clone(), + ) + .await; + + let partial_cache = match result { + Err(ScannerError::PartialCache(partial_cache)) => partial_cache, + other => panic!("expected partial cache after directory budget cancellation, got {other:?}"), + }; + + assert!(partial_cache.info.last_update.is_some()); + assert_eq!(partial_cache.info.next_cycle, 7); + assert!(partial_cache.root().is_some(), "partial cache should keep completed scan progress"); + assert!(budget.budget_elapsed()); + assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Directories)); + } + #[tokio::test] #[serial] #[cfg(unix)] diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 4c070b880..6231ed438 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -12,11 +12,12 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::scanner_budget::ScannerCycleBudget; use crate::scanner_folder::{ScannerItem, scan_data_folder}; use crate::sleeper::SCANNER_SLEEPER; use crate::{ DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, DataUsageCache, DataUsageCacheInfo, DataUsageEntry, DataUsageEntryInfo, - DataUsageInfo, SizeSummary, TierStats, + DataUsageInfo, ScannerError, SizeSummary, TierStats, }; use futures::future::join_all; use metrics::counter; @@ -66,6 +67,41 @@ const METRIC_SCANNER_SET_SCANS_QUEUED: &str = "rustfs_scanner_set_scans_queued"; const METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE: &str = "rustfs_scanner_disk_bucket_scans_active"; const METRIC_SCANNER_DISK_BUCKET_SCANS_QUEUED: &str = "rustfs_scanner_disk_bucket_scans_queued"; +fn record_set_scan_concurrency_limit(limit: usize) { + metrics::gauge!(METRIC_SCANNER_SET_SCAN_CONCURRENCY_LIMIT).set(limit as f64); + global_metrics().record_scanner_set_scan_state(Some(limit), None, None); +} + +fn record_set_scans_queued(count: usize) { + metrics::gauge!(METRIC_SCANNER_SET_SCANS_QUEUED).set(count as f64); + global_metrics().record_scanner_set_scan_state(None, Some(count), None); +} + +fn record_set_scans_active(count: usize) { + metrics::gauge!(METRIC_SCANNER_SET_SCANS_ACTIVE).set(count as f64); + global_metrics().record_scanner_set_scan_state(None, None, Some(count)); +} + +fn record_disk_scan_concurrency_limit(pool: &str, set: &str, limit: usize) { + metrics::gauge!( + METRIC_SCANNER_DISK_SCAN_CONCURRENCY_LIMIT, + "pool" => pool.to_owned(), + "set" => set.to_owned() + ) + .set(limit as f64); + global_metrics().record_scanner_disk_bucket_scan_state(pool, set, Some(limit), None, None); +} + +fn record_disk_bucket_scans_active(count: usize, pool: &str, set: &str) { + metrics::gauge!( + METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE, + "pool" => pool.to_owned(), + "set" => set.to_owned() + ) + .set(count as f64); + global_metrics().record_scanner_disk_bucket_scan_state(pool, set, None, None, Some(count)); +} + struct SetScanActiveGuard { active: Arc, } @@ -73,7 +109,7 @@ struct SetScanActiveGuard { impl SetScanActiveGuard { fn new(active: Arc) -> Self { let active_count = active.fetch_add(1, Ordering::Relaxed) + 1; - metrics::gauge!(METRIC_SCANNER_SET_SCANS_ACTIVE).set(active_count as f64); + record_set_scans_active(active_count); Self { active } } } @@ -81,7 +117,7 @@ impl SetScanActiveGuard { impl Drop for SetScanActiveGuard { fn drop(&mut self) { let active_count = decrement_atomic_usize(&self.active); - metrics::gauge!(METRIC_SCANNER_SET_SCANS_ACTIVE).set(active_count as f64); + record_set_scans_active(active_count); } } @@ -94,12 +130,7 @@ struct DiskBucketScanActiveGuard { impl DiskBucketScanActiveGuard { fn new(active: Arc, pool: String, set: String) -> Self { let active_count = active.fetch_add(1, Ordering::Relaxed) + 1; - metrics::gauge!( - METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE, - "pool" => pool.clone(), - "set" => set.clone() - ) - .set(active_count as f64); + record_disk_bucket_scans_active(active_count, &pool, &set); Self { active, pool, set } } } @@ -107,12 +138,7 @@ impl DiskBucketScanActiveGuard { impl Drop for DiskBucketScanActiveGuard { fn drop(&mut self) { let active_count = decrement_atomic_usize(&self.active); - metrics::gauge!( - METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE, - "pool" => self.pool.clone(), - "set" => self.set.clone() - ) - .set(active_count as f64); + record_disk_bucket_scans_active(active_count, &self.pool, &self.set); } } @@ -138,6 +164,23 @@ impl Drop for BucketDriveFailureGuard { } } +struct DiskBucketScanGaugeReset { + pool: String, + set: String, +} + +impl DiskBucketScanGaugeReset { + fn new(pool: String, set: String) -> Self { + Self { pool, set } + } +} + +impl Drop for DiskBucketScanGaugeReset { + fn drop(&mut self) { + reset_disk_bucket_scan_gauges(&self.pool, &self.set); + } +} + fn decrement_atomic_usize(counter: &AtomicUsize) -> usize { counter .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(1))) @@ -152,6 +195,7 @@ fn record_disk_bucket_scans_queued(count: usize, pool: &str, set: &str) { "set" => set.to_owned() ) .set(count as f64); + global_metrics().record_scanner_disk_bucket_scan_state(pool, set, None, Some(count), None); } fn decrement_disk_bucket_scans_queued(counter: &AtomicUsize, pool: &str, set: &str) { @@ -160,25 +204,16 @@ fn decrement_disk_bucket_scans_queued(counter: &AtomicUsize, pool: &str, set: &s } fn reset_set_scan_gauges() { - metrics::gauge!(METRIC_SCANNER_SET_SCAN_CONCURRENCY_LIMIT).set(0.0); - metrics::gauge!(METRIC_SCANNER_SET_SCANS_QUEUED).set(0.0); - metrics::gauge!(METRIC_SCANNER_SET_SCANS_ACTIVE).set(0.0); + record_set_scan_concurrency_limit(0); + record_set_scans_queued(0); + record_set_scans_active(0); + global_metrics().reset_scanner_set_scan_state(); } fn reset_disk_bucket_scan_gauges(pool: &str, set: &str) { - metrics::gauge!( - METRIC_SCANNER_DISK_SCAN_CONCURRENCY_LIMIT, - "pool" => pool.to_owned(), - "set" => set.to_owned() - ) - .set(0.0); + record_disk_scan_concurrency_limit(pool, set, 0); record_disk_bucket_scans_queued(0, pool, set); - metrics::gauge!( - METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE, - "pool" => pool.to_owned(), - "set" => set.to_owned() - ) - .set(0.0); + record_disk_bucket_scans_active(0, pool, set); } fn scanner_concurrency_limit(configured: usize, available: usize) -> usize { @@ -232,6 +267,23 @@ fn is_xl_meta_path(path: &str) -> bool { .is_some_and(|name| name == STORAGE_FORMAT_FILE) } +fn cache_root_entry_info(cache: &DataUsageCache) -> DataUsageEntryInfo { + let entry = cache.root().map(|root| cache.flatten(&root)).unwrap_or_default(); + + DataUsageEntryInfo { + name: cache.info.name.clone(), + parent: DATA_USAGE_ROOT.to_string(), + entry, + } +} + +async fn send_cache_root_entry_info( + bucket_result_tx: &Arc>>, + cache: &DataUsageCache, +) -> std::result::Result<(), mpsc::error::SendError> { + bucket_result_tx.lock().await.send(cache_root_entry_info(cache)).await +} + async fn persist_and_publish_cache_snapshot( store: Arc, updates: &mpsc::Sender, @@ -257,6 +309,7 @@ pub trait ScannerIO: Send + Sync + Debug + 'static { async fn nsscanner( &self, ctx: CancellationToken, + budget: Arc, updates: mpsc::Sender, want_cycle: u64, scan_mode: HealScanMode, @@ -268,6 +321,7 @@ pub trait ScannerIOCache: Send + Sync + Debug + 'static { async fn nsscanner_cache( self: Arc, ctx: CancellationToken, + budget: Arc, buckets: Vec, updates: mpsc::Sender, want_cycle: u64, @@ -280,20 +334,28 @@ pub trait ScannerIODisk: Send + Sync + Debug + 'static { async fn nsscanner_disk( &self, ctx: CancellationToken, + budget: Arc, cache: DataUsageCache, updates: Option>, scan_mode: HealScanMode, - ) -> Result; + ) -> Result; async fn get_size(&self, item: ScannerItem) -> Result; } +#[derive(Debug)] +pub enum ScannerDiskScanOutcome { + Complete(DataUsageCache), + Partial(DataUsageCache), +} + #[async_trait::async_trait] impl ScannerIO for ECStore { - #[tracing::instrument(skip(self, updates))] + #[tracing::instrument(skip(self, budget, updates))] async fn nsscanner( &self, ctx: CancellationToken, + budget: Arc, updates: mpsc::Sender, want_cycle: u64, scan_mode: HealScanMode, @@ -321,7 +383,7 @@ impl ScannerIO for ECStore { } let set_scan_limit = scanner_max_concurrent_set_scans(total_results); - metrics::gauge!(METRIC_SCANNER_SET_SCAN_CONCURRENCY_LIMIT).set(set_scan_limit as f64); + record_set_scan_concurrency_limit(set_scan_limit); debug!( total_sets = total_results, concurrency_limit = set_scan_limit, @@ -330,8 +392,8 @@ impl ScannerIO for ECStore { let set_scan_semaphore = Arc::new(Semaphore::new(set_scan_limit)); let queued_set_scans = Arc::new(AtomicUsize::new(total_results)); let active_set_scans = Arc::new(AtomicUsize::new(0)); - metrics::gauge!(METRIC_SCANNER_SET_SCANS_QUEUED).set(total_results as f64); - metrics::gauge!(METRIC_SCANNER_SET_SCANS_ACTIVE).set(0.0); + record_set_scans_queued(total_results); + record_set_scans_active(0); let results = vec![DataUsageCache::default(); total_results]; let results_mutex: Arc>> = Arc::new(Mutex::new(results)); @@ -350,6 +412,7 @@ impl ScannerIO for ECStore { let set_label = set.set_index.to_string(); let child_token_clone = child_token.clone(); + let budget_clone = budget.clone(); let want_cycle_clone = want_cycle; let scan_mode_clone = scan_mode; let results_mutex_clone = results_mutex.clone(); @@ -388,11 +451,18 @@ impl ScannerIO for ECStore { ) .record(permit_wait_start.elapsed().as_secs_f64()); let queued_count = decrement_atomic_usize(&queued_set_scans_clone); - metrics::gauge!(METRIC_SCANNER_SET_SCANS_QUEUED).set(queued_count as f64); + record_set_scans_queued(queued_count); let _active_guard = SetScanActiveGuard::new(active_set_scans_clone); if let Err(e) = set_clone - .nsscanner_cache(child_token_clone.clone(), all_buckets_clone, tx, want_cycle_clone, scan_mode_clone) + .nsscanner_cache( + child_token_clone.clone(), + budget_clone, + all_buckets_clone, + tx, + want_cycle_clone, + scan_mode_clone, + ) .await { if child_token_clone.is_cancelled() { @@ -488,8 +558,9 @@ impl ScannerIO for ECStore { }); let _ = join_all(wait_futs).await; - metrics::gauge!(METRIC_SCANNER_SET_SCANS_QUEUED).set(0.0); - metrics::gauge!(METRIC_SCANNER_SET_SCANS_ACTIVE).set(0.0); + record_set_scan_concurrency_limit(0); + record_set_scans_queued(0); + record_set_scans_active(0); let _ = update_tx.send(()); @@ -501,10 +572,11 @@ impl ScannerIO for ECStore { #[async_trait::async_trait] impl ScannerIOCache for SetDisks { - #[tracing::instrument(skip(self, updates))] + #[tracing::instrument(skip(self, budget, updates))] async fn nsscanner_cache( self: Arc, ctx: CancellationToken, + budget: Arc, buckets: Vec, updates: mpsc::Sender, want_cycle: u64, @@ -525,12 +597,7 @@ impl ScannerIOCache for SetDisks { return Ok(()); } let disk_scan_limit = scanner_max_concurrent_disk_scans(disks.len()); - metrics::gauge!( - METRIC_SCANNER_DISK_SCAN_CONCURRENCY_LIMIT, - "pool" => self.pool_index.to_string(), - "set" => self.set_index.to_string() - ) - .set(disk_scan_limit as f64); + record_disk_scan_concurrency_limit(&pool_label, &set_label, disk_scan_limit); debug!( pool = self.pool_index, set = self.set_index, @@ -542,12 +609,8 @@ impl ScannerIOCache for SetDisks { let queued_disk_bucket_scans = Arc::new(AtomicUsize::new(buckets.len())); let active_disk_bucket_scans = Arc::new(AtomicUsize::new(0)); record_disk_bucket_scans_queued(buckets.len(), &pool_label, &set_label); - metrics::gauge!( - METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE, - "pool" => pool_label.clone(), - "set" => set_label.clone() - ) - .set(0.0); + record_disk_bucket_scans_active(0, &pool_label, &set_label); + let _reset_disk_bucket_scan_gauges = DiskBucketScanGaugeReset::new(pool_label.clone(), set_label.clone()); let mut old_cache = DataUsageCache::default(); old_cache.load(self.clone(), DATA_USAGE_CACHE_NAME).await?; @@ -597,13 +660,14 @@ impl ScannerIOCache for SetDisks { let mut ticker = tokio::time::interval(Duration::from_secs(3 + rand::random::() % 10)); let mut last_update = None; + let mut cancelled = false; loop { tokio::select! { - _ = ctx_clone.cancelled() => { - break; + _ = ctx_clone.cancelled(), if !cancelled => { + cancelled = true; } - _ = ticker.tick() => { + _ = ticker.tick(), if !cancelled => { let cache_snapshot = { let cache = cache_mutex_clone.lock().await; if cache.info.last_update == last_update { @@ -648,6 +712,7 @@ impl ScannerIOCache for SetDisks { for disk in disks.into_iter() { let bucket_rx_mutex_clone = bucket_rx_mutex.clone(); let ctx_clone = ctx.clone(); + let budget_clone = budget.clone(); let store_clone_clone = self.clone(); let bucket_result_tx_clone_clone = bucket_result_tx_clone.clone(); let disk_clone = disk.clone(); @@ -756,11 +821,11 @@ impl ScannerIOCache for SetDisks { let before = cache.info.last_update; - cache = match disk_clone - .nsscanner_disk(ctx_clone.clone(), cache.clone(), Some(updates_tx), scan_mode) + let scan_outcome = match disk_clone + .nsscanner_disk(ctx_clone.clone(), budget_clone.clone(), cache.clone(), Some(updates_tx), scan_mode) .await { - Ok(cache) => cache, + Ok(scan_outcome) => scan_outcome, Err(e) => { if ctx_clone.is_cancelled() { debug!("Scanner disk scan stopped after cancellation: {}", e); @@ -785,34 +850,39 @@ impl ScannerIOCache for SetDisks { } }; + cache = match scan_outcome { + ScannerDiskScanOutcome::Complete(cache) => cache, + ScannerDiskScanOutcome::Partial(cache) => { + let done_save = Metrics::time(Metric::SaveUsage); + if let Err(e) = cache.save(store_clone_clone.clone(), cache_name.as_str()).await { + error!("Failed to save partial data usage cache: {}", e); + } + done_save(); + + if let Err(e) = update_fut.await { + error!("Failed to update partial data usage cache: {}", e); + } + + if let Err(e) = send_cache_root_entry_info(&bucket_result_tx_clone_clone, &cache).await { + error!("Failed to send partial data usage entry info: {}", e); + } + continue; + } + }; + debug!("nsscanner_disk: got cache: {}", cache.info.name); if let Err(e) = update_fut.await { error!("nsscanner_disk: Failed to update data usage cache: {}", e); } - let root = if let Some(r) = cache.root() { - cache.flatten(&r) - } else { - DataUsageEntry::default() - }; - if ctx_clone.is_cancelled() { break; } debug!("nsscanner_disk: sending data usage entry info: {}", cache.info.name); - if let Err(e) = bucket_result_tx_clone_clone - .lock() - .await - .send(DataUsageEntryInfo { - name: cache.info.name.clone(), - parent: DATA_USAGE_ROOT.to_string(), - entry: root, - }) - .await - { + if let Err(e) = send_cache_root_entry_info(&bucket_result_tx_clone_clone, &cache).await { error!("nsscanner_disk: Failed to send data usage entry info: {}", e); } @@ -826,13 +896,9 @@ impl ScannerIOCache for SetDisks { } let _ = join_all(futs).await; + record_disk_scan_concurrency_limit(&pool_label, &set_label, 0); record_disk_bucket_scans_queued(0, &pool_label, &set_label); - metrics::gauge!( - METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE, - "pool" => pool_label.clone(), - "set" => set_label.clone() - ) - .set(0.0); + record_disk_bucket_scans_active(0, &pool_label, &set_label); drop(bucket_result_tx_clone); @@ -931,14 +997,15 @@ impl ScannerIODisk for Disk { Ok(size_summary) } - #[tracing::instrument(skip(self, updates, cache))] + #[tracing::instrument(skip(self, budget, updates, cache))] async fn nsscanner_disk( &self, ctx: CancellationToken, + budget: Arc, cache: DataUsageCache, updates: Option>, scan_mode: HealScanMode, - ) -> Result { + ) -> Result { let done_drive = Metrics::time(Metric::ScanBucketDrive); let drive_start = std::time::Instant::now(); let bucket = cache.info.name.clone(); @@ -1004,7 +1071,8 @@ impl ScannerIODisk for Disk { let disks = disks_result.into_iter().flatten().collect::>>(); - let result = scan_data_folder(ctx.clone(), disks, local_disk, cache, updates, scan_mode, SCANNER_SLEEPER.clone()).await; + let result = + scan_data_folder(ctx.clone(), budget, disks, local_disk, cache, updates, scan_mode, SCANNER_SLEEPER.clone()).await; match result { Ok(mut data_usage_info) => { @@ -1012,7 +1080,14 @@ impl ScannerIODisk for Disk { emit_scan_bucket_drive_complete(true, &bucket, &disk_path, drive_start.elapsed()); data_usage_info.info.last_update = Some(SystemTime::now()); failure_guard.mark_not_failed(); - Ok(data_usage_info) + Ok(ScannerDiskScanOutcome::Complete(data_usage_info)) + } + Err(ScannerError::PartialCache(mut partial_cache)) => { + done_drive(); + emit_scan_bucket_drive_partial(&bucket, &disk_path, drive_start.elapsed()); + partial_cache.info.last_update.get_or_insert_with(SystemTime::now); + failure_guard.mark_not_failed(); + Ok(ScannerDiskScanOutcome::Partial(*partial_cache)) } Err(e) => { if ctx.is_cancelled() { @@ -1115,4 +1190,78 @@ mod tests { fn is_xl_meta_path_accepts_forward_separator() { assert!(is_xl_meta_path("/data/bucket/object/xl.meta")); } + + #[test] + fn cache_root_entry_info_flattens_bucket_children() { + let mut cache = DataUsageCache { + info: DataUsageCacheInfo { + name: "bucket".to_string(), + ..Default::default() + }, + ..Default::default() + }; + cache.replace( + "bucket", + DATA_USAGE_ROOT, + DataUsageEntry { + size: 10, + objects: 1, + ..Default::default() + }, + ); + cache.replace( + "bucket/prefix", + "bucket", + DataUsageEntry { + size: 20, + objects: 2, + ..Default::default() + }, + ); + + let info = cache_root_entry_info(&cache); + + assert_eq!(info.name, "bucket"); + assert_eq!(info.parent, DATA_USAGE_ROOT); + assert_eq!(info.entry.size, 30); + assert_eq!(info.entry.objects, 3); + assert!(info.entry.children.is_empty()); + } + + #[tokio::test] + async fn send_cache_root_entry_info_sends_after_budget_cancellation() { + let ctx = CancellationToken::new(); + ctx.cancel(); + assert!(ctx.is_cancelled()); + + let mut cache = DataUsageCache { + info: DataUsageCacheInfo { + name: "bucket".to_string(), + ..Default::default() + }, + ..Default::default() + }; + cache.replace( + "bucket", + DATA_USAGE_ROOT, + DataUsageEntry { + size: 10, + objects: 1, + ..Default::default() + }, + ); + + let (tx, mut rx) = mpsc::channel(1); + let tx = Arc::new(Mutex::new(tx)); + + send_cache_root_entry_info(&tx, &cache) + .await + .expect("partial cache should be sent even after budget cancellation"); + + let info = rx.recv().await.expect("partial cache entry should be received"); + assert_eq!(info.name, "bucket"); + assert_eq!(info.parent, DATA_USAGE_ROOT); + assert_eq!(info.entry.size, 10); + assert_eq!(info.entry.objects, 1); + } }