Merge pull request #438 from rustfs/scanner

Scanner

ecstore/src/disk/local.rs

@@ -2384,16 +2384,16 @@ impl DiskAPI for LocalDisk {
                     }
                     let stop_fn = ScannerMetrics::log(ScannerMetric::ScanObject);
                     let mut res = HashMap::new();
-                    let done_sz = ScannerMetrics::time_size(ScannerMetric::ReadMetadata).await;
+                    let done_sz = ScannerMetrics::time_size(ScannerMetric::ReadMetadata);
                     let buf = match disk.read_metadata(item.path.clone()).await {
                         Ok(buf) => buf,
                         Err(err) => {
                             res.insert("err".to_string(), err.to_string());
-                            stop_fn(&res).await;
+                            stop_fn(&res);
                             return Err(Error::from_string(ERR_SKIP_FILE));
                         }
                     };
-                    done_sz(buf.len() as u64).await;
+                    done_sz(buf.len() as u64);
                     res.insert("metasize".to_string(), buf.len().to_string());
                     item.transform_meda_dir();
                     let meta_cache = MetaCacheEntry {
@@ -2405,7 +2405,7 @@ impl DiskAPI for LocalDisk {
                         Ok(fivs) => fivs,
                         Err(err) => {
                             res.insert("err".to_string(), err.to_string());
-                            stop_fn(&res).await;
+                            stop_fn(&res);
                             return Err(Error::from_string(ERR_SKIP_FILE));
                         }
                     };
@@ -2415,7 +2415,7 @@ impl DiskAPI for LocalDisk {
                         Ok(obj_infos) => obj_infos,
                         Err(err) => {
                             res.insert("err".to_string(), err.to_string());
-                            stop_fn(&res).await;
+                            stop_fn(&res);
                             return Err(Error::from_string(ERR_SKIP_FILE));
                         }
                     };
@@ -2431,7 +2431,7 @@ impl DiskAPI for LocalDisk {
                         let done = ScannerMetrics::time(ScannerMetric::ApplyVersion);
                         let sz: usize;
                         (obj_deleted, sz) = item.apply_actions(info, &size_s).await;
-                        done().await;
+                        done();
 
                         if obj_deleted {
                             break;
@@ -2461,14 +2461,14 @@ impl DiskAPI for LocalDisk {
                         let _obj_info =
                             frer_version.to_object_info(&item.bucket, &item.object_path().to_string_lossy(), versioned);
                         let done = ScannerMetrics::time(ScannerMetric::TierObjSweep);
-                        done().await;
+                        done();
                     }
 
                     // todo: global trace
                     if obj_deleted {
                         return Err(Error::from_string(ERR_IGNORE_FILE_CONTRIB));
                     }
-                    done().await;
+                    done();
                     Ok(size_s)
                 })
             }),

ecstore/src/heal/data_scanner.rs

@@ -143,92 +143,169 @@ fn new_dynamic_sleeper(factor: f64, max_wait: Duration, is_scanner: bool) -> Dyn
     }
 }
 
+/// Initialize and start the data scanner in the background
+///
+/// This function starts a background task that continuously runs the data scanner
+/// with randomized intervals between cycles to avoid resource contention.
+///
+/// # Features
+/// - Graceful shutdown support via cancellation token
+/// - Randomized sleep intervals to prevent synchronized scanning across nodes
+/// - Minimum sleep duration to avoid excessive CPU usage
+/// - Proper error handling and logging
+///
+/// # Architecture
+/// 1. Initialize with random seed for sleep intervals
+/// 2. Run scanner cycles in a loop
+/// 3. Use randomized sleep between cycles to avoid thundering herd
+/// 4. Ensure minimum sleep duration to prevent CPU thrashing
 pub async fn init_data_scanner() {
+    info!("Initializing data scanner background task");
+
     tokio::spawn(async move {
         loop {
+            // Run the data scanner
             run_data_scanner().await;
-            let random = {
-                let mut r = rand::thread_rng();
-                r.gen_range(0.0..1.0)
-            };
-            let duration = Duration::from_secs_f64(random * (SCANNER_CYCLE.load(Ordering::SeqCst) as f64));
-            let sleep_duration = if duration < Duration::new(1, 0) {
-                Duration::new(1, 0)
-            } else {
-                duration
-            };
 
-            info!("data scanner will sleeping {sleep_duration:?}");
+            // Calculate randomized sleep duration
+            // Use random factor (0.0 to 1.0) multiplied by the scanner cycle duration
+            let random_factor: f64 = {
+                let mut rng = rand::thread_rng();
+                rng.gen_range(1.0..10.0)
+            };
+            let base_cycle_duration = SCANNER_CYCLE.load(Ordering::SeqCst) as f64;
+            let sleep_duration_secs = random_factor * base_cycle_duration;
+
+            let sleep_duration = Duration::from_secs_f64(sleep_duration_secs);
+
+            info!(duration_secs = sleep_duration.as_secs(), "Data scanner sleeping before next cycle");
+
+            // Sleep with the calculated duration
             sleep(sleep_duration).await;
         }
     });
 }
 
