From 4ae070c8ccb5c1e06a7c6f9ccb32734b6159161a Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Sun, 7 Jun 2026 14:40:04 +0800 Subject: [PATCH] feat(scanner): expose scan partial source status (#3247) * feat(scanner): expose scan partial source status * fix(scanner): expose partial source in aggregated metrics --------- Co-authored-by: Henry Guo Co-authored-by: cxymds Co-authored-by: houseme --- crates/common/src/metrics.rs | 106 ++++++++++++++++++++++++- crates/ecstore/src/metrics_realtime.rs | 35 +++++++- crates/madmin/src/metrics.rs | 86 +++++++++++++++++++- crates/scanner/src/scanner.rs | 38 ++++++++- 4 files changed, 258 insertions(+), 7 deletions(-) diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index bbd447857..546a7fcc7 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -267,6 +267,31 @@ impl ScannerWorkSource { } } + fn code(self) -> u8 { + match self { + Self::Usage => 1, + Self::Lifecycle => 2, + Self::BucketReplication => 3, + Self::SiteReplication => 4, + Self::Heal => 5, + Self::Bitrot => 6, + Self::Alerts => 7, + } + } + + fn from_code(code: u8) -> Option { + match code { + 1 => Some(Self::Usage), + 2 => Some(Self::Lifecycle), + 3 => Some(Self::BucketReplication), + 4 => Some(Self::SiteReplication), + 5 => Some(Self::Heal), + 6 => Some(Self::Bitrot), + 7 => Some(Self::Alerts), + _ => None, + } + } + fn all() -> &'static [Self] { &Self::ALL } @@ -521,6 +546,7 @@ pub struct Metrics { scanner_disk_bucket_scan_states: Mutex>, last_scan_cycle_result: AtomicU8, last_scan_cycle_partial_reason: AtomicU8, + last_scan_cycle_partial_source: AtomicU8, last_scan_cycle_duration_millis: AtomicU64, last_scan_cycle_objects_scanned: AtomicU64, last_scan_cycle_directories_scanned: AtomicU64, @@ -539,6 +565,7 @@ pub struct Metrics { partial_scan_cycles_runtime: AtomicU64, partial_scan_cycles_objects: AtomicU64, partial_scan_cycles_directories: AtomicU64, + partial_scan_cycles_by_source: Vec, scanner_yield_duration_millis: AtomicU64, scanner_throttle_sleep_duration_millis: AtomicU64, scanner_ilm_actions: AtomicU64, @@ -650,6 +677,12 @@ pub struct ScannerSourceWorkSnapshot { pub skipped: u64, } +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct ScannerSourceCycleSnapshot { + pub source: String, + pub cycles: u64, +} + #[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct ScannerLastMinute { pub actions: HashMap, @@ -714,6 +747,10 @@ pub struct ScannerMetricsReport { pub last_cycle_partial_reason: String, #[serde(default)] pub last_cycle_partial_reason_code: u64, + #[serde(default)] + pub last_cycle_partial_source: String, + #[serde(default)] + pub last_cycle_partial_source_code: u64, pub last_cycle_duration_seconds: f64, #[serde(default)] pub last_cycle_objects_scanned: u64, @@ -749,6 +786,8 @@ pub struct ScannerMetricsReport { #[serde(default)] pub partial_cycles_directories: u64, #[serde(default)] + pub partial_cycles_by_source: Vec, + #[serde(default)] pub throttle_idle_mode_enabled: bool, #[serde(default)] pub throttle_sleep_factor: f64, @@ -863,6 +902,15 @@ pub fn emit_scan_cycle_partial(duration: Duration, reason: ScanCyclePartialReaso metrics::counter!(OTEL_SCANNER_CYCLES, "result" => SCAN_CYCLE_RESULT_PARTIAL_LABEL).increment(1); } +pub fn emit_scan_cycle_partial_with_source( + duration: Duration, + reason: ScanCyclePartialReason, + source: Option, +) { + global_metrics().record_scan_cycle_partial_with_source(duration, reason, source); + metrics::counter!(OTEL_SCANNER_CYCLES, "result" => SCAN_CYCLE_RESULT_PARTIAL_LABEL).increment(1); +} + pub fn emit_scan_bucket_drive_complete(success: bool, bucket: &str, disk: &str, duration: Duration) { let result = if success { "success" } else { "error" }; metrics::counter!( @@ -931,6 +979,7 @@ impl Metrics { 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_partial_source: AtomicU8::new(0), last_scan_cycle_duration_millis: AtomicU64::new(0), last_scan_cycle_objects_scanned: AtomicU64::new(0), last_scan_cycle_directories_scanned: AtomicU64::new(0), @@ -949,6 +998,7 @@ impl Metrics { partial_scan_cycles_runtime: AtomicU64::new(0), partial_scan_cycles_objects: AtomicU64::new(0), partial_scan_cycles_directories: AtomicU64::new(0), + partial_scan_cycles_by_source: ScannerWorkSource::all().iter().map(|_| AtomicU64::new(0)).collect(), scanner_yield_duration_millis: AtomicU64::new(0), scanner_throttle_sleep_duration_millis: AtomicU64::new(0), scanner_ilm_actions: AtomicU64::new(0), @@ -1358,11 +1408,21 @@ impl Metrics { 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_partial_source.store(0, Ordering::Relaxed); self.last_scan_cycle_duration_millis .store(duration_millis_saturated(duration), Ordering::Relaxed); } pub fn record_scan_cycle_partial(&self, duration: Duration, reason: ScanCyclePartialReason) { + self.record_scan_cycle_partial_with_source(duration, reason, None); + } + + pub fn record_scan_cycle_partial_with_source( + &self, + duration: Duration, + reason: ScanCyclePartialReason, + source: Option, + ) { self.partial_scan_cycles.fetch_add(1, Ordering::Relaxed); match reason { ScanCyclePartialReason::Unknown => &self.partial_scan_cycles_unknown, @@ -1374,6 +1434,13 @@ impl Metrics { 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_partial_source + .store(source.map(ScannerWorkSource::code).unwrap_or_default(), Ordering::Relaxed); + if let Some(source) = source + && let Some(cycles) = self.partial_scan_cycles_by_source.get(source.index()) + { + cycles.fetch_add(1, Ordering::Relaxed); + } self.last_scan_cycle_duration_millis .store(duration_millis_saturated(duration), Ordering::Relaxed); } @@ -1647,6 +1714,12 @@ impl Metrics { 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; + let partial_source = ScannerWorkSource::from_code(self.last_scan_cycle_partial_source.load(Ordering::Relaxed)); + m.last_cycle_partial_source = partial_source + .map(ScannerWorkSource::as_str) + .unwrap_or(SCAN_CYCLE_RESULT_UNKNOWN_LABEL) + .to_string(); + m.last_cycle_partial_source_code = partial_source.map(|source| source.code().into()).unwrap_or_default(); 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); @@ -1667,6 +1740,17 @@ impl Metrics { 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.partial_cycles_by_source = ScannerWorkSource::all() + .iter() + .filter_map(|source| { + self.partial_scan_cycles_by_source + .get(source.index()) + .map(|cycles| ScannerSourceCycleSnapshot { + source: source.as_str().to_string(), + cycles: cycles.load(Ordering::Relaxed), + }) + }) + .collect(); 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; @@ -2081,7 +2165,11 @@ 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_partial_with_source( + Duration::from_millis(500), + ScanCyclePartialReason::Runtime, + Some(ScannerWorkSource::Heal), + ); metrics.record_scan_cycle_complete(true, Duration::from_secs(2)); let report = metrics.report().await; @@ -2090,6 +2178,8 @@ mod tests { 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_partial_source, SCAN_CYCLE_RESULT_UNKNOWN_LABEL); + assert_eq!(report.last_cycle_partial_source_code, 0); assert_eq!(report.last_cycle_duration_seconds, 2.0); assert_eq!(report.failed_cycles, 0); assert_eq!(report.partial_cycles, 1); @@ -2098,7 +2188,11 @@ mod tests { #[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), ScanCyclePartialReason::Directories); + metrics.record_scan_cycle_partial_with_source( + Duration::from_millis(2500), + ScanCyclePartialReason::Directories, + Some(ScannerWorkSource::Usage), + ); let report = metrics.report().await; @@ -2106,6 +2200,8 @@ mod tests { 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_partial_source, ScannerWorkSource::Usage.as_str()); + assert_eq!(report.last_cycle_partial_source_code, 1); assert_eq!(report.last_cycle_duration_seconds, 2.5); assert_eq!(report.failed_cycles, 0); assert_eq!(report.partial_cycles, 1); @@ -2113,6 +2209,12 @@ mod tests { assert_eq!(report.partial_cycles_runtime, 0); assert_eq!(report.partial_cycles_objects, 0); assert_eq!(report.partial_cycles_unknown, 0); + let usage = report + .partial_cycles_by_source + .iter() + .find(|source| source.source == ScannerWorkSource::Usage.as_str()) + .expect("usage partial source count should be visible"); + assert_eq!(usage.cycles, 1); } #[tokio::test] diff --git a/crates/ecstore/src/metrics_realtime.rs b/crates/ecstore/src/metrics_realtime.rs index 1944eb8ad..64b8bea9a 100644 --- a/crates/ecstore/src/metrics_realtime.rs +++ b/crates/ecstore/src/metrics_realtime.rs @@ -18,7 +18,8 @@ use rustfs_common::{GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_ADDR, heal_channel::Dr use rustfs_io_metrics::internode_metrics::global_internode_metrics; use rustfs_madmin::metrics::{ DiskIOStats, DiskMetric, LastMinute as MadminLastMinute, NetDevLine, NetMetrics, RPCMetrics, RealtimeMetrics, - ScannerMetrics as MadminScannerMetrics, TimedAction as MadminTimedAction, + ScannerMetrics as MadminScannerMetrics, ScannerSourceCycleSnapshot as MadminScannerSourceCycleSnapshot, + TimedAction as MadminTimedAction, }; use rustfs_utils::os::get_drive_stats; use serde::{Deserialize, Serialize}; @@ -99,6 +100,16 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo .collect(), }, active_paths: metrics.active_paths, + last_cycle_partial_source: metrics.last_cycle_partial_source, + last_cycle_partial_source_code: metrics.last_cycle_partial_source_code, + partial_cycles_by_source: metrics + .partial_cycles_by_source + .into_iter() + .map(|source| MadminScannerSourceCycleSnapshot { + source: source.source, + cycles: source.cycles, + }) + .collect(), } } @@ -343,4 +354,26 @@ mod test { metrics.reset_for_test(); } + + #[test] + fn scanner_metrics_mapping_preserves_partial_source_status() { + let scanner = to_madmin_scanner_metrics(rustfs_common::metrics::ScannerMetricsReport { + last_cycle_partial_source: "usage".to_string(), + last_cycle_partial_source_code: 1, + partial_cycles_by_source: vec![rustfs_common::metrics::ScannerSourceCycleSnapshot { + source: "usage".to_string(), + cycles: 2, + }], + ..Default::default() + }); + + assert_eq!(scanner.last_cycle_partial_source, "usage"); + assert_eq!(scanner.last_cycle_partial_source_code, 1); + let usage = scanner + .partial_cycles_by_source + .iter() + .find(|source| source.source == "usage") + .expect("usage partial source should be mapped"); + assert_eq!(usage.cycles, 2); + } } diff --git a/crates/madmin/src/metrics.rs b/crates/madmin/src/metrics.rs index 7c72224bf..f5b5c0c30 100644 --- a/crates/madmin/src/metrics.rs +++ b/crates/madmin/src/metrics.rs @@ -120,6 +120,14 @@ pub struct LastMinute { pub ilm: HashMap, } +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct ScannerSourceCycleSnapshot { + #[serde(rename = "source", default)] + pub source: String, + #[serde(rename = "cycles", default)] + pub cycles: u64, +} + #[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct ScannerMetrics { #[serde(rename = "collected")] @@ -140,12 +148,21 @@ pub struct ScannerMetrics { pub last_minute: LastMinute, #[serde(rename = "active")] pub active_paths: Vec, + #[serde(rename = "last_cycle_partial_source", default)] + pub last_cycle_partial_source: String, + #[serde(rename = "last_cycle_partial_source_code", default)] + pub last_cycle_partial_source_code: u64, + #[serde(rename = "partial_cycles_by_source", default)] + pub partial_cycles_by_source: Vec, } impl ScannerMetrics { pub fn merge(&mut self, other: &Self) { - if self.collected_at < other.collected_at { + let other_is_newer = self.collected_at < other.collected_at; + if other_is_newer { self.collected_at = other.collected_at; + self.last_cycle_partial_source = other.last_cycle_partial_source.clone(); + self.last_cycle_partial_source_code = other.last_cycle_partial_source_code; } if self.ongoing_buckets < other.ongoing_buckets { @@ -182,6 +199,20 @@ impl ScannerMetrics { self.last_minute.ilm.entry(k.clone()).or_default().merge(v); } + for source in other.partial_cycles_by_source.iter() { + if let Some(existing) = self + .partial_cycles_by_source + .iter_mut() + .find(|existing| existing.source == source.source) + { + existing.cycles += source.cycles; + } else { + self.partial_cycles_by_source.push(source.clone()); + } + } + self.partial_cycles_by_source + .sort_by(|left, right| left.source.cmp(&right.source)); + self.active_paths.extend(other.active_paths.clone()); self.active_paths.sort(); @@ -668,3 +699,56 @@ pub struct Operations { #[serde(rename = "operations")] pub operations: HashMap, } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn scanner_metrics_merge_aggregates_partial_cycles_by_source() { + let collected_at = Utc::now(); + let mut scanner = ScannerMetrics { + collected_at, + last_cycle_partial_source: "usage".to_string(), + last_cycle_partial_source_code: 1, + partial_cycles_by_source: vec![ScannerSourceCycleSnapshot { + source: "usage".to_string(), + cycles: 1, + }], + ..Default::default() + }; + + scanner.merge(&ScannerMetrics { + collected_at: collected_at + chrono::Duration::seconds(1), + last_cycle_partial_source: "lifecycle".to_string(), + last_cycle_partial_source_code: 2, + partial_cycles_by_source: vec![ + ScannerSourceCycleSnapshot { + source: "usage".to_string(), + cycles: 2, + }, + ScannerSourceCycleSnapshot { + source: "lifecycle".to_string(), + cycles: 1, + }, + ], + ..Default::default() + }); + + assert_eq!(scanner.last_cycle_partial_source, "lifecycle"); + assert_eq!(scanner.last_cycle_partial_source_code, 2); + assert_eq!( + scanner.partial_cycles_by_source, + vec![ + ScannerSourceCycleSnapshot { + source: "lifecycle".to_string(), + cycles: 1, + }, + ScannerSourceCycleSnapshot { + source: "usage".to_string(), + cycles: 3, + }, + ] + ); + } +} diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 2b17268ad..76aff02aa 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -27,7 +27,8 @@ use crate::{DataUsageInfo, ScannerActivityGuard, ScannerError}; use chrono::{DateTime, Utc}; use rustfs_common::heal_channel::HealScanMode; use rustfs_common::metrics::{ - CurrentCycle, Metric, Metrics, ScanCyclePartialReason, emit_scan_cycle_complete, emit_scan_cycle_partial, global_metrics, + CurrentCycle, Metric, Metrics, ScanCyclePartialReason, ScannerWorkSource, emit_scan_cycle_complete, + emit_scan_cycle_partial_with_source, global_metrics, }; use rustfs_config::ScannerSpeed; #[cfg(test)] @@ -96,6 +97,13 @@ fn scan_cycle_partial_reason(reason: Option) -> ScanCy } } +fn scan_cycle_partial_source(reason: Option) -> Option { + match reason { + Some(ScannerCycleBudgetReason::Objects | ScannerCycleBudgetReason::Directories) => Some(ScannerWorkSource::Usage), + Some(ScannerCycleBudgetReason::Runtime) | None => None, + } +} + /// Compute a randomized inter-cycle sleep. // Delay is scan interval +- 10%, with a floor of 1 second. fn randomized_cycle_delay() -> Duration { @@ -505,7 +513,12 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc max_directories = ?cycle_budget.max_directories(), "Data scanner cycle stopped after reaching its cycle budget" ); - emit_scan_cycle_partial(cycle_start.elapsed(), scan_cycle_partial_reason(cycle_budget.reason())); + let budget_reason = cycle_budget.reason(); + emit_scan_cycle_partial_with_source( + cycle_start.elapsed(), + scan_cycle_partial_reason(budget_reason), + scan_cycle_partial_source(budget_reason), + ); mark_scan_cycle_idle(cycle_info).await; return; } @@ -526,7 +539,12 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc "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(), scan_cycle_partial_reason(cycle_budget.reason())); + let budget_reason = cycle_budget.reason(); + emit_scan_cycle_partial_with_source( + cycle_start.elapsed(), + scan_cycle_partial_reason(budget_reason), + scan_cycle_partial_source(budget_reason), + ); mark_scan_cycle_idle(cycle_info).await; return; } @@ -855,6 +873,20 @@ mod tests { assert_eq!(scan_cycle_partial_reason(None), ScanCyclePartialReason::Unknown); } + #[test] + fn test_scan_cycle_partial_source_maps_budget_reason() { + assert_eq!(scan_cycle_partial_source(Some(ScannerCycleBudgetReason::Runtime)), None); + assert_eq!( + scan_cycle_partial_source(Some(ScannerCycleBudgetReason::Objects)), + Some(ScannerWorkSource::Usage) + ); + assert_eq!( + scan_cycle_partial_source(Some(ScannerCycleBudgetReason::Directories)), + Some(ScannerWorkSource::Usage) + ); + assert_eq!(scan_cycle_partial_source(None), None); + } + #[tokio::test] #[serial] async fn test_mark_scan_cycle_idle_clears_published_cycle_state() {