support multi node

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

ecstore/src/disk/mod.rs

@@ -28,7 +28,6 @@ use tokio::{
     io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt},
 };
 use tonic::{transport::Channel, Request};
-use tower::timeout::Timeout;
 use uuid::Uuid;
 
 pub type DiskStore = Arc<Box<dyn DiskAPI>>;
@@ -96,7 +95,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>>;
 }
 
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
 pub struct FileInfoVersions {
     // Name of the volume.
     pub volume: String,
@@ -350,17 +349,11 @@ pub struct RemoteFileWriter {
     pub volume: String,
     pub path: String,
     pub is_append: bool,
-    client: NodeServiceClient<Timeout<Channel>>,
+    client: NodeServiceClient<Channel>,
 }
 
 impl RemoteFileWriter {
-    pub fn new(
-        root: PathBuf,
-        volume: String,
-        path: String,
-        is_append: bool,
-        client: NodeServiceClient<Timeout<Channel>>,
-    ) -> Self {
+    pub fn new(root: PathBuf, volume: String, path: String, is_append: bool, client: NodeServiceClient<Channel>) -> Self {
         Self {
             root,
             volume,
@@ -433,11 +426,11 @@ pub struct RemoteFileReader {
     pub root: PathBuf,
     pub volume: String,
     pub path: String,
-    client: NodeServiceClient<Timeout<Channel>>,
+    client: NodeServiceClient<Channel>,
 }
 
 impl RemoteFileReader {
-    pub fn new(root: PathBuf, volume: String, path: String, client: NodeServiceClient<Timeout<Channel>>) -> Self {
+    pub fn new(root: PathBuf, volume: String, path: String, client: NodeServiceClient<Channel>) -> Self {
         Self {
             root,
             volume,

ecstore/src/disk/remote.rs

@@ -1,17 +1,18 @@
-use std::{path::PathBuf, sync::RwLock, time::Duration};
+use std::{path::PathBuf, sync::Arc, time::Duration};
 
 use bytes::Bytes;
 use futures::lock::Mutex;
 use protos::{
     node_service_time_out_client,
     proto_gen::node_service::{
-        node_service_client::NodeServiceClient, DeleteRequest, DeleteVolumeRequest, ListDirRequest, ListVolumesRequest,
-        MakeVolumeRequest, MakeVolumesRequest, ReadAllRequest, ReadMultipleRequest, ReadVersionRequest, ReadXlRequest,
-        RenameDataRequest, RenameFileRequst, StatVolumeRequest, WalkDirRequest, WriteAllRequest, WriteMetadataRequest,
+        node_service_client::NodeServiceClient, DeleteRequest, DeleteVersionsRequest, DeleteVolumeRequest, ListDirRequest,
+        ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, ReadAllRequest, ReadMultipleRequest, ReadVersionRequest,
+        ReadXlRequest, RenameDataRequest, RenameFileRequst, StatVolumeRequest, WalkDirRequest, WriteAllRequest,
+        WriteMetadataRequest,
     },
     DEFAULT_GRPC_SERVER_MESSAGE_LEN,
 };
-use tokio::fs;
+use tokio::{fs, sync::RwLock};
 use tonic::{
     transport::{Channel, Endpoint as tonic_Endpoint},
     Request,
@@ -21,6 +22,7 @@ use tracing::info;
 use uuid::Uuid;
 
 use crate::{
+    disk::error::DiskError,
     error::{Error, Result},
     store_api::{FileInfo, RawFileInfo},
 };
@@ -34,7 +36,7 @@ use super::{
 #[derive(Debug)]
 pub struct RemoteDisk {
     id: Mutex<Option<Uuid>>,
-    channel: RwLock<Option<Channel>>,
+    channel: Arc<RwLock<Option<Channel>>>,
     url: url::Url,
     pub root: PathBuf,
 }
@@ -44,39 +46,49 @@ impl RemoteDisk {
         let root = fs::canonicalize(ep.url.path()).await?;
 
         Ok(Self {
-            channel: RwLock::new(None),
+            channel: Arc::new(RwLock::new(None)),
             url: ep.url.clone(),
             root,
             id: Mutex::new(None),
         })
     }
 
-    fn get_client(&self) -> NodeServiceClient<Timeout<Channel>> {
+    #[allow(dead_code)]
+    async fn get_client(&self) -> Result<NodeServiceClient<Timeout<Channel>>> {
+        let channel_clone = self.channel.clone();
         let channel = {
-            let read_lock = self.channel.read().unwrap();
+            let read_lock = channel_clone.read().await;
 
             if let Some(ref channel) = *read_lock {
                 channel.clone()
             } else {
                 let addr = format!("{}://{}:{}", self.url.scheme(), self.url.host_str().unwrap(), self.url.port().unwrap());
-                info!("disk url: {:?}", addr);
-                let connector = tonic_Endpoint::from_shared(addr).unwrap();
+                info!("disk url: {}", addr);
+                let connector = tonic_Endpoint::from_shared(addr.clone())?;
 
-                let new_channel = tokio::runtime::Runtime::new().unwrap().block_on(connector.connect()).unwrap();
+                let new_channel = connector.connect().await.map_err(|_err| DiskError::DiskNotFound)?;
 
-                *self.channel.write().unwrap() = Some(new_channel.clone());
+                info!("get channel success");
+
+                *self.channel.write().await = Some(new_channel.clone());
 
                 new_channel
             }
         };
 
-        node_service_time_out_client(
+        Ok(node_service_time_out_client(
             channel,
             Duration::new(30, 0), // TODO: use config setting
             DEFAULT_GRPC_SERVER_MESSAGE_LEN,
             // grpc_enable_gzip,
             false, // TODO: use config setting
-        )
+        ))
+    }
+
+    async fn get_client_v2(&self) -> Result<NodeServiceClient<tonic::transport::Channel>> {
+        // Ok(NodeServiceClient::connect("http://220.181.1.138:9000").await?)
+        let addr = format!("{}://{}:{}", self.url.scheme(), self.url.host_str().unwrap(), self.url.port().unwrap());
+        Ok(NodeServiceClient::connect(addr).await?)
     }
 }
 
@@ -104,7 +116,8 @@ impl DiskAPI for RemoteDisk {
     }
 
     async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
-        let mut client = self.get_client();
+        info!("read_all");
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(ReadAllRequest {
             disk: self.root.to_string_lossy().to_string(),
             volume: volume.to_string(),
@@ -113,11 +126,18 @@ impl DiskAPI for RemoteDisk {
 
         let response = client.read_all(request).await?.into_inner();
 
+        info!("read_all success");
+
+        if !response.success {
+            return Err(DiskError::FileNotFound.into());
+        }
+
         Ok(Bytes::from(response.data))
     }
 
     async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()> {
-        let mut client = self.get_client();
+        info!("write_all");
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(WriteAllRequest {
             disk: self.root.to_string_lossy().to_string(),
             volume: volume.to_string(),
@@ -125,14 +145,19 @@ impl DiskAPI for RemoteDisk {
             data,
         });
 
-        let _response = client.write_all(request).await?.into_inner();
+        let response = client.write_all(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
 
         Ok(())
     }
 
     async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> {
+        info!("delete");
         let options = serde_json::to_string(&opt)?;
-        let mut client = self.get_client();
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(DeleteRequest {
             disk: self.root.to_string_lossy().to_string(),
             volume: volume.to_string(),
@@ -140,13 +165,18 @@ impl DiskAPI for RemoteDisk {
             options,
         });
 
-        let _response = client.delete(request).await?.into_inner();
+        let response = client.delete(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
 
         Ok(())
     }
 
     async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
-        let mut client = self.get_client();
+        info!("rename_file");
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(RenameFileRequst {
             disk: self.root.to_string_lossy().to_string(),
             src_volume: src_volume.to_string(),
@@ -155,42 +185,50 @@ impl DiskAPI for RemoteDisk {
             dst_path: dst_path.to_string(),
         });
 
-        let _response = client.rename_file(request).await?.into_inner();
+        let response = client.rename_file(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
 
         Ok(())
     }
 
     async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result<FileWriter> {
+        info!("create_file");
         Ok(FileWriter::Remote(RemoteFileWriter::new(
             self.root.clone(),
             volume.to_string(),
             path.to_string(),
             false,
-            self.get_client(),
+            self.get_client_v2().await?,
         )))
     }
 
     async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter> {
+        info!("append_file");
         Ok(FileWriter::Remote(RemoteFileWriter::new(
             self.root.clone(),
             volume.to_string(),
             path.to_string(),
             true,
-            self.get_client(),
+            self.get_client_v2().await?,
         )))
     }
 
     async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader> {
+        info!("read_file");
         Ok(FileReader::Remote(RemoteFileReader::new(
             self.root.clone(),
             volume.to_string(),
             path.to_string(),
-            self.get_client(),
+            self.get_client_v2().await?,
         )))
     }
 
     async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result<Vec<String>> {
-        let mut client = self.get_client();
+        info!("list_dir");
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(ListDirRequest {
             disk: self.root.to_string_lossy().to_string(),
             volume: volume.to_string(),
@@ -198,18 +236,28 @@ impl DiskAPI for RemoteDisk {
 
         let response = client.list_dir(request).await?.into_inner();
 
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
+
         Ok(response.volumes)
     }
 
     async fn walk_dir(&self, opts: WalkDirOptions) -> Result<Vec<MetaCacheEntry>> {
+        info!("walk_dir");
         let walk_dir_options = serde_json::to_string(&opts)?;
-        let mut client = self.get_client();
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(WalkDirRequest {
             disk: self.root.to_string_lossy().to_string(),
             walk_dir_options,
         });
 
         let response = client.walk_dir(request).await?.into_inner();
+        
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
+        
         let entries = response
             .meta_cache_entry
             .into_iter()
@@ -227,8 +275,9 @@ impl DiskAPI for RemoteDisk {
         dst_volume: &str,
         dst_path: &str,
     ) -> Result<RenameDataResp> {
+        info!("rename_data");
         let file_info = serde_json::to_string(&fi)?;
-        let mut client = self.get_client();
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(RenameDataRequest {
             disk: self.root.to_string_lossy().to_string(),
             src_volume: src_volume.to_string(),
@@ -239,42 +288,63 @@ impl DiskAPI for RemoteDisk {
         });
 
         let response = client.rename_data(request).await?.into_inner();
+        
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
+
         let rename_data_resp = serde_json::from_str::<RenameDataResp>(&response.rename_data_resp)?;
 
         Ok(rename_data_resp)
     }
 
     async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> {
-        let mut client = self.get_client();
+        info!("make_volumes");
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(MakeVolumesRequest {
             disk: self.root.to_string_lossy().to_string(),
             volumes: volumes.iter().map(|s| (*s).to_string()).collect(),
         });
 
-        let _response = client.make_volumes(request).await?.into_inner();
+        let response = client.make_volumes(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
 
         Ok(())
     }
 
     async fn make_volume(&self, volume: &str) -> Result<()> {
-        let mut client = self.get_client();
+        info!("make_volume");
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(MakeVolumeRequest {
             disk: self.root.to_string_lossy().to_string(),
             volume: volume.to_string(),
         });
 
-        let _response = client.make_volume(request).await?.into_inner();
+        let response = client.make_volume(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
 
         Ok(())
     }
 
     async fn list_volumes(&self) -> Result<Vec<VolumeInfo>> {
-        let mut client = self.get_client();
+        info!("list_volumes");
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(ListVolumesRequest {
             disk: self.root.to_string_lossy().to_string(),
         });
 
         let response = client.list_volumes(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
+
         let infos = response
             .volume_infos
             .into_iter()
@@ -285,21 +355,28 @@ impl DiskAPI for RemoteDisk {
     }
 
     async fn stat_volume(&self, volume: &str) -> Result<VolumeInfo> {
-        let mut client = self.get_client();
+        info!("stat_volume");
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(StatVolumeRequest {
             disk: self.root.to_string_lossy().to_string(),
             volume: volume.to_string(),
         });
 
         let response = client.stat_volume(request).await?.into_inner();
+        
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
+
         let volume_info = serde_json::from_str::<VolumeInfo>(&response.volume_info)?;
 
         Ok(volume_info)
     }
 
     async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
+        info!("write_metadata");
         let file_info = serde_json::to_string(&fi)?;
-        let mut client = self.get_client();
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(WriteMetadataRequest {
             disk: self.root.to_string_lossy().to_string(),
             volume: volume.to_string(),
@@ -307,7 +384,11 @@ impl DiskAPI for RemoteDisk {
             file_info,
         });
 
-        let _response = client.write_metadata(request).await?.into_inner();
+        let response = client.write_metadata(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
 
         Ok(())
     }
@@ -320,8 +401,9 @@ impl DiskAPI for RemoteDisk {
         version_id: &str,
         opts: &ReadOptions,
     ) -> Result<FileInfo> {
+        info!("read_version");
         let opts = serde_json::to_string(opts)?;
-        let mut client = self.get_client();
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(ReadVersionRequest {
             disk: self.root.to_string_lossy().to_string(),
             volume: volume.to_string(),
@@ -331,13 +413,19 @@ impl DiskAPI for RemoteDisk {
         });
 
         let response = client.read_version(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
+
         let file_info = serde_json::from_str::<FileInfo>(&response.file_info)?;
 
         Ok(file_info)
     }
 
     async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo> {
-        let mut client = self.get_client();
+        info!("read_xl");
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(ReadXlRequest {
             disk: self.root.to_string_lossy().to_string(),
             volume: volume.to_string(),
@@ -346,6 +434,11 @@ impl DiskAPI for RemoteDisk {
         });
 
         let response = client.read_xl(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
+
         let raw_file_info = serde_json::from_str::<RawFileInfo>(&response.raw_file_info)?;
 
         Ok(raw_file_info)
@@ -353,22 +446,61 @@ impl DiskAPI for RemoteDisk {
 
     async fn delete_versions(
         &self,
-        _volume: &str,
-        _versions: Vec<FileInfoVersions>,
-        _opts: DeleteOptions,
+        volume: &str,
+        versions: Vec<FileInfoVersions>,
+        opts: DeleteOptions,
     ) -> Result<Vec<Option<Error>>> {
-        unimplemented!()
+        info!("delete_versions");
+        let opts = serde_json::to_string(&opts)?;
+        let mut versions_str = Vec::with_capacity(versions.len());
+        for file_info_versions in versions.iter() {
+            versions_str.push(serde_json::to_string(file_info_versions)?);
+        }
+        let mut client = self.get_client_v2().await?;
+        let request = Request::new(DeleteVersionsRequest {
+            disk: self.root.to_string_lossy().to_string(),
+            volume: volume.to_string(),
+            versions: versions_str,
+            opts,
+        });
+
+        let response = client.delete_versions(request).await?.into_inner();
+        if !response.success {
+            return Err(Error::from_string(format!(
+                "delete versions remote err: {}",
+                response.error_info.unwrap_or("None".to_string())
+            )));
+        }
+        let errors = response
+            .errors
+            .iter()
+            .map(|error| {
+                if error.is_empty() {
+                    None
+                } else {
+                    Some(Error::from_string(error))
+                }
+            })
+            .collect();
+
+        Ok(errors)
     }
 
     async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>> {
+        info!("read_multiple");
         let read_multiple_req = serde_json::to_string(&req)?;
-        let mut client = self.get_client();
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(ReadMultipleRequest {
             disk: self.root.to_string_lossy().to_string(),
             read_multiple_req,
         });
 
         let response = client.read_multiple(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
+
         let read_multiple_resps = response
             .read_multiple_resps
             .into_iter()
@@ -379,13 +511,18 @@ impl DiskAPI for RemoteDisk {
     }
 
     async fn delete_volume(&self, volume: &str) -> Result<()> {
-        let mut client = self.get_client();
+        info!("delete_volume");
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(DeleteVolumeRequest {
             disk: self.root.to_string_lossy().to_string(),
             volume: volume.to_string(),
         });
 
-        let _response = client.delete_volume(request).await?.into_inner();
+        let response = client.delete_volume(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
 
         Ok(())
     }

ecstore/src/peer.rs

@@ -1,12 +1,15 @@
 use async_trait::async_trait;
 use futures::future::join_all;
+use protos::proto_gen::node_service::node_service_client::NodeServiceClient;
 use protos::proto_gen::node_service::{DeleteBucketRequest, GetBucketInfoRequest, ListBucketRequest, MakeBucketRequest};
 use protos::{node_service_time_out_client, DEFAULT_GRPC_SERVER_MESSAGE_LEN};
 use regex::Regex;
 use std::{collections::HashMap, fmt::Debug, sync::Arc, time::Duration};
+use tokio::sync::RwLock;
 use tonic::transport::{Channel, Endpoint};
 use tonic::Request;
-use tracing::warn;
+use tower::timeout::Timeout;
+use tracing::{info, warn};
 
 use crate::store::all_local_disk;
 use crate::{
@@ -386,22 +389,58 @@ impl PeerS3Client for LocalPeerS3Client {
 
 #[derive(Debug)]
 pub struct RemotePeerS3Client {
-    pub _node: Option<Node>,
+    pub node: Option<Node>,
     pub pools: Option<Vec<usize>>,
-    pub channel: Channel,
+    connector: Endpoint,
+    channel: Arc<RwLock<Option<Channel>>>,
 }
 
 impl RemotePeerS3Client {
     fn new(node: Option<Node>, pools: Option<Vec<usize>>) -> Self {
         let connector =
             Endpoint::from_shared(format!("{}", node.as_ref().map(|v| { v.url.to_string() }).unwrap_or_default())).unwrap();
-        let channel = tokio::runtime::Runtime::new().unwrap().block_on(connector.connect()).unwrap();
         Self {
-            _node: node,
+            node,
             pools,
-            channel,
+            connector,
+            channel: Arc::new(RwLock::new(None)),
         }
     }
+
+    #[allow(dead_code)]
+    async fn get_client(&self) -> Result<NodeServiceClient<Timeout<Channel>>> {
+        let channel_clone = self.channel.clone();
+        let channel = {
+            let read_lock = channel_clone.read().await;
+
+            if let Some(ref channel) = *read_lock {
+                channel.clone()
+            } else {
+                let new_channel = self.connector.connect().await?;
+
+                info!("get channel success");
+
+                *self.channel.write().await = Some(new_channel.clone());
+
+                new_channel
+            }
+        };
+
+        Ok(node_service_time_out_client(
+            channel,
+            Duration::new(30, 0), // TODO: use config setting
+            DEFAULT_GRPC_SERVER_MESSAGE_LEN,
+            // grpc_enable_gzip,
+            false, // TODO: use config setting
+        ))
+    }
+
+    async fn get_client_v2(&self) -> Result<NodeServiceClient<tonic::transport::Channel>> {
+        // Ok(NodeServiceClient::connect("http://220.181.1.138:9000").await?)
+        // let addr = format!("{}://{}:{}", self.url.scheme(), self.url.host_str().unwrap(), self.url.port().unwrap());
+        let addr = format!("{}", self.node.as_ref().map(|v| { v.url.to_string() }).unwrap_or_default());
+        Ok(NodeServiceClient::connect(addr).await?)
+    }
 }
 
 #[async_trait]
@@ -411,13 +450,7 @@ impl PeerS3Client for RemotePeerS3Client {
     }
     async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
         let options = serde_json::to_string(opts)?;
-        let mut client = node_service_time_out_client(
-            self.channel.clone(),
-            Duration::new(30, 0), // TODO: use config setting
-            DEFAULT_GRPC_SERVER_MESSAGE_LEN,
-            // grpc_enable_gzip,
-            false, // TODO: use config setting
-        );
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(ListBucketRequest { options });
         let response = client.list_bucket(request).await?.into_inner();
         let bucket_infos = response
@@ -430,32 +463,23 @@ impl PeerS3Client for RemotePeerS3Client {
     }
     async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
         let options = serde_json::to_string(opts)?;
-        let mut client = node_service_time_out_client(
-            self.channel.clone(),
-            Duration::new(30, 0), // TODO: use config setting
-            DEFAULT_GRPC_SERVER_MESSAGE_LEN,
-            // grpc_enable_gzip,
-            false, // TODO: use config setting
-        );
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(MakeBucketRequest {
             name: bucket.to_string(),
             options,
         });
-        let _response = client.make_bucket(request).await?.into_inner();
+        let response = client.make_bucket(request).await?.into_inner();
 
         // TODO: deal with error
+        if !response.success {
+            warn!("make bucket error: {:?}", response.error_info);
+        }
 
         Ok(())
     }
     async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
         let options = serde_json::to_string(opts)?;
-        let mut client = node_service_time_out_client(
-            self.channel.clone(),
-            Duration::new(30, 0), // TODO: use config setting
-            DEFAULT_GRPC_SERVER_MESSAGE_LEN,
-            // grpc_enable_gzip,
-            false, // TODO: use config setting
-        );
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(GetBucketInfoRequest {
             bucket: bucket.to_string(),
             options,
@@ -467,13 +491,7 @@ impl PeerS3Client for RemotePeerS3Client {
     }
 
     async fn delete_bucket(&self, bucket: &str) -> Result<()> {
-        let mut client = node_service_time_out_client(
-            self.channel.clone(),
-            Duration::new(30, 0), // TODO: use config setting
-            DEFAULT_GRPC_SERVER_MESSAGE_LEN,
-            // grpc_enable_gzip,
-            false, // TODO: use config setting
-        );
+        let mut client = self.get_client_v2().await?;
         let request = Request::new(DeleteBucketRequest {
             bucket: bucket.to_string(),
         });

ecstore/src/store.rs

@@ -14,16 +14,18 @@ use crate::{
     },
     store_init, utils,
 };
+use backon::{ExponentialBuilder, Retryable};
 use futures::future::join_all;
 use http::HeaderMap;
 use s3s::{dto::StreamingBlob, Body};
 use std::{
     collections::{HashMap, HashSet},
     sync::Arc,
+    time::Duration,
 };
 use time::OffsetDateTime;
 use tokio::{fs, sync::RwLock};
-use tracing::{debug, warn};
+use tracing::{debug, info, warn};
 use uuid::Uuid;
 
 use lazy_static::lazy_static;
@@ -168,6 +170,8 @@ impl ECStore {
 
         let mut local_disks = Vec::new();
 
+        info!("endpoint_pools: {:?}", endpoint_pools);
+
         for (i, pool_eps) in endpoint_pools.as_ref().iter().enumerate() {
             // TODO: read from config parseStorageClass
             let partiy_count = store_init::default_partiy_count(pool_eps.drives_per_set);
@@ -183,17 +187,23 @@ impl ECStore {
 
             DiskError::check_disk_fatal_errs(&errs)?;
 
-            let fm = store_init::connect_load_init_formats(
-                first_is_local,
-                &disks,
-                pool_eps.set_count,
-                pool_eps.drives_per_set,
-                deployment_id,
-            )
+            let fm = (|| async {
+                store_init::connect_load_init_formats(
+                    first_is_local,
+                    &disks,
+                    pool_eps.set_count,
+                    pool_eps.drives_per_set,
+                    deployment_id,
+                )
+                .await
+            })
+            .retry(ExponentialBuilder::default().with_max_times(usize::MAX))
+            .sleep(tokio::time::sleep)
+            .notify(|err, dur: Duration| {
+                info!("retrying get formats {:?} after {:?}", err, dur);
+            })
             .await?;
 
-            // TODO: 失败 重试 等试 3次
-
             if deployment_id.is_none() {
                 deployment_id = Some(fm.id);
             }

ecstore/src/store_init.rs

@@ -133,7 +133,7 @@ fn get_format_erasure_in_quorum(formats: &[Option<FormatV3>]) -> Result<FormatV3
 
     warn!("get_format_erasure_in_quorum fi: {:?}", &formats);
 
-    if *max_drives == 0 || *max_count < formats.len() / 2 {
+    if *max_drives == 0 || *max_count <= formats.len() / 2 {
         warn!(
             "*max_drives == 0 || *max_count < formats.len() / 2, {} || {}<{}",
             max_drives,