rpc add local_peer

ecstore/src/lib.rs

@@ -6,7 +6,7 @@ mod endpoints;
 mod erasure;
 pub mod error;
 mod file_meta;
-mod peer;
+pub mod peer;
 pub mod set_disk;
 mod sets;
 pub mod store;

ecstore/src/peer.rs

@@ -26,7 +26,7 @@ pub trait PeerS3Client: Debug + Sync + Send + 'static {
     async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>>;
     async fn delete_bucket(&self, bucket: &str) -> Result<()>;
     async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo>;
-    fn get_pools(&self) -> Vec<usize>;
+    fn get_pools(&self) -> Option<Vec<usize>>;
 }
 
 #[derive(Debug)]
@@ -50,10 +50,10 @@ impl S3PeerSys {
             .map(|e| {
                 if e.is_local {
                     let cli: Box<dyn PeerS3Client> =
-                        Box::new(LocalPeerS3Client::new(local_disks.clone(), e.clone(), e.pools.clone()));
+                        Box::new(LocalPeerS3Client::new(local_disks.clone(), Some(e.clone()), Some(e.pools.clone())));
                     Arc::new(cli)
                 } else {
-                    let cli: Box<dyn PeerS3Client> = Box::new(RemotePeerS3Client::new(e.clone(), e.pools.clone()));
+                    let cli: Box<dyn PeerS3Client> = Box::new(RemotePeerS3Client::new(Some(e.clone()), Some(e.pools.clone())));
                     Arc::new(cli)
                 }
             })
@@ -89,7 +89,7 @@ impl S3PeerSys {
             for (j, cli) in self.clients.iter().enumerate() {
                 let pools = cli.get_pools();
                 let idx = i;
-                if pools.contains(&idx) {
+                if pools.unwrap_or_default().contains(&idx) {
                     per_pool_errs.push(errors[j].as_ref());
                 }
 
@@ -199,7 +199,7 @@ impl S3PeerSys {
             for (j, cli) in self.clients.iter().enumerate() {
                 let pools = cli.get_pools();
                 let idx = i;
-                if pools.contains(&idx) {
+                if pools.unwrap_or_default().contains(&idx) {
                     per_pool_errs.push(errors[j].as_ref());
                 }
 
@@ -212,7 +212,7 @@ impl S3PeerSys {
             .ok_or(Error::new(DiskError::VolumeNotFound))
     }
 
-    pub fn get_pools(&self) -> Vec<usize> {
+    pub fn get_pools(&self) -> Option<Vec<usize>> {
         unimplemented!()
     }
 }
@@ -221,11 +221,11 @@ impl S3PeerSys {
 pub struct LocalPeerS3Client {
     pub local_disks: Vec<DiskStore>,
     // pub node: Node,
-    pub pools: Vec<usize>,
+    pub pools: Option<Vec<usize>>,
 }
 
 impl LocalPeerS3Client {
-    fn new(local_disks: Vec<DiskStore>, _node: Node, pools: Vec<usize>) -> Self {
+    pub fn new(local_disks: Vec<DiskStore>, _node: Option<Node>, pools: Option<Vec<usize>>) -> Self {
         Self {
             local_disks,
             // node,
@@ -236,7 +236,7 @@ impl LocalPeerS3Client {
 
 #[async_trait]
 impl PeerS3Client for LocalPeerS3Client {
-    fn get_pools(&self) -> Vec<usize> {
+    fn get_pools(&self) -> Option<Vec<usize>> {
         self.pools.clone()
     }
     async fn list_bucket(&self, _opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
@@ -384,14 +384,15 @@ impl PeerS3Client for LocalPeerS3Client {
 
 #[derive(Debug)]
 pub struct RemotePeerS3Client {
-    pub _node: Node,
-    pub pools: Vec<usize>,
+    pub _node: Option<Node>,
+    pub pools: Option<Vec<usize>>,
     pub channel: Channel,
 }
 
 impl RemotePeerS3Client {
-    fn new(node: Node, pools: Vec<usize>) -> Self {
-        let connector = Endpoint::from_shared(format!("{}", node.url.clone().as_str())).unwrap();
+    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,
@@ -403,7 +404,7 @@ impl RemotePeerS3Client {
 
 #[async_trait]
 impl PeerS3Client for RemotePeerS3Client {
-    fn get_pools(&self) -> Vec<usize> {
+    fn get_pools(&self) -> Option<Vec<usize>> {
         self.pools.clone()
     }
     async fn list_bucket(&self, _opts: &BucketOptions) -> Result<Vec<BucketInfo>> {

ecstore/src/store.rs

@@ -4,7 +4,7 @@ use crate::{
     disks_layout::DisksLayout,
     endpoints::EndpointServerPools,
     error::{Error, Result},
-    peer::{PeerS3Client, S3PeerSys},
+    peer::S3PeerSys,
     sets::Sets,
     store_api::{
         BucketInfo, BucketOptions, CompletePart, GetObjectReader, HTTPRangeSpec, ListObjectsInfo, ListObjectsV2Info,
@@ -104,6 +104,10 @@ impl ECStore {
         })
     }
 
+    pub fn local_disks(&self) -> Vec<DiskStore> {
+        self.local_disks.clone()
+    }
+
     fn single_pool(&self) -> bool {
         self.pools.len() == 1
     }

ecstore/src/store_api.rs

@@ -242,6 +242,7 @@ pub enum BitrotAlgorithm {
     BLAKE2b512,
 }
 
+#[derive(Debug, Default)]
 pub struct MakeBucketOptions {
     pub force_create: bool,
 }

rustfs/src/grpc.rs

@@ -1,3 +1,4 @@
+use ecstore::{disk::DiskStore, peer::LocalPeerS3Client};
 use tonic::{Request, Response, Status};
 use tracing::{debug, error, info};
 
@@ -10,10 +11,13 @@ use protos::{
 };
 
 #[derive(Debug)]
-struct NodeService {}
+struct NodeService {
+    pub local_peer: LocalPeerS3Client,
+}
 
-pub fn make_server() -> NodeServer<impl Node> {
-    NodeServer::new(NodeService {})
+pub fn make_server(local_disks: Vec<DiskStore>) -> NodeServer<impl Node> {
+    let local_peer = LocalPeerS3Client::new(local_disks, None, None);
+    NodeServer::new(NodeService { local_peer })
 }
 
 #[tonic::async_trait]
@@ -48,8 +52,9 @@ impl Node for NodeService {
     async fn make_bucket(&self, request: Request<MakeBucketRequest>) -> Result<Response<MakeBucketResponse>, Status> {
         debug!("make bucket");
 
-        let req = request.into_inner();
+        let _req = request.into_inner();
 
+        // match self.local_peer.make_bucket(&name, &MakeBucketOptions::default()).await {}
         unimplemented!()
     }
 }

rustfs/src/main.rs

@@ -4,7 +4,7 @@ mod service;
 mod storage;
 
 use clap::Parser;
-use ecstore::error::Result;
+use ecstore::{error::Result, store::ECStore};
 use grpc::make_server;
 use hyper_util::{
     rt::{TokioExecutor, TokioIo},
@@ -66,9 +66,12 @@ async fn run(opt: config::Opt) -> Result<()> {
     //         })
     // };
 
+    let store: ECStore = ECStore::new(opt.address.clone(), opt.volumes.clone()).await?;
+    let local_disks = store.local_disks();
+
     // Setup S3 service
     let service = {
-        let mut b = S3ServiceBuilder::new(storage::ecfs::FS::new(opt.address.clone(), opt.volumes.clone()).await?);
+        let mut b = S3ServiceBuilder::new(storage::ecfs::FS::new(store).await?);
 
         let mut access_key = String::from_str(config::DEFAULT_ACCESS_KEY).unwrap();
         let mut secret_key = String::from_str(config::DEFAULT_SECRET_KEY).unwrap();
@@ -101,7 +104,7 @@ async fn run(opt: config::Opt) -> Result<()> {
 
     let hyper_service = service.into_shared();
 
-    let hybrid_service = TowerToHyperService::new(hybrid(hyper_service, make_server()));
+    let hybrid_service = TowerToHyperService::new(hybrid(hyper_service, make_server(local_disks)));
 
     let http_server = ConnBuilder::new(TokioExecutor::new());
     let graceful = hyper_util::server::graceful::GracefulShutdown::new();

rustfs/src/storage/ecfs.rs

@@ -43,8 +43,7 @@ pub struct FS {
 }
 
 impl FS {
-    pub async fn new(address: String, endpoints: Vec<String>) -> Result<Self> {
-        let store: ECStore = ECStore::new(address, endpoints).await?;
+    pub async fn new(store: ECStore) -> Result<Self> {
         Ok(Self { store })
     }
 }