From 644698e7ca5cff9279ab57356de3d15f7bcb103f Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Wed, 28 May 2025 10:17:41 +0000 Subject: [PATCH] improve scanner metric Signed-off-by: junxiang Mu <1948535941@qq.com> --- ecstore/src/heal/data_scanner.rs | 55 ++- ecstore/src/heal/data_scanner_metric.rs | 570 ++++++++++++++++-------- ecstore/src/metrics_realtime.rs | 2 +- 3 files changed, 419 insertions(+), 208 deletions(-) diff --git a/ecstore/src/heal/data_scanner.rs b/ecstore/src/heal/data_scanner.rs index 0e3f21074..7e9f5f9da 100644 --- a/ecstore/src/heal/data_scanner.rs +++ b/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.write().await.set_cycle(Some(cycle_info.clone())).await; - } + 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; + 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 { @@ -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) } diff --git a/ecstore/src/heal/data_scanner_metric.rs b/ecstore/src/heal/data_scanner_metric.rs index dedd4417f..44fe8550b 100644 --- a/ecstore/src/heal/data_scanner_metric.rs +++ b/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, + sync::{ + atomic::{AtomicU64, Ordering}, + Arc, + }, + time::{Duration, SystemTime}, + pin::Pin, }; -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> = Arc::new(RwLock::new(ScannerMetrics::new())); + pub static ref globalScannerMetrics: Arc = 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 { + 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, - latency: LastMinuteLatency, + latency: Arc>, } 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) -> Pin + Send>> + Send + Sync + 'static>; -pub type TimeSizeFn = Arc Pin + Send>> + Send + Sync + 'static>; -pub type TimeFn = Arc Pin + Send>> + Send + Sync + 'static>; +/// Current path tracker for monitoring active scan paths +struct CurrentPathTracker { + current_path: Arc>, +} +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, latency: Vec, - cycle_info: RwLock>, - current_paths: HashMap, + + // Current paths contains disk -> tracker mappings + current_paths: Arc>>>, + + // Cycle information + cycle_info: Arc>>, +} + +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) { + let start_time = SystemTime::now(); + move |_custom: &HashMap| { + 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 Box + 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) { + *self.cycle_info.write().await = cycle; + } + + /// Get current cycle information + pub async fn get_cycle(&self) -> Option { + self.cycle_info.read().await.clone() + } + + /// Get current active paths + pub async fn get_current_paths(&self) -> Vec { + 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 Pin + Send>> + Send + Sync>; +pub type CloseDiskFn = Arc Pin + 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 + 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 + 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) { - 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| { - 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 { - 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 { - 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 Pin + 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); - }) - }), - ) -} +} \ No newline at end of file diff --git a/ecstore/src/metrics_realtime.rs b/ecstore/src/metrics_realtime.rs index 5bfc189b0..509bc76b3 100644 --- a/ecstore/src/metrics_realtime.rs +++ b/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); }