gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-20 19:42:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into bucketmeta

Browse Source
This commit is contained in:
weisd
2024-09-30 21:25:31 +08:00
parent cbaea166c4 9421da7b22
commit 80248981dd
34 changed files with 3149 additions and 295 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/disk/local.rs
+2 -1
View File
@@ -10,10 +10,10 @@ use crate::disk::error::{
};
use crate::disk::os::check_path_length;
use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE};
use crate::error::{Error, Result};
use crate::utils::fs::{lstat, O_APPEND, O_CREATE, O_RDONLY, O_WRONLY};
use crate::utils::path::{has_suffix, SLASH_SEPARATOR};
use crate::{
error::{Error, Result},
file_meta::FileMeta,
store_api::{FileInfo, RawFileInfo},
utils,
@@ -1322,6 +1322,7 @@ impl DiskAPI for LocalDisk {
if let Err(e) = utils::fs::access(&volume_dir).await {
if os_is_not_exist(&e) {
os::make_dir_all(&volume_dir, self.root.as_path()).await?;
return Ok(());
}
if os_is_permission(&e) {
return Err(Error::new(DiskError::DiskAccessDenied));
ecstore/src/disk/mod.rs
+101 -20
View File
@@ -3,7 +3,7 @@ pub mod error;
pub mod format;
mod local;
pub mod os;
mod remote;
pub mod remote;
pub const RUSTFS_META_BUCKET: &str = ".rustfs.sys";
pub const RUSTFS_META_MULTIPART_BUCKET: &str = ".rustfs.sys/multipart";
@@ -21,15 +21,21 @@ use crate::{
};
use endpoint::Endpoint;
use protos::proto_gen::node_service::{node_service_client::NodeServiceClient, ReadAtRequest, WriteRequest};
use futures::StreamExt;
use protos::proto_gen::node_service::{
node_service_client::NodeServiceClient, ReadAtRequest, ReadAtResponse, WriteRequest, WriteResponse,
};
use serde::{Deserialize, Serialize};
use std::{fmt::Debug, io::SeekFrom, path::PathBuf, sync::Arc};
use time::OffsetDateTime;
use tokio::{
fs::File,
io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt},
sync::mpsc::{self, Sender},
};
use tonic::{transport::Channel, Request};
use tokio_stream::wrappers::ReceiverStream;
use tonic::{service::interceptor::InterceptedService, transport::Channel, Request, Status, Streaming};
use tracing::info;
use uuid::Uuid;
pub type DiskStore = Arc<Box<dyn DiskAPI>>;
@@ -406,32 +412,77 @@ pub struct RemoteFileWriter {
pub volume: String,
pub path: String,
pub is_append: bool,
client: NodeServiceClient<Channel>,
tx: Sender<WriteRequest>,
resp_stream: Streaming<WriteResponse>,
}
impl RemoteFileWriter {
pub fn new(root: PathBuf, volume: String, path: String, is_append: bool, client: NodeServiceClient<Channel>) -> Self {
Self {
pub async fn new(
root: PathBuf,
volume: String,
path: String,
is_append: bool,
mut client: NodeServiceClient<
InterceptedService<Channel, Box<dyn Fn(Request<()>) -> Result<Request<()>, Status> + Send + Sync + 'static>>,
>,
) -> Result<Self> {
let (tx, rx) = mpsc::channel(128);
let in_stream = ReceiverStream::new(rx);
let response = client.write_stream(in_stream).await.unwrap();
let resp_stream = response.into_inner();
Ok(Self {
root,
volume,
path,
is_append,
client,
}
tx,
resp_stream,
})
}
}
#[async_trait::async_trait]
impl Write for RemoteFileWriter {
async fn write(&mut self, buf: &[u8]) -> Result<()> {
let request = Request::new(WriteRequest {
let request = WriteRequest {
disk: self.root.to_string_lossy().to_string(),
volume: self.volume.to_string(),
path: self.path.to_string(),
is_append: self.is_append,
data: buf.to_vec(),
});
let _response = self.client.write(request).await?.into_inner();
};
self.tx.send(request).await?;
if let Some(resp) = self.resp_stream.next().await {
// match resp {
// Ok(resp) => {
// if resp.success {
// info!("write stream success");
// } else {
// info!("write stream failed: {}", resp.error_info.unwrap_or("".to_string()));
// }
// }
// Err(_err) => {
// }
// }
let resp = resp?;
if resp.success {
info!("write stream success");
} else {
let error_info = resp.error_info.unwrap_or("".to_string());
info!("write stream failed: {}", error_info);
return Err(Error::from_string(error_info));
}
} else {
let error_info = "can not get response";
info!("write stream failed: {}", error_info);
return Err(Error::from_string(error_info));
}
Ok(())
}
}
@@ -483,32 +534,62 @@ pub struct RemoteFileReader {
pub root: PathBuf,
pub volume: String,
pub path: String,
client: NodeServiceClient<Channel>,
tx: Sender<ReadAtRequest>,
resp_stream: Streaming<ReadAtResponse>,
}
impl RemoteFileReader {
pub fn new(root: PathBuf, volume: String, path: String, client: NodeServiceClient<Channel>) -> Self {
Self {
pub async fn new(
root: PathBuf,
volume: String,
path: String,
mut client: NodeServiceClient<
InterceptedService<Channel, Box<dyn Fn(Request<()>) -> Result<Request<()>, Status> + Send + Sync + 'static>>,
>,
) -> Result<Self> {
let (tx, rx) = mpsc::channel(128);
let in_stream = ReceiverStream::new(rx);
let response = client.read_at(in_stream).await.unwrap();
let resp_stream = response.into_inner();
Ok(Self {
root,
volume,
path,
client,
}
tx,
resp_stream,
})
}
}
#[async_trait::async_trait]
impl ReadAt for RemoteFileReader {
async fn read_at(&mut self, offset: usize, length: usize) -> Result<(Vec<u8>, usize)> {
let request = Request::new(ReadAtRequest {
let request = ReadAtRequest {
disk: self.root.to_string_lossy().to_string(),
volume: self.volume.to_string(),
path: self.path.to_string(),
offset: offset.try_into().unwrap(),
length: length.try_into().unwrap(),
});
let response = self.client.read_at(request).await?.into_inner();
};
self.tx.send(request).await?;
Ok((response.data, response.read_size.try_into().unwrap()))
if let Some(resp) = self.resp_stream.next().await {
let resp = resp?;
if resp.success {
info!("read at stream success");
Ok((resp.data, resp.read_size.try_into().unwrap()))
} else {
let error_info = resp.error_info.unwrap_or("".to_string());
info!("read at stream failed: {}", error_info);
Err(Error::from_string(error_info))
}
} else {
let error_info = "can not get response";
info!("read at stream failed: {}", error_info);
Err(Error::from_string(error_info))
}
}
}
ecstore/src/disk/remote.rs
+119 -105
View File
@@ -1,96 +1,53 @@
use std::{path::PathBuf, sync::Arc, time::Duration};
use std::path::PathBuf;
use futures::lock::Mutex;
use protos::{
node_service_time_out_client,
proto_gen::node_service::{
node_service_client::NodeServiceClient, DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest,
DeleteVolumeRequest, ListDirRequest, ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, ReadAllRequest,
ReadMultipleRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest, RenameFileRequst, RenamePartRequst,
StatVolumeRequest, UpdateMetadataRequest, WalkDirRequest, WriteAllRequest, WriteMetadataRequest,
DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, ListDirRequest,
ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, ReadAllRequest, ReadMultipleRequest, ReadVersionRequest,
ReadXlRequest, RenameDataRequest, RenameFileRequst, StatVolumeRequest, UpdateMetadataRequest, WalkDirRequest,
WriteAllRequest, WriteMetadataRequest,
},
DEFAULT_GRPC_SERVER_MESSAGE_LEN,
};
use tokio::{fs, sync::RwLock};
use tonic::{
transport::{Channel, Endpoint as tonic_Endpoint},
Request,
};
use tower::timeout::Timeout;
use tonic::Request;
use tracing::info;
use uuid::Uuid;
use crate::{
disk::error::DiskError,
error::{Error, Result},
store_api::{FileInfo, RawFileInfo},
};
use super::{
endpoint::Endpoint, DeleteOptions, DiskAPI, DiskLocation, DiskOption, FileInfoVersions, FileReader, FileWriter,
MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader, RemoteFileWriter, RenameDataResp,
UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
};
use crate::{
disk::error::DiskError,
error::{Error, Result},
store_api::{FileInfo, RawFileInfo},
};
use protos::proto_gen::node_service::RenamePartRequst;
#[derive(Debug)]
pub struct RemoteDisk {
id: Mutex<Option<Uuid>>,
channel: Arc<RwLock<Option<Channel>>>,
url: url::Url,
pub id: Mutex<Option<Uuid>>,
pub addr: String,
pub url: url::Url,
pub root: PathBuf,
endpoint: Endpoint,
}
impl RemoteDisk {
pub async fn new(ep: &Endpoint, _opt: &DiskOption) -> Result<Self> {
let root = fs::canonicalize(ep.url.path()).await?;
// let root = fs::canonicalize(ep.url.path()).await?;
let root = PathBuf::from(ep.url.path());
let addr = format!("{}://{}:{}", ep.url.scheme(), ep.url.host_str().unwrap(), ep.url.port().unwrap());
Ok(Self {
channel: Arc::new(RwLock::new(None)),
id: Mutex::new(None),
addr,
url: ep.url.clone(),
root,
id: Mutex::new(None),
endpoint: ep.clone(),
})
}
#[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 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.clone())?;
let new_channel = connector.connect().await.map_err(|_err| DiskError::DiskNotFound)?;
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());
Ok(NodeServiceClient::connect(addr).await?)
}
}
// TODO: all api need to handle errors
@@ -109,7 +66,7 @@ impl DiskAPI for RemoteDisk {
}
async fn is_online(&self) -> bool {
// TODO: 连接状态
if let Ok(_) = self.get_client_v2().await {
if let Ok(_) = node_service_time_out_client(&self.addr).await {
return true;
}
false
@@ -144,7 +101,9 @@ impl DiskAPI for RemoteDisk {
async fn read_all(&self, volume: &str, path: &str) -> Result<Vec<u8>> {
info!("read_all");
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(ReadAllRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
@@ -164,7 +123,9 @@ impl DiskAPI for RemoteDisk {
async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()> {
info!("write_all");
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(WriteAllRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
@@ -184,7 +145,9 @@ impl DiskAPI for RemoteDisk {
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_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(DeleteRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
@@ -202,7 +165,9 @@ impl DiskAPI for RemoteDisk {
}
async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec<u8>) -> Result<()> {
info!("rename_part");
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(RenamePartRequst {
disk: self.root.to_string_lossy().to_string(),
src_volume: src_volume.to_string(),
@@ -222,7 +187,9 @@ impl DiskAPI for RemoteDisk {
}
async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
info!("rename_file");
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(RenameFileRequst {
disk: self.root.to_string_lossy().to_string(),
src_volume: src_volume.to_string(),
@@ -242,39 +209,56 @@ impl DiskAPI for RemoteDisk {
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_v2().await?,
)))
Ok(FileWriter::Remote(
RemoteFileWriter::new(
self.root.clone(),
volume.to_string(),
path.to_string(),
false,
node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?,
)
.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_v2().await?,
)))
Ok(FileWriter::Remote(
RemoteFileWriter::new(
self.root.clone(),
volume.to_string(),
path.to_string(),
true,
node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?,
)
.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_v2().await?,
)))
Ok(FileReader::Remote(
RemoteFileReader::new(
self.root.clone(),
volume.to_string(),
path.to_string(),
node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?,
)
.await?,
))
}
async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result<Vec<String>> {
info!("list_dir");
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(ListDirRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
@@ -292,7 +276,9 @@ impl DiskAPI for RemoteDisk {
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_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(WalkDirRequest {
disk: self.root.to_string_lossy().to_string(),
walk_dir_options,
@@ -323,7 +309,9 @@ impl DiskAPI for RemoteDisk {
) -> Result<RenameDataResp> {
info!("rename_data");
let file_info = serde_json::to_string(&fi)?;
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(RenameDataRequest {
disk: self.root.to_string_lossy().to_string(),
src_volume: src_volume.to_string(),
@@ -346,7 +334,9 @@ impl DiskAPI for RemoteDisk {
async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> {
info!("make_volumes");
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(MakeVolumesRequest {
disk: self.root.to_string_lossy().to_string(),
volumes: volumes.iter().map(|s| (*s).to_string()).collect(),
@@ -363,7 +353,9 @@ impl DiskAPI for RemoteDisk {
async fn make_volume(&self, volume: &str) -> Result<()> {
info!("make_volume");
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(MakeVolumeRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
@@ -380,7 +372,9 @@ impl DiskAPI for RemoteDisk {
async fn list_volumes(&self) -> Result<Vec<VolumeInfo>> {
info!("list_volumes");
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(ListVolumesRequest {
disk: self.root.to_string_lossy().to_string(),
});
@@ -402,7 +396,9 @@ impl DiskAPI for RemoteDisk {
async fn stat_volume(&self, volume: &str) -> Result<VolumeInfo> {
info!("stat_volume");
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(StatVolumeRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
@@ -422,7 +418,9 @@ impl DiskAPI for RemoteDisk {
async fn delete_paths(&self, volume: &str, paths: &[&str]) -> Result<()> {
info!("delete_paths");
let paths = paths.iter().map(|s| s.to_string()).collect::<Vec<String>>();
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(DeletePathsRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
@@ -442,7 +440,9 @@ impl DiskAPI for RemoteDisk {
let file_info = serde_json::to_string(&fi)?;
let opts = serde_json::to_string(&opts)?;
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(UpdateMetadataRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
@@ -463,7 +463,9 @@ impl DiskAPI for RemoteDisk {
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_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(WriteMetadataRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
@@ -490,7 +492,9 @@ impl DiskAPI for RemoteDisk {
) -> Result<FileInfo> {
info!("read_version");
let opts = serde_json::to_string(opts)?;
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(ReadVersionRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
@@ -512,7 +516,9 @@ impl DiskAPI for RemoteDisk {
async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo> {
info!("read_xl");
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(ReadXlRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
@@ -542,7 +548,9 @@ impl DiskAPI for RemoteDisk {
let file_info = serde_json::to_string(&fi)?;
let opts = serde_json::to_string(&opts)?;
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(DeleteVersionRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
@@ -574,7 +582,9 @@ impl DiskAPI for RemoteDisk {
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 mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(DeleteVersionsRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
@@ -607,7 +617,9 @@ impl DiskAPI for RemoteDisk {
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_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(ReadMultipleRequest {
disk: self.root.to_string_lossy().to_string(),
read_multiple_req,
@@ -630,7 +642,9 @@ impl DiskAPI for RemoteDisk {
async fn delete_volume(&self, volume: &str) -> Result<()> {
info!("delete_volume");
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(DeleteVolumeRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
ecstore/src/peer.rs
+19 -56
View File
@@ -1,15 +1,11 @@
use async_trait::async_trait;
use futures::future::join_all;
use protos::proto_gen::node_service::node_service_client::NodeServiceClient;
use protos::node_service_time_out_client;
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 std::{collections::HashMap, fmt::Debug, sync::Arc};
use tonic::Request;
use tower::timeout::Timeout;
use tracing::{info, warn};
use tracing::warn;
use crate::store::all_local_disk;
use crate::{
@@ -412,55 +408,13 @@ impl PeerS3Client for LocalPeerS3Client {
pub struct RemotePeerS3Client {
pub node: Option<Node>,
pub pools: Option<Vec<usize>>,
connector: Endpoint,
channel: Arc<RwLock<Option<Channel>>>,
addr: String,
}
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();
Self {
node,
pools,
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?)
let addr = format!("{}", node.as_ref().map(|v| { v.url.to_string() }).unwrap_or_default());
Self { node, pools, addr }
}
}
@@ -471,7 +425,9 @@ impl PeerS3Client for RemotePeerS3Client {
}
async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
let options = serde_json::to_string(opts)?;
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(ListBucketRequest { options });
let response = client.list_bucket(request).await?.into_inner();
let bucket_infos = response
@@ -484,7 +440,9 @@ impl PeerS3Client for RemotePeerS3Client {
}
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
let options = serde_json::to_string(opts)?;
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(MakeBucketRequest {
name: bucket.to_string(),
options,
@@ -500,7 +458,9 @@ impl PeerS3Client for RemotePeerS3Client {
}
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
let options = serde_json::to_string(opts)?;
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(GetBucketInfoRequest {
bucket: bucket.to_string(),
options,
@@ -512,7 +472,10 @@ impl PeerS3Client for RemotePeerS3Client {
}
async fn delete_bucket(&self, bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> {
let mut client = self.get_client_v2().await?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
let request = Request::new(DeleteBucketRequest {
bucket: bucket.to_string(),
});
ecstore/src/sets.rs
+31 -4
View File
@@ -1,6 +1,13 @@
#![allow(clippy::map_entry)]
use std::{collections::HashMap, sync::Arc};
use common::globals::GLOBAL_Local_Node_Name;
use futures::future::join_all;
use http::HeaderMap;
use lock::{namespace_lock::NsLockMap, new_lock_api, LockApi};
use tokio::sync::RwLock;
use uuid::Uuid;
use crate::{
disk::{
format::{DistributionAlgoVersion, FormatV3},
@@ -17,20 +24,18 @@ use crate::{
},
utils::hash,
};
use futures::future::join_all;
use http::HeaderMap;
use tokio::sync::RwLock;
use tokio::time::Duration;
use tokio_util::sync::CancellationToken;
use tracing::info;
use tracing::warn;
use uuid::Uuid;
#[derive(Debug, Clone)]
pub struct Sets {
pub id: Uuid,
// pub sets: Vec<Objects>,
// pub disk_set: Vec<Vec<Option<DiskStore>>>, // [set_count_idx][set_drive_count_idx] = disk_idx
pub lockers: Vec<Vec<LockApi>>,
pub disk_set: Vec<Arc<SetDisks>>, // [set_count_idx][set_drive_count_idx] = disk_idx
pub pool_idx: usize,
pub endpoints: PoolEndpoints,
@@ -53,6 +58,24 @@ impl Sets {
let set_count = fm.erasure.sets.len();
let set_drive_count = fm.erasure.sets[0].len();
let mut unique: Vec<Vec<String>> = vec![vec![]; set_count];
let mut lockers: Vec<Vec<LockApi>> = vec![vec![]; set_count];
endpoints.endpoints.as_ref().iter().enumerate().for_each(|(idx, endpoint)| {
let set_idx = idx / set_drive_count;
if endpoint.is_local && !unique[set_idx].contains(&"local".to_string()) {
unique[set_idx].push("local".to_string());
lockers[set_idx].push(new_lock_api(true, None));
}
if !endpoint.is_local {
let host_port = format!("{}:{}", endpoint.url.host_str().unwrap(), endpoint.url.port().unwrap().to_string());
if !unique[set_idx].contains(&host_port) {
unique[set_idx].push(host_port);
lockers[set_idx].push(new_lock_api(false, Some(endpoint.url.clone())));
}
}
});
let mut disk_set = Vec::with_capacity(set_count);
for i in 0..set_count {
@@ -99,6 +122,9 @@ impl Sets {
// warn!("sets new set_drive {:?}", &set_drive);
let set_disks = SetDisks {
lockers: lockers[i].clone(),
locker_owner: GLOBAL_Local_Node_Name.read().await.to_string(),
ns_mutex: Arc::new(RwLock::new(NsLockMap::new(*GLOBAL_IsDistErasure.read().await))),
disks: RwLock::new(set_drive),
set_drive_count,
default_parity_count: partiy_count,
@@ -115,6 +141,7 @@ impl Sets {
id: fm.id,
// sets: todo!(),
disk_set,
lockers,
pool_idx,
endpoints: endpoints.clone(),
format: fm.clone(),
ecstore/src/store.rs
+32
View File
@@ -1,4 +1,5 @@
#![allow(clippy::map_entry)]
use crate::disk::endpoint::EndpointType;
use crate::{
bucket::BucketMetadata,
disk::{error::DiskError, new_disk, DiskOption, DiskStore, WalkDirOptions, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
@@ -15,6 +16,7 @@ use crate::{
store_init, utils,
};
use backon::{ExponentialBuilder, Retryable};
use common::globals::{GLOBAL_Local_Node_Name, GLOBAL_Rustfs_Host, GLOBAL_Rustfs_Port};
use futures::future::join_all;
use http::HeaderMap;
use s3s::{dto::StreamingBlob, Body};
@@ -173,6 +175,13 @@ impl ECStore {
let mut local_disks = Vec::new();
init_local_peer(
&endpoint_pools,
&GLOBAL_Rustfs_Host.read().await.to_string(),
&GLOBAL_Rustfs_Port.read().await.to_string(),
)
.await;
debug!("endpoint_pools: {:?}", endpoint_pools);
for (i, pool_eps) in endpoint_pools.as_ref().iter().enumerate() {
@@ -805,3 +814,26 @@ impl StorageAPI for ECStore {
unimplemented!()
}
}
async fn init_local_peer(endpoint_pools: &EndpointServerPools, host: &String, port: &String) {
let mut peer_set = Vec::new();
endpoint_pools.as_ref().iter().for_each(|endpoints| {
endpoints.endpoints.as_ref().iter().for_each(|endpoint| {
if endpoint.get_type() == EndpointType::Url && endpoint.is_local && endpoint.url.has_host() {
peer_set.push(endpoint.url.host_str().unwrap().to_string());
}
});
});
if peer_set.is_empty() {
if !host.is_empty() {
*GLOBAL_Local_Node_Name.write().await = format!("{}:{}", host, port);
return;
}
*GLOBAL_Local_Node_Name.write().await = format!("127.0.0.1:{}", port);
return;
}
*GLOBAL_Local_Node_Name.write().await = peer_set[0].clone();
}
ecstore/src/store_api.rs
+1
View File
@@ -385,6 +385,7 @@ pub struct ObjectOptions {
pub delete_prefix: bool,
pub version_id: String,
pub no_lock: bool,
}
// impl Default for ObjectOptions {