feat(scanner): expose cycle progress metrics (#3152)

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>

crates/common/src/metrics.rs

@@ -23,7 +23,7 @@ use std::{
     pin::Pin,
     sync::{
         Arc, Mutex, OnceLock,
-        atomic::{AtomicU8, AtomicU64, Ordering},
+        atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering},
     },
     time::{Duration, SystemTime},
 };
@@ -342,8 +342,15 @@ pub struct Metrics {
     current_paths: Arc<RwLock<HashMap<String, Arc<CurrentPathTracker>>>>,
     cycle_info: Arc<RwLock<Option<CurrentCycle>>>,
     current_scan_mode: AtomicU8,
+    current_scan_cycle_work_active: AtomicBool,
+    current_scan_cycle_objects_start: AtomicU64,
+    current_scan_cycle_directories_start: AtomicU64,
+    current_scan_cycle_bucket_drive_scans_start: AtomicU64,
     last_scan_cycle_result: AtomicU8,
     last_scan_cycle_duration_millis: AtomicU64,
+    last_scan_cycle_objects_scanned: AtomicU64,
+    last_scan_cycle_directories_scanned: AtomicU64,
+    last_scan_cycle_bucket_drive_scans: AtomicU64,
     failed_scan_cycles: AtomicU64,
 }
 
@@ -362,6 +369,13 @@ pub struct CurrentCycle {
     pub started: DateTime<Utc>,
 }
 
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct ScanCycleWorkSnapshot {
+    objects_scanned: u64,
+    directories_scanned: u64,
+    bucket_drive_scans: u64,
+}
+
 #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
 pub struct ScannerTimedAction {
     pub count: u64,
@@ -389,9 +403,21 @@ pub struct ScannerMetricsReport {
     pub last_minute: ScannerLastMinute,
     pub active_paths: Vec<String>,
     pub current_scan_mode: String,
+    #[serde(default)]
+    pub current_cycle_objects_scanned: u64,
+    #[serde(default)]
+    pub current_cycle_directories_scanned: u64,
+    #[serde(default)]
+    pub current_cycle_bucket_drive_scans: u64,
     pub last_cycle_result: String,
     pub last_cycle_result_code: u64,
     pub last_cycle_duration_seconds: f64,
+    #[serde(default)]
+    pub last_cycle_objects_scanned: u64,
+    #[serde(default)]
+    pub last_cycle_directories_scanned: u64,
+    #[serde(default)]
+    pub last_cycle_bucket_drive_scans: u64,
     pub failed_cycles: u64,
 }
 
@@ -484,8 +510,15 @@ impl Metrics {
             current_paths: Arc::new(RwLock::new(HashMap::new())),
             cycle_info: Arc::new(RwLock::new(None)),
             current_scan_mode: AtomicU8::new(HealScanMode::Unknown as u8),
+            current_scan_cycle_work_active: AtomicBool::new(false),
+            current_scan_cycle_objects_start: AtomicU64::new(0),
+            current_scan_cycle_directories_start: AtomicU64::new(0),
+            current_scan_cycle_bucket_drive_scans_start: AtomicU64::new(0),
             last_scan_cycle_result: AtomicU8::new(SCAN_CYCLE_RESULT_UNKNOWN),
             last_scan_cycle_duration_millis: AtomicU64::new(0),
+            last_scan_cycle_objects_scanned: AtomicU64::new(0),
+            last_scan_cycle_directories_scanned: AtomicU64::new(0),
+            last_scan_cycle_bucket_drive_scans: AtomicU64::new(0),
             failed_scan_cycles: AtomicU64::new(0),
         }
     }
@@ -645,6 +678,59 @@ impl Metrics {
             .store(duration_millis_saturated(duration), Ordering::Relaxed);
     }
 
