From 5880f9fd30e47a8ab141f077bcc8469965066f63 Mon Sep 17 00:00:00 2001 From: weisd Date: Mon, 8 Jul 2024 11:39:43 +0800 Subject: [PATCH] test:mc cp --- ecstore/src/disk.rs | 131 +++++-------------------------------- ecstore/src/disk_api.rs | 123 +++++++++++++++++++++++++++++++++- ecstore/src/format.rs | 4 +- ecstore/src/lib.rs | 2 +- ecstore/src/peer.rs | 99 +++++++++++++++++++++++++--- ecstore/src/sets.rs | 6 +- ecstore/src/store.rs | 10 ++- ecstore/src/store_api.rs | 9 +++ ecstore/src/store_init.rs | 25 ++----- rustfs/src/storage/ecfs.rs | 11 +++- 10 files changed, 269 insertions(+), 151 deletions(-) diff --git a/ecstore/src/disk.rs b/ecstore/src/disk.rs index 7c0201363..3f0ba8281 100644 --- a/ecstore/src/disk.rs +++ b/ecstore/src/disk.rs @@ -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 { + 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(mut stream: S, writer: &mut W) -> Result @@ -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>) -> 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>, err: &DiskError) -> usize { - return errs - .iter() - .filter(|&e| { - if e.is_some() { - let e = e.as_ref().unwrap(); - let cast = e.downcast_ref::(); - if cast.is_some() { - let cast = cast.unwrap(); - return cast == err; - } - } - false - }) - .count(); - } - - pub fn quorum_unformatted_disks(errs: &Vec>) -> 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::(); - if cast.is_none() { - return false; - } - - let e = cast.unwrap(); - - e == disk_err - } - - // pub fn match_err(err: Error, matchs: Vec) -> bool { - // let cast = err.downcast_ref::(); - // 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, diff --git a/ecstore/src/disk_api.rs b/ecstore/src/disk_api.rs index cf3da89d0..3995a4cf3 100644 --- a/ecstore/src/disk_api.rs +++ b/ecstore/src/disk_api.rs @@ -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; +} + +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>) -> 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>, err: &DiskError) -> usize { + return errs + .iter() + .filter(|&e| { + if e.is_some() { + let e = e.as_ref().unwrap(); + let cast = e.downcast_ref::(); + if cast.is_some() { + let cast = cast.unwrap(); + return cast == err; + } + } + false + }) + .count(); + } + + pub fn quorum_unformatted_disks(errs: &Vec>) -> 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::(); + if cast.is_none() { + return false; + } + + let e = cast.unwrap(); + + e == disk_err + } + + // pub fn match_err(err: Error, matchs: Vec) -> bool { + // let cast = err.downcast_ref::(); + // 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) + } } diff --git a/ecstore/src/format.rs b/ecstore/src/format.rs index 4781fa4e3..1bc3e9f64 100644 --- a/ecstore/src/format.rs +++ b/ecstore/src/format.rs @@ -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")); diff --git a/ecstore/src/lib.rs b/ecstore/src/lib.rs index 7335ba9a7..8deeec19e 100644 --- a/ecstore/src/lib.rs +++ b/ecstore/src/lib.rs @@ -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; diff --git a/ecstore/src/peer.rs b/ecstore/src/peer.rs index ddfbe12e1..1af3a7b96 100644 --- a/ecstore/src/peer.rs +++ b/ecstore/src/peer.rs @@ -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>; @@ -14,6 +16,7 @@ type Client = Arc>; #[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; fn get_pools(&self) -> Vec; } @@ -37,15 +40,11 @@ impl S3PeerSys { .iter() .map(|e| { if e.is_local { - let cli: Box = Box::new(LocalPeerS3Client::new( - local_disks.clone(), - e.clone(), - e.pools.clone(), - )); + let cli: Box = + Box::new(LocalPeerS3Client::new(local_disks.clone(), e.clone(), 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(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 { + 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 { + 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 { + unimplemented!() + } } diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index 4516e64e9..8795b14d6 100644 --- a/ecstore/src/sets.rs +++ b/ecstore/src/sets.rs @@ -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 { + unimplemented!() + } + async fn put_object(&self, bucket: &str, object: &str, data: PutObjReader, opts: ObjectOptions) -> Result<()> { let disks = self.get_disks_by_key(object); diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 60d7c28ab..3655a6bde 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -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 { + 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 diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index d6acf6241..999aa4928 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -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; async fn put_object(&self, bucket: &str, object: &str, data: PutObjReader, opts: ObjectOptions) -> Result<()>; } diff --git a/ecstore/src/store_init.rs b/ecstore/src/store_init.rs index 9b54a7a61..0f63501b6 100644 --- a/ecstore/src/store_init.rs +++ b/ecstore/src/store_init.rs @@ -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>, - heal: bool, -) -> (Vec>, Vec>) { +async fn read_format_file_all(disks: &Vec>, heal: bool) -> (Vec>, Vec>) { let mut futures = Vec::with_capacity(disks.len()); for ep in disks.iter() { @@ -216,10 +214,7 @@ async fn read_format_file(disk: &Option, _heal: bool) -> Result>, - formats: &Vec>, -) -> Vec> { +async fn save_format_file_all(disks: &Vec>, formats: &Vec>) -> Vec> { 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, format: &Option) - 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; diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 29e3d238e..ef2bd1a27 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -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) -> S3Result> { 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())) }