From 76da2a48d0934daf82e816f8be84b377cb88421c Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Mon, 1 Jun 2026 05:55:46 +0800 Subject: [PATCH] feat(scanner): expose cycle observability controls (#3147) Co-authored-by: Henry Guo --- crates/common/src/heal_channel.rs | 26 ++++- crates/common/src/metrics.rs | 108 +++++++++++++++++- crates/config/src/constants/scanner.rs | 8 ++ crates/obs/src/metrics/collectors/scanner.rs | 73 +++++++++++- .../src/metrics/schema/entry/metric_name.rs | 14 +++ crates/obs/src/metrics/schema/scanner.rs | 63 ++++++++++ crates/obs/src/metrics/stats_collector.rs | 66 ++++++++++- crates/scanner/src/scanner.rs | 17 +++ crates/scanner/src/scanner_folder.rs | 38 +++++- 9 files changed, 398 insertions(+), 15 deletions(-) diff --git a/crates/common/src/heal_channel.rs b/crates/common/src/heal_channel.rs index f940d29ef..f19d6f9ff 100644 --- a/crates/common/src/heal_channel.rs +++ b/crates/common/src/heal_channel.rs @@ -110,6 +110,25 @@ pub enum HealScanMode { Deep = 2, } +impl HealScanMode { + pub const fn as_str(self) -> &'static str { + match self { + Self::Unknown => "unknown", + Self::Normal => "normal", + Self::Deep => "deep", + } + } + + pub const fn from_u8(value: u8) -> Option { + match value { + 0 => Some(Self::Unknown), + 1 => Some(Self::Normal), + 2 => Some(Self::Deep), + _ => None, + } + } +} + impl Serialize for HealScanMode { fn serialize(&self, serializer: S) -> Result where @@ -137,12 +156,7 @@ impl<'de> Deserialize<'de> for HealScanMode { where E: serde::de::Error, { - match value { - 0 => Ok(HealScanMode::Unknown), - 1 => Ok(HealScanMode::Normal), - 2 => Ok(HealScanMode::Deep), - _ => Err(E::custom(format!("invalid HealScanMode value: {value}"))), - } + HealScanMode::from_u8(value).ok_or_else(|| E::custom(format!("invalid HealScanMode value: {value}"))) } fn visit_u64(self, value: u64) -> Result diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index 9fa097914..8bdee2e1f 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::heal_channel::HealScanMode; use crate::last_minute::{AccElem, LastMinuteLatency}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; @@ -22,7 +23,7 @@ use std::{ pin::Pin, sync::{ Arc, Mutex, OnceLock, - atomic::{AtomicU64, Ordering}, + atomic::{AtomicU8, AtomicU64, Ordering}, }, time::{Duration, SystemTime}, }; @@ -340,8 +341,19 @@ pub struct Metrics { actions_latency: Vec, current_paths: Arc>>>, cycle_info: Arc>>, + current_scan_mode: AtomicU8, + last_scan_cycle_result: AtomicU8, + last_scan_cycle_duration_millis: AtomicU64, + failed_scan_cycles: AtomicU64, } +const SCAN_CYCLE_RESULT_UNKNOWN: u8 = 0; +const SCAN_CYCLE_RESULT_SUCCESS: u8 = 1; +const SCAN_CYCLE_RESULT_ERROR: u8 = 2; +const SCAN_CYCLE_RESULT_UNKNOWN_LABEL: &str = "unknown"; +const SCAN_CYCLE_RESULT_SUCCESS_LABEL: &str = "success"; +const SCAN_CYCLE_RESULT_ERROR_LABEL: &str = "error"; + #[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct CurrentCycle { pub current: u64, @@ -376,6 +388,11 @@ pub struct ScannerMetricsReport { pub life_time_ilm: HashMap, pub last_minute: ScannerLastMinute, pub active_paths: Vec, + pub current_scan_mode: String, + pub last_cycle_result: String, + pub last_cycle_result_code: u64, + pub last_cycle_duration_seconds: f64, + pub failed_cycles: u64, } impl CurrentCycle { @@ -409,8 +426,25 @@ fn emit_otel_counter(metric: usize, count: u64) { } } +fn scan_cycle_result_label(result: u8) -> &'static str { + match result { + SCAN_CYCLE_RESULT_SUCCESS => SCAN_CYCLE_RESULT_SUCCESS_LABEL, + SCAN_CYCLE_RESULT_ERROR => SCAN_CYCLE_RESULT_ERROR_LABEL, + _ => SCAN_CYCLE_RESULT_UNKNOWN_LABEL, + } +} + +fn duration_millis_saturated(duration: Duration) -> u64 { + duration.as_millis().min(u64::MAX as u128) as u64 +} + pub fn emit_scan_cycle_complete(success: bool, duration: Duration) { - let result = if success { "success" } else { "error" }; + let result = if success { + SCAN_CYCLE_RESULT_SUCCESS_LABEL + } else { + SCAN_CYCLE_RESULT_ERROR_LABEL + }; + global_metrics().record_scan_cycle_complete(success, duration); metrics::counter!(OTEL_SCANNER_CYCLES, "result" => result).increment(1); if success { metrics::gauge!(OTEL_SCANNER_CYCLE_DURATION_SECONDS).set(duration.as_secs_f64()); @@ -449,6 +483,10 @@ impl Metrics { .collect(), current_paths: Arc::new(RwLock::new(HashMap::new())), cycle_info: Arc::new(RwLock::new(None)), + current_scan_mode: AtomicU8::new(HealScanMode::Unknown as u8), + last_scan_cycle_result: AtomicU8::new(SCAN_CYCLE_RESULT_UNKNOWN), + last_scan_cycle_duration_millis: AtomicU64::new(0), + failed_scan_cycles: AtomicU64::new(0), } } @@ -583,6 +621,30 @@ impl Metrics { self.cycle_info.read().await.clone() } + pub fn set_current_scan_mode(&self, scan_mode: HealScanMode) { + self.current_scan_mode.store(scan_mode as u8, Ordering::Relaxed); + } + + pub fn clear_current_scan_mode(&self) { + self.set_current_scan_mode(HealScanMode::Unknown); + } + + pub fn current_scan_mode(&self) -> HealScanMode { + HealScanMode::from_u8(self.current_scan_mode.load(Ordering::Relaxed)).unwrap_or(HealScanMode::Unknown) + } + + pub fn record_scan_cycle_complete(&self, success: bool, duration: Duration) { + let result = if success { + SCAN_CYCLE_RESULT_SUCCESS + } else { + self.failed_scan_cycles.fetch_add(1, Ordering::Relaxed); + SCAN_CYCLE_RESULT_ERROR + }; + self.last_scan_cycle_result.store(result, Ordering::Relaxed); + self.last_scan_cycle_duration_millis + .store(duration_millis_saturated(duration), Ordering::Relaxed); + } + /// Snapshot of every path currently being scanned. pub async fn get_current_paths(&self) -> Vec { let paths = self.current_paths.read().await; @@ -618,6 +680,12 @@ impl Metrics { m.collected_at = Utc::now(); m.active_paths = self.get_current_paths().await; m.active_scan_paths = m.active_paths.len(); + m.current_scan_mode = self.current_scan_mode().as_str().to_string(); + 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; + m.last_cycle_duration_seconds = self.last_scan_cycle_duration_millis.load(Ordering::Relaxed) as f64 / 1000.0; + m.failed_cycles = self.failed_scan_cycles.load(Ordering::Relaxed); // Lifetime operation counts for i in 0..Metric::Last as usize { @@ -795,4 +863,40 @@ mod tests { assert_eq!(report.current_started, cycle_started); } + + #[tokio::test] + async fn report_includes_current_scan_mode() { + let metrics = Metrics::new(); + metrics.set_current_scan_mode(HealScanMode::Deep); + + let report = metrics.report().await; + + assert_eq!(report.current_scan_mode, HealScanMode::Deep.as_str()); + } + + #[tokio::test] + async fn report_includes_last_scan_cycle_result() { + let metrics = Metrics::new(); + metrics.record_scan_cycle_complete(false, Duration::from_millis(1500)); + + let report = metrics.report().await; + + assert_eq!(report.last_cycle_result, SCAN_CYCLE_RESULT_ERROR_LABEL); + assert_eq!(report.last_cycle_result_code, SCAN_CYCLE_RESULT_ERROR as u64); + assert_eq!(report.last_cycle_duration_seconds, 1.5); + assert_eq!(report.failed_cycles, 1); + } + + #[tokio::test] + async fn report_tracks_successful_scan_cycle_without_failed_increment() { + let metrics = Metrics::new(); + 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_duration_seconds, 2.0); + assert_eq!(report.failed_cycles, 0); + } } diff --git a/crates/config/src/constants/scanner.rs b/crates/config/src/constants/scanner.rs index 6875bfc6d..4a15c9eae 100644 --- a/crates/config/src/constants/scanner.rs +++ b/crates/config/src/constants/scanner.rs @@ -93,6 +93,11 @@ pub const ENV_SCANNER_MAX_CONCURRENT_SET_SCANS: &str = "RUSTFS_SCANNER_MAX_CONCU /// - Example: `export RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS=1` pub const ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS: &str = "RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS"; +/// Environment variable that controls how often scanner object loops yield to the async runtime. +/// A value of `0` disables this extra object-count yield. +/// - Example: `export RUSTFS_SCANNER_YIELD_EVERY_N_OBJECTS=32` +pub const ENV_SCANNER_YIELD_EVERY_N_OBJECTS: &str = "RUSTFS_SCANNER_YIELD_EVERY_N_OBJECTS"; + /// Default scanner idle mode. pub const DEFAULT_SCANNER_IDLE_MODE: bool = true; @@ -104,6 +109,9 @@ pub const DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS: usize = 0; /// `0` means no additional limit beyond available disks in the set. pub const DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS: usize = 0; +/// Default object interval for cooperative scanner yields. +pub const DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS: u64 = 128; + /// 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 8f3737d20..fdf56802c 100644 --- a/crates/obs/src/metrics/collectors/scanner.rs +++ b/crates/obs/src/metrics/collectors/scanner.rs @@ -21,8 +21,10 @@ use crate::metrics::report::PrometheusMetric; use crate::metrics::schema::scanner::{ - 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, + SCANNER_ACTIVE_PATHS_MD, SCANNER_BUCKET_SCANS_FINISHED_MD, SCANNER_BUCKET_SCANS_STARTED_MD, SCANNER_COMPLETED_CYCLES_MD, + SCANNER_CURRENT_CYCLE_AGE_SECONDS_MD, SCANNER_CURRENT_CYCLE_MD, SCANNER_CURRENT_SCAN_MODE_MD, SCANNER_DIRECTORIES_SCANNED_MD, + SCANNER_FAILED_CYCLES_MD, SCANNER_LAST_ACTIVITY_SECONDS_MD, SCANNER_LAST_CYCLE_DURATION_SECONDS_MD, + SCANNER_LAST_CYCLE_RESULT_MD, SCANNER_OBJECTS_SCANNED_MD, SCANNER_VERSIONS_SCANNED_MD, }; /// Scanner statistics. @@ -42,6 +44,20 @@ pub struct ScannerStats { pub last_activity_seconds: u64, /// Number of scanner paths currently being processed pub active_paths: u64, + /// Current scanner cycle number, or zero when idle + pub current_cycle: u64, + /// Number of scanner cycles completed since server start + pub completed_cycles: u64, + /// Seconds elapsed since the current scanner cycle started + pub current_cycle_age_seconds: u64, + /// Current scanner mode: 0 unknown or idle, 1 normal, 2 deep bitrot scan + pub current_scan_mode: u64, + /// Last scanner cycle result: 0 unknown, 1 success, 2 error + pub last_cycle_result: u64, + /// Duration in seconds of the last finished scanner cycle + pub last_cycle_duration_seconds: f64, + /// Number of scanner cycles that failed since server start + pub failed_cycles: u64, } /// Collects scanner metrics from the given stats. @@ -57,6 +73,13 @@ 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_CURRENT_CYCLE_MD, stats.current_cycle as f64), + PrometheusMetric::from_descriptor(&SCANNER_COMPLETED_CYCLES_MD, stats.completed_cycles as f64), + PrometheusMetric::from_descriptor(&SCANNER_CURRENT_CYCLE_AGE_SECONDS_MD, stats.current_cycle_age_seconds 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_DURATION_SECONDS_MD, stats.last_cycle_duration_seconds), + PrometheusMetric::from_descriptor(&SCANNER_FAILED_CYCLES_MD, stats.failed_cycles as f64), ] } @@ -75,12 +98,19 @@ mod tests { versions_scanned: 1500000, last_activity_seconds: 30, active_paths: 4, + current_cycle: 12, + completed_cycles: 11, + current_cycle_age_seconds: 90, + current_scan_mode: 2, + last_cycle_result: 1, + last_cycle_duration_seconds: 42.5, + failed_cycles: 3, }; let metrics = collect_scanner_metrics(&stats); report_metrics(&metrics); - assert_eq!(metrics.len(), 7); + assert_eq!(metrics.len(), 14); let objects = metrics.iter().find(|m| m.value == 1000000.0); assert!(objects.is_some()); @@ -92,6 +122,41 @@ mod tests { .iter() .find(|m| m.name == SCANNER_ACTIVE_PATHS_MD.get_full_metric_name()); assert_eq!(active_paths.map(|m| m.value), Some(4.0)); + + let current_cycle = metrics + .iter() + .find(|m| m.name == SCANNER_CURRENT_CYCLE_MD.get_full_metric_name()); + assert_eq!(current_cycle.map(|m| m.value), Some(12.0)); + + let completed_cycles = metrics + .iter() + .find(|m| m.name == SCANNER_COMPLETED_CYCLES_MD.get_full_metric_name()); + assert_eq!(completed_cycles.map(|m| m.value), Some(11.0)); + + let current_cycle_age = metrics + .iter() + .find(|m| m.name == SCANNER_CURRENT_CYCLE_AGE_SECONDS_MD.get_full_metric_name()); + assert_eq!(current_cycle_age.map(|m| m.value), Some(90.0)); + + let current_scan_mode = metrics + .iter() + .find(|m| m.name == SCANNER_CURRENT_SCAN_MODE_MD.get_full_metric_name()); + assert_eq!(current_scan_mode.map(|m| m.value), Some(2.0)); + + let last_cycle_result = metrics + .iter() + .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_duration = metrics + .iter() + .find(|m| m.name == SCANNER_LAST_CYCLE_DURATION_SECONDS_MD.get_full_metric_name()); + assert_eq!(last_cycle_duration.map(|m| m.value), Some(42.5)); + + let failed_cycles = metrics + .iter() + .find(|m| m.name == SCANNER_FAILED_CYCLES_MD.get_full_metric_name()); + assert_eq!(failed_cycles.map(|m| m.value), Some(3.0)); } #[test] @@ -99,7 +164,7 @@ mod tests { let stats = ScannerStats::default(); let metrics = collect_scanner_metrics(&stats); - assert_eq!(metrics.len(), 7); + assert_eq!(metrics.len(), 14); 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 55531ff70..7c11e7495 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, + ScannerCurrentCycle, + ScannerCompletedCycles, + ScannerCurrentCycleAgeSeconds, + ScannerCurrentScanMode, + ScannerLastCycleResult, + ScannerLastCycleDurationSeconds, + ScannerFailedCycles, // CPU system-related metrics SysCPUAvgIdle, @@ -620,6 +627,13 @@ impl MetricName { Self::ScannerVersionsScanned => "versions_scanned".to_string(), Self::ScannerLastActivitySeconds => "last_activity_seconds".to_string(), Self::ScannerActivePaths => "active_paths".to_string(), + Self::ScannerCurrentCycle => "current_cycle".to_string(), + Self::ScannerCompletedCycles => "completed_cycles".to_string(), + Self::ScannerCurrentCycleAgeSeconds => "current_cycle_age_seconds".to_string(), + Self::ScannerCurrentScanMode => "current_scan_mode".to_string(), + Self::ScannerLastCycleResult => "last_cycle_result".to_string(), + Self::ScannerLastCycleDurationSeconds => "last_cycle_duration_seconds".to_string(), + Self::ScannerFailedCycles => "failed_cycles".to_string(), // CPU system-related metrics Self::SysCPUAvgIdle => "avg_idle".to_string(), diff --git a/crates/obs/src/metrics/schema/scanner.rs b/crates/obs/src/metrics/schema/scanner.rs index 4ad9d96e9..d25c5fcda 100644 --- a/crates/obs/src/metrics/schema/scanner.rs +++ b/crates/obs/src/metrics/schema/scanner.rs @@ -79,3 +79,66 @@ pub static SCANNER_ACTIVE_PATHS_MD: LazyLock = LazyLock::new(| subsystems::SCANNER, ) }); + +pub static SCANNER_CURRENT_CYCLE_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCurrentCycle, + "Current scanner cycle number, or zero when no cycle is running.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_COMPLETED_CYCLES_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCompletedCycles, + "Total number of scanner cycles completed since server start.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_CURRENT_CYCLE_AGE_SECONDS_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCurrentCycleAgeSeconds, + "Time elapsed (in seconds) since the current scanner cycle started, or zero when no cycle is running.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_CURRENT_SCAN_MODE_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerCurrentScanMode, + "Current scanner mode: 0 unknown or idle, 1 normal, 2 deep bitrot scan.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_LAST_CYCLE_RESULT_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerLastCycleResult, + "Last scanner cycle result: 0 unknown, 1 success, 2 error.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_LAST_CYCLE_DURATION_SECONDS_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::ScannerLastCycleDurationSeconds, + "Duration in seconds of the last finished scanner cycle.", + &[], + subsystems::SCANNER, + ) +}); + +pub static SCANNER_FAILED_CYCLES_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::ScannerFailedCycles, + "Total number of scanner cycles that failed since server start.", + &[], + subsystems::SCANNER, + ) +}); diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 6cff3dcd2..439a432e8 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -27,6 +27,7 @@ use crate::metrics::collectors::{ ProcessStatusType, ReplicationStats, ResourceStats, ScannerStats, }; use chrono::Utc; +use rustfs_common::heal_channel::HealScanMode; use rustfs_common::metrics::global_metrics; use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::{GLOBAL_ExpiryState, GLOBAL_TransitionState}; use rustfs_ecstore::bucket::metadata_sys::get_quota_config; @@ -44,6 +45,26 @@ use std::time::Duration; use sysinfo::{Networks, System}; use tracing::{instrument, warn}; +fn current_scanner_cycle_age_seconds( + current_cycle: u64, + current_started: chrono::DateTime, + now: chrono::DateTime, +) -> u64 { + if current_cycle == 0 { + 0 + } else { + now.signed_duration_since(current_started).num_seconds().max(0) as u64 + } +} + +fn scanner_scan_mode_code(scan_mode: &str) -> u64 { + match scan_mode { + mode if mode == HealScanMode::Normal.as_str() => HealScanMode::Normal as u8 as u64, + mode if mode == HealScanMode::Deep.as_str() => HealScanMode::Deep as u8 as u64, + _ => HealScanMode::Unknown as u8 as u64, + } +} + const DRIVE_STATE_OK: &str = "ok"; const DRIVE_STATE_ONLINE: &str = "online"; const DRIVE_STATE_UNFORMATTED: &str = "unformatted"; @@ -869,13 +890,17 @@ pub async fn collect_ilm_metric_stats() -> Option { /// rustfs-obs scanner collector shape. pub async fn collect_scanner_metric_stats() -> Option { let metrics = global_metrics().report().await; + let now = Utc::now(); let bucket_scans_finished = metrics.life_time_ops.get("scan_bucket_drive").copied().unwrap_or_default(); + let completed_cycles = metrics.life_time_ops.get("scan_cycle").copied().unwrap_or_default(); let directories_scanned = metrics.life_time_ops.get("scan_folder").copied().unwrap_or_default(); let objects_scanned = metrics.life_time_ops.get("scan_object").copied().unwrap_or_default(); 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 last_activity_seconds = now.signed_duration_since(reference_time).num_seconds().max(0) as u64; let active_paths = metrics.active_scan_paths as u64; + let current_cycle_age_seconds = current_scanner_cycle_age_seconds(metrics.current_cycle, metrics.current_started, now); + let current_scan_mode = scanner_scan_mode_code(&metrics.current_scan_mode); Some(ScannerStats { bucket_scans_finished, @@ -888,6 +913,13 @@ pub async fn collect_scanner_metric_stats() -> Option { versions_scanned, last_activity_seconds, active_paths, + current_cycle: metrics.current_cycle, + completed_cycles, + current_cycle_age_seconds, + current_scan_mode, + last_cycle_result: metrics.last_cycle_result_code, + last_cycle_duration_seconds: metrics.last_cycle_duration_seconds, + failed_cycles: metrics.failed_cycles, }) } @@ -957,4 +989,36 @@ mod tests { assert_eq!(stats.write_health, 1); assert_eq!(stats.health, 1); } + + #[test] + fn current_scanner_cycle_age_seconds_returns_zero_when_idle() { + let now = Utc::now(); + + assert_eq!(current_scanner_cycle_age_seconds(0, now - chrono::Duration::seconds(30), now), 0); + } + + #[test] + fn current_scanner_cycle_age_seconds_clamps_future_start() { + let now = Utc::now(); + + assert_eq!(current_scanner_cycle_age_seconds(4, now + chrono::Duration::seconds(30), now), 0); + } + + #[test] + fn current_scanner_cycle_age_seconds_reports_active_elapsed_time() { + let now = Utc::now(); + + assert_eq!(current_scanner_cycle_age_seconds(4, now - chrono::Duration::seconds(45), now), 45); + } + + #[test] + fn scanner_scan_mode_code_maps_known_modes() { + assert_eq!(scanner_scan_mode_code(HealScanMode::Normal.as_str()), HealScanMode::Normal as u8 as u64); + assert_eq!(scanner_scan_mode_code(HealScanMode::Deep.as_str()), HealScanMode::Deep as u8 as u64); + } + + #[test] + fn scanner_scan_mode_code_maps_unknown_mode() { + assert_eq!(scanner_scan_mode_code(""), HealScanMode::Unknown as u8 as u64); + } } diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 1c67d666c..2e0396d57 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -438,6 +438,7 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc background_heal_info.bitrot_start_cycle, background_heal_info.bitrot_start_time, ); + let _scan_mode_guard = ScannerScanModeGuard::new(scan_mode); if let Some(new_heal_info) = background_heal_info_for_scan_start(background_heal_info.clone(), cycle_info.current, scan_mode, Utc::now()) { @@ -475,6 +476,7 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc cycle_info.next += 1; cycle_info.current = 0; cycle_info.cycle_completed.push(Utc::now()); + global_metrics().clear_current_scan_mode(); info!(duration = ?now.elapsed(), cycles_total=cycle_info.cycle_completed.len(), "Success run data scanner cycle"); @@ -553,6 +555,21 @@ pub async fn run_data_scanner(ctx: CancellationToken, storeapi: Arc) -> Ok(()) } +struct ScannerScanModeGuard; + +impl ScannerScanModeGuard { + fn new(scan_mode: HealScanMode) -> Self { + global_metrics().set_current_scan_mode(scan_mode); + Self + } +} + +impl Drop for ScannerScanModeGuard { + fn drop(&mut self) { + global_metrics().clear_current_scan_mode(); + } +} + /// Store data usage info in backend. Will store all objects sent on the receiver until closed. #[instrument(skip(ctx, storeapi))] pub async fn store_data_usage_in_backend( diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 76bd0b406..ee904b8dc 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -57,7 +57,6 @@ use tracing::{debug, error, info, warn}; const DATA_USAGE_UPDATE_DIR_CYCLES: u32 = 16; const DATA_SCANNER_COMPACT_LEAST_OBJECT: usize = 500; -const YIELD_EVERY_N_OBJECTS: u64 = 128; const DATA_SCANNER_COMPACT_AT_CHILDREN: usize = 10000; const DATA_SCANNER_COMPACT_AT_FOLDERS: usize = DATA_SCANNER_COMPACT_AT_CHILDREN / 4; const DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS: usize = 250_000; @@ -144,6 +143,17 @@ fn scanner_excess_folders_threshold() -> u64 { ) } +fn scanner_yield_every_n_objects() -> u64 { + rustfs_utils::get_env_u64( + rustfs_config::ENV_SCANNER_YIELD_EVERY_N_OBJECTS, + rustfs_config::DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, + ) +} + +fn should_yield_after_object(object_count: u64, yield_every: u64) -> bool { + yield_every > 0 && object_count.is_multiple_of(yield_every) +} + fn should_alert_excessive_versions(remaining_versions: usize, cumulative_size: i64) -> (bool, bool) { let too_many_versions = remaining_versions as u64 >= scanner_excess_versions_threshold(); let too_large_versions = cumulative_size > 0 && cumulative_size as u64 >= scanner_excess_version_size_threshold(); @@ -916,6 +926,7 @@ impl FolderScanner { let mut new_folders: Vec = Vec::new(); let mut found_objects = false; let mut object_count: u64 = 0; + let yield_every_objects = scanner_yield_every_n_objects(); let dir_path = path_join_buf(&[&self.root, &folder.name]); @@ -1098,7 +1109,7 @@ impl FolderScanner { timer.sleep().await; - if object_count.is_multiple_of(YIELD_EVERY_N_OBJECTS) { + if should_yield_after_object(object_count, yield_every_objects) { tokio::task::yield_now().await; } } @@ -1816,6 +1827,29 @@ mod tests { }); } + #[test] + #[serial] + fn test_scanner_yield_every_n_objects_uses_env() { + with_var(rustfs_config::ENV_SCANNER_YIELD_EVERY_N_OBJECTS, Some("32"), || { + assert_eq!(scanner_yield_every_n_objects(), 32); + }); + } + + #[test] + #[serial] + fn test_scanner_yield_every_n_objects_uses_default() { + with_var_unset(rustfs_config::ENV_SCANNER_YIELD_EVERY_N_OBJECTS, || { + assert_eq!(scanner_yield_every_n_objects(), rustfs_config::DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS); + }); + } + + #[test] + fn test_should_yield_after_object_respects_interval_and_disable() { + assert!(!should_yield_after_object(127, 128)); + assert!(should_yield_after_object(128, 128)); + assert!(!should_yield_after_object(128, 0)); + } + #[tokio::test] #[serial] async fn test_record_failed_prunes_to_max_entries() {