improve scanner metric

Signed-off-by: junxiang Mu <1948535941@qq.com>

ecstore/src/heal/data_scanner.rs

@@ -144,16 +144,16 @@ 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
@@ -161,12 +161,12 @@ fn new_dynamic_sleeper(factor: f64, max_wait: Duration, is_scanner: bool) -> Dyn
 /// 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;
-
+            
             // Calculate randomized sleep duration
             // Use random factor (0.0 to 1.0) multiplied by the scanner cycle duration
             let random_factor: f64 = {
@@ -175,7 +175,7 @@ pub async fn init_data_scanner() {
             };
             let base_cycle_duration = SCANNER_CYCLE.load(Ordering::SeqCst) as f64;
             let sleep_duration_secs = random_factor * base_cycle_duration;
-
+            
             // Ensure minimum sleep duration of 1 second to avoid high CPU usage
             let sleep_duration = if sleep_duration_secs < 1.0 {
                 Duration::from_secs(1)
@@ -183,8 +183,11 @@ pub async fn init_data_scanner() {
                 Duration::from_secs_f64(sleep_duration_secs)
             };
 
-            info!(duration_secs = sleep_duration.as_secs(), "Data scanner sleeping before next cycle");
+            info!(
-
+                duration_secs = sleep_duration.as_secs(),
+                "Data scanner sleeping before next cycle"
+            );
+            
             // Sleep with the calculated duration
             sleep(sleep_duration).await;
         }
@@ -192,20 +195,20 @@ pub async fn init_data_scanner() {
 }
 
 /// 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!("Starting data scanner cycle");
-
+    
     // Get the object layer, return early if not available
     let Some(store) = new_object_layer_fn() else {
         error!("Object layer not initialized, skipping scanner cycle");
@@ -232,20 +235,22 @@ async fn run_data_scanner() {
 
     // 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;
-        globalScannerMetrics.write().await.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 =
+    let scan_mode = get_cycle_scan_mode(
-        get_cycle_scan_mode(cycle_info.current, bg_heal_info.bitrot_start_cycle, bg_heal_info.bitrot_start_time).await;
+        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 {
@@ -275,22 +280,26 @@ async fn run_data_scanner() {
     );
 
     // Run the namespace scanner
-    match store.clone().ns_scanner(tx, cycle_info.current as usize, scan_mode).await {
+    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);
             }
 
             // Update global metrics with completion info
-            globalScannerMetrics.write().await.set_cycle(Some(cycle_info.clone())).await;
+            globalScannerMetrics.set_cycle(Some(cycle_info.clone())).await;
 
             // Persist updated cycle information
             // ignore error, continue.
@@ -312,7 +321,7 @@ async fn run_data_scanner() {
     }
 
     // Complete metrics collection for this cycle
-    stop_fn(&scan_result).await;
+    stop_fn(&scan_result);
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -560,7 +569,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},
+    sync::{
-    time::SystemTime,
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    },
+    time::{Duration, SystemTime},
+    pin::Pin,
 };
-use tokio::sync::RwLock;
+use tokio::sync::{Mutex, RwLock};
-use tracing::{debug, info};
+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,202 +58,404 @@ 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,
+    latency: Arc<Mutex<LastMinuteLatency>>,
-    cached: AccElem,
-    mu: RwLock<bool>,
-    latency: LastMinuteLatency,
 }
 
 impl Clone for LockedLastMinuteLatency {
     fn clone(&self) -> Self {
         Self {
-            cached_sec: AtomicU64::new(0),
+            latency: Arc::clone(&self.latency),
-            cached: self.cached.clone(),
-            mu: RwLock::new(true),
-            latency: self.latency.clone(),
         }
     }
 }
 
 impl LockedLastMinuteLatency {
-    pub async fn add(&mut self, value: &Duration) {
+    pub fn new() -> Self {
-        self.add_size(value, 0).await;
+        Self {
-    }
+            latency: Arc::new(Mutex::new(LastMinuteLatency::default())),
-
-    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);
         }
-        self.cached.n += 1;
-        self.cached.total += value.as_secs();
-        self.cached.size += sz;
     }
 
-    pub async fn total(&mut self) -> AccElem {
+    /// Add a duration measurement
-        let _ = self.mu.read().await;
+    pub async fn add(&self, duration: Duration) {
-        self.latency.get_total()
+        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>;
+/// Current path tracker for monitoring active scan paths
-pub type TimeSizeFn = Arc<dyn Fn(u64) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
+struct CurrentPathTracker {
-pub type TimeFn = Arc<dyn Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
+    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 {
     fn default() -> Self {
         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);
     }