tmp(1)

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

ecstore/src/disk/local.rs

@@ -19,7 +19,7 @@ use crate::disk::os::{check_path_length, is_empty_dir};
 use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE};
 use crate::error::{Error, Result};
 use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold};
-use crate::heal::data_scanner::{has_active_rules, scan_data_folder, ScannerItem, SizeSummary};
+use crate::heal::data_scanner::{has_active_rules, scan_data_folder, ScannerItem, ShouldSleepFn, SizeSummary};
 use crate::heal::data_scanner_metric::{ScannerMetric, ScannerMetrics};
 use crate::heal::data_usage_cache::{DataUsageCache, DataUsageEntry};
 use crate::heal::error::{ERR_IGNORE_FILE_CONTRIB, ERR_SKIP_FILE};
@@ -1974,6 +1974,7 @@ impl DiskAPI for LocalDisk {
         cache: &DataUsageCache,
         updates: Sender<DataUsageEntry>,
         scan_mode: HealScanMode,
+        we_sleep: ShouldSleepFn,
     ) -> Result<DataUsageCache> {
         self.scanning.fetch_add(1, Ordering::SeqCst);
         defer!(|| { self.scanning.fetch_sub(1, Ordering::SeqCst) });
@@ -2089,6 +2090,7 @@ impl DiskAPI for LocalDisk {
                 })
             }),
             scan_mode,
+            we_sleep,
         )
         .await?;
         data_usage_info.info.last_update = Some(SystemTime::now());

ecstore/src/disk/mod.rs

@@ -19,6 +19,7 @@ use crate::{
     error::{Error, Result},
     file_meta::{merge_file_meta_versions, FileMeta, FileMetaShallowVersion},
     heal::{
+        data_scanner::ShouldSleepFn,
         data_usage_cache::{DataUsageCache, DataUsageEntry},
         heal_commands::{HealScanMode, HealingTracker},
     },
@@ -350,11 +351,12 @@ impl DiskAPI for Disk {
         cache: &DataUsageCache,
         updates: Sender<DataUsageEntry>,
         scan_mode: HealScanMode,
+        we_sleep: ShouldSleepFn,
     ) -> Result<DataUsageCache> {
         info!("ns_scanner");
         match self {
-            Disk::Local(local_disk) => local_disk.ns_scanner(cache, updates, scan_mode).await,
-            Disk::Remote(remote_disk) => remote_disk.ns_scanner(cache, updates, scan_mode).await,
+            Disk::Local(local_disk) => local_disk.ns_scanner(cache, updates, scan_mode, we_sleep).await,
+            Disk::Remote(remote_disk) => remote_disk.ns_scanner(cache, updates, scan_mode, we_sleep).await,
         }
     }
 
@@ -467,6 +469,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
         cache: &DataUsageCache,
         updates: Sender<DataUsageEntry>,
         scan_mode: HealScanMode,
+        we_sleep: ShouldSleepFn,
     ) -> Result<DataUsageCache>;
     async fn healing(&self) -> Option<HealingTracker>;
 }

ecstore/src/disk/remote.rs

