From a99ef64db2749041102076ac16680e4a4a963932 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Mon, 1 Jun 2026 00:42:38 +0800 Subject: [PATCH] feat(scanner): add scanner budgets and progress metrics (#3145) * fix(scanner): preserve maintenance scan cadence * feat(scanner): add scanner concurrency budget * feat(scanner): expose scanner runtime progress * fix(scanner): address scanner review feedback --------- Co-authored-by: Henry Guo Co-authored-by: houseme --- crates/common/src/metrics.rs | 55 +++- crates/config/src/constants/scanner.rs | 18 + crates/obs/src/metrics/collectors/scanner.rs | 15 +- .../src/metrics/schema/entry/metric_name.rs | 2 + crates/obs/src/metrics/schema/scanner.rs | 9 + crates/obs/src/metrics/stats_collector.rs | 2 + crates/scanner/src/scanner.rs | 144 +++++++- crates/scanner/src/scanner_io.rs | 309 +++++++++++++++++- 8 files changed, 540 insertions(+), 14 deletions(-) diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index 744435ccd..9fa097914 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -370,6 +370,8 @@ pub struct ScannerMetricsReport { pub current_started: DateTime, pub cycles_completed_at: Vec>, pub ongoing_buckets: usize, + #[serde(default)] + pub active_scan_paths: usize, pub life_time_ops: HashMap, pub life_time_ilm: HashMap, pub last_minute: ScannerLastMinute, @@ -600,18 +602,22 @@ impl Metrics { pub async fn report(&self) -> ScannerMetricsReport { let mut m = ScannerMetricsReport::default(); - if let Some(cycle) = self.get_cycle().await { + let has_cycle = if let Some(cycle) = self.get_cycle().await { m.current_cycle = cycle.current; m.cycles_completed_at = cycle.cycle_completed; m.current_started = cycle.started; - } + true + } else { + false + }; - if let Some(init_time) = crate::get_global_init_time().await { + if !has_cycle && let Some(init_time) = crate::get_global_init_time().await { m.current_started = init_time; } m.collected_at = Utc::now(); m.active_paths = self.get_current_paths().await; + m.active_scan_paths = m.active_paths.len(); // Lifetime operation counts for i in 0..Metric::Last as usize { @@ -747,3 +753,46 @@ impl Drop for CloseDiskGuard { // If there is no runtime we are in a test or shutdown path; skip cleanup. } } + +#[cfg(test)] +mod tests { + use super::*; + + #[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 report = metrics.report().await; + + assert_eq!(report.active_scan_paths, 1); + assert_eq!(report.ongoing_buckets, 0); + assert_eq!(report.active_paths, vec!["disk-a/bucket-a".to_string()]); + } + + #[tokio::test] + async fn report_preserves_current_cycle_started_time() { + let previous_init_time = *crate::globals::GLOBAL_INIT_TIME.read().await; + let init_time = Utc::now() - chrono::Duration::hours(1); + let cycle_started = Utc::now(); + *crate::globals::GLOBAL_INIT_TIME.write().await = Some(init_time); + + let metrics = Metrics::new(); + metrics + .set_cycle(Some(CurrentCycle { + current: 7, + started: cycle_started, + ..Default::default() + })) + .await; + + let report = metrics.report().await; + *crate::globals::GLOBAL_INIT_TIME.write().await = previous_init_time; + + assert_eq!(report.current_started, cycle_started); + } +} diff --git a/crates/config/src/constants/scanner.rs b/crates/config/src/constants/scanner.rs index 3b8b5e45a..6875bfc6d 100644 --- a/crates/config/src/constants/scanner.rs +++ b/crates/config/src/constants/scanner.rs @@ -83,9 +83,27 @@ pub const ENV_SCANNER_IDLE_MODE: &str = "RUSTFS_SCANNER_IDLE_MODE"; /// - Example: `export RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS=30` pub const ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS: &str = "RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS"; +/// Environment variable that caps concurrent scanner set tasks. +/// A value of `0` keeps the existing topology-based concurrency. +/// - Example: `export RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS=2` +pub const ENV_SCANNER_MAX_CONCURRENT_SET_SCANS: &str = "RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS"; + +/// Environment variable that caps concurrent scanner disk bucket walks per set. +/// A value of `0` keeps the existing disk-count-based concurrency. +/// - Example: `export RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS=1` +pub const ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS: &str = "RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS"; + /// Default scanner idle mode. pub const DEFAULT_SCANNER_IDLE_MODE: bool = true; +/// Default set scan concurrency budget. +/// `0` means no additional limit beyond deployment topology. +pub const DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS: usize = 0; + +/// Default disk scan concurrency budget. +/// `0` means no additional limit beyond available disks in the set. +pub const DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS: usize = 0; + /// Compatibility flag kept for Patch 3 rollback windows. /// /// Inline scanner heal execution has been removed in favor of heal-candidate enqueue. diff --git a/crates/obs/src/metrics/collectors/scanner.rs b/crates/obs/src/metrics/collectors/scanner.rs index 37af62689..8f3737d20 100644 --- a/crates/obs/src/metrics/collectors/scanner.rs +++ b/crates/obs/src/metrics/collectors/scanner.rs @@ -21,7 +21,7 @@ use crate::metrics::report::PrometheusMetric; use crate::metrics::schema::scanner::{ - SCANNER_BUCKET_SCANS_FINISHED_MD, SCANNER_BUCKET_SCANS_STARTED_MD, SCANNER_DIRECTORIES_SCANNED_MD, + SCANNER_ACTIVE_PATHS_MD, SCANNER_BUCKET_SCANS_FINISHED_MD, SCANNER_BUCKET_SCANS_STARTED_MD, SCANNER_DIRECTORIES_SCANNED_MD, SCANNER_LAST_ACTIVITY_SECONDS_MD, SCANNER_OBJECTS_SCANNED_MD, SCANNER_VERSIONS_SCANNED_MD, }; @@ -40,6 +40,8 @@ pub struct ScannerStats { pub versions_scanned: u64, /// Seconds since last scanner activity pub last_activity_seconds: u64, + /// Number of scanner paths currently being processed + pub active_paths: u64, } /// Collects scanner metrics from the given stats. @@ -54,6 +56,7 @@ pub fn collect_scanner_metrics(stats: &ScannerStats) -> Vec { PrometheusMetric::from_descriptor(&SCANNER_OBJECTS_SCANNED_MD, stats.objects_scanned as f64), 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), ] } @@ -71,18 +74,24 @@ mod tests { objects_scanned: 1000000, versions_scanned: 1500000, last_activity_seconds: 30, + active_paths: 4, }; let metrics = collect_scanner_metrics(&stats); report_metrics(&metrics); - assert_eq!(metrics.len(), 6); + assert_eq!(metrics.len(), 7); let objects = metrics.iter().find(|m| m.value == 1000000.0); assert!(objects.is_some()); let last_activity = metrics.iter().find(|m| m.value == 30.0); assert!(last_activity.is_some()); + + let active_paths = metrics + .iter() + .find(|m| m.name == SCANNER_ACTIVE_PATHS_MD.get_full_metric_name()); + assert_eq!(active_paths.map(|m| m.value), Some(4.0)); } #[test] @@ -90,7 +99,7 @@ mod tests { let stats = ScannerStats::default(); let metrics = collect_scanner_metrics(&stats); - assert_eq!(metrics.len(), 6); + assert_eq!(metrics.len(), 7); for metric in &metrics { assert_eq!(metric.value, 0.0); assert!(metric.labels.is_empty()); diff --git a/crates/obs/src/metrics/schema/entry/metric_name.rs b/crates/obs/src/metrics/schema/entry/metric_name.rs index 6ab12af1f..55531ff70 100644 --- a/crates/obs/src/metrics/schema/entry/metric_name.rs +++ b/crates/obs/src/metrics/schema/entry/metric_name.rs @@ -278,6 +278,7 @@ pub enum MetricName { ScannerObjectsScanned, ScannerVersionsScanned, ScannerLastActivitySeconds, + ScannerActivePaths, // CPU system-related metrics SysCPUAvgIdle, @@ -618,6 +619,7 @@ impl MetricName { Self::ScannerObjectsScanned => "objects_scanned".to_string(), Self::ScannerVersionsScanned => "versions_scanned".to_string(), Self::ScannerLastActivitySeconds => "last_activity_seconds".to_string(), + Self::ScannerActivePaths => "active_paths".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 47ab02264..4ad9d96e9 100644 --- a/crates/obs/src/metrics/schema/scanner.rs +++ b/crates/obs/src/metrics/schema/scanner.rs @@ -70,3 +70,12 @@ pub static SCANNER_LAST_ACTIVITY_SECONDS_MD: LazyLock = LazyLo subsystems::SCANNER, ) }); + +pub static SCANNER_ACTIVE_PATHS_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerActivePaths, + "Current number of scanner paths being processed.", + &[], + subsystems::SCANNER, + ) +}); diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 616a51b70..6cff3dcd2 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -875,6 +875,7 @@ pub async fn collect_scanner_metric_stats() -> Option { let versions_scanned = metrics.life_time_ilm.values().copied().sum(); let reference_time = metrics.cycles_completed_at.last().copied().unwrap_or(metrics.current_started); let last_activity_seconds = Utc::now().signed_duration_since(reference_time).num_seconds().max(0) as u64; + let active_paths = metrics.active_scan_paths as u64; Some(ScannerStats { bucket_scans_finished, @@ -886,6 +887,7 @@ pub async fn collect_scanner_metric_stats() -> Option { objects_scanned, versions_scanned, last_activity_seconds, + active_paths, }) } diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 308045491..1c67d666c 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -31,11 +31,15 @@ use rustfs_config::{ ENV_SCANNER_START_DELAY_SECS, }; use rustfs_ecstore::StorageAPI as _; +use rustfs_ecstore::bucket::lifecycle::lifecycle::Lifecycle as _; +use rustfs_ecstore::bucket::metadata_sys::{get_lifecycle_config, get_replication_config}; +use rustfs_ecstore::bucket::replication::ReplicationConfigurationExt as _; use rustfs_ecstore::config::com::{read_config, save_config}; use rustfs_ecstore::disk::RUSTFS_META_BUCKET; use rustfs_ecstore::error::Error as EcstoreError; use rustfs_ecstore::global::is_erasure_sd; use rustfs_ecstore::store::ECStore; +use rustfs_ecstore::store_api::{BucketOperations, BucketOptions}; use serde::{Deserialize, Serialize}; use tokio::sync::mpsc; use tokio::time::{Duration, Instant}; @@ -108,7 +112,7 @@ fn initial_scanner_delay_for(start_delay_secs: Option) -> Duration { } pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { - configure_scanner_defaults().await; + configure_scanner_defaults(&storeapi).await; // Force init global sleeper so config is read once at startup. let _ = &*SCANNER_SLEEPER; @@ -136,16 +140,105 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { }); } -async fn configure_scanner_defaults() { +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +struct ScannerMaintenanceFeatures { + lifecycle: bool, + replication: bool, + inspection_failed: bool, +} + +impl ScannerMaintenanceFeatures { + fn needs_regular_cycle(self) -> bool { + self.lifecycle || self.replication || self.inspection_failed + } +} + +fn single_disk_default_cycle_secs(features: ScannerMaintenanceFeatures) -> Option { + if features.needs_regular_cycle() { + None + } else { + Some(SINGLE_DISK_SCANNER_CYCLE_SECS) + } +} + +async fn detect_scanner_maintenance_features(storeapi: &Arc) -> ScannerMaintenanceFeatures { + let mut features = ScannerMaintenanceFeatures::default(); + let buckets = match storeapi + .list_bucket(&BucketOptions { + no_metadata: true, + ..Default::default() + }) + .await + { + Ok(buckets) => buckets, + Err(err) => { + warn!( + error = %err, + "Failed to inspect scanner maintenance features; preserving speed-based scanner cycle" + ); + features.inspection_failed = true; + return features; + } + }; + + for bucket in buckets { + if !features.lifecycle { + match get_lifecycle_config(&bucket.name).await { + Ok((lifecycle, _)) => { + features.lifecycle = lifecycle.has_active_rules(""); + } + Err(EcstoreError::ConfigNotFound) => {} + Err(err) => { + warn!( + bucket = %bucket.name, + error = %err, + "Failed to inspect bucket lifecycle config; preserving speed-based scanner cycle" + ); + features.inspection_failed = true; + } + } + } + + if !features.replication { + match get_replication_config(&bucket.name).await { + Ok((replication, _)) => { + features.replication = replication.has_active_rules("", true); + } + Err(EcstoreError::ConfigNotFound) => {} + Err(err) => { + warn!( + bucket = %bucket.name, + error = %err, + "Failed to inspect bucket replication config; preserving speed-based scanner cycle" + ); + features.inspection_failed = true; + } + } + } + + if features.needs_regular_cycle() { + break; + } + } + + features +} + +async fn configure_scanner_defaults(storeapi: &Arc) { if is_erasure_sd().await { + let features = detect_scanner_maintenance_features(storeapi).await; + let default_cycle_secs = single_disk_default_cycle_secs(features); set_scanner_default_speed(ScannerSpeed::Slowest); - set_scanner_default_cycle_secs(Some(SINGLE_DISK_SCANNER_CYCLE_SECS)); + set_scanner_default_cycle_secs(default_cycle_secs); info!( env_speed = ENV_SCANNER_SPEED, env_cycle = ENV_SCANNER_CYCLE, env_start_delay = ENV_SCANNER_START_DELAY_SECS, - default_cycle_secs = SINGLE_DISK_SCANNER_CYCLE_SECS, - "Using slower scanner defaults for single-disk deployments; explicit scanner cycle or start-delay settings still take precedence" + ?default_cycle_secs, + lifecycle_active = features.lifecycle, + replication_active = features.replication, + feature_inspection_failed = features.inspection_failed, + "Using slower scanner defaults for single-disk deployments; regular scanner cycles are preserved when maintenance features are active" ); } else { set_scanner_default_speed(ScannerSpeed::Default); @@ -641,6 +734,47 @@ mod tests { }); } + #[test] + fn test_single_disk_default_cycle_uses_long_interval_without_maintenance_features() { + assert_eq!( + single_disk_default_cycle_secs(ScannerMaintenanceFeatures::default()), + Some(SINGLE_DISK_SCANNER_CYCLE_SECS) + ); + } + + #[test] + fn test_single_disk_default_cycle_preserves_regular_cycle_for_lifecycle() { + assert_eq!( + single_disk_default_cycle_secs(ScannerMaintenanceFeatures { + lifecycle: true, + ..Default::default() + }), + None + ); + } + + #[test] + fn test_single_disk_default_cycle_preserves_regular_cycle_for_replication() { + assert_eq!( + single_disk_default_cycle_secs(ScannerMaintenanceFeatures { + replication: true, + ..Default::default() + }), + None + ); + } + + #[test] + fn test_single_disk_default_cycle_preserves_regular_cycle_on_inspection_failure() { + assert_eq!( + single_disk_default_cycle_secs(ScannerMaintenanceFeatures { + inspection_failed: true, + ..Default::default() + }), + None + ); + } + #[test] #[serial] fn test_cycle_interval_keeps_single_disk_cycle_with_explicit_speed() { diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 76816ed7b..8d33af5a6 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -23,6 +23,10 @@ use metrics::counter; use rand::seq::SliceRandom as _; use rustfs_common::heal_channel::HealScanMode; use rustfs_common::metrics::{Metric, Metrics, emit_scan_bucket_drive_complete}; +use rustfs_config::{ + DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS, ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, + ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, +}; use rustfs_ecstore::bucket::bucket_target_sys::BucketTargetSys; use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::GLOBAL_ExpiryState; use rustfs_ecstore::bucket::lifecycle::lifecycle::Lifecycle; @@ -44,14 +48,143 @@ use rustfs_utils::path::path_join_buf; use s3s::dto::{BucketLifecycleConfiguration, ReplicationConfiguration}; use std::collections::HashMap; use std::path::Path; -use std::time::SystemTime; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::{Instant, SystemTime}; use std::{fmt::Debug, sync::Arc}; use time::OffsetDateTime; -use tokio::sync::{Mutex, mpsc}; +use tokio::sync::{Mutex, Semaphore, mpsc}; use tokio::time::Duration; use tokio_util::sync::CancellationToken; use tracing::{debug, error, warn}; +const METRIC_SCANNER_SET_SCAN_CONCURRENCY_LIMIT: &str = "rustfs_scanner_set_scan_concurrency_limit"; +const METRIC_SCANNER_DISK_SCAN_CONCURRENCY_LIMIT: &str = "rustfs_scanner_disk_scan_concurrency_limit"; +const METRIC_SCANNER_SET_SCAN_WAIT_SECONDS: &str = "rustfs_scanner_set_scan_wait_seconds"; +const METRIC_SCANNER_DISK_SCAN_WAIT_SECONDS: &str = "rustfs_scanner_disk_scan_wait_seconds"; +const METRIC_SCANNER_SET_SCANS_ACTIVE: &str = "rustfs_scanner_set_scans_active"; +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"; + +struct SetScanActiveGuard { + active: Arc, +} + +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); + Self { active } + } +} + +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); + } +} + +struct DiskBucketScanActiveGuard { + active: Arc, + pool: String, + set: String, +} + +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); + Self { active, pool, set } + } +} + +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); + } +} + +fn decrement_atomic_usize(counter: &AtomicUsize) -> usize { + counter + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(1))) + .map(|previous| previous.saturating_sub(1)) + .unwrap_or_else(|current| current) +} + +fn record_disk_bucket_scans_queued(count: usize, pool: &str, set: &str) { + metrics::gauge!( + METRIC_SCANNER_DISK_BUCKET_SCANS_QUEUED, + "pool" => pool.to_owned(), + "set" => set.to_owned() + ) + .set(count as f64); +} + +fn decrement_disk_bucket_scans_queued(counter: &AtomicUsize, pool: &str, set: &str) { + let queued_count = decrement_atomic_usize(counter); + record_disk_bucket_scans_queued(queued_count, pool, set); +} + +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); +} + +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_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); +} + +fn scanner_concurrency_limit(configured: usize, available: usize) -> usize { + if available == 0 { + return 0; + } + + if configured == 0 { + available + } else { + configured.min(available).max(1) + } +} + +fn scanner_max_concurrent_set_scans(available: usize) -> usize { + scanner_concurrency_limit( + rustfs_utils::get_env_usize(ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS), + available, + ) +} + +fn scanner_max_concurrent_disk_scans(available: usize) -> usize { + scanner_concurrency_limit( + rustfs_utils::get_env_usize(ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS), + available, + ) +} + fn record_set_scan_failure(first_err: &mut Option, err: Error) { if first_err.is_none() { *first_err = Some(err); @@ -146,6 +279,7 @@ impl ScannerIO for ECStore { let all_buckets = self.list_bucket(&BucketOptions::default()).await?; if all_buckets.is_empty() { + reset_set_scan_gauges(); if let Err(e) = updates.send(DataUsageInfo::default()).await { error!("Failed to send data usage info: {}", e); } @@ -156,6 +290,24 @@ impl ScannerIO for ECStore { for pool in self.pools.iter() { total_results += pool.disk_set.len(); } + if total_results == 0 { + warn!("nsscanner: no disk sets available for non-empty bucket list"); + reset_set_scan_gauges(); + return Ok(()); + } + + 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); + debug!( + total_sets = total_results, + concurrency_limit = set_scan_limit, + "nsscanner: scanner set concurrency budget" + ); + 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); let results = vec![DataUsageCache::default(); total_results]; let results_mutex: Arc>> = Arc::new(Mutex::new(results)); @@ -178,6 +330,9 @@ impl ScannerIO for ECStore { let scan_mode_clone = scan_mode; let results_mutex_clone = results_mutex.clone(); let first_err_mutex_clone = first_err_mutex.clone(); + let set_scan_semaphore_clone = set_scan_semaphore.clone(); + let queued_set_scans_clone = queued_set_scans.clone(); + let active_set_scans_clone = active_set_scans.clone(); let (tx, mut rx) = mpsc::channel::(1); @@ -193,6 +348,25 @@ impl ScannerIO for ECStore { let all_buckets_clone = all_buckets.clone(); // Spawn task to run the scanner let scanner_fut = tokio::spawn(async move { + let permit_wait = child_token_clone.clone(); + let permit_wait_start = Instant::now(); + let _permit = tokio::select! { + permit = set_scan_semaphore_clone.acquire_owned() => match permit { + Ok(permit) => permit, + Err(_) => return, + }, + _ = permit_wait.cancelled() => return, + }; + metrics::histogram!( + METRIC_SCANNER_SET_SCAN_WAIT_SECONDS, + "pool" => pool_label.clone(), + "set" => set_label.clone() + ) + .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); + 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) .await @@ -280,6 +454,8 @@ 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); let _ = update_tx.send(()); @@ -300,15 +476,44 @@ impl ScannerIOCache for SetDisks { want_cycle: u64, scan_mode: HealScanMode, ) -> Result<()> { + let pool_label = self.pool_index.to_string(); + let set_label = self.set_index.to_string(); + if buckets.is_empty() { + reset_disk_bucket_scan_gauges(&pool_label, &set_label); return Ok(()); } let (disks, healing) = self.get_online_disks_with_healing(false).await; if disks.is_empty() { debug!("nsscanner_cache: no online disks available for set"); + reset_disk_bucket_scan_gauges(&pool_label, &set_label); 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); + debug!( + pool = self.pool_index, + set = self.set_index, + online_disks = disks.len(), + concurrency_limit = disk_scan_limit, + "nsscanner_cache: scanner disk concurrency budget" + ); + let disk_scan_semaphore = Arc::new(Semaphore::new(disk_scan_limit)); + 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); let mut old_cache = DataUsageCache::default(); old_cache.load(self.clone(), DATA_USAGE_CACHE_NAME).await?; @@ -412,12 +617,58 @@ impl ScannerIOCache for SetDisks { let store_clone_clone = self.clone(); let bucket_result_tx_clone_clone = bucket_result_tx_clone.clone(); let disk_clone = disk.clone(); + let disk_scan_semaphore_clone = disk_scan_semaphore.clone(); + let queued_disk_bucket_scans_clone = queued_disk_bucket_scans.clone(); + let active_disk_bucket_scans_clone = active_disk_bucket_scans.clone(); + let pool_label_clone = pool_label.clone(); + let set_label_clone = set_label.clone(); futs.push(tokio::spawn(async move { - while let Some(bucket) = bucket_rx_mutex_clone.lock().await.recv().await { + loop { + let Some(bucket) = bucket_rx_mutex_clone.lock().await.recv().await else { + break; + }; + if ctx_clone.is_cancelled() { + decrement_disk_bucket_scans_queued(&queued_disk_bucket_scans_clone, &pool_label_clone, &set_label_clone); break; } + let permit_wait = ctx_clone.clone(); + let permit_wait_start = Instant::now(); + let _permit = tokio::select! { + permit = disk_scan_semaphore_clone.clone().acquire_owned() => match permit { + Ok(permit) => permit, + Err(_) => { + decrement_disk_bucket_scans_queued( + &queued_disk_bucket_scans_clone, + &pool_label_clone, + &set_label_clone, + ); + break; + }, + }, + _ = permit_wait.cancelled() => { + decrement_disk_bucket_scans_queued( + &queued_disk_bucket_scans_clone, + &pool_label_clone, + &set_label_clone, + ); + break; + }, + }; + metrics::histogram!( + METRIC_SCANNER_DISK_SCAN_WAIT_SECONDS, + "pool" => pool_label_clone.clone(), + "set" => set_label_clone.clone() + ) + .record(permit_wait_start.elapsed().as_secs_f64()); + decrement_disk_bucket_scans_queued(&queued_disk_bucket_scans_clone, &pool_label_clone, &set_label_clone); + let _active_guard = DiskBucketScanActiveGuard::new( + active_disk_bucket_scans_clone.clone(), + pool_label_clone.clone(), + set_label_clone.clone(), + ); + debug!("nsscanner_disk: got bucket: {}", bucket.name); let cache_name = path_join_buf(&[&bucket.name, DATA_USAGE_CACHE_NAME]); @@ -532,6 +783,13 @@ impl ScannerIOCache for SetDisks { } let _ = join_all(futs).await; + 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); drop(bucket_result_tx_clone); @@ -721,6 +979,8 @@ impl ScannerIODisk for Disk { #[cfg(test)] mod tests { use super::*; + use serial_test::serial; + use temp_env::with_var; #[test] fn record_set_scan_failure_preserves_first_error() { @@ -750,6 +1010,49 @@ mod tests { assert!(err.to_string().contains("set failed")); } + #[test] + fn scanner_concurrency_limit_preserves_available_when_unconfigured() { + assert_eq!(scanner_concurrency_limit(0, 4), 4); + } + + #[test] + fn scanner_concurrency_limit_caps_to_configured_value() { + assert_eq!(scanner_concurrency_limit(2, 4), 2); + } + + #[test] + fn scanner_concurrency_limit_never_exceeds_available_work() { + assert_eq!(scanner_concurrency_limit(8, 4), 4); + } + + #[test] + fn scanner_concurrency_limit_handles_no_available_work() { + assert_eq!(scanner_concurrency_limit(2, 0), 0); + } + + #[test] + fn decrement_atomic_usize_saturates_at_zero() { + let counter = AtomicUsize::new(1); + assert_eq!(decrement_atomic_usize(&counter), 0); + assert_eq!(decrement_atomic_usize(&counter), 0); + } + + #[test] + #[serial] + fn scanner_max_concurrent_set_scans_uses_env_cap() { + with_var(ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, Some("2"), || { + assert_eq!(scanner_max_concurrent_set_scans(4), 2); + }); + } + + #[test] + #[serial] + fn scanner_max_concurrent_disk_scans_uses_env_cap() { + with_var(ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, Some("1"), || { + assert_eq!(scanner_max_concurrent_disk_scans(4), 1); + }); + } + #[test] #[cfg(windows)] fn is_xl_meta_path_accepts_windows_separator() {