mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-13 00:26:53 +00:00
add delete_bucket opts
This commit is contained in:
@@ -31,6 +31,9 @@ pub enum DiskError {
|
|||||||
|
|
||||||
#[error("volume not found")]
|
#[error("volume not found")]
|
||||||
VolumeNotFound,
|
VolumeNotFound,
|
||||||
|
|
||||||
|
#[error("volume not empty")]
|
||||||
|
VolumeNotEmpty,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl DiskError {
|
impl DiskError {
|
||||||
|
|||||||
@@ -972,7 +972,20 @@ impl DiskAPI for LocalDisk {
|
|||||||
let p = self.get_bucket_path(volume)?;
|
let p = self.get_bucket_path(volume)?;
|
||||||
|
|
||||||
// TODO: 不能用递归删除,如果目录下面有文件,返回errVolumeNotEmpty
|
// TODO: 不能用递归删除,如果目录下面有文件,返回errVolumeNotEmpty
|
||||||
fs::remove_dir_all(&p).await?;
|
|
||||||
|
if let Err(err) = fs::remove_dir_all(&p).await {
|
||||||
|
match err.kind() {
|
||||||
|
ErrorKind::NotFound => (),
|
||||||
|
// ErrorKind::DirectoryNotEmpty => (),
|
||||||
|
kind => {
|
||||||
|
if kind.to_string() == "directory not empty" {
|
||||||
|
return Err(Error::new(DiskError::VolumeNotEmpty));
|
||||||
|
}
|
||||||
|
|
||||||
|
return Err(Error::from(err));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|||||||
+28
-8
@@ -8,7 +8,7 @@ use crate::{
|
|||||||
disk::{self, error::DiskError, DiskStore, VolumeInfo},
|
disk::{self, error::DiskError, DiskStore, VolumeInfo},
|
||||||
endpoints::{EndpointServerPools, Node},
|
endpoints::{EndpointServerPools, Node},
|
||||||
error::{Error, Result},
|
error::{Error, Result},
|
||||||
store_api::{BucketInfo, BucketOptions, MakeBucketOptions},
|
store_api::{BucketInfo, BucketOptions, DeleteBucketOptions, MakeBucketOptions},
|
||||||
};
|
};
|
||||||
|
|
||||||
type Client = Arc<Box<dyn PeerS3Client>>;
|
type Client = Arc<Box<dyn PeerS3Client>>;
|
||||||
@@ -17,7 +17,7 @@ type Client = Arc<Box<dyn PeerS3Client>>;
|
|||||||
pub trait PeerS3Client: Debug + Sync + Send + 'static {
|
pub trait PeerS3Client: Debug + Sync + Send + 'static {
|
||||||
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>;
|
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>;
|
||||||
async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>>;
|
async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>>;
|
||||||
async fn delete_bucket(&self, bucket: &str) -> Result<()>;
|
async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()>;
|
||||||
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo>;
|
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo>;
|
||||||
fn get_pools(&self) -> Vec<usize>;
|
fn get_pools(&self) -> Vec<usize>;
|
||||||
}
|
}
|
||||||
@@ -140,10 +140,10 @@ impl PeerS3Client for S3PeerSys {
|
|||||||
|
|
||||||
Ok(buckets)
|
Ok(buckets)
|
||||||
}
|
}
|
||||||
async fn delete_bucket(&self, bucket: &str) -> Result<()> {
|
async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()> {
|
||||||
let mut futures = Vec::with_capacity(self.clients.len());
|
let mut futures = Vec::with_capacity(self.clients.len());
|
||||||
for cli in self.clients.iter() {
|
for cli in self.clients.iter() {
|
||||||
futures.push(cli.delete_bucket(bucket));
|
futures.push(cli.delete_bucket(bucket, &opts));
|
||||||
}
|
}
|
||||||
|
|
||||||
let mut errors = Vec::with_capacity(self.clients.len());
|
let mut errors = Vec::with_capacity(self.clients.len());
|
||||||
@@ -350,7 +350,7 @@ impl PeerS3Client for LocalPeerS3Client {
|
|||||||
.ok_or(Error::new(DiskError::VolumeNotFound))
|
.ok_or(Error::new(DiskError::VolumeNotFound))
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn delete_bucket(&self, bucket: &str) -> Result<()> {
|
async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()> {
|
||||||
let mut futures = Vec::with_capacity(self.local_disks.len());
|
let mut futures = Vec::with_capacity(self.local_disks.len());
|
||||||
|
|
||||||
for disk in self.local_disks.iter() {
|
for disk in self.local_disks.iter() {
|
||||||
@@ -361,14 +361,34 @@ impl PeerS3Client for LocalPeerS3Client {
|
|||||||
|
|
||||||
let mut errs = Vec::new();
|
let mut errs = Vec::new();
|
||||||
|
|
||||||
|
let mut recreate = false;
|
||||||
|
|
||||||
for res in results {
|
for res in results {
|
||||||
match res {
|
match res {
|
||||||
Ok(_) => errs.push(None),
|
Ok(_) => errs.push(None),
|
||||||
Err(e) => errs.push(Some(e)),
|
Err(e) => {
|
||||||
|
if DiskError::VolumeNotEmpty.is(&e) {
|
||||||
|
recreate = true;
|
||||||
|
}
|
||||||
|
errs.push(Some(e))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: errVolumeNotEmpty 不删除,把已经删除的重新创建
|
// errVolumeNotEmpty 不删除,把已经删除的重新创建
|
||||||
|
|
||||||
|
let mut idx = 0;
|
||||||
|
for err in errs {
|
||||||
|
if err.is_none() && recreate {
|
||||||
|
let _ = self.local_disks[idx].make_volume(bucket).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
idx += 1;
|
||||||
|
}
|
||||||
|
|
||||||
|
if recreate {
|
||||||
|
return Err(Error::new(DiskError::VolumeNotEmpty));
|
||||||
|
}
|
||||||
|
|
||||||
// TODO: reduceWriteQuorumErrs
|
// TODO: reduceWriteQuorumErrs
|
||||||
|
|
||||||
@@ -404,7 +424,7 @@ impl PeerS3Client for RemotePeerS3Client {
|
|||||||
unimplemented!()
|
unimplemented!()
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn delete_bucket(&self, _bucket: &str) -> Result<()> {
|
async fn delete_bucket(&self, _bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> {
|
||||||
unimplemented!()
|
unimplemented!()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+4
-3
@@ -13,8 +13,9 @@ use crate::{
|
|||||||
error::{Error, Result},
|
error::{Error, Result},
|
||||||
set_disk::SetDisks,
|
set_disk::SetDisks,
|
||||||
store_api::{
|
store_api::{
|
||||||
BucketInfo, BucketOptions, CompletePart, DeletedObject, GetObjectReader, HTTPRangeSpec, ListObjectsV2Info,
|
BucketInfo, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, HTTPRangeSpec,
|
||||||
MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, ObjectToDelete, PartInfo, PutObjReader, StorageAPI,
|
ListObjectsV2Info, MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, ObjectToDelete, PartInfo,
|
||||||
|
PutObjReader, StorageAPI,
|
||||||
},
|
},
|
||||||
utils::hash,
|
utils::hash,
|
||||||
};
|
};
|
||||||
@@ -297,7 +298,7 @@ impl StorageAPI for Sets {
|
|||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn delete_bucket(&self, _bucket: &str) -> Result<()> {
|
async fn delete_bucket(&self, _bucket: &str, opts: &DeleteBucketOptions) -> Result<()> {
|
||||||
unimplemented!()
|
unimplemented!()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,9 +7,9 @@ use crate::{
|
|||||||
peer::{PeerS3Client, S3PeerSys},
|
peer::{PeerS3Client, S3PeerSys},
|
||||||
sets::Sets,
|
sets::Sets,
|
||||||
store_api::{
|
store_api::{
|
||||||
BucketInfo, BucketOptions, CompletePart, DeletedObject, GetObjectReader, HTTPRangeSpec, ListObjectsInfo,
|
BucketInfo, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, HTTPRangeSpec,
|
||||||
ListObjectsV2Info, MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, ObjectToDelete, PartInfo,
|
ListObjectsInfo, ListObjectsV2Info, MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, ObjectToDelete,
|
||||||
PutObjReader, StorageAPI,
|
PartInfo, PutObjReader, StorageAPI,
|
||||||
},
|
},
|
||||||
store_init, utils,
|
store_init, utils,
|
||||||
};
|
};
|
||||||
@@ -630,8 +630,8 @@ impl StorageAPI for ECStore {
|
|||||||
unimplemented!()
|
unimplemented!()
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn delete_bucket(&self, bucket: &str) -> Result<()> {
|
async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()> {
|
||||||
self.peer_sys.delete_bucket(bucket).await?;
|
self.peer_sys.delete_bucket(bucket, opts).await?;
|
||||||
|
|
||||||
// 删除meta
|
// 删除meta
|
||||||
self.delete_all(RUSTFS_META_BUCKET, format!("{}/{}", BUCKET_META_PREFIX, bucket).as_str())
|
self.delete_all(RUSTFS_META_BUCKET, format!("{}/{}", BUCKET_META_PREFIX, bucket).as_str())
|
||||||
|
|||||||
@@ -243,6 +243,10 @@ pub struct MakeBucketOptions {
|
|||||||
pub force_create: bool,
|
pub force_create: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub struct DeleteBucketOptions {
|
||||||
|
pub force: bool, // Force deletion
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub struct PutObjReader {
|
pub struct PutObjReader {
|
||||||
pub stream: StreamingBlob,
|
pub stream: StreamingBlob,
|
||||||
@@ -488,7 +492,7 @@ pub struct DeletedObject {
|
|||||||
#[async_trait::async_trait]
|
#[async_trait::async_trait]
|
||||||
pub trait StorageAPI {
|
pub trait StorageAPI {
|
||||||
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>;
|
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>;
|
||||||
async fn delete_bucket(&self, bucket: &str) -> Result<()>;
|
async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()>;
|
||||||
async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>>;
|
async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>>;
|
||||||
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo>;
|
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo>;
|
||||||
async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo>;
|
async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo>;
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ use bytes::Bytes;
|
|||||||
use ecstore::disk::error::DiskError;
|
use ecstore::disk::error::DiskError;
|
||||||
use ecstore::store_api::BucketOptions;
|
use ecstore::store_api::BucketOptions;
|
||||||
use ecstore::store_api::CompletePart;
|
use ecstore::store_api::CompletePart;
|
||||||
|
use ecstore::store_api::DeleteBucketOptions;
|
||||||
use ecstore::store_api::HTTPRangeSpec;
|
use ecstore::store_api::HTTPRangeSpec;
|
||||||
use ecstore::store_api::MakeBucketOptions;
|
use ecstore::store_api::MakeBucketOptions;
|
||||||
use ecstore::store_api::MultipartUploadResult;
|
use ecstore::store_api::MultipartUploadResult;
|
||||||
@@ -86,8 +87,12 @@ impl S3 for FS {
|
|||||||
#[tracing::instrument(level = "debug", skip(self, req))]
|
#[tracing::instrument(level = "debug", skip(self, req))]
|
||||||
async fn delete_bucket(&self, req: S3Request<DeleteBucketInput>) -> S3Result<S3Response<DeleteBucketOutput>> {
|
async fn delete_bucket(&self, req: S3Request<DeleteBucketInput>) -> S3Result<S3Response<DeleteBucketOutput>> {
|
||||||
let input = req.input;
|
let input = req.input;
|
||||||
|
// TODO: DeleteBucketInput 没有force参数?
|
||||||
try_!(self.store.delete_bucket(&input.bucket).await);
|
try_!(
|
||||||
|
self.store
|
||||||
|
.delete_bucket(&input.bucket, &DeleteBucketOptions { force: false })
|
||||||
|
.await
|
||||||
|
);
|
||||||
|
|
||||||
Ok(S3Response::new(DeleteBucketOutput {}))
|
Ok(S3Response::new(DeleteBucketOutput {}))
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user