+    pub fn start_scan_cycle_work(&self) -> ScanCycleWorkSnapshot {
+        let snapshot = self.scan_cycle_work_snapshot();
+        self.current_scan_cycle_objects_start
+            .store(snapshot.objects_scanned, Ordering::Relaxed);
+        self.current_scan_cycle_directories_start
+            .store(snapshot.directories_scanned, Ordering::Relaxed);
+        self.current_scan_cycle_bucket_drive_scans_start
+            .store(snapshot.bucket_drive_scans, Ordering::Relaxed);
+        self.current_scan_cycle_work_active.store(true, Ordering::Relaxed);
+        snapshot
+    }
+
+    pub fn finish_scan_cycle_work(&self, start: ScanCycleWorkSnapshot) {
+        let work = self.scan_cycle_work_since(start);
+        self.record_scan_cycle_work(work.objects_scanned, work.directories_scanned, work.bucket_drive_scans);
+        self.current_scan_cycle_work_active.store(false, Ordering::Relaxed);
+    }
+
+    fn scan_cycle_work_snapshot(&self) -> ScanCycleWorkSnapshot {
+        ScanCycleWorkSnapshot {
+            objects_scanned: self.lifetime(Metric::ScanObject),
+            directories_scanned: self.lifetime(Metric::ScanFolder),
+            bucket_drive_scans: self.lifetime(Metric::ScanBucketDrive),
+        }
+    }
+
+    fn current_scan_cycle_work_start(&self) -> ScanCycleWorkSnapshot {
+        ScanCycleWorkSnapshot {
+            objects_scanned: self.current_scan_cycle_objects_start.load(Ordering::Relaxed),
+            directories_scanned: self.current_scan_cycle_directories_start.load(Ordering::Relaxed),
+            bucket_drive_scans: self.current_scan_cycle_bucket_drive_scans_start.load(Ordering::Relaxed),
+        }
+    }
+
+    fn scan_cycle_work_since(&self, start: ScanCycleWorkSnapshot) -> ScanCycleWorkSnapshot {
+        let current = self.scan_cycle_work_snapshot();
+        ScanCycleWorkSnapshot {
+            objects_scanned: current.objects_scanned.saturating_sub(start.objects_scanned),
+            directories_scanned: current.directories_scanned.saturating_sub(start.directories_scanned),
+            bucket_drive_scans: current.bucket_drive_scans.saturating_sub(start.bucket_drive_scans),
+        }
+    }
+
+    pub fn record_scan_cycle_work(&self, objects_scanned: u64, directories_scanned: u64, bucket_drive_scans: u64) {
+        // Telemetry-only gauges: readers may observe a transient mixed snapshot
+        // while these independent atomic fields are updated.
+        self.last_scan_cycle_objects_scanned.store(objects_scanned, Ordering::Relaxed);
+        self.last_scan_cycle_directories_scanned
+            .store(directories_scanned, Ordering::Relaxed);
+        self.last_scan_cycle_bucket_drive_scans
+            .store(bucket_drive_scans, Ordering::Relaxed);
+    }
+
     /// Snapshot of every path currently being scanned.
     pub async fn get_current_paths(&self) -> Vec<String> {
         let paths = self.current_paths.read().await;
@@ -681,10 +767,19 @@ impl Metrics {
         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();
+        if self.current_scan_cycle_work_active.load(Ordering::Relaxed) {
+            let current_work = self.scan_cycle_work_since(self.current_scan_cycle_work_start());
+            m.current_cycle_objects_scanned = current_work.objects_scanned;
+            m.current_cycle_directories_scanned = current_work.directories_scanned;
+            m.current_cycle_bucket_drive_scans = current_work.bucket_drive_scans;
+        }
         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.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);
+        m.last_cycle_bucket_drive_scans = self.last_scan_cycle_bucket_drive_scans.load(Ordering::Relaxed);
         m.failed_cycles = self.failed_scan_cycles.load(Ordering::Relaxed);
 
         // Lifetime operation counts
@@ -899,4 +994,45 @@ mod tests {
         assert_eq!(report.last_cycle_duration_seconds, 2.0);
         assert_eq!(report.failed_cycles, 0);
     }
+
+    #[tokio::test]
+    async fn report_includes_last_scan_cycle_work() {
+        let metrics = Metrics::new();
+        metrics.record_scan_cycle_work(11, 7, 3);
+
+        let report = metrics.report().await;
+
+        assert_eq!(report.last_cycle_objects_scanned, 11);
+        assert_eq!(report.last_cycle_directories_scanned, 7);
+        assert_eq!(report.last_cycle_bucket_drive_scans, 3);
+    }
+
+    #[tokio::test]
+    async fn report_includes_active_scan_cycle_work() {
+        let metrics = Metrics::new();
+        metrics.operations[Metric::ScanObject as usize].store(10, Ordering::Relaxed);
+        metrics.operations[Metric::ScanFolder as usize].store(5, Ordering::Relaxed);
+        metrics.operations[Metric::ScanBucketDrive as usize].store(1, Ordering::Relaxed);
+
+        let start = metrics.start_scan_cycle_work();
+        metrics.operations[Metric::ScanObject as usize].store(17, Ordering::Relaxed);
+        metrics.operations[Metric::ScanFolder as usize].store(8, Ordering::Relaxed);
+        metrics.operations[Metric::ScanBucketDrive as usize].store(3, Ordering::Relaxed);
+
+        let report = metrics.report().await;
+
+        assert_eq!(report.current_cycle_objects_scanned, 7);
+        assert_eq!(report.current_cycle_directories_scanned, 3);
+        assert_eq!(report.current_cycle_bucket_drive_scans, 2);
+
+        metrics.finish_scan_cycle_work(start);
+        let report = metrics.report().await;
+
+        assert_eq!(report.current_cycle_objects_scanned, 0);
+        assert_eq!(report.current_cycle_directories_scanned, 0);
+        assert_eq!(report.current_cycle_bucket_drive_scans, 0);
+        assert_eq!(report.last_cycle_objects_scanned, 7);
+        assert_eq!(report.last_cycle_directories_scanned, 3);
+        assert_eq!(report.last_cycle_bucket_drive_scans, 2);
+    }
 }