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 <marshawcoco@users.noreply.github.com>
Co-authored-by: cxymds <Cxymds@qq.com>
Co-authored-by: houseme <housemecn@gmail.com>

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<Self> {
+        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<HashMap<String, ScannerDiskBucketScanState>>,
     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<AtomicU64>,
     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<String, ScannerTimedAction>,
@@ -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<ScannerSourceCycleSnapshot>,
+    #[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<ScannerWorkSource>,
+) {
+    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<ScannerWorkSource>,
+    ) {
         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]

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);
+    }
 }

crates/madmin/src/metrics.rs

@@ -120,6 +120,14 @@ pub struct LastMinute {
     pub ilm: HashMap<String, TimedAction>,
 }
 
+#[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<String>,
+    #[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<ScannerSourceCycleSnapshot>,
 }
 
 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<String, TimedAction>,
 }
+
+#[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,
+                },
+            ]
+        );
+    }
+}

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<ScannerCycleBudgetReason>) -> ScanCy
     }
 }
 
+fn scan_cycle_partial_source(reason: Option<ScannerCycleBudgetReason>) -> Option<ScannerWorkSource> {
+    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<ECStore>
                 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<ECStore>
             "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() {