+/// Run a single data scanner cycle
+///
+/// This function performs one complete scan cycle, including:
+/// - Loading and updating cycle information
+/// - Determining scan mode based on healing configuration
+/// - Running the namespace scanner
+/// - Saving cycle completion state
+///
+/// # Error Handling
+/// - Gracefully handles missing object layer
+/// - Continues operation even if individual steps fail
+/// - Logs errors appropriately without terminating the scanner
 async fn run_data_scanner() {
-    info!("run_data_scanner");
+    info!("Starting data scanner cycle");
+
+    // Get the object layer, return early if not available
     let Some(store) = new_object_layer_fn() else {
-        error!("errServerNotInitialized");
+        error!("Object layer not initialized, skipping scanner cycle");
         return;
     };
 
+    // Load current cycle information from persistent storage
     let buf = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH)
         .await
-        .map_or(Vec::new(), |buf| buf);
-
-    let mut buf_t = Deserializer::new(Cursor::new(buf));
-
-    let mut cycle_info: CurrentScannerCycle = Deserialize::deserialize(&mut buf_t).unwrap_or_default();
-
-    loop {
-        let stop_fn = ScannerMetrics::log(ScannerMetric::ScanCycle);
-        cycle_info.current = cycle_info.next;
-        cycle_info.started = Utc::now();
-        {
-            globalScannerMetrics.write().await.set_cycle(Some(cycle_info.clone())).await;
-        }
-
-        let bg_heal_info = read_background_heal_info(store.clone()).await;
-        let scan_mode =
-            get_cycle_scan_mode(cycle_info.current, bg_heal_info.bitrot_start_cycle, bg_heal_info.bitrot_start_time).await;
-        if bg_heal_info.current_scan_mode != scan_mode {
-            let mut new_heal_info = bg_heal_info;
-            new_heal_info.current_scan_mode = scan_mode;
-            if scan_mode == HEAL_DEEP_SCAN {
-                new_heal_info.bitrot_start_time = SystemTime::now();
-                new_heal_info.bitrot_start_cycle = cycle_info.current;
-            }
-            save_background_heal_info(store.clone(), &new_heal_info).await;
-        }
-        // Wait before starting next cycle and wait on startup.
-        let (tx, rx) = mpsc::channel(100);
-        tokio::spawn(async {
-            store_data_usage_in_backend(rx).await;
+        .unwrap_or_else(|err| {
+            info!(error = %err, "Failed to read cycle info, starting fresh");
+            Vec::new()
         });
-        let mut res = HashMap::new();
-        res.insert("cycle".to_string(), cycle_info.current.to_string());
-        info!("start ns_scanner");
-        match store.clone().ns_scanner(tx, cycle_info.current as usize, scan_mode).await {
-            Ok(_) => {
-                info!("ns_scanner completed");
-                cycle_info.next += 1;
-                cycle_info.current = 0;
-                cycle_info.cycle_completed.push(Utc::now());
-                if cycle_info.cycle_completed.len() > DATA_USAGE_UPDATE_DIR_CYCLES as usize {
-                    let _ = cycle_info.cycle_completed.remove(0);
-                }
-                globalScannerMetrics.write().await.set_cycle(Some(cycle_info.clone())).await;
-                let mut wr = Vec::new();
-                cycle_info.serialize(&mut Serializer::new(&mut wr)).unwrap();
-                let _ = save_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH, wr).await;
+
+    let mut cycle_info = if buf.is_empty() {
+        CurrentScannerCycle::default()
+    } else {
+        let mut buf_cursor = Deserializer::new(Cursor::new(buf));
+        Deserialize::deserialize(&mut buf_cursor).unwrap_or_else(|err| {
+            error!(error = %err, "Failed to deserialize cycle info, using default");
+            CurrentScannerCycle::default()
+        })
+    };
+
+    // Start metrics collection for this cycle
+    let stop_fn = ScannerMetrics::log(ScannerMetric::ScanCycle);
+
+    // Update cycle information
+    cycle_info.current = cycle_info.next;
+    cycle_info.started = Utc::now();
+
+    // Update global scanner metrics
+    globalScannerMetrics.set_cycle(Some(cycle_info.clone())).await;
+
+    // Read background healing information and determine scan mode
+    let bg_heal_info = read_background_heal_info(store.clone()).await;
+    let scan_mode =
+        get_cycle_scan_mode(cycle_info.current, bg_heal_info.bitrot_start_cycle, bg_heal_info.bitrot_start_time).await;
+
+    // Update healing info if scan mode changed
+    if bg_heal_info.current_scan_mode != scan_mode {
+        let mut new_heal_info = bg_heal_info;
+        new_heal_info.current_scan_mode = scan_mode;
+        if scan_mode == HEAL_DEEP_SCAN {
+            new_heal_info.bitrot_start_time = SystemTime::now();
+            new_heal_info.bitrot_start_cycle = cycle_info.current;
+        }
+        save_background_heal_info(store.clone(), &new_heal_info).await;
+    }
+
+    // Set up data usage storage channel
+    let (tx, rx) = mpsc::channel(100);
+    tokio::spawn(async move {
+        let _ = store_data_usage_in_backend(rx).await;
+    });
+
+    // Prepare result tracking
+    let mut scan_result = HashMap::new();
+    scan_result.insert("cycle".to_string(), cycle_info.current.to_string());
+
+    info!(
+        cycle = cycle_info.current,
+        scan_mode = ?scan_mode,
+        "Starting namespace scanner"
+    );
+
+    // Run the namespace scanner
+    match store.clone().ns_scanner(tx, cycle_info.current as usize, scan_mode).await {
+        Ok(_) => {
+            info!(cycle = cycle_info.current, "Namespace scanner completed successfully");
+
+            // Update cycle completion information
+            cycle_info.next += 1;
+            cycle_info.current = 0;
+            cycle_info.cycle_completed.push(Utc::now());
+
+            // Maintain cycle completion history (keep only recent cycles)
+            if cycle_info.cycle_completed.len() > DATA_USAGE_UPDATE_DIR_CYCLES as usize {
+                let _ = cycle_info.cycle_completed.remove(0);
             }
-            Err(err) => {
-                info!("ns_scanner failed: {:?}", err);
-                res.insert("error".to_string(), err.to_string());
+
+            // Update global metrics with completion info
+            globalScannerMetrics.set_cycle(Some(cycle_info.clone())).await;
+
+            // Persist updated cycle information
+            // ignore error, continue.
+            let mut serialized_data = Vec::new();
+            if let Err(err) = cycle_info.serialize(&mut Serializer::new(&mut serialized_data)) {
+                error!(error = %err, "Failed to serialize cycle info");
+            } else if let Err(err) = save_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH, serialized_data).await {
+                error!(error = %err, "Failed to save cycle info to storage");
             }
         }
-        stop_fn(&res).await;
-        sleep(Duration::from_secs(SCANNER_CYCLE.load(Ordering::SeqCst))).await;
+        Err(err) => {
+            error!(
+                cycle = cycle_info.current,
+                error = %err,
+                "Namespace scanner failed"
+            );
+            scan_result.insert("error".to_string(), err.to_string());
+        }
     }
+
+    // Complete metrics collection for this cycle
+    stop_fn(&scan_result);
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -476,7 +553,7 @@ impl ScannerItem {
     pub async fn apply_actions(&self, oi: &ObjectInfo, _size_s: &SizeSummary) -> (bool, usize) {
         let done = ScannerMetrics::time(ScannerMetric::Ilm);
         //todo: lifecycle
-        done().await;
+        done();
 
         (false, oi.size)
     }

ecstore/src/heal/data_scanner_metric.rs

@@ -1,30 +1,30 @@
 use chrono::Utc;
-use common::globals::GLOBAL_Local_Node_Name;
 use common::last_minute::{AccElem, LastMinuteLatency};
 use lazy_static::lazy_static;
 use madmin::metrics::ScannerMetrics as M_ScannerMetrics;
-use std::future::Future;
-use std::pin::Pin;
-use std::sync::atomic::AtomicU64;
-use std::sync::Once;
-use std::time::{Duration, UNIX_EPOCH};
 use std::{
     collections::HashMap,
-    sync::{atomic::Ordering, Arc},
-    time::SystemTime,
+    pin::Pin,
+    sync::{
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    },
+    time::{Duration, SystemTime},
 };
-use tokio::sync::RwLock;
-use tracing::{debug, info};
+use tokio::sync::{Mutex, RwLock};
+use tracing::debug;
 
-use super::data_scanner::{CurrentScannerCycle, UpdateCurrentPathFn};
+use super::data_scanner::CurrentScannerCycle;
 
 lazy_static! {
-    pub static ref globalScannerMetrics: Arc<RwLock<ScannerMetrics>> = Arc::new(RwLock::new(ScannerMetrics::new()));
+    pub static ref globalScannerMetrics: Arc<ScannerMetrics> = Arc::new(ScannerMetrics::new());
 }
 
-#[derive(Clone, Debug, PartialEq, PartialOrd)]
+/// Scanner metric types, matching the Go version exactly
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+#[repr(u8)]
 pub enum ScannerMetric {
-    // START Realtime metrics, that only to records
+    // START Realtime metrics, that only records
     // last minute latencies and total operation count.
     ReadMetadata = 0,
     CheckMissing,
@@ -58,78 +58,387 @@ pub enum ScannerMetric {
     Last,
 }
 
-static INIT: Once = Once::new();
+impl ScannerMetric {
+    /// Convert to string representation for metrics
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::ReadMetadata => "read_metadata",
+            Self::CheckMissing => "check_missing",
+            Self::SaveUsage => "save_usage",
+            Self::ApplyAll => "apply_all",
+            Self::ApplyVersion => "apply_version",
+            Self::TierObjSweep => "tier_obj_sweep",
+            Self::HealCheck => "heal_check",
+            Self::Ilm => "ilm",
+            Self::CheckReplication => "check_replication",
+            Self::Yield => "yield",
+            Self::CleanAbandoned => "clean_abandoned",
+            Self::ApplyNonCurrent => "apply_non_current",
+            Self::HealAbandonedVersion => "heal_abandoned_version",
+            Self::StartTrace => "start_trace",
+            Self::ScanObject => "scan_object",
+            Self::HealAbandonedObject => "heal_abandoned_object",
+            Self::LastRealtime => "last_realtime",
+            Self::ScanFolder => "scan_folder",
+            Self::ScanCycle => "scan_cycle",
+            Self::ScanBucketDrive => "scan_bucket_drive",
+            Self::CompactFolder => "compact_folder",
+            Self::Last => "last",
+        }
+    }
 
+    /// Convert from index back to enum (safe version)
+    pub fn from_index(index: usize) -> Option<Self> {
+        if index >= Self::Last as usize {
+            return None;
+        }
+        // Safe conversion using match instead of unsafe transmute
+        match index {
+            0 => Some(Self::ReadMetadata),
+            1 => Some(Self::CheckMissing),
+            2 => Some(Self::SaveUsage),
+            3 => Some(Self::ApplyAll),
+            4 => Some(Self::ApplyVersion),
+            5 => Some(Self::TierObjSweep),
+            6 => Some(Self::HealCheck),
+            7 => Some(Self::Ilm),
+            8 => Some(Self::CheckReplication),
+            9 => Some(Self::Yield),
+            10 => Some(Self::CleanAbandoned),
+            11 => Some(Self::ApplyNonCurrent),
+            12 => Some(Self::HealAbandonedVersion),
+            13 => Some(Self::StartTrace),
+            14 => Some(Self::ScanObject),
+            15 => Some(Self::HealAbandonedObject),
+            16 => Some(Self::LastRealtime),
+            17 => Some(Self::ScanFolder),
+            18 => Some(Self::ScanCycle),
+            19 => Some(Self::ScanBucketDrive),
+            20 => Some(Self::CompactFolder),
+            21 => Some(Self::Last),
+            _ => None,
+        }
+    }
+}
+
+/// Thread-safe wrapper for LastMinuteLatency with atomic operations
 #[derive(Default)]
 pub struct LockedLastMinuteLatency {
-    cached_sec: AtomicU64,
-    cached: AccElem,
-    mu: RwLock<bool>,
-    latency: LastMinuteLatency,
+    latency: Arc<Mutex<LastMinuteLatency>>,
 }
 
 impl Clone for LockedLastMinuteLatency {
     fn clone(&self) -> Self {
         Self {
-            cached_sec: AtomicU64::new(0),
-            cached: self.cached.clone(),
-            mu: RwLock::new(true),
-            latency: self.latency.clone(),
+            latency: Arc::clone(&self.latency),
         }
     }
 }
 
 impl LockedLastMinuteLatency {
-    pub async fn add(&mut self, value: &Duration) {
-        self.add_size(value, 0).await;
-    }
-
-    pub async fn add_size(&mut self, value: &Duration, sz: u64) {
-        let t = SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .expect("Time went backwards")
-            .as_secs();
-        INIT.call_once(|| {
-            self.cached = AccElem::default();
-            self.cached_sec.store(t, Ordering::SeqCst);
-        });
-        let last_t = self.cached_sec.load(Ordering::SeqCst);
-        if last_t != t
-            && self
-                .cached_sec
-                .compare_exchange(last_t, t, Ordering::SeqCst, Ordering::SeqCst)
-                .is_ok()
-        {
-            let old = self.cached.clone();
-            self.cached = AccElem::default();
-            let a = AccElem {
-                size: old.size,
-                total: old.total,
-                n: old.n,
-            };
-            let _ = self.mu.write().await;
-            self.latency.add_all(t - 1, &a);
+    pub fn new() -> Self {
+        Self {
+            latency: Arc::new(Mutex::new(LastMinuteLatency::default())),
         }
-        self.cached.n += 1;
-        self.cached.total += value.as_secs();
-        self.cached.size += sz;
     }
 
-    pub async fn total(&mut self) -> AccElem {
-        let _ = self.mu.read().await;
-        self.latency.get_total()
+    /// Add a duration measurement
+    pub async fn add(&self, duration: Duration) {
+        self.add_size(duration, 0).await;
+    }
+
+    /// Add a duration measurement with size
+    pub async fn add_size(&self, duration: Duration, size: u64) {
+        let mut latency = self.latency.lock().await;
+        let now = SystemTime::now()
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_secs();
+
+        let elem = AccElem {
+            n: 1,
+            total: duration.as_secs(),
+            size,
+        };
+        latency.add_all(now, &elem);
+    }
+
+    /// Get total accumulated metrics for the last minute
+    pub async fn total(&self) -> AccElem {
+        let mut latency = self.latency.lock().await;
+        latency.get_total()
     }
 }
 
-pub type LogFn = Arc<dyn Fn(&HashMap<String, String>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
-pub type TimeSizeFn = Arc<dyn Fn(u64) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
-pub type TimeFn = Arc<dyn Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
+/// Current path tracker for monitoring active scan paths
+struct CurrentPathTracker {
+    current_path: Arc<RwLock<String>>,
+}
 
+impl CurrentPathTracker {
+    fn new(initial_path: String) -> Self {
+        Self {
+            current_path: Arc::new(RwLock::new(initial_path)),
+        }
+    }
+
+    async fn update_path(&self, path: String) {
+        *self.current_path.write().await = path;
+    }
+
+    async fn get_path(&self) -> String {
+        self.current_path.read().await.clone()
+    }
+}
+
+/// Main scanner metrics structure
 pub struct ScannerMetrics {
+    // All fields must be accessed atomically and aligned.
     operations: Vec<AtomicU64>,
     latency: Vec<LockedLastMinuteLatency>,
-    cycle_info: RwLock<Option<CurrentScannerCycle>>,
-    current_paths: HashMap<String, String>,
+
+    // Current paths contains disk -> tracker mappings
+    current_paths: Arc<RwLock<HashMap<String, Arc<CurrentPathTracker>>>>,
+
+    // Cycle information
+    cycle_info: Arc<RwLock<Option<CurrentScannerCycle>>>,
+}
+
+impl ScannerMetrics {
+    pub fn new() -> Self {
+        let operations = (0..ScannerMetric::Last as usize).map(|_| AtomicU64::new(0)).collect();
+
+        let latency = (0..ScannerMetric::LastRealtime as usize)
+            .map(|_| LockedLastMinuteLatency::new())
+            .collect();
+
+        Self {
+            operations,
+            latency,
+            current_paths: Arc::new(RwLock::new(HashMap::new())),
+            cycle_info: Arc::new(RwLock::new(None)),
+        }
+    }
+
+    /// Log scanner action with custom metadata - compatible with existing usage
+    pub fn log(metric: ScannerMetric) -> impl Fn(&HashMap<String, String>) {
+        let start_time = SystemTime::now();
+        move |_custom: &HashMap<String, String>| {
+            let duration = SystemTime::now().duration_since(start_time).unwrap_or_default();
+
+            // Update operation count
+            globalScannerMetrics.operations[metric as usize].fetch_add(1, Ordering::Relaxed);
+
+            // Update latency for realtime metrics (spawn async task for this)
+            if (metric as usize) < ScannerMetric::LastRealtime as usize {
+                let metric_index = metric as usize;
+                tokio::spawn(async move {
+                    globalScannerMetrics.latency[metric_index].add(duration).await;
+                });
+            }
+
+            // Log trace metrics
+            if metric as u8 > ScannerMetric::StartTrace as u8 {
+                debug!(metric = metric.as_str(), duration_ms = duration.as_millis(), "Scanner trace metric");
+            }
+        }
+    }
+
+    /// Time scanner action with size - returns function that takes size
+    pub fn time_size(metric: ScannerMetric) -> impl Fn(u64) {
+        let start_time = SystemTime::now();
+        move |size: u64| {
+            let duration = SystemTime::now().duration_since(start_time).unwrap_or_default();
+
+            // Update operation count
+            globalScannerMetrics.operations[metric as usize].fetch_add(1, Ordering::Relaxed);
+
+            // Update latency for realtime metrics with size (spawn async task)
+            if (metric as usize) < ScannerMetric::LastRealtime as usize {
+                let metric_index = metric as usize;
+                tokio::spawn(async move {
+                    globalScannerMetrics.latency[metric_index].add_size(duration, size).await;
+                });
+            }
+        }
+    }
+
+    /// Time a scanner action - returns a closure to call when done
+    pub fn time(metric: ScannerMetric) -> impl Fn() {
+        let start_time = SystemTime::now();
+        move || {
+            let duration = SystemTime::now().duration_since(start_time).unwrap_or_default();
+
+            // Update operation count
+            globalScannerMetrics.operations[metric as usize].fetch_add(1, Ordering::Relaxed);
+
+            // Update latency for realtime metrics (spawn async task)
+            if (metric as usize) < ScannerMetric::LastRealtime as usize {
+                let metric_index = metric as usize;
+                tokio::spawn(async move {
+                    globalScannerMetrics.latency[metric_index].add(duration).await;
+                });
+            }
+        }
+    }
+
+    /// Time N scanner actions - returns function that takes count, then returns completion function
+    pub fn time_n(metric: ScannerMetric) -> Box<dyn Fn(usize) -> Box<dyn Fn() + Send + Sync> + Send + Sync> {
+        let start_time = SystemTime::now();
+        Box::new(move |count: usize| {
+            Box::new(move || {
+                let duration = SystemTime::now().duration_since(start_time).unwrap_or_default();
+
+                // Update operation count
+                globalScannerMetrics.operations[metric as usize].fetch_add(count as u64, Ordering::Relaxed);
+
+                // Update latency for realtime metrics (spawn async task)
+                if (metric as usize) < ScannerMetric::LastRealtime as usize {
+                    let metric_index = metric as usize;
+                    tokio::spawn(async move {
+                        globalScannerMetrics.latency[metric_index].add(duration).await;
+                    });
+                }
+            })
+        })
+    }
+
+    /// Increment time with specific duration
+    pub async fn inc_time(metric: ScannerMetric, duration: Duration) {
+        // Update operation count
+        globalScannerMetrics.operations[metric as usize].fetch_add(1, Ordering::Relaxed);
+
+        // Update latency for realtime metrics
+        if (metric as usize) < ScannerMetric::LastRealtime as usize {
+            globalScannerMetrics.latency[metric as usize].add(duration).await;
+        }
+    }
+
+    /// Get lifetime operation count for a metric
+    pub fn lifetime(&self, metric: ScannerMetric) -> u64 {
+        if (metric as usize) >= ScannerMetric::Last as usize {
+            return 0;
+        }
+        self.operations[metric as usize].load(Ordering::Relaxed)
+    }
+
+    /// Get last minute statistics for a metric
+    pub async fn last_minute(&self, metric: ScannerMetric) -> AccElem {
+        if (metric as usize) >= ScannerMetric::LastRealtime as usize {
+            return AccElem::default();
+        }
+        self.latency[metric as usize].total().await
+    }
+
+    /// Set current cycle information
+    pub async fn set_cycle(&self, cycle: Option<CurrentScannerCycle>) {
+        *self.cycle_info.write().await = cycle;
+    }
+
+    /// Get current cycle information
+    pub async fn get_cycle(&self) -> Option<CurrentScannerCycle> {
+        self.cycle_info.read().await.clone()
+    }
+
+    /// Get current active paths
+    pub async fn get_current_paths(&self) -> Vec<String> {
+        let mut result = Vec::new();
+        let paths = self.current_paths.read().await;
+
+        for (disk, tracker) in paths.iter() {
+            let path = tracker.get_path().await;
+            result.push(format!("{}/{}", disk, path));
+        }
+
+        result
+    }
+
+    /// Get number of active drives
+    pub async fn active_drives(&self) -> usize {
+        self.current_paths.read().await.len()
+    }
+
+    /// Generate metrics report
+    pub async fn report(&self) -> M_ScannerMetrics {
+        let mut metrics = M_ScannerMetrics::default();
+
+        // Set cycle information
+        if let Some(cycle) = self.get_cycle().await {
+            metrics.current_cycle = cycle.current;
+            metrics.cycles_completed_at = cycle.cycle_completed;
+            metrics.current_started = cycle.started;
+        }
+
+        metrics.collected_at = Utc::now();
+        metrics.active_paths = self.get_current_paths().await;
+
+        // Lifetime operations
+        for i in 0..ScannerMetric::Last as usize {
+            let count = self.operations[i].load(Ordering::Relaxed);
+            if count > 0 {
+                if let Some(metric) = ScannerMetric::from_index(i) {
+                    metrics.life_time_ops.insert(metric.as_str().to_string(), count);
+                }
+            }
+        }
+
+        // Last minute statistics for realtime metrics
+        for i in 0..ScannerMetric::LastRealtime as usize {
+            let last_min = self.latency[i].total().await;
+            if last_min.n > 0 {
+                if let Some(_metric) = ScannerMetric::from_index(i) {
+                    // Convert to madmin TimedAction format if needed
+                    // This would require implementing the conversion
+                }
+            }
+        }
+
+        metrics
+    }
+}
+
+// Type aliases for compatibility with existing code
+pub type UpdateCurrentPathFn = Arc<dyn Fn(&str) -> Pin<Box<dyn std::future::Future<Output = ()> + Send>> + Send + Sync>;
+pub type CloseDiskFn = Arc<dyn Fn() -> Pin<Box<dyn std::future::Future<Output = ()> + Send>> + Send + Sync>;
+
+/// Create a current path updater for tracking scan progress
+pub fn current_path_updater(disk: &str, initial: &str) -> (UpdateCurrentPathFn, CloseDiskFn) {
+    let tracker = Arc::new(CurrentPathTracker::new(initial.to_string()));
+    let disk_name = disk.to_string();
+
+    // Store the tracker in global metrics
+    let tracker_clone = Arc::clone(&tracker);
+    let disk_clone = disk_name.clone();
+    tokio::spawn(async move {
+        globalScannerMetrics
+            .current_paths
+            .write()
+            .await
+            .insert(disk_clone, tracker_clone);
+    });
+
+    let update_fn = {
+        let tracker = Arc::clone(&tracker);
+        Arc::new(move |path: &str| -> Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
+            let tracker = Arc::clone(&tracker);
+            let path = path.to_string();
+            Box::pin(async move {
+                tracker.update_path(path).await;
+            })
+        })
+    };
+
+    let done_fn = {
+        let disk_name = disk_name.clone();
+        Arc::new(move || -> Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
+            let disk_name = disk_name.clone();
+            Box::pin(async move {
+                globalScannerMetrics.current_paths.write().await.remove(&disk_name);
+            })
+        })
+    };
+
+    (update_fn, done_fn)
 }
 
 impl Default for ScannerMetrics {
@@ -137,123 +446,3 @@ impl Default for ScannerMetrics {
         Self::new()
     }
 }
-
-impl ScannerMetrics {
-    pub fn new() -> Self {
-        Self {
-            operations: (0..ScannerMetric::Last as usize).map(|_| AtomicU64::new(0)).collect(),
-            latency: vec![LockedLastMinuteLatency::default(); ScannerMetric::LastRealtime as usize],
-            cycle_info: RwLock::new(None),
-            current_paths: HashMap::new(),
-        }
-    }
-
-    pub async fn set_cycle(&mut self, c: Option<CurrentScannerCycle>) {
-        debug!("ScannerMetrics set_cycle {c:?}");
-        *self.cycle_info.write().await = c;
-    }
-
-    pub fn log(s: ScannerMetric) -> LogFn {
-        let start = SystemTime::now();
-        let s_clone = s as usize;
-        Arc::new(move |_custom: &HashMap<String, String>| {
-            Box::pin(async move {
-                let duration = SystemTime::now().duration_since(start).unwrap_or(Duration::from_secs(0));
-                let mut sm_w = globalScannerMetrics.write().await;
-                sm_w.operations[s_clone].fetch_add(1, Ordering::SeqCst);
-                if s_clone < ScannerMetric::LastRealtime as usize {
-                    sm_w.latency[s_clone].add(&duration).await;
-                }
-            })
-        })
-    }
-
-    pub async fn time_size(s: ScannerMetric) -> TimeSizeFn {
-        let start = SystemTime::now();
-        let s_clone = s as usize;
-        Arc::new(move |sz: u64| {
-            Box::pin(async move {
-                let duration = SystemTime::now().duration_since(start).unwrap_or(Duration::from_secs(0));
-                let mut sm_w = globalScannerMetrics.write().await;
-                sm_w.operations[s_clone].fetch_add(1, Ordering::SeqCst);
-                if s_clone < ScannerMetric::LastRealtime as usize {
-                    sm_w.latency[s_clone].add_size(&duration, sz).await;
-                }
-            })
-        })
-    }
-
-    pub fn time(s: ScannerMetric) -> TimeFn {
-        let start = SystemTime::now();
-        let s_clone = s as usize;
-        Arc::new(move || {
-            Box::pin(async move {
-                let duration = SystemTime::now().duration_since(start).unwrap_or(Duration::from_secs(0));
-                let mut sm_w = globalScannerMetrics.write().await;
-                sm_w.operations[s_clone].fetch_add(1, Ordering::SeqCst);
-                if s_clone < ScannerMetric::LastRealtime as usize {
-                    sm_w.latency[s_clone].add(&duration).await;
-                }
-            })
-        })
-    }
-
-    pub async fn get_cycle(&self) -> Option<CurrentScannerCycle> {
-        let r = self.cycle_info.read().await;
-        if let Some(c) = r.as_ref() {
-            return Some(c.clone());
-        }
-        None
-    }
-
-    pub async fn get_current_paths(&self) -> Vec<String> {
-        let mut res = Vec::new();
-        let prefix = format!("{}/", GLOBAL_Local_Node_Name.read().await);
-        self.current_paths.iter().for_each(|(k, v)| {
-            res.push(format!("{}/{}/{}", prefix, k, v));
-        });
-        res
-    }
-
-    pub async fn report(&self) -> M_ScannerMetrics {
-        let mut m = M_ScannerMetrics::default();
-        if let Some(cycle) = self.get_cycle().await {
-            info!("cycle: {cycle:?}");
-            m.current_cycle = cycle.current;
-            m.cycles_completed_at = cycle.cycle_completed;
-            m.current_started = cycle.started;
-        }
-        m.collected_at = Utc::now();
-        m.active_paths = self.get_current_paths().await;
-        for (i, v) in self.operations.iter().enumerate() {
-            m.life_time_ops.insert(i.to_string(), v.load(Ordering::SeqCst));
-        }
-
-        m
-    }
-}
-
-pub type CloseDiskFn = Arc<dyn Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
-pub fn current_path_updater(disk: &str, _initial: &str) -> (UpdateCurrentPathFn, CloseDiskFn) {
-    let disk_1 = disk.to_string();
-    let disk_2 = disk.to_string();
-    (
-        Arc::new(move |path: &str| {
-            let disk_inner = disk_1.clone();
-            let path = path.to_string();
-            Box::pin(async move {
-                globalScannerMetrics
-                    .write()
-                    .await
-                    .current_paths
-                    .insert(disk_inner, path.to_string());
-            })
-        }),
-        Arc::new(move || {
-            let disk_inner = disk_2.clone();
-            Box::pin(async move {
-                globalScannerMetrics.write().await.current_paths.remove(&disk_inner);
-            })
-        }),
-    )
-}

ecstore/src/metrics_realtime.rs

@@ -93,7 +93,7 @@ pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts)
 
     if types.contains(&MetricType::SCANNER) {
         info!("start get scanner metrics");
-        let metrics = globalScannerMetrics.read().await.report().await;
+        let metrics = globalScannerMetrics.report().await;
         real_time_metrics.aggregated.scanner = Some(metrics);
     }
 

ecstore/src/set_disk.rs

@@ -2893,7 +2893,7 @@ impl SetDisks {
     }
 
     pub async fn ns_scanner(
-        &self,
+        self: Arc<Self>,
         buckets: &[BucketInfo],
         want_cycle: u32,
         updates: Sender<DataUsageCache>,
@@ -2911,7 +2911,7 @@ impl SetDisks {
             return Ok(());
         }
 
-        let old_cache = DataUsageCache::load(self, DATA_USAGE_CACHE_NAME).await?;
+        let old_cache = DataUsageCache::load(&self, DATA_USAGE_CACHE_NAME).await?;
         let mut cache = DataUsageCache {
             info: DataUsageCacheInfo {
                 name: DATA_USAGE_ROOT.to_string(),
@@ -2934,6 +2934,7 @@ impl SetDisks {
             permutes.shuffle(&mut rng);
             permutes
         };
+
         // Add new buckets first
         for idx in permutes.iter() {
             let b = buckets[*idx].clone();
@@ -2954,6 +2955,7 @@ impl SetDisks {
             Duration::from_secs(30) + Duration::from_secs_f64(10.0 * rng.gen_range(0.0..1.0))
         };
         let mut ticker = interval(update_time);
+
         let task = tokio::spawn(async move {
             let last_save = Some(SystemTime::now());
             let mut need_loop = true;
@@ -2983,8 +2985,8 @@ impl SetDisks {
                 }
             }
         });
+
         // Restrict parallelism for disk usage scanner
-        // upto GOMAXPROCS if GOMAXPROCS is < len(disks)
         let max_procs = num_cpus::get();
         if max_procs < disks.len() {
             disks = disks[0..max_procs].to_vec();
@@ -2997,6 +2999,7 @@ impl SetDisks {
                 Some(disk) => disk.clone(),
                 None => continue,
             };
+            let self_clone = Arc::clone(&self);
             let bucket_rx_clone = bucket_rx.clone();
             let buckets_results_tx_clone = buckets_results_tx.clone();
             futures.push(async move {
@@ -3005,7 +3008,7 @@ impl SetDisks {
                         Err(_) => return,
                         Ok(bucket_info) => {
                             let cache_name = Path::new(&bucket_info.name).join(DATA_USAGE_CACHE_NAME);
-                            let mut cache = match DataUsageCache::load(self, &cache_name.to_string_lossy()).await {
+                            let mut cache = match DataUsageCache::load(&self_clone, &cache_name.to_string_lossy()).await {
                                 Ok(cache) => cache,
                                 Err(_) => continue,
                             };
@@ -3022,6 +3025,7 @@ impl SetDisks {
                                     ..Default::default()
                                 };
                             }
+
                             // Collect updates.
                             let (tx, mut rx) = mpsc::channel(1);
                             let buckets_results_tx_inner_clone = buckets_results_tx_clone.clone();
@@ -3042,9 +3046,10 @@ impl SetDisks {
                                     }
                                 }
                             });
+
                             // Calc usage
                             let before = cache.info.last_update;
-                            let mut cache = match disk.clone().ns_scanner(&cache, tx, heal_scan_mode, None).await {
+                            let mut cache = match disk.ns_scanner(&cache, tx, heal_scan_mode, None).await {
                                 Ok(cache) => cache,
                                 Err(_) => {
                                     if cache.info.last_update > before {
@@ -3080,9 +3085,9 @@ impl SetDisks {
                 }
             });
         }
+
         info!("ns_scanner start");
         let _ = join_all(futures).await;
-        drop(buckets_results_tx);
         let _ = task.await;
         info!("ns_scanner completed");
         Ok(())

ecstore/src/store.rs

@@ -828,6 +828,7 @@ impl ECStore {
                         }
                     });
                     if let Err(err) = set
+                        .clone()
                         .ns_scanner(&all_buckets_clone, want_cycle as u32, tx, heal_scan_mode)
                         .await
                     {