merge heal

ecstore/src/store.rs

@@ -1,18 +1,19 @@
 #![allow(clippy::map_entry)]
 
-use crate::bucket::metadata;
 use crate::bucket::metadata_sys::{self, init_bucket_metadata_sys, set_bucket_metadata};
 use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname};
 use crate::config::GLOBAL_StorageClass;
 use crate::config::{self, storageclass, GLOBAL_ConfigSys};
-use crate::disk::endpoint::EndpointType;
+use crate::disk::endpoint::{Endpoint, EndpointType};
 use crate::disk::{DiskAPI, DiskInfo, DiskInfoOptions, MetaCacheEntry};
 use crate::global::{
     is_dist_erasure, is_erasure_sd, set_global_deployment_id, set_object_layer, DISK_ASSUME_UNKNOWN_SIZE, DISK_FILL_FRACTION,
     DISK_MIN_INODES, DISK_RESERVE_FRACTION, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES,
 };
-use crate::heal::heal_commands::{HealOpts, HealResultItem, HealScanMode};
-use crate::heal::heal_ops::HealObjectFn;
+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_ops::{HealEntryFn, HealSequence};
 use crate::new_object_layer_fn;
 use crate::pools::PoolMeta;
 use crate::store_api::{BackendByte, BackendDisks, BackendInfo, ListMultipartsInfo, ObjectIO, StorageInfo};
@@ -45,18 +46,21 @@ use http::HeaderMap;
 use lazy_static::lazy_static;
 use rand::Rng;
 use s3s::dto::{BucketVersioningStatus, ObjectLockConfiguration, ObjectLockEnabled, VersioningConfiguration};
+use s3s::{S3Error, S3ErrorCode};
 use std::cmp::Ordering;
 use std::slice::Iter;
+use std::time::SystemTime;
 use std::{
     collections::{HashMap, HashSet},
     sync::Arc,
     time::Duration,
 };
 use time::OffsetDateTime;
-use tokio::fs;
-use tokio::sync::{RwLock, Semaphore};
-
-use tracing::{debug, info, warn};
+use tokio::sync::mpsc::Sender;
+use tokio::sync::{broadcast, mpsc, RwLock, Semaphore};
+use tokio::time::interval;
+use tokio::{fs, select};
+use tracing::{debug, info};
 use uuid::Uuid;
 
 const MAX_UPLOADS_LIST: usize = 10000;
@@ -494,6 +498,103 @@ impl ECStore {
         internal_get_pool_info_existing_with_opts(&self.pools, bucket, object, opts).await
     }
 
