move admin router to rustfs, add pool stats/list admin api

ecstore/src/bucket/metadata_sys.rs

@@ -149,8 +149,11 @@ impl BucketMetadataSys {
     }
     async fn init_internal(&self, buckets: Vec<String>) -> Result<()> {
         let count = {
-            let endpoints = GLOBAL_Endpoints.read().await;
-            endpoints.es_count() * 10
+            if let Some(endpoints) = GLOBAL_Endpoints.get() {
+                endpoints.es_count() * 10
+            } else {
+                return Err(Error::msg("GLOBAL_Endpoints not init"));
+            }
         };
 
         let mut failed_buckets: HashSet<String> = HashSet::new();

ecstore/src/endpoints.rs

@@ -405,7 +405,7 @@ pub struct PoolEndpoints {
 }
 
 /// list of list of endpoints
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Default)]
 pub struct EndpointServerPools(pub Vec<PoolEndpoints>);
 
 impl From<Vec<PoolEndpoints>> for EndpointServerPools {
@@ -430,6 +430,17 @@ impl EndpointServerPools {
     pub fn reset(&mut self, eps: Vec<PoolEndpoints>) {
         self.0 = eps;
     }
+    pub fn legacy(&self) -> bool {
+        self.0.len() == 1 && self.0[0].legacy
+    }
+    pub fn get_pool_idx(&self, cmd_line: &str) -> Option<usize> {
+        for (idx, eps) in self.0.iter().enumerate() {
+            if eps.cmd_line.as_str() == cmd_line {
+                return Some(idx);
+            }
+        }
+        None
+    }
     pub fn from_volumes(server_addr: &str, endpoints: Vec<String>) -> Result<(EndpointServerPools, SetupType)> {
         let layouts = DisksLayout::from_volumes(endpoints.as_slice())?;
 

ecstore/src/global.rs

@@ -1,11 +1,15 @@
 use lazy_static::lazy_static;
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::HashMap,
+    sync::{Arc, OnceLock},
+};
 use tokio::sync::RwLock;
 use uuid::Uuid;
 
 use crate::{
     disk::DiskStore,
     endpoints::{EndpointServerPools, PoolEndpoints, SetupType},
+    error::{Error, Result},
     heal::{background_heal_ops::HealRoutine, heal_ops::AllHealState},
     store::ECStore,
 };
@@ -23,7 +27,7 @@ lazy_static! {
     pub static ref GLOBAL_IsErasureSD: RwLock<bool> = RwLock::new(false);
     pub static ref GLOBAL_LOCAL_DISK_MAP: Arc<RwLock<HashMap<String, Option<DiskStore>>>> = Arc::new(RwLock::new(HashMap::new()));
     pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc<RwLock<TypeLocalDiskSetDrives>> = Arc::new(RwLock::new(Vec::new()));
-    pub static ref GLOBAL_Endpoints: RwLock<EndpointServerPools> = RwLock::new(EndpointServerPools(Vec::new()));
+    pub static ref GLOBAL_Endpoints: OnceLock<EndpointServerPools> = OnceLock::new();
     pub static ref GLOBAL_RootDiskThreshold: RwLock<u64> = RwLock::new(0);
     pub static ref GLOBAL_BackgroundHealRoutine: Arc<RwLock<HealRoutine>> = HealRoutine::new();
     pub static ref GLOBAL_BackgroundHealState: Arc<RwLock<AllHealState>> = AllHealState::new(false);
@@ -40,9 +44,11 @@ pub async fn get_global_deployment_id() -> Uuid {
     *id_ptr
 }
 
-pub async fn set_global_endpoints(eps: Vec<PoolEndpoints>) {
-    let mut endpoints = GLOBAL_Endpoints.write().await;
-    endpoints.reset(eps);
+pub fn set_global_endpoints(eps: Vec<PoolEndpoints>) -> Result<()> {
+    GLOBAL_Endpoints
+        .set(EndpointServerPools::from(eps))
+        .map_err(|_| Error::msg("GLOBAL_Endpoints set faild"))?;
+    Ok(())
 }
 
 pub fn new_object_layer_fn() -> Arc<RwLock<Option<ECStore>>> {
@@ -84,10 +90,12 @@ pub async fn update_erasure_type(setup_type: SetupType) {
     *is_erasure_sd = setup_type == SetupType::ErasureSD;
 }
 
-pub async fn is_legacy() -> bool {
-    let lock = GLOBAL_Endpoints.read().await;
-    let endpoints = lock.as_ref();
-    endpoints.len() == 1 && endpoints[0].legacy
-}
+// pub fn is_legacy() -> bool {
+//     if let Some(endpoints) = GLOBAL_Endpoints.get() {
+//         endpoints.as_ref().len() == 1 && endpoints.as_ref()[0].legacy
+//     } else {
+//         false
+//     }
+// }
 
 type TypeLocalDiskSetDrives = Vec<Vec<Vec<Option<DiskStore>>>>;

ecstore/src/lib.rs

@@ -26,7 +26,7 @@ pub mod pools;
 pub mod store_err;
 pub mod xhttp;
 
-pub use global::is_legacy;
 pub use global::new_object_layer_fn;
 pub use global::set_global_endpoints;
 pub use global::update_erasure_type;
+pub use global::GLOBAL_Endpoints;

ecstore/src/pools.rs

@@ -1,10 +1,12 @@
 use crate::error::{Error, Result};
 use crate::store_api::{StorageAPI, StorageDisk, StorageInfo};
+use crate::store_err::StorageError;
 use crate::{sets::Sets, store::ECStore};
+use serde::Serialize;
 use std::sync::Arc;
 use time::OffsetDateTime;
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct PoolStatus {
     pub id: usize,
     pub cmd_line: String,
@@ -43,11 +45,36 @@ impl PoolMeta {
 
         self.pools[idx].decommission.is_some()
     }
+
+    pub fn decommission_cancel(&mut self, idx: usize) -> bool {
+        if let Some(stats) = self.pools.get_mut(idx) {
+            if let Some(d) = &stats.decommission {
+                if !d.canceled {
+                    stats.last_update = OffsetDateTime::now_utc();
+
+                    let mut pd = d.clone();
+                    pd.start_time = None;
+                    pd.canceled = true;
+                    pd.failed = false;
+                    pd.complete = false;
+
+                    stats.decommission = Some(pd);
+                    true
+                } else {
+                    false
+                }
+            } else {
+                false
+            }
+        } else {
+            false
+        }
+    }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Default)]
 pub struct PoolDecommissionInfo {
-    pub start_time: OffsetDateTime,
+    pub start_time: Option<OffsetDateTime>,
     pub start_size: usize,
     pub total_size: usize,
     pub current_size: usize,
@@ -74,27 +101,84 @@ pub struct PoolSpaceInfo {
 }
 
 impl ECStore {
-    pub fn status(&self, _idx: usize) -> Result<PoolStatus> {
-        unimplemented!()
+    pub async fn status(&self, idx: usize) -> Result<PoolStatus> {
+        let space_info = self.get_decommission_pool_space_info(idx).await?;
+        let mut pool_info = self.pool_meta.pools[idx].clone();
+        if let Some(d) = pool_info.decommission.as_mut() {
+            d.total_size = space_info.total;
+            d.current_size = space_info.free;
+        } else {
+            pool_info.decommission = Some(PoolDecommissionInfo {
+                total_size: space_info.total,
+                current_size: space_info.free,
+                ..Default::default()
+            });
+        }
+
+        Ok(pool_info)
     }
 
-    async fn _get_decommission_pool_space_info(&self, idx: usize) -> Result<PoolSpaceInfo> {
+    async fn get_decommission_pool_space_info(&self, idx: usize) -> Result<PoolSpaceInfo> {
         if let Some(sets) = self.pools.get(idx) {
             let mut info = sets.storage_info().await;
             info.backend = self.backend_info().await;
 
-            unimplemented!()
+            let total = get_total_usable_capacity(&info.disks, &info);
+            let free = get_total_usable_capacity_free(&info.disks, &info);
+
+            Ok(PoolSpaceInfo {
+                free,
+                total,
+                used: total - free,
+            })
         } else {
             Err(Error::msg("InvalidArgument"))
         }
     }
+
+    pub async fn decommission_cancel(&mut self, idx: usize) -> Result<()> {
+        if self.single_pool() {
+            return Err(Error::msg("InvalidArgument"));
+        }
+
+        let Some(has_canceler) = self.decommission_cancelers.get(idx) else {
+            return Err(Error::msg("InvalidArgument"));
+        };
+
+        if has_canceler.is_none() {
+            return Err(Error::new(StorageError::DecommissionNotStarted));
+        }
+
+        if self.pool_meta.decommission_cancel(idx) {
+            // FIXME:
+        }
+
+        unimplemented!()
+    }
 }
 
-fn _get_total_usable_capacity(disks: &Vec<StorageDisk>, _info: &StorageInfo) -> usize {
-    for _disk in disks.iter() {
-        // if disk.pool_index < 0 || info.backend.standard_scdata.len() <= disk.pool_index {
-        //     continue;
-        // }
+fn get_total_usable_capacity(disks: &Vec<StorageDisk>, info: &StorageInfo) -> usize {
+    let mut capacity = 0;
+    for disk in disks.iter() {
+        if disk.pool_index < 0 || info.backend.standard_sc_data.len() <= disk.pool_index as usize {
+            continue;
+        }
+        if (disk.disk_index as usize) < info.backend.standard_sc_data[disk.pool_index as usize] {
+            capacity += disk.total_space as usize;
+        }
     }
-    unimplemented!()
+    capacity
+}
+
+fn get_total_usable_capacity_free(disks: &Vec<StorageDisk>, info: &StorageInfo) -> usize {
+    let mut capacity = 0;
+    for disk in disks.iter() {
+        if disk.pool_index < 0 || info.backend.standard_sc_data.len() <= disk.pool_index as usize {
+            continue;
+        }
+        if (disk.disk_index as usize) < info.backend.standard_sc_data[disk.pool_index as usize] {
+            capacity += disk.available_space as usize;
+        }
+    }
+    capacity
 }

ecstore/src/store.rs

@@ -77,6 +77,7 @@ pub struct ECStore {
     pub peer_sys: S3PeerSys,
     // pub local_disks: Vec<DiskStore>,
     pub pool_meta: PoolMeta,
+    pub decommission_cancelers: Vec<Option<usize>>,
 }
 
 impl ECStore {
@@ -197,12 +198,14 @@ impl ECStore {
         let mut pool_meta = PoolMeta::new(pools.clone());
         pool_meta.dont_save = true;
 
+        let decommission_cancelers = vec![None; pools.len()];
         let ec = ECStore {
             id: deployment_id.unwrap(),
             disk_map,
             pools,
             peer_sys,
             pool_meta,
+            decommission_cancelers,
         };
 
         set_object_layer(ec.clone()).await;
@@ -239,7 +242,7 @@ impl ECStore {
     //     self.local_disks.clone()
     // }
 
-    fn single_pool(&self) -> bool {
+    pub fn single_pool(&self) -> bool {
         self.pools.len() == 1
     }
 

ecstore/src/store_err.rs

@@ -71,6 +71,9 @@ pub enum StorageError {
 
     #[error("Storage resources are insufficient for the write operation")]
     InsufficientWriteQuorum,
+
+    #[error("Decommission not started")]
+    DecommissionNotStarted,
 }
 
 pub fn to_object_err(err: Error, params: Vec<&str>) -> Error {