@@ -25,6 +25,7 @@ use crate::{
     disk::error::DiskError,
     error::{Error, Result},
     heal::{
+        data_scanner::ShouldSleepFn,
         data_usage_cache::{DataUsageCache, DataUsageEntry},
         heal_commands::{HealScanMode, HealingTracker},
     },
@@ -759,6 +760,7 @@ impl DiskAPI for RemoteDisk {
         cache: &DataUsageCache,
         updates: Sender<DataUsageEntry>,
         scan_mode: HealScanMode,
+        _we_sleep: ShouldSleepFn,
     ) -> Result<DataUsageCache> {
         info!("ns_scanner");
         let cache = serde_json::to_string(cache)?;

ecstore/src/endpoints.rs

@@ -1,4 +1,5 @@
-use tracing::warn;
+use tracing::{info, warn};
+use url::Url;
 
 use crate::{
     disk::endpoint::{Endpoint, EndpointType},

ecstore/src/heal/background_heal_ops.rs

@@ -1,3 +1,4 @@
+use madmin::heal_commands::HealResultItem;
 use std::{cmp::Ordering, env, path::PathBuf, sync::Arc, time::Duration};
 use tokio::{
     sync::{
@@ -10,7 +11,7 @@ use tracing::{error, info};
 use uuid::Uuid;
 
 use super::{
-    heal_commands::{HealOpts, HealResultItem},
+    heal_commands::HealOpts,
     heal_ops::{new_bg_heal_sequence, HealSequence},
 };
 use crate::heal::error::ERR_RETRY_HEALING;

ecstore/src/heal/data_scanner.rs

@@ -62,7 +62,7 @@ use crate::{
     store_api::{FileInfo, ObjectInfo},
 };
 
-const _DATA_SCANNER_SLEEP_PER_FOLDER: Duration = Duration::from_millis(1); // Time to wait between folders.
+const DATA_SCANNER_SLEEP_PER_FOLDER: Duration = Duration::from_millis(1); // Time to wait between folders.
 const DATA_USAGE_UPDATE_DIR_CYCLES: u32 = 16; // Visit all folders every n cycles.
 const DATA_SCANNER_COMPACT_LEAST_OBJECT: u64 = 500; // Compact when there are less than this many objects in a branch.
 const DATA_SCANNER_COMPACT_AT_CHILDREN: u64 = 10000; // Compact when there are this many children in a branch.
@@ -73,7 +73,6 @@ const DATA_SCANNER_START_DELAY: Duration = Duration::from_secs(60); // Time to w
 pub const HEAL_DELETE_DANGLING: bool = true;
 const HEAL_OBJECT_SELECT_PROB: u64 = 1024; // Overall probability of a file being scanned; one in n.
 
-// static SCANNER_SLEEPER: () = new_dynamic_sleeper(2, Duration::from_secs(1), true); // Keep defaults same as config defaults
 static SCANNER_CYCLE: AtomicU64 = AtomicU64::new(DATA_SCANNER_START_DELAY.as_secs());
 static _SCANNER_IDLE_MODE: AtomicU32 = AtomicU32::new(0); // default is throttled when idle
 static SCANNER_EXCESS_OBJECT_VERSIONS: AtomicU64 = AtomicU64::new(100);
@@ -81,9 +80,67 @@ static SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL_SIZE: AtomicU64 = AtomicU64::new(102
 static SCANNER_EXCESS_FOLDERS: AtomicU64 = AtomicU64::new(50_000);
 
 lazy_static! {
+    static ref SCANNER_SLEEPER: RwLock<DynamicSleeper> = RwLock::new(new_dynamic_sleeper(2.0, Duration::from_secs(1), true));
     pub static ref globalHealConfig: Arc<RwLock<Config>> = Arc::new(RwLock::new(Config::default()));
 }
 
+struct DynamicSleeper {
+    factor: f64,
+    max_sleep: Duration,
+    min_sleep: Duration,
+    _is_scanner: bool,
+}
+
+type TimerFn = Pin<Box<dyn Future<Output = ()> + Send>>;
+impl DynamicSleeper {
+    fn timer() -> TimerFn {
+        let t = SystemTime::now();
+        Box::pin(async move {
+            let done_at = SystemTime::now().duration_since(t).unwrap_or_default();
+            SCANNER_SLEEPER.read().await.sleep(done_at).await;
+        })
+    }
+
+    async fn sleep(&self, base: Duration) {
+        let (min_wait, max_wait) = (self.min_sleep, self.max_sleep);
+        let factor = self.factor;
+
+        let want_sleep = {
+            let tmp = base.mul_f64(factor);
+            if tmp < min_wait {
+                return;
+            }
+
+            if max_wait > Duration::from_secs(0) && tmp > max_wait {
+                max_wait
+            } else {
+                tmp
+            }
+        };
+        sleep(want_sleep).await;
+    }
+
+    fn _update(&mut self, factor: f64, max_wait: Duration) -> Result<()> {
+        if (self.factor - factor).abs() < 1e-10 && self.max_sleep == max_wait {
+            return Ok(());
+        }
+
+        self.factor = factor;
+        self.max_sleep = max_wait;
+
+        Ok(())
+    }
+}
+
+fn new_dynamic_sleeper(factor: f64, max_wait: Duration, is_scanner: bool) -> DynamicSleeper {
+    DynamicSleeper {
+        factor,
+        max_sleep: max_wait,
+        min_sleep: Duration::from_micros(100),
+        _is_scanner: is_scanner,
+    }
+}
+
 pub async fn init_data_scanner() {
     tokio::spawn(async move {
         loop {
@@ -457,6 +514,7 @@ struct CachedFolder {
 pub type GetSizeFn =
     Box<dyn Fn(&ScannerItem) -> Pin<Box<dyn Future<Output = Result<SizeSummary>> + Send>> + Send + Sync + 'static>;
 pub type UpdateCurrentPathFn = Arc<dyn Fn(&str) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
+pub type ShouldSleepFn = Option<Arc<dyn Fn() -> bool + Send + Sync + 'static>>;
 
 struct FolderScanner {
     root: String,
@@ -474,6 +532,7 @@ struct FolderScanner {
     update_current_path: UpdateCurrentPathFn,
     skip_heal: AtomicBool,
     drive: LocalDrive,
+    we_sleep: ShouldSleepFn,
 }
 
 impl FolderScanner {
@@ -514,6 +573,12 @@ impl FolderScanner {
                 None
             };
 
+            if let Some(should_sleep) = &self.we_sleep {
+                if should_sleep() {
+                    SCANNER_SLEEPER.read().await.sleep(DATA_SCANNER_SLEEP_PER_FOLDER).await;
+                }
+            }
+
             let mut existing_folders = Vec::new();
             let mut new_folders = Vec::new();
             let mut found_objects: bool = false;
@@ -553,6 +618,16 @@ impl FolderScanner {
                         continue;
                     }
 
+                    let _wait = if let Some(should_sleep) = &self.we_sleep {
+                        if should_sleep() {
+                            DynamicSleeper::timer()
+                        } else {
+                            Box::pin(async {})
+                        }
+                    } else {
+                        Box::pin(async {})
+                    };
+
                     let mut item = ScannerItem {
                         path: Path::new(&self.root).join(&ent_name).to_string_lossy().to_string(),
                         bucket,
@@ -1001,6 +1076,7 @@ pub async fn scan_data_folder(
     cache: &DataUsageCache,
     get_size_fn: GetSizeFn,
     heal_scan_mode: HealScanMode,
+    should_sleep: ShouldSleepFn,
 ) -> Result<DataUsageCache> {
     if cache.info.name.is_empty() || cache.info.name == DATA_USAGE_ROOT {
         return Err(Error::from_string("internal error: root scan attempted"));
@@ -1029,6 +1105,7 @@ pub async fn scan_data_folder(
         disks_quorum: disks.len() / 2,
         skip_heal,
         drive: drive.clone(),
+        we_sleep: should_sleep,
     };
 
     if *GLOBAL_IsErasure.read().await || !cache.info.skip_healing {

ecstore/src/heal/heal_commands.rs

@@ -24,7 +24,6 @@ use crate::{
 use super::{background_heal_ops::get_local_disks_to_heal, heal_ops::BG_HEALING_UUID};
 
 pub type HealScanMode = usize;
-pub type HealItemType = String;
 
 pub const HEAL_UNKNOWN_SCAN: HealScanMode = 0;
 pub const HEAL_NORMAL_SCAN: HealScanMode = 1;
@@ -66,49 +65,6 @@ pub struct HealOpts {
     pub set: Option<usize>,
 }
 
-#[derive(Clone, Debug, Default, Serialize, Deserialize)]
-pub struct HealDriveInfo {
-    pub uuid: String,
-    pub endpoint: String,
-    pub state: String,
-}
-
-#[derive(Clone, Debug, Default, Serialize, Deserialize)]
-pub struct Infos {
-    #[serde(rename = "drives")]
-    pub drives: Vec<HealDriveInfo>,
-}
-
-#[derive(Clone, Debug, Default, Serialize, Deserialize)]
-pub struct HealResultItem {
-    #[serde(rename = "resultId")]
-    pub result_index: usize,
-    #[serde(rename = "type")]
-    pub heal_item_type: HealItemType,
-    #[serde(rename = "bucket")]
-    pub bucket: String,
-    #[serde(rename = "object")]
-    pub object: String,
-    #[serde(rename = "versionId")]
-    pub version_id: String,
-    #[serde(rename = "detail")]
-    pub detail: String,
-    #[serde(rename = "parityBlocks")]
-    pub parity_blocks: usize,
-    #[serde(rename = "dataBlocks")]
-    pub data_blocks: usize,
-    #[serde(rename = "diskCount")]
-    pub disk_count: usize,
-    #[serde(rename = "setCount")]
-    pub set_count: usize,
-    #[serde(rename = "before")]
-    pub before: Infos,
-    #[serde(rename = "after")]
-    pub after: Infos,
-    #[serde(rename = "objectSize")]
-    pub object_size: usize,
-}
-
 #[derive(Debug, Serialize, Deserialize)]
 pub struct HealStartSuccess {
     #[serde(rename = "clientToken")]

ecstore/src/heal/heal_ops.rs

@@ -2,19 +2,14 @@ use super::{
     background_heal_ops::HealTask,
     data_scanner::HEAL_DELETE_DANGLING,
     error::ERR_SKIP_FILE,
-    heal_commands::{
-        HealItemType, HealOpts, HealResultItem, HealScanMode, HealStopSuccess, HealingTracker, HEAL_ITEM_BUCKET_METADATA,
-    },
+    heal_commands::{HealOpts, HealScanMode, HealStopSuccess, HealingDisk, HealingTracker, HEAL_ITEM_BUCKET_METADATA},
 };
 use crate::store_api::StorageAPI;
 use crate::{
     config::common::CONFIG_PREFIX,
     disk::RUSTFS_META_BUCKET,
     global::GLOBAL_BackgroundHealRoutine,
-    heal::{
-        error::ERR_HEAL_STOP_SIGNALLED,
-        heal_commands::{HealDriveInfo, DRIVE_STATE_OK},
-    },
+    heal::{error::ERR_HEAL_STOP_SIGNALLED, heal_commands::DRIVE_STATE_OK},
 };
 use crate::{
     disk::{endpoint::Endpoint, MetaCacheEntry},
@@ -32,6 +27,7 @@ use crate::{
 use chrono::Utc;
 use futures::join;
 use lazy_static::lazy_static;
+use madmin::heal_commands::{HealDriveInfo, HealItemType, HealResultItem};
 use serde::{Deserialize, Serialize};
 use std::{
     collections::HashMap,

ecstore/src/lib.rs

@@ -14,7 +14,7 @@ pub mod heal;
 pub mod metrics_realtime;
 pub mod notification_sys;
 pub mod peer;
-mod peer_rest_client;
+pub mod peer_rest_client;
 mod quorum;
 pub mod set_disk;
 mod sets;

ecstore/src/notification_sys.rs

@@ -25,13 +25,13 @@ pub fn get_global_notification_sys() -> Option<&'static NotificationSys> {
 }
 
 pub struct NotificationSys {
-    peer_clients: Vec<Option<PeerRestClient>>,
-    all_peer_clients: Vec<Option<PeerRestClient>>,
+    pub peer_clients: Vec<Option<PeerRestClient>>,
+    pub all_peer_clients: Vec<Option<PeerRestClient>>,
 }
 
 impl NotificationSys {
     pub async fn new(eps: EndpointServerPools) -> Self {
-        let (peer_clients, all_peer_clients) = PeerRestClient::new_clients(eps).await;
+        let (peer_clients, all_peer_clients) = PeerRestClient::new_clients(&eps).await;
         Self {
             peer_clients,
             all_peer_clients,

ecstore/src/peer.rs

@@ -1,5 +1,6 @@
 use async_trait::async_trait;
 use futures::future::join_all;
+use madmin::heal_commands::{HealDriveInfo, HealResultItem};
 use protos::node_service_time_out_client;
 use protos::proto_gen::node_service::{
     DeleteBucketRequest, GetBucketInfoRequest, HealBucketRequest, ListBucketRequest, MakeBucketRequest,
@@ -14,8 +15,7 @@ use crate::disk::error::is_all_buckets_not_found;
 use crate::disk::{DiskAPI, DiskStore};
 use crate::global::GLOBAL_LOCAL_DISK_MAP;
 use crate::heal::heal_commands::{
-    HealDriveInfo, HealOpts, HealResultItem, DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING, DRIVE_STATE_OFFLINE, DRIVE_STATE_OK,
-    HEAL_ITEM_BUCKET,
+    HealOpts, DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING, DRIVE_STATE_OFFLINE, DRIVE_STATE_OK, HEAL_ITEM_BUCKET,
 };
 use crate::heal::heal_ops::RUESTFS_RESERVED_BUCKET;
 use crate::quorum::{bucket_op_ignored_errs, reduce_write_quorum_errs};

ecstore/src/peer_rest_client.rs

@@ -35,7 +35,7 @@ pub const PEER_RESTSIGNAL: &str = "signal";
 pub const PEER_RESTSUB_SYS: &str = "sub-sys";
 pub const PEER_RESTDRY_RUN: &str = "dry-run";
 
-#[derive(Debug, Clone)]
+#[derive(Clone, Debug)]
 pub struct PeerRestClient {
     pub host: XHost,
     pub grid_host: String,

ecstore/src/set_disk.rs

@@ -31,8 +31,8 @@ use crate::{
         data_usage::{DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT},
         data_usage_cache::{DataUsageCacheInfo, DataUsageEntry, DataUsageEntryInfo},
         heal_commands::{
-            HealDriveInfo, HealOpts, HealResultItem, HealScanMode, HealingTracker, DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING,
-            DRIVE_STATE_OFFLINE, DRIVE_STATE_OK, HEAL_DEEP_SCAN, HEAL_ITEM_OBJECT, HEAL_NORMAL_SCAN,
+            HealOpts, HealScanMode, HealingTracker, DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING, DRIVE_STATE_OFFLINE,
+            DRIVE_STATE_OK, HEAL_DEEP_SCAN, HEAL_ITEM_OBJECT, HEAL_NORMAL_SCAN,
         },
         heal_ops::BG_HEALING_UUID,
     },
@@ -64,6 +64,7 @@ use lock::{
     namespace_lock::{new_nslock, NsLockMap},
     LockApi,
 };
+use madmin::heal_commands::{HealDriveInfo, HealResultItem};
 use rand::{
     thread_rng,
     {seq::SliceRandom, Rng},
@@ -2811,7 +2812,7 @@ impl SetDisks {
                             });
                             // Calc usage
                             let before = cache.info.last_update;
-                            let cache = match disk.clone().ns_scanner(&cache, tx, heal_scan_mode).await {
+                            let cache = match disk.clone().ns_scanner(&cache, tx, heal_scan_mode, None).await {
                                 Ok(cache) => cache,
                                 Err(_) => {
                                     if cache.info.last_update > before {

ecstore/src/sets.rs

@@ -5,6 +5,7 @@ use common::globals::GLOBAL_Local_Node_Name;
 use futures::future::join_all;
 use http::HeaderMap;
 use lock::{namespace_lock::NsLockMap, new_lock_api, LockApi};
+use madmin::heal_commands::{HealDriveInfo, HealResultItem};
 use tokio::sync::RwLock;
 use uuid::Uuid;
 
@@ -18,8 +19,7 @@ use crate::{
     error::{Error, Result},
     global::{is_dist_erasure, GLOBAL_LOCAL_DISK_SET_DRIVES},
     heal::heal_commands::{
-        HealDriveInfo, HealOpts, HealResultItem, DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING, DRIVE_STATE_OFFLINE, DRIVE_STATE_OK,
-        HEAL_ITEM_METADATA,
+        HealOpts, DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING, DRIVE_STATE_OFFLINE, DRIVE_STATE_OK, HEAL_ITEM_METADATA,
     },
     set_disk::SetDisks,
     store_api::{

ecstore/src/store.rs

@@ -12,7 +12,7 @@ use crate::global::{
 };
 use crate::heal::data_usage::{DataUsageInfo, DATA_USAGE_ROOT};
 use crate::heal::data_usage_cache::{DataUsageCache, DataUsageCacheInfo};
-use crate::heal::heal_commands::{HealOpts, HealResultItem, HealScanMode, HEAL_ITEM_METADATA};
+use crate::heal::heal_commands::{HealOpts, HealScanMode, HEAL_ITEM_METADATA};
 use crate::heal::heal_ops::{HealEntryFn, HealSequence};
 use crate::new_object_layer_fn;
 use crate::notification_sys::get_global_notification_sys;
@@ -45,6 +45,7 @@ use futures::future::join_all;
 use glob::Pattern;
 use http::HeaderMap;
 use lazy_static::lazy_static;
+use madmin::heal_commands::HealResultItem;
 use rand::Rng;
 use s3s::dto::{BucketVersioningStatus, ObjectLockConfiguration, ObjectLockEnabled, VersioningConfiguration};
 use std::cmp::Ordering;

ecstore/src/store_api.rs

@@ -2,12 +2,14 @@ use crate::heal::heal_ops::HealSequence;
 use crate::{
     disk::DiskStore,
     error::{Error, Result},
-    heal::heal_commands::{HealOpts, HealResultItem},
+    heal::heal_commands::HealOpts,
     utils::path::decode_dir_object,
     xhttp,
 };
 use futures::StreamExt;
 use http::HeaderMap;
+use madmin::heal_commands::HealResultItem;
+use madmin::info_commands::DiskMetrics;
 use rmp_serde::Serializer;
 use s3s::{dto::StreamingBlob, Body};
 use serde::{Deserialize, Serialize};

ecstore/src/utils/mod.rs

@@ -6,6 +6,5 @@ pub mod hash;
 pub mod net;
 pub mod os;
 pub mod path;
-pub mod time;
 pub mod wildcard;
 pub mod xml;

ecstore/src/utils/time.rs

@@ -1,55 +0,0 @@
-use std::time::Duration;
-
-use tracing::info;
-
-pub fn parse_duration(s: &str) -> Option<Duration> {
-    if s.ends_with("ms") {
-        if let Ok(s) = s.trim_end_matches("ms").parse::<u64>() {
-            return Some(Duration::from_millis(s));
-        }
-    } else if s.ends_with("s") {
-        if let Ok(s) = s.trim_end_matches('s').parse::<u64>() {
-            return Some(Duration::from_secs(s));
-        }
-    } else if s.ends_with("m") {
-        if let Ok(s) = s.trim_end_matches('m').parse::<u64>() {
-            return Some(Duration::from_secs(s * 60));
-        }
-    } else if s.ends_with("h") {
-        if let Ok(s) = s.trim_end_matches('h').parse::<u64>() {
-            return Some(Duration::from_secs(s * 60 * 60));
-        }
-    }
-    info!("can not parse duration, s: {}", s);
-    None
-}
-
-#[cfg(test)]
-mod test {
-    use std::time::Duration;
-
-    use super::parse_duration;
-
-    #[test]
-    fn test_parse_dur() {
-        let s = String::from("3s");
-        let dur = parse_duration(&s);
-        println!("{:?}", dur);
-        assert_eq!(Some(Duration::from_secs(3)), dur);
-
-        let s = String::from("3ms");
-        let dur = parse_duration(&s);
-        println!("{:?}", dur);
-        assert_eq!(Some(Duration::from_millis(3)), dur);
-
-        let s = String::from("3m");
-        let dur = parse_duration(&s);
-        println!("{:?}", dur);
-        assert_eq!(Some(Duration::from_secs(3 * 60)), dur);
-
-        let s = String::from("3h");
-        let dur = parse_duration(&s);
-        println!("{:?}", dur);
-        assert_eq!(Some(Duration::from_secs(3 * 60 * 60)), dur);
-    }
-}