gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-27 16:48:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

test:mc cp

Browse Source
This commit is contained in:
weisd
2024-07-08 11:39:43 +08:00
parent c71e62e810
commit 5880f9fd30
10 changed files with 269 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/disk.rs
+16 -115
View File
@@ -20,7 +20,7 @@ use tracing::debug;
use uuid::Uuid;
use crate::{
disk_api::DiskAPI,
disk_api::{DiskAPI, DiskError, VolumeInfo},
endpoint::{Endpoint, Endpoints},
file_meta::FileMeta,
format::{DistributionAlgoVersion, FormatV3},
@@ -477,6 +477,21 @@ impl DiskAPI for LocalDisk {
Err(Error::new(DiskError::VolumeExists))
}
async fn stat_volume(&self, volume: &str) -> Result<VolumeInfo> {
let p = self.get_bucket_path(volume)?;
let m = read_file_metadata(&p).await?;
let modtime = match m.modified() {
Ok(md) => OffsetDateTime::from(md),
Err(_) => return Err(Error::msg("Not supported modified on this platform")),
};
Ok(VolumeInfo {
name: volume.to_string(),
created: modtime,
})
}
}
// pub async fn copy_bytes<S, W>(mut stream: S, writer: &mut W) -> Result<u64>
@@ -505,120 +520,6 @@ impl DiskAPI for LocalDisk {
// }
// }
#[derive(Debug, thiserror::Error)]
pub enum DiskError {
#[error("file not found")]
FileNotFound,
#[error("disk not found")]
DiskNotFound,
#[error("disk access denied")]
FileAccessDenied,
#[error("InconsistentDisk")]
InconsistentDisk,
#[error("volume already exists")]
VolumeExists,
#[error("unformatted disk error")]
UnformattedDisk,
#[error("unsupport disk")]
UnsupportedDisk,
#[error("disk not a dir")]
DiskNotDir,
#[error("volume not found")]
VolumeNotFound,
}
impl DiskError {
pub fn check_disk_fatal_errs(errs: &Vec<Option<Error>>) -> Result<()> {
println!("errs: {:?}", errs);
if Self::count_errs(errs, &DiskError::UnsupportedDisk) == errs.len() {
return Err(Error::new(DiskError::UnsupportedDisk));
}
// if count_errs(errs, &DiskError::DiskAccessDenied) == errs.len() {
// return Err(Error::new(DiskError::DiskAccessDenied));
// }
if Self::count_errs(errs, &DiskError::FileAccessDenied) == errs.len() {
return Err(Error::new(DiskError::FileAccessDenied));
}
// if count_errs(errs, &DiskError::FaultyDisk) == errs.len() {
// return Err(Error::new(DiskError::FaultyDisk));
// }
if Self::count_errs(errs, &DiskError::DiskNotDir) == errs.len() {
return Err(Error::new(DiskError::DiskNotDir));
}
// if count_errs(errs, &DiskError::XLBackend) == errs.len() {
// return Err(Error::new(DiskError::XLBackend));
// }
Ok(())
}
pub fn count_errs(errs: &Vec<Option<Error>>, err: &DiskError) -> usize {
return errs
.iter()
.filter(|&e| {
if e.is_some() {
let e = e.as_ref().unwrap();
let cast = e.downcast_ref::<DiskError>();
if cast.is_some() {
let cast = cast.unwrap();
return cast == err;
}
}
false
})
.count();
}
pub fn quorum_unformatted_disks(errs: &Vec<Option<Error>>) -> bool {
Self::count_errs(errs, &DiskError::UnformattedDisk) >= (errs.len() / 2) + 1
}
pub fn is_err(err: &Error, disk_err: &DiskError) -> bool {
let cast = err.downcast_ref::<DiskError>();
if cast.is_none() {
return false;
}
let e = cast.unwrap();
e == disk_err
}
// pub fn match_err(err: Error, matchs: Vec<DiskError>) -> bool {
// let cast = err.downcast_ref::<DiskError>();
// if cast.is_none() {
// return false;
// }
// let e = cast.unwrap();
// for i in matchs.iter() {
// if e == i {
// return true;
// }
// }
// return false;
// }
}
impl PartialEq for DiskError {
fn eq(&self, other: &Self) -> bool {
core::mem::discriminant(self) == core::mem::discriminant(other)
}
}
pub(crate) struct FileWriter<'a> {
tmp_path: PathBuf,
dest_path: &'a Path,
ecstore/src/disk_api.rs
+122 -1
View File
@@ -1,7 +1,8 @@
use std::fmt::Debug;
use anyhow::Result;
use anyhow::{Error, Result};
use bytes::Bytes;
use time::OffsetDateTime;
use tokio::io::DuplexStream;
use crate::store_api::FileInfo;
@@ -25,4 +26,124 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
async fn make_volumes(&self, volume: Vec<&str>) -> Result<()>;
async fn make_volume(&self, volume: &str) -> Result<()>;
async fn stat_volume(&self, volume: &str) -> Result<VolumeInfo>;
}
pub struct VolumeInfo {
pub name: String,
pub created: OffsetDateTime,
}
#[derive(Debug, thiserror::Error)]
pub enum DiskError {
#[error("file not found")]
FileNotFound,
#[error("disk not found")]
DiskNotFound,
#[error("disk access denied")]
FileAccessDenied,
#[error("InconsistentDisk")]
InconsistentDisk,
#[error("volume already exists")]
VolumeExists,
#[error("unformatted disk error")]
UnformattedDisk,
#[error("unsupport disk")]
UnsupportedDisk,
#[error("disk not a dir")]
DiskNotDir,
#[error("volume not found")]
VolumeNotFound,
}
impl DiskError {
pub fn check_disk_fatal_errs(errs: &Vec<Option<Error>>) -> Result<()> {
println!("errs: {:?}", errs);
if Self::count_errs(errs, &DiskError::UnsupportedDisk) == errs.len() {
return Err(Error::new(DiskError::UnsupportedDisk));
}
// if count_errs(errs, &DiskError::DiskAccessDenied) == errs.len() {
// return Err(Error::new(DiskError::DiskAccessDenied));
// }
if Self::count_errs(errs, &DiskError::FileAccessDenied) == errs.len() {
return Err(Error::new(DiskError::FileAccessDenied));
}
// if count_errs(errs, &DiskError::FaultyDisk) == errs.len() {
// return Err(Error::new(DiskError::FaultyDisk));
// }
if Self::count_errs(errs, &DiskError::DiskNotDir) == errs.len() {
return Err(Error::new(DiskError::DiskNotDir));
}
// if count_errs(errs, &DiskError::XLBackend) == errs.len() {
// return Err(Error::new(DiskError::XLBackend));
// }
Ok(())
}
pub fn count_errs(errs: &Vec<Option<Error>>, err: &DiskError) -> usize {
return errs
.iter()
.filter(|&e| {
if e.is_some() {
let e = e.as_ref().unwrap();
let cast = e.downcast_ref::<DiskError>();
if cast.is_some() {
let cast = cast.unwrap();
return cast == err;
}
}
false
})
.count();
}
pub fn quorum_unformatted_disks(errs: &Vec<Option<Error>>) -> bool {
Self::count_errs(errs, &DiskError::UnformattedDisk) >= (errs.len() / 2) + 1
}
pub fn is_err(err: &Error, disk_err: &DiskError) -> bool {
let cast = err.downcast_ref::<DiskError>();
if cast.is_none() {
return false;
}
let e = cast.unwrap();
e == disk_err
}
// pub fn match_err(err: Error, matchs: Vec<DiskError>) -> bool {
// let cast = err.downcast_ref::<DiskError>();
// if cast.is_none() {
// return false;
// }
// let e = cast.unwrap();
// for i in matchs.iter() {
// if e == i {
// return true;
// }
// }
// return false;
// }
}
impl PartialEq for DiskError {
fn eq(&self, other: &Self) -> bool {
core::mem::discriminant(self) == core::mem::discriminant(other)
}
}
ecstore/src/format.rs
+2 -2
View File
@@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize};
use serde_json::Error as JsonError;
use uuid::Uuid;
use crate::disk;
use crate::{disk, disk_api::DiskError};
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
pub enum FormatMetaVersion {
@@ -167,7 +167,7 @@ impl FormatV3 {
/// - j'th position is the disk index in the current set
pub fn find_disk_index_by_disk_id(&self, disk_id: Uuid) -> Result<(usize, usize)> {
if disk_id == Uuid::nil() {
return Err(Error::new(disk::DiskError::DiskNotFound));
return Err(Error::new(DiskError::DiskNotFound));
}
if disk_id == Uuid::max() {
return Err(Error::msg("disk offline"));
ecstore/src/lib.rs
+1 -1
View File
@@ -1,7 +1,7 @@
mod bucket_meta;
mod chunk_stream;
mod disk;
mod disk_api;
pub mod disk_api;
mod disks_layout;
mod ellipses;
mod endpoint;
ecstore/src/peer.rs
+89 -10
View File
@@ -1,12 +1,14 @@
use anyhow::Result;
use anyhow::{Error, Result};
use async_trait::async_trait;
use futures::future::join_all;
use std::{fmt::Debug, sync::Arc};
use tracing::debug;
use crate::{
disk::{DiskError, DiskStore},
disk::DiskStore,
disk_api::DiskError,
endpoint::{EndpointServerPools, Node},
store_api::MakeBucketOptions,
store_api::{BucketInfo, BucketOptions, MakeBucketOptions},
};
type Client = Arc<Box<dyn PeerS3Client>>;
@@ -14,6 +16,7 @@ type Client = Arc<Box<dyn PeerS3Client>>;
#[async_trait]
pub trait PeerS3Client: Debug + Sync + Send + 'static {
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>;
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo>;
fn get_pools(&self) -> Vec<i32>;
}
@@ -37,15 +40,11 @@ impl S3PeerSys {
.iter()
.map(|e| {
if e.is_local {
let cli: Box<dyn PeerS3Client> = Box::new(LocalPeerS3Client::new(
local_disks.clone(),
e.clone(),
e.pools.clone(),
));
let cli: Box<dyn PeerS3Client> =
Box::new(LocalPeerS3Client::new(local_disks.clone(), e.clone(), 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(e.clone(), e.pools.clone()));
Arc::new(cli)
}
})
@@ -97,6 +96,46 @@ impl PeerS3Client for S3PeerSys {
Ok(())
}
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
let mut futures = Vec::with_capacity(self.clients.len());
for cli in self.clients.iter() {
futures.push(cli.get_bucket_info(bucket, opts));
}
let mut ress = Vec::with_capacity(self.clients.len());
let mut errors = Vec::with_capacity(self.clients.len());
let results = join_all(futures).await;
for result in results {
match result {
Ok(res) => {
ress.push(Some(res));
errors.push(None);
}
Err(e) => {
ress.push(None);
errors.push(Some(e));
}
}
}
for i in 0..self.pools_count {
let mut per_pool_errs = Vec::with_capacity(self.clients.len());
for (j, cli) in self.clients.iter().enumerate() {
let pools = cli.get_pools();
let idx = i as i32;
if pools.contains(&idx) {
per_pool_errs.push(errors[j].as_ref());
}
// TODO: reduceWriteQuorumErrs
}
}
ress.iter()
.find_map(|op| op.as_ref().map(|v| v.clone()))
.ok_or(Error::new(DiskError::VolumeNotFound))
}
}
#[derive(Debug)]
@@ -153,6 +192,43 @@ impl PeerS3Client for LocalPeerS3Client {
Ok(())
}
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
let mut futures = Vec::with_capacity(self.local_disks.len());
for disk in self.local_disks.iter() {
futures.push(disk.stat_volume(bucket));
}
let results = join_all(futures).await;
let mut ress = Vec::with_capacity(self.local_disks.len());
let mut errs = Vec::with_capacity(self.local_disks.len());
for res in results {
match res {
Ok(r) => {
errs.push(None);
ress.push(Some(r));
}
Err(e) => {
errs.push(Some(e));
ress.push(None);
}
}
}
// TODO: reduceWriteQuorumErrs
// debug!("get_bucket_info errs:{:?}", errs);
ress.iter()
.find_map(|op| {
op.as_ref().map(|v| BucketInfo {
name: v.name.clone(),
created: v.created,
})
})
.ok_or(Error::new(DiskError::VolumeNotFound))
}
}
#[derive(Debug)]
@@ -175,4 +251,7 @@ impl PeerS3Client for RemotePeerS3Client {
async fn make_bucket(&self, _bucket: &str, _opts: &MakeBucketOptions) -> Result<()> {
unimplemented!()
}
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
unimplemented!()
}
}
ecstore/src/sets.rs
+5 -1
View File
@@ -12,7 +12,7 @@ use crate::{
endpoint::PoolEndpoints,
erasure::Erasure,
format::{DistributionAlgoVersion, FormatV3},
store_api::{FileInfo, MakeBucketOptions, ObjectOptions, PutObjReader, StorageAPI},
store_api::{BucketInfo, BucketOptions, FileInfo, MakeBucketOptions, ObjectOptions, PutObjReader, StorageAPI},
utils::hash,
};
@@ -158,6 +158,10 @@ impl StorageAPI for Sets {
unimplemented!()
}
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
unimplemented!()
}
async fn put_object(&self, bucket: &str, object: &str, data: PutObjReader, opts: ObjectOptions) -> Result<()> {
let disks = self.get_disks_by_key(object);
ecstore/src/store.rs
+8 -2
View File
@@ -8,12 +8,13 @@ use uuid::Uuid;
use crate::{
bucket_meta::BucketMetadata,
disk::{self, DiskError, DiskOption, DiskStore, RUSTFS_META_BUCKET},
disk::{self, DiskOption, DiskStore, RUSTFS_META_BUCKET},
disk_api::DiskError,
disks_layout::DisksLayout,
endpoint::EndpointServerPools,
peer::{PeerS3Client, S3PeerSys},
sets::Sets,
store_api::{MakeBucketOptions, ObjectOptions, PutObjReader, StorageAPI},
store_api::{BucketInfo, BucketOptions, MakeBucketOptions, ObjectOptions, PutObjReader, StorageAPI},
store_init, utils,
};
@@ -134,6 +135,11 @@ impl StorageAPI for ECStore {
Ok(())
}
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
let info = self.peer_sys.get_bucket_info(bucket, opts).await?;
Ok(info)
}
async fn put_object(&self, bucket: &str, object: &str, data: PutObjReader, opts: ObjectOptions) -> Result<()> {
// checkPutObjectArgs
ecstore/src/store_api.rs
+9
View File
@@ -149,9 +149,18 @@ pub struct ObjectOptions {
pub max_parity: bool,
}
pub struct BucketOptions {}
#[derive(Debug, Clone)]
pub struct BucketInfo {
pub name: String,
pub created: OffsetDateTime,
}
#[async_trait::async_trait]
pub trait StorageAPI {
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>;
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo>;
async fn put_object(&self, bucket: &str, object: &str, data: PutObjReader, opts: ObjectOptions) -> Result<()>;
}
ecstore/src/store_init.rs
+7 -18
View File
@@ -3,7 +3,8 @@ use futures::future::join_all;
use uuid::Uuid;
use crate::{
disk::{DiskError, DiskStore, FORMAT_CONFIG_FILE, RUSTFS_META_BUCKET},
disk::{DiskStore, FORMAT_CONFIG_FILE, RUSTFS_META_BUCKET},
disk_api::DiskError,
format::{FormatErasureVersion, FormatMetaVersion, FormatV3},
};
@@ -163,10 +164,7 @@ pub fn default_partiy_count(drive: usize) -> usize {
}
}
// read_format_file_all 读取所有foramt.json
async fn read_format_file_all(
disks: &Vec<Option<DiskStore>>,
heal: bool,
) -> (Vec<Option<FormatV3>>, Vec<Option<Error>>) {
async fn read_format_file_all(disks: &Vec<Option<DiskStore>>, heal: bool) -> (Vec<Option<FormatV3>>, Vec<Option<Error>>) {
let mut futures = Vec::with_capacity(disks.len());
for ep in disks.iter() {
@@ -216,10 +214,7 @@ async fn read_format_file(disk: &Option<DiskStore>, _heal: bool) -> Result<Forma
Ok(fm)
}
async fn save_format_file_all(
disks: &Vec<Option<DiskStore>>,
formats: &Vec<Option<FormatV3>>,
) -> Vec<Option<Error>> {
async fn save_format_file_all(disks: &Vec<Option<DiskStore>>, formats: &Vec<Option<FormatV3>>) -> Vec<Option<Error>> {
let mut futures = Vec::with_capacity(disks.len());
for (i, ep) in disks.iter().enumerate() {
@@ -255,16 +250,10 @@ async fn save_format_file(disk: &Option<DiskStore>, format: &Option<FormatV3>) -
let tmpfile = Uuid::new_v4().to_string();
let disk = disk.as_ref().unwrap();
disk.write_all(RUSTFS_META_BUCKET, tmpfile.as_str(), json_data)
.await?;
disk.write_all(RUSTFS_META_BUCKET, tmpfile.as_str(), json_data).await?;
disk.rename_file(
RUSTFS_META_BUCKET,
tmpfile.as_str(),
RUSTFS_META_BUCKET,
FORMAT_CONFIG_FILE,
)
.await?;
disk.rename_file(RUSTFS_META_BUCKET, tmpfile.as_str(), RUSTFS_META_BUCKET, FORMAT_CONFIG_FILE)
.await?;
// let mut disk = disk;
rustfs/src/storage/ecfs.rs
+10 -1
View File
@@ -1,5 +1,7 @@
use std::fmt::Debug;
use ecstore::disk_api::DiskError;
use ecstore::store_api::BucketOptions;
use ecstore::store_api::MakeBucketOptions;
use ecstore::store_api::ObjectOptions;
use ecstore::store_api::PutObjReader;
@@ -121,7 +123,14 @@ impl S3 for FS {
async fn head_bucket(&self, req: S3Request<HeadBucketInput>) -> S3Result<S3Response<HeadBucketOutput>> {
let input = req.input;
// mc cp step 2
if let Err(e) = self.store.get_bucket_info(&input.bucket, &BucketOptions {}).await {
if DiskError::is_err(&e, &DiskError::VolumeNotFound) {
return Err(s3_error!(NoSuchBucket));
} else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("{}", e)));
}
}
// mc cp step 2 GetBucketInfo
Ok(S3Response::new(HeadBucketOutput::default()))
}