support some peer rest api

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

ecstore/src/heal/background_heal_ops.rs

@@ -65,7 +65,7 @@ async fn init_background_healing() {
         .await;
 }
 
-async fn get_local_disks_to_heal() -> Vec<Endpoint> {
+pub async fn get_local_disks_to_heal() -> Vec<Endpoint> {
     let mut disks_to_heal = Vec::new();
     for (_, disk) in GLOBAL_LOCAL_DISK_MAP.read().await.iter() {
         if let Some(disk) = disk {

ecstore/src/heal/heal_ops.rs

@@ -190,19 +190,19 @@ impl HealSequence {
 }
 
 impl HealSequence {
-    fn _get_scanned_items_count(&self) -> usize {
+    pub fn get_scanned_items_count(&self) -> usize {
         self.scanned_items_map.values().sum()
     }
 
-    fn _get_scanned_items_map(&self) -> ItemsMap {
+    pub fn _get_scanned_items_map(&self) -> ItemsMap {
         self.scanned_items_map.clone()
     }
 
-    fn _get_healed_items_map(&self) -> ItemsMap {
+    pub fn _get_healed_items_map(&self) -> ItemsMap {
         self.healed_items_map.clone()
     }
 
-    fn _get_heal_failed_items_map(&self) -> ItemsMap {
+    pub fn _get_heal_failed_items_map(&self) -> ItemsMap {
         self.heal_failed_items_map.clone()
     }
 

ecstore/src/lib.rs

@@ -9,7 +9,7 @@ pub mod endpoints;
 pub mod erasure;
 pub mod error;
 mod file_meta;
-mod global;
+pub mod global;
 pub mod heal;
 pub mod peer;
 mod quorum;

ecstore/src/pools.rs

@@ -1,12 +1,22 @@
+use crate::config::common::{read_config, save_config};
+use crate::config::error::ConfigError;
 use crate::error::{Error, Result};
+use crate::new_object_layer_fn;
 use crate::store_api::{StorageAPI, StorageDisk, StorageInfo};
 use crate::store_err::StorageError;
 use crate::{sets::Sets, store::ECStore};
-use serde::Serialize;
+use byteorder::{ByteOrder, LittleEndian, WriteBytesExt};
+use rmp_serde::{Deserializer, Serializer};
+use serde::{Deserialize, Serialize};
+use std::io::{Cursor, Write};
 use std::sync::Arc;
 use time::OffsetDateTime;
 
-#[derive(Debug, Clone, Serialize)]
+pub const POOL_META_NAME: &str = "pool.bin";
+pub const POOL_META_FORMAT: u16 = 1;
+pub const POOL_META_VERSION: u16 = 1;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct PoolStatus {
     pub id: usize,
     pub cmd_line: String,
@@ -14,8 +24,9 @@ pub struct PoolStatus {
     pub decommission: Option<PoolDecommissionInfo>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct PoolMeta {
+    pub version: u16,
     pub pools: Vec<PoolStatus>,
     pub dont_save: bool,
 }
@@ -33,6 +44,7 @@ impl PoolMeta {
         }
 
         Self {
+            version: POOL_META_VERSION,
             pools: status,
             dont_save: false,
         }
@@ -46,6 +58,62 @@ impl PoolMeta {
         self.pools[idx].decommission.is_some()
     }
 
+    pub async fn load(&mut self, store: &ECStore) -> Result<()> {
+        let data = match read_config(store, POOL_META_NAME).await {
+            Ok(data) => {
+                if data.is_empty() {
+                    return Ok(());
+                } else if data.len() <= 4 {
+                    return Err(Error::from_string("poolMeta: no data"));
+                }
+                data
+            }
+            Err(err) => {
+                if let Some(ConfigError::NotFound) = err.downcast_ref::<ConfigError>() {
+                    return Ok(());
+                }
+                return Err(err);
+            }
+        };
+        let format = LittleEndian::read_u16(&data[0..2]);
+        if format != POOL_META_FORMAT {
+            return Err(Error::msg(format!("PoolMeta: unknown format: {}", format)));
+        }
+        let version = LittleEndian::read_u16(&data[2..4]);
+        if version != POOL_META_VERSION {
+            return Err(Error::msg(format!("PoolMeta: unknown version: {}", version)));
+        }
+
+        let mut buf = Deserializer::new(Cursor::new(&data[4..]));
+        let meta: PoolMeta = Deserialize::deserialize(&mut buf).unwrap();
+        *self = meta;
+
+        if self.version != POOL_META_VERSION {
+            return Err(Error::msg(format!("unexpected PoolMeta version: {}", self.version)));
+        }
+        Ok(())
+    }
+
+    pub async fn save(&self) -> Result<()> {
+        if self.dont_save {
+            return Ok(());
+        }
+        let mut data = Vec::new();
+        data.write_u16::<LittleEndian>(POOL_META_FORMAT).unwrap();
+        data.write_u16::<LittleEndian>(POOL_META_VERSION).unwrap();
+        let mut buf = Vec::new();
+        self.serialize(&mut Serializer::new(&mut buf))?;
+        data.write_all(&buf)?;
+
+        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_string("errServerNotInitialized".to_string())),
+        };
+        save_config(store, &POOL_META_NAME, &data).await
+    }
+
     pub fn decommission_cancel(&mut self, idx: usize) -> bool {
         if let Some(stats) = self.pools.get_mut(idx) {
             if let Some(d) = &stats.decommission {
@@ -72,7 +140,7 @@ impl PoolMeta {
     }
 }
 
-#[derive(Debug, Clone, Serialize, Default)]
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct PoolDecommissionInfo {
     pub start_time: Option<OffsetDateTime>,
     pub start_size: usize,
@@ -157,15 +225,11 @@ impl ECStore {
     }
 }
 
-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;
-        }
+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;
+        // }
     }
     capacity
 }
@@ -182,3 +246,33 @@ fn get_total_usable_capacity_free(disks: &Vec<StorageDisk>, info: &StorageInfo)
     }
     capacity
 }
+
+#[test]
+fn test_pool_meta() -> Result<()> {
+    let meta = PoolMeta::new(vec![]);
+    let mut data = Vec::new();
+    data.write_u16::<LittleEndian>(POOL_META_FORMAT).unwrap();
+    data.write_u16::<LittleEndian>(POOL_META_VERSION).unwrap();
+    let mut buf = Vec::new();
+    meta.serialize(&mut Serializer::new(&mut buf))?;
+    data.write_all(&buf)?;
+
+    let format = LittleEndian::read_u16(&data[0..2]);
+    if format != POOL_META_FORMAT {
+        return Err(Error::msg(format!("PoolMeta: unknown format: {}", format)));
+    }
+    let version = LittleEndian::read_u16(&data[2..4]);
+    if version != POOL_META_VERSION {
+        return Err(Error::msg(format!("PoolMeta: unknown version: {}", version)));
+    }
+
+    let mut buf = Deserializer::new(Cursor::new(&data[4..]));
+    let de_meta: PoolMeta = Deserialize::deserialize(&mut buf).unwrap();
+
+    if de_meta.version != POOL_META_VERSION {
+        return Err(Error::msg(format!("unexpected PoolMeta version: {}", de_meta.version)));
+    }
+
+    println!("meta: {:?}", de_meta);
+    Ok(())
+}

ecstore/src/store.rs

@@ -55,7 +55,7 @@ use std::slice::Iter;
 use std::time::SystemTime;
 use std::{
     collections::{HashMap, HashSet},
-    sync::Arc,
+    sync::{Arc, RwLock as std_RwLock},
     time::Duration,
 };
 use time::OffsetDateTime;
@@ -68,7 +68,7 @@ use uuid::Uuid;
 
 const MAX_UPLOADS_LIST: usize = 10000;
 
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub struct ECStore {
     pub id: uuid::Uuid,
     // pub disks: Vec<DiskStore>,
@@ -76,10 +76,27 @@ pub struct ECStore {
     pub pools: Vec<Arc<Sets>>,
     pub peer_sys: S3PeerSys,
     // pub local_disks: Vec<DiskStore>,
-    pub pool_meta: PoolMeta,
+    pub pool_meta: std_RwLock<PoolMeta>,
     pub decommission_cancelers: Vec<Option<usize>>,
 }
 
+impl Clone for ECStore {
+    fn clone(&self) -> Self {
+        let pool_meta = match self.pool_meta.read() {
+            Ok(pool_meta) => pool_meta.clone(),
+            Err(_) => PoolMeta::default(),
+        };
+        Self {
+            id: self.id.clone(),
+            disk_map: self.disk_map.clone(),
+            pools: self.pools.clone(),
+            peer_sys: self.peer_sys.clone(),
+            pool_meta: std_RwLock::new(pool_meta),
+            decommission_cancelers: self.decommission_cancelers.clone(),
+        }
+    }
+}
+
 impl ECStore {
     #[allow(clippy::new_ret_no_self)]
     pub async fn new(_address: String, endpoint_pools: EndpointServerPools) -> Result<Self> {
@@ -204,7 +221,7 @@ impl ECStore {
             disk_map,
             pools,
             peer_sys,
-            pool_meta,
+            pool_meta: pool_meta.into(),
             decommission_cancelers,
         };
 
@@ -480,7 +497,10 @@ impl ECStore {
 
     fn is_suspended(&self, idx: usize) -> bool {
         // TODO: LOCK
-        self.pool_meta.is_suspended(idx)
+        match self.pool_meta.read() {
+            Ok(pool_meta) => pool_meta.is_suspended(idx),
+            Err(_) => false,
+        }
     }
 
     async fn get_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Result<usize> {
@@ -575,7 +595,7 @@ impl ECStore {
         let mut has_def_pool = false;
 
         for pinfo in ress.iter() {
-            if opts.skip_decommissioned && self.pool_meta.is_suspended(pinfo.index) {
+            if opts.skip_decommissioned && self.pool_meta.read().unwrap().is_suspended(pinfo.index) {
                 continue;
             }
 
@@ -616,7 +636,7 @@ impl ECStore {
     fn pools_with_object(&self, pools: &Vec<PoolObjInfo>, opts: &ObjectOptions) -> Vec<PoolErr> {
         let mut errs = Vec::new();
         for pool in pools.iter() {
-            if opts.skip_decommissioned && self.pool_meta.is_suspended(pool.index) {
+            if opts.skip_decommissioned && self.pool_meta.read().unwrap().is_suspended(pool.index) {
                 continue;
             }
             // TODO:SkipRebalancing
@@ -869,6 +889,13 @@ impl ECStore {
 
         Ok(objs[0].as_ref().unwrap().clone())
     }
+
+    pub async fn reload_pool_meta(&self) -> Result<()> {
+        let mut meta = PoolMeta::default();
+        meta.load(self).await?;
+        *self.pool_meta.write().unwrap() = meta;
+        Ok(())
+    }
 }
 
 async fn update_scan(