diff --git a/ecstore/src/lib.rs b/ecstore/src/lib.rs index 1371d1f14..82741b44c 100644 --- a/ecstore/src/lib.rs +++ b/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; diff --git a/ecstore/src/peer.rs b/ecstore/src/peer.rs index 0c9dd58e0..4274c2977 100644 --- a/ecstore/src/peer.rs +++ b/ecstore/src/peer.rs @@ -26,7 +26,7 @@ pub trait PeerS3Client: Debug + Sync + Send + 'static { async fn list_bucket(&self, opts: &BucketOptions) -> Result>; async fn delete_bucket(&self, bucket: &str) -> Result<()>; async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result; - fn get_pools(&self) -> Vec; + fn get_pools(&self) -> Option>; } #[derive(Debug)] @@ -50,10 +50,10 @@ impl S3PeerSys { .map(|e| { if e.is_local { let cli: Box = - 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 = Box::new(RemotePeerS3Client::new(e.clone(), e.pools.clone())); + let cli: Box = 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 { + pub fn get_pools(&self) -> Option> { unimplemented!() } } @@ -221,11 +221,11 @@ impl S3PeerSys { pub struct LocalPeerS3Client { pub local_disks: Vec, // pub node: Node, - pub pools: Vec, + pub pools: Option>, } impl LocalPeerS3Client { - fn new(local_disks: Vec, _node: Node, pools: Vec) -> Self { + pub fn new(local_disks: Vec, _node: Option, pools: Option>) -> Self { Self { local_disks, // node, @@ -236,7 +236,7 @@ impl LocalPeerS3Client { #[async_trait] impl PeerS3Client for LocalPeerS3Client { - fn get_pools(&self) -> Vec { + fn get_pools(&self) -> Option> { self.pools.clone() } async fn list_bucket(&self, _opts: &BucketOptions) -> Result> { @@ -384,14 +384,15 @@ impl PeerS3Client for LocalPeerS3Client { #[derive(Debug)] pub struct RemotePeerS3Client { - pub _node: Node, - pub pools: Vec, + pub _node: Option, + pub pools: Option>, pub channel: Channel, } impl RemotePeerS3Client { - fn new(node: Node, pools: Vec) -> Self { - let connector = Endpoint::from_shared(format!("{}", node.url.clone().as_str())).unwrap(); + fn new(node: Option, pools: Option>) -> 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 { + fn get_pools(&self) -> Option> { self.pools.clone() } async fn list_bucket(&self, _opts: &BucketOptions) -> Result> { diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index bbcc49fa2..b31a5fc4d 100644 --- a/ecstore/src/store.rs +++ b/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 { + self.local_disks.clone() + } + fn single_pool(&self) -> bool { self.pools.len() == 1 } diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 3aa50f569..a1556b103 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -242,6 +242,7 @@ pub enum BitrotAlgorithm { BLAKE2b512, } +#[derive(Debug, Default)] pub struct MakeBucketOptions { pub force_create: bool, } diff --git a/rustfs/src/grpc.rs b/rustfs/src/grpc.rs index 63403b7e9..890e6cba6 100644 --- a/rustfs/src/grpc.rs +++ b/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 { - NodeServer::new(NodeService {}) +pub fn make_server(local_disks: Vec) -> NodeServer { + 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) -> Result, 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!() } } diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index 113f81a7d..c0567d5e1 100644 --- a/rustfs/src/main.rs +++ b/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(); diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 8e8f53db5..76b091aca 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -43,8 +43,7 @@ pub struct FS { } impl FS { - pub async fn new(address: String, endpoints: Vec) -> Result { - let store: ECStore = ECStore::new(address, endpoints).await?; + pub async fn new(store: ECStore) -> Result { Ok(Self { store }) } }