+    pub async fn ns_scanner(
+        &self,
+        updates: Sender<DataUsageInfo>,
+        want_cycle: usize,
+        heal_scan_mode: HealScanMode,
+    ) -> Result<()> {
+        let all_buckets = self.list_bucket(&BucketOptions::default()).await?;
+        if all_buckets.is_empty() {
+            let _ = updates.send(DataUsageInfo::default()).await;
+            return Ok(());
+        }
+
+        let mut total_results = 0;
+        let mut result_index = 0;
+        self.pools.iter().for_each(|pool| {
+            total_results += pool.disk_set.len();
+        });
+        let results = Arc::new(RwLock::new(vec![DataUsageCache::default(); total_results]));
+        let (cancel, _) = broadcast::channel(100);
+        let first_err = Arc::new(RwLock::new(None));
+        let mut futures = Vec::new();
+        for pool in self.pools.iter() {
+            for set in pool.disk_set.iter() {
+                let index = result_index;
+                let results_clone = results.clone();
+                let first_err_clone = first_err.clone();
+                let cancel_clone = cancel.clone();
+                let all_buckets_clone = all_buckets.clone();
+                futures.push(async move {
+                    let (tx, mut rx) = mpsc::channel(100);
+                    let task = tokio::spawn(async move {
+                        loop {
+                            match rx.recv().await {
+                                Some(info) => {
+                                    results_clone.write().await[index] = info;
+                                }
+                                None => {
+                                    return;
+                                }
+                            }
+                        }
+                    });
+                    if let Err(err) = set
+                        .ns_scanner(&all_buckets_clone, want_cycle as u32, tx, heal_scan_mode)
+                        .await
+                    {
+                        let mut f_w = first_err_clone.write().await;
+                        if f_w.is_none() {
+                            *f_w = Some(err);
+                        }
+                        let _ = cancel_clone.send(true);
+                        return;
+                    }
+                    let _ = task.await;
+                });
+                result_index += 1;
+            }
+        }
+        let (update_closer_tx, mut update_close_rx) = mpsc::channel(10);
+        let mut ctx_clone = cancel.subscribe();
+        let all_buckets_clone = all_buckets.clone();
+        let task = tokio::spawn(async move {
+            let mut last_update: Option<SystemTime> = None;
+            let mut interval = interval(Duration::from_secs(30));
+            let all_merged = Arc::new(RwLock::new(DataUsageCache::default()));
+            loop {
+                select! {
+                    _ = ctx_clone.recv() => {
+                        return;
+                    }
+                    _ = update_close_rx.recv() => {
+                        update_scan(all_merged.clone(), results.clone(), &mut last_update, all_buckets_clone.clone(), updates.clone()).await;
+                        return;
+                    }
+                    _ = interval.tick() => {
+                        update_scan(all_merged.clone(), results.clone(), &mut last_update, all_buckets_clone.clone(), updates.clone()).await;
+                    }
+                }
+            }
+        });
+        let _ = join_all(futures).await;
+        let mut ctx_closer = cancel.subscribe();
+        select! {
+            _ = update_closer_tx.send(true) => {
+
+            }
+            _ = ctx_closer.recv() => {
+
+            }
+        }
+        let _ = task.await;
+        if let Some(err) = first_err.read().await.as_ref() {
+            return Err(err.clone());
+        }
+        Ok(())
+    }
+
     async fn get_latest_object_info_with_idx(
         &self,
         bucket: &str,
@@ -515,8 +616,7 @@ impl ECStore {
 
         let mut idx_res = Vec::with_capacity(self.pools.len());
 
-        let mut idx = 0;
-        for result in results {
+        for (idx, result) in results.into_iter().enumerate() {
             match result {
                 Ok(res) => {
                     idx_res.push(IndexRes {
@@ -533,8 +633,6 @@ impl ECStore {
                     });
                 }
             }
-
-            idx += 1;
         }
 
         // TODO: test order
@@ -589,6 +687,33 @@ impl ECStore {
     }
 }
 
+async fn update_scan(
+    all_merged: Arc<RwLock<DataUsageCache>>,
+    results: Arc<RwLock<Vec<DataUsageCache>>>,
+    last_update: &mut Option<SystemTime>,
+    all_buckets: Vec<BucketInfo>,
+    updates: Sender<DataUsageInfo>,
+) {
+    let mut w = all_merged.write().await;
+    *w = DataUsageCache {
+        info: DataUsageCacheInfo {
+            name: DATA_USAGE_ROOT.to_string(),
+            ..Default::default()
+        },
+        ..Default::default()
+    };
+    for info in results.read().await.iter() {
+        if info.info.last_update.is_none() {
+            return;
+        }
+        w.merge(info);
+    }
+    if w.info.last_update > *last_update && w.root().is_none() {
+        let _ = updates.send(w.dui(&w.info.name, &all_buckets)).await;
+        *last_update = w.info.last_update;
+    }
+}
+
 pub async fn find_local_disk(disk_path: &String) -> Option<DiskStore> {
     let disk_path = match fs::canonicalize(disk_path).await {
         Ok(disk_path) => disk_path,
@@ -606,6 +731,14 @@ pub async fn find_local_disk(disk_path: &String) -> Option<DiskStore> {
     None
 }
 
+pub async fn get_disk_via_endpoint(endpoint: &Endpoint) -> Option<DiskStore> {
+    let global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.read().await;
+    if global_set_drives.is_empty() {
+        return GLOBAL_LOCAL_DISK_MAP.read().await[&endpoint.to_string()].clone();
+    }
+    global_set_drives[endpoint.pool_idx as usize][endpoint.set_idx as usize][endpoint.disk_idx as usize].clone()
+}
+
 pub async fn all_local_disk_path() -> Vec<String> {
     let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await;
     disk_map.keys().cloned().collect()
@@ -1306,7 +1439,7 @@ impl StorageAPI for ECStore {
         Ok(ListMultipartsInfo {
             key_marker: key_marker.to_owned(),
             upload_id_marker: upload_id_marker.to_owned(),
-            max_uploads: max_uploads,
+            max_uploads,
             uploads,
             prefix: prefix.to_owned(),
             delimiter: delimiter.to_owned(),
@@ -1439,53 +1572,187 @@ impl StorageAPI for ECStore {
         }
         counts
     }
-    async fn heal_format(&self, dry_run: bool) -> Result<HealResultItem> {
-        unimplemented!()
+    async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
+        let mut r = HealResultItem {
+            heal_item_type: HEAL_ITEM_METADATA.to_string(),
+            detail: "disk-format".to_string(),
+            ..Default::default()
+        };
+
+        let mut count_no_heal = 0;
+        for pool in self.pools.iter() {
+            let (mut result, err) = pool.heal_format(dry_run).await?;
+            if let Some(err) = err {
+                match err.downcast_ref::<DiskError>() {
+                    Some(DiskError::NoHealRequired) => {
+                        count_no_heal += 1;
+                    }
+                    _ => {
+                        continue;
+                    }
+                }
+            }
+            r.disk_count += result.disk_count;
+            r.set_count += result.set_count;
+            r.before.append(&mut result.before);
+            r.after.append(&mut result.after);
+        }
+        if count_no_heal == self.pools.len() {
+            return Ok((r, Some(Error::new(DiskError::NoHealRequired))));
+        }
+        Ok((r, None))
     }
 
     async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
-        unimplemented!()
+        self.peer_sys.heal_bucket(bucket, opts).await
     }
-    async fn heal_object(&self, bucket: &str, object: &str, version_id: &str, opts: &HealOpts) -> Result<HealResultItem> {
+    async fn heal_object(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: &str,
+        opts: &HealOpts,
+    ) -> Result<(HealResultItem, Option<Error>)> {
         let object = utils::path::encode_dir_object(object);
-        let mut errs = HashMap::new();
-        let mut results = HashMap::new();
+        let errs = Arc::new(RwLock::new(vec![None; self.pools.len()]));
+        let results = Arc::new(RwLock::new(vec![HealResultItem::default(); self.pools.len()]));
+        let mut futures = Vec::with_capacity(self.pools.len());
         for (idx, pool) in self.pools.iter().enumerate() {
             //TODO: IsSuspended
-            match pool.heal_object(bucket, &object, version_id, opts).await {
-                Ok(mut result) => {
-                    result.object = utils::path::decode_dir_object(&result.object);
-                    results.insert(idx, result);
+            let object = object.clone();
+            let results = results.clone();
+            let errs = errs.clone();
+            futures.push(async move {
+                match pool.heal_object(bucket, &object, version_id, opts).await {
+                    Ok((mut result, err)) => {
+                        result.object = utils::path::decode_dir_object(&result.object);
+                        results.write().await.insert(idx, result);
+                        errs.write().await.insert(idx, err);
+                    }
+                    Err(err) => {
+                        errs.write().await.insert(idx, Some(err));
+                    }
                 }
-                Err(err) => {
-                    errs.insert(idx, err);
-                }
-            }
+            });
         }
+        let _ = join_all(futures).await;
 
         // Return the first nil error
-        for i in 0..self.pools.len() {
-            if !errs.contains_key(&i) {
-                return Ok(results.remove(&i).unwrap());
+        for (index, err) in errs.read().await.iter().enumerate() {
+            if err.is_none() {
+                return Ok((results.write().await.remove(index), None));
             }
         }
 
         // No pool returned a nil error, return the first non 'not found' error
-        for (k, err) in errs.iter() {
-            match err.downcast_ref::<DiskError>() {
-                Some(DiskError::FileNotFound) | Some(DiskError::FileVersionNotFound) => {}
-                _ => return Ok(results.remove(k).unwrap()),
+        for (index, err) in errs.read().await.iter().enumerate() {
+            match err {
+                Some(err) => match err.downcast_ref::<DiskError>() {
+                    Some(DiskError::FileNotFound) | Some(DiskError::FileVersionNotFound) => {}
+                    _ => return Ok((results.write().await.remove(index), Some(err.clone()))),
+                },
+                None => {
+                    return Ok((results.write().await.remove(index), None));
+                }
             }
         }
 
         // At this stage, all errors are 'not found'
         if !version_id.is_empty() {
-            return Err(Error::new(DiskError::FileVersionNotFound));
+            return Ok((HealResultItem::default(), Some(Error::new(DiskError::FileVersionNotFound))));
         }
 
-        Err(Error::new(DiskError::FileNotFound))
+        Ok((HealResultItem::default(), Some(Error::new(DiskError::FileNotFound))))
     }
-    async fn heal_objects(&self, bucket: &str, prefix: &str, opts: &HealOpts, func: HealObjectFn) -> Result<()> {
+    async fn heal_objects(
+        &self,
+        bucket: &str,
+        prefix: &str,
+        opts: &HealOpts,
+        hs: Arc<RwLock<HealSequence>>,
+        is_meta: bool,
+    ) -> Result<()> {
+        let opts_clone = *opts;
+        let heal_entry: HealEntryFn = Arc::new(move |bucket: String, entry: MetaCacheEntry, scan_mode: HealScanMode| {
+            let opts_clone = opts_clone;
+            let hs_clone = hs.clone();
+            Box::pin(async move {
+                if entry.is_dir() {
+                    return Ok(());
+                }
+
+                if bucket == RUSTFS_META_BUCKET
+                    && Pattern::new("buckets/*/.metacache/*")
+                        .map(|p| p.matches(&entry.name))
+                        .unwrap_or(false)
+                    || Pattern::new("tmp/*").map(|p| p.matches(&entry.name)).unwrap_or(false)
+                    || Pattern::new("multipart/*").map(|p| p.matches(&entry.name)).unwrap_or(false)
+                    || Pattern::new("tmp-old/*").map(|p| p.matches(&entry.name)).unwrap_or(false)
+                {
+                    return Ok(());
+                }
+                let fivs = match entry.file_info_versions(&bucket) {
+                    Ok(fivs) => fivs,
+                    Err(_) => {
+                        if is_meta {
+                            return HealSequence::heal_meta_object(hs_clone.clone(), &bucket, &entry.name, "", scan_mode).await;
+                        } else {
+                            return HealSequence::heal_object(hs_clone.clone(), &bucket, &entry.name, "", scan_mode).await;
+                        }
+                    }
+                };
+
+                if opts_clone.remove && !opts_clone.dry_run {
+                    let layer = new_object_layer_fn();
+                    let lock = layer.read().await;
+                    let store = match lock.as_ref() {
+                        Some(s) => s,
+                        None => {
+                            return Err(Error::from(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())))
+                        }
+                    };
+                    if let Err(err) = store.check_abandoned_parts(&bucket, &entry.name, &opts_clone).await {
+                        info!("unable to check object {}/{} for abandoned data: {}", bucket, entry.name, err.to_string());
+                    }
+                }
+                for version in fivs.versions.iter() {
+                    if is_meta {
+                        if let Err(err) = HealSequence::heal_meta_object(
+                            hs_clone.clone(),
+                            &bucket,
+                            &version.name,
+                            &version.version_id.map(|v| v.to_string()).unwrap_or("".to_string()),
+                            scan_mode,
+                        )
+                        .await
+                        {
+                            match err.downcast_ref() {
+                                Some(DiskError::FileNotFound) | Some(DiskError::FileVersionNotFound) => {}
+                                _ => {
+                                    return Err(err);
+                                }
+                            }
+                        }
+                    } else if let Err(err) = HealSequence::heal_object(
+                        hs_clone.clone(),
+                        &bucket,
+                        &version.name,
+                        &version.version_id.map(|v| v.to_string()).unwrap_or("".to_string()),
+                        scan_mode,
+                    )
+                    .await
+                    {
+                        match err.downcast_ref() {
+                            Some(DiskError::FileNotFound) | Some(DiskError::FileVersionNotFound) => {}
+                            _ => {
+                                return Err(err);
+                            }
+                        }
+                    }
+                }
+                Ok(())
+            })
+        });
         let mut first_err = None;
         for (idx, pool) in self.pools.iter().enumerate() {
             if opts.pool.is_some() && opts.pool.unwrap() != idx {
@@ -1498,7 +1765,7 @@ impl StorageAPI for ECStore {
                     continue;
                 }
 
-                if let Err(err) = set.list_and_heal(bucket, prefix, opts, func.clone()).await {
+                if let Err(err) = set.list_and_heal(bucket, prefix, opts, heal_entry.clone()).await {
                     if first_err.is_none() {
                         first_err = Some(err)
                     }
@@ -1736,67 +2003,6 @@ fn check_put_object_args(bucket: &str, object: &str) -> Result<()> {
     Ok(())
 }
 
-pub async fn heal_entry(
-    bucket: String,
-    entry: MetaCacheEntry,
-    scan_mode: HealScanMode,
-    opts: HealOpts,
-    func: HealObjectFn,
-) -> Result<()> {
-    if entry.is_dir() {
-        return Ok(());
-    }
-
-    // We might land at .metacache, .trash, .multipart
-    // no need to heal them skip, only when bucket
-    // is '.rustfs.sys'
-    if bucket == RUSTFS_META_BUCKET {
-        if Pattern::new("buckets/*/.metacache/*")
-            .map(|p| p.matches(&entry.name))
-            .unwrap_or(false)
-            || Pattern::new("tmp/*").map(|p| p.matches(&entry.name)).unwrap_or(false)
-            || Pattern::new("multipart/*").map(|p| p.matches(&entry.name)).unwrap_or(false)
-            || Pattern::new("tmp-old/*").map(|p| p.matches(&entry.name)).unwrap_or(false)
-        {
-            return Ok(());
-        }
-    }
-
-    let layer = new_object_layer_fn();
-    let lock = layer.read().await;
-    let store = match lock.as_ref() {
-        Some(s) => s,
-        None => return Err(Error::msg("errServerNotInitialized")),
-    };
-
-    match entry.file_info_versions(&bucket) {
-        Ok(fivs) => {
-            if opts.remove && !opts.dry_run {
-                if let Err(err) = store.check_abandoned_parts(&bucket, &entry.name, &opts).await {
-                    return Err(Error::from_string(format!(
-                        "unable to check object {}/{} for abandoned data: {}",
-                        bucket, entry.name, err
-                    )));
-                }
-            }
-
-            for version in fivs.versions.iter() {
-                let version_id = version.version_id.map_or("".to_string(), |version_id| version_id.to_string());
-                if let Err(err) = func(&bucket, &entry.name, &version_id, scan_mode) {
-                    match err.downcast_ref::<DiskError>() {
-                        Some(DiskError::FileNotFound) | Some(DiskError::FileVersionNotFound) => {}
-                        _ => return Err(err),
-                    }
-                }
-            }
-        }
-        Err(_) => {
-            return func(&bucket, &entry.name, "", scan_mode);
-        }
-    }
-    Ok(())
-}
-
 async fn get_disk_infos(disks: &[Option<DiskStore>]) -> Vec<Option<DiskInfo>> {
     let opts = &DiskInfoOptions::default();
     let mut res = vec![None; disks.len()];