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

refactor(ecstore): move List/Bucket Operations into set_disk::ops (backlog#819) (#4282)

Browse Source
This commit is contained in:
Zhengchao An
2026-07-05 15:53:16 +08:00
committed by GitHub
parent 5f2359de45
commit 1f51d21d33
5 changed files with 304 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/set_disk/lock.rs
+1 -1
View File
@@ -659,7 +659,7 @@ mod tests {
// the List operation family must run identically through it.
#[tokio::test]
async fn set_disks_ctx_mirrors_core_and_drives_list_operations() {
use crate::set_disk::list::ListOperations;
use crate::set_disk::ops::list::ListOperations;
let disk_count = 4;
let format = FormatV3::new(1, disk_count);
crates/ecstore/src/set_disk/mod.rs
-269
View File
@@ -444,7 +444,6 @@ const DEFAULT_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: bool = false;
static OBJECT_LOCK_DIAG_ENABLED: OnceLock<bool> = OnceLock::new();
mod ctx;
mod list;
mod lock;
mod metadata;
mod ops;
@@ -3532,213 +3531,6 @@ impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks {
}
}
#[async_trait::async_trait]
impl BucketOperations for SetDisks {
type Error = Error;
#[tracing::instrument(skip(self))]
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
let disks = self.disk_inventory().await;
let write_quorum = (disks.len() / 2) + 1;
let force_create = opts.force_create;
let mut futures = Vec::with_capacity(disks.len());
for disk in disks {
let bucket = bucket.to_string();
futures.push(async move {
match disk {
Some(disk) => match disk.make_volume(&bucket).await {
Ok(()) => Ok(()),
Err(err) if force_create && matches!(err, DiskError::VolumeExists) => Ok(()),
Err(err) => Err(err),
},
None => Err(DiskError::DiskNotFound),
}
});
}
let results = join_all(futures).await;
let errs = results
.into_iter()
.map(|result| result.err())
.collect::<Vec<Option<DiskError>>>();
if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) {
return Err(err.into());
}
Ok(())
}
#[tracing::instrument(skip(self))]
async fn get_bucket_info(&self, bucket: &str, _opts: &BucketOptions) -> Result<BucketInfo> {
let disks = self.disk_inventory().await;
let write_quorum = (disks.len() / 2) + 1;
let mut futures = Vec::with_capacity(disks.len());
for disk in disks {
let bucket = bucket.to_string();
futures.push(async move {
match disk {
Some(disk) => disk.stat_volume(&bucket).await,
None => Err(DiskError::DiskNotFound),
}
});
}
let results = join_all(futures).await;
let mut infos = Vec::with_capacity(results.len());
let mut errs = Vec::with_capacity(results.len());
for result in results {
match result {
Ok(info) => {
infos.push(Some(info));
errs.push(None);
}
Err(err) => {
infos.push(None);
errs.push(Some(err));
}
}
}
if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) {
return Err(err.into());
}
let mut versioning = false;
let mut object_locking = false;
if let Ok(sys) = metadata_sys::get(bucket).await {
versioning = sys.versioning();
object_locking = sys.object_locking();
}
infos
.into_iter()
.flatten()
.next()
.map(|info| BucketInfo {
name: info.name,
created: info.created,
versioning,
object_locking,
..Default::default()
})
.ok_or(Error::VolumeNotFound)
}
#[tracing::instrument(skip(self))]
async fn list_bucket(&self, _opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
let disks = self.disk_inventory().await;
let write_quorum = (disks.len() / 2) + 1;
let mut futures = Vec::with_capacity(disks.len());
for disk in disks {
futures.push(async move {
match disk {
Some(disk) => disk.list_volumes().await,
None => Err(DiskError::DiskNotFound),
}
});
}
let results = join_all(futures).await;
let mut infos = Vec::with_capacity(results.len());
let mut errs = Vec::with_capacity(results.len());
for result in results {
match result {
Ok(volumes) => {
infos.push(Some(volumes));
errs.push(None);
}
Err(err) => {
infos.push(None);
errs.push(Some(err));
}
}
}
if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) {
return Err(err.into());
}
let mut counts: HashMap<String, (usize, BucketInfo)> = HashMap::new();
for volumes in infos.into_iter().flatten() {
for volume in volumes {
if is_reserved_or_invalid_bucket(&volume.name, false) {
continue;
}
let entry = counts.entry(volume.name.clone()).or_insert((
0,
BucketInfo {
name: volume.name.clone(),
created: volume.created,
..Default::default()
},
));
entry.0 += 1;
}
}
let mut buckets = counts
.into_values()
.filter_map(|(count, bucket)| (count >= write_quorum).then_some(bucket))
.collect::<Vec<_>>();
buckets.sort_by(|left, right| left.name.cmp(&right.name));
Ok(buckets)
}
#[tracing::instrument(skip(self))]
async fn delete_bucket(&self, bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> {
let disks = self.disk_inventory().await;
let write_quorum = (disks.len() / 2) + 1;
let mut futures = Vec::with_capacity(disks.len());
for disk in disks.iter().cloned() {
let bucket = bucket.to_string();
futures.push(async move {
match disk {
Some(disk) => disk.delete_volume(&bucket).await,
None => Err(DiskError::DiskNotFound),
}
});
}
let results = join_all(futures).await;
let mut errs = Vec::with_capacity(results.len());
let mut recreate = false;
for result in results {
match result {
Ok(()) => errs.push(None),
Err(err) => {
if matches!(err, DiskError::VolumeNotEmpty) {
recreate = true;
}
errs.push(Some(err));
}
}
}
if recreate {
for (index, err) in errs.iter().enumerate() {
if err.is_none()
&& let Some(Some(disk)) = disks.get(index)
{
let _ = disk.make_volume(bucket).await;
}
}
return Err(Error::VolumeNotEmpty);
}
if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) {
return Err(err.into());
}
Ok(())
}
}
fn check_object_lock_retention_update(bucket: &str, object: &str, obj_info: &ObjectInfo, opts: &ObjectOptions) -> Result<()> {
if let Some(retention) = &opts.object_lock_retention
&& check_retention_for_modification(
@@ -5007,67 +4799,6 @@ impl SetDisks {
}
}
#[async_trait::async_trait]
impl crate::storage_api_contracts::list::ListOperations for SetDisks {
type Error = Error;
type ListObjectsV2Info = ListObjectsV2Info;
type ListObjectVersionsInfo = ListObjectVersionsInfo;
type ObjectInfoOrErr = ObjectInfoOrErr;
type WalkOptions = WalkOptions;
type WalkCancellation = CancellationToken;
type WalkResultSender = Sender<ObjectInfoOrErr>;
#[tracing::instrument(skip(self))]
async fn list_objects_v2(
self: Arc<Self>,
bucket: &str,
prefix: &str,
continuation_token: Option<String>,
delimiter: Option<String>,
max_keys: i32,
fetch_owner: bool,
start_after: Option<String>,
incl_deleted: bool,
) -> Result<ListObjectsV2Info> {
self.inner_list_objects_v2(
bucket,
prefix,
continuation_token,
delimiter,
max_keys,
fetch_owner,
start_after,
incl_deleted,
)
.await
}
#[tracing::instrument(skip(self))]
async fn list_object_versions(
self: Arc<Self>,
bucket: &str,
prefix: &str,
marker: Option<String>,
version_marker: Option<String>,
delimiter: Option<String>,
max_keys: i32,
) -> Result<ListObjectVersionsInfo> {
self.inner_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys)
.await
}
async fn walk(
self: Arc<Self>,
rx: CancellationToken,
bucket: &str,
prefix: &str,
result: Sender<ObjectInfoOrErr>,
opts: WalkOptions,
) -> Result<()> {
self.walk_internal(rx, bucket, prefix, result, opts).await
}
}
#[derive(Debug, PartialEq, Eq)]
struct ObjProps {
successor_mod_time: Option<OffsetDateTime>,
crates/ecstore/src/set_disk/ops/bucket.rs
+229
View File
@@ -0,0 +1,229 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! `BucketOperations` for `SetDisks`.
//!
//! P4 of the SetDisks God-Object split (tracking backlog#815, issue #819).
//! Relocated verbatim from `set_disk/mod.rs`; the contract stays implemented
//! `for SetDisks`, so its associated-type bounds are unchanged and runtime
//! behavior is the same.
use super::super::*;
#[async_trait::async_trait]
impl BucketOperations for SetDisks {
type Error = Error;
#[tracing::instrument(skip(self))]
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
let disks = self.disk_inventory().await;
let write_quorum = (disks.len() / 2) + 1;
let force_create = opts.force_create;
let mut futures = Vec::with_capacity(disks.len());
for disk in disks {
let bucket = bucket.to_string();
futures.push(async move {
match disk {
Some(disk) => match disk.make_volume(&bucket).await {
Ok(()) => Ok(()),
Err(err) if force_create && matches!(err, DiskError::VolumeExists) => Ok(()),
Err(err) => Err(err),
},
None => Err(DiskError::DiskNotFound),
}
});
}
let results = join_all(futures).await;
let errs = results
.into_iter()
.map(|result| result.err())
.collect::<Vec<Option<DiskError>>>();
if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) {
return Err(err.into());
}
Ok(())
}
#[tracing::instrument(skip(self))]
async fn get_bucket_info(&self, bucket: &str, _opts: &BucketOptions) -> Result<BucketInfo> {
let disks = self.disk_inventory().await;
let write_quorum = (disks.len() / 2) + 1;
let mut futures = Vec::with_capacity(disks.len());
for disk in disks {
let bucket = bucket.to_string();
futures.push(async move {
match disk {
Some(disk) => disk.stat_volume(&bucket).await,
None => Err(DiskError::DiskNotFound),
}
});
}
let results = join_all(futures).await;
let mut infos = Vec::with_capacity(results.len());
let mut errs = Vec::with_capacity(results.len());
for result in results {
match result {
Ok(info) => {
infos.push(Some(info));
errs.push(None);
}
Err(err) => {
infos.push(None);
errs.push(Some(err));
}
}
}
if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) {
return Err(err.into());
}
let mut versioning = false;
let mut object_locking = false;
if let Ok(sys) = metadata_sys::get(bucket).await {
versioning = sys.versioning();
object_locking = sys.object_locking();
}
infos
.into_iter()
.flatten()
.next()
.map(|info| BucketInfo {
name: info.name,
created: info.created,
versioning,
object_locking,
..Default::default()
})
.ok_or(Error::VolumeNotFound)
}
#[tracing::instrument(skip(self))]
async fn list_bucket(&self, _opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
let disks = self.disk_inventory().await;
let write_quorum = (disks.len() / 2) + 1;
let mut futures = Vec::with_capacity(disks.len());
for disk in disks {
futures.push(async move {
match disk {
Some(disk) => disk.list_volumes().await,
None => Err(DiskError::DiskNotFound),
}
});
}
let results = join_all(futures).await;
let mut infos = Vec::with_capacity(results.len());
let mut errs = Vec::with_capacity(results.len());
for result in results {
match result {
Ok(volumes) => {
infos.push(Some(volumes));
errs.push(None);
}
Err(err) => {
infos.push(None);
errs.push(Some(err));
}
}
}
if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) {
return Err(err.into());
}
let mut counts: HashMap<String, (usize, BucketInfo)> = HashMap::new();
for volumes in infos.into_iter().flatten() {
for volume in volumes {
if is_reserved_or_invalid_bucket(&volume.name, false) {
continue;
}
let entry = counts.entry(volume.name.clone()).or_insert((
0,
BucketInfo {
name: volume.name.clone(),
created: volume.created,
..Default::default()
},
));
entry.0 += 1;
}
}
let mut buckets = counts
.into_values()
.filter_map(|(count, bucket)| (count >= write_quorum).then_some(bucket))
.collect::<Vec<_>>();
buckets.sort_by(|left, right| left.name.cmp(&right.name));
Ok(buckets)
}
#[tracing::instrument(skip(self))]
async fn delete_bucket(&self, bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> {
let disks = self.disk_inventory().await;
let write_quorum = (disks.len() / 2) + 1;
let mut futures = Vec::with_capacity(disks.len());
for disk in disks.iter().cloned() {
let bucket = bucket.to_string();
futures.push(async move {
match disk {
Some(disk) => disk.delete_volume(&bucket).await,
None => Err(DiskError::DiskNotFound),
}
});
}
let results = join_all(futures).await;
let mut errs = Vec::with_capacity(results.len());
let mut recreate = false;
for result in results {
match result {
Ok(()) => errs.push(None),
Err(err) => {
if matches!(err, DiskError::VolumeNotEmpty) {
recreate = true;
}
errs.push(Some(err));
}
}
}
if recreate {
for (index, err) in errs.iter().enumerate() {
if err.is_none()
&& let Some(Some(disk)) = disks.get(index)
{
let _ = disk.make_volume(bucket).await;
}
}
return Err(Error::VolumeNotEmpty);
}
if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) {
return Err(err.into());
}
Ok(())
}
}
crates/ecstore/src/set_disk/list.rs → crates/ecstore/src/set_disk/ops/list.rs
+72 -2
View File
@@ -12,8 +12,17 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::ctx::SetDisksCtx;
use super::*;
//! `ListOperations` for `SetDisks` and the borrow-based prefix-maintenance
//! service unit.
//!
//! P3 of the SetDisks God-Object split (tracking backlog#815, issue #819).
//! Consolidates the `ListOperations` storage-api contract impl (relocated from
//! `set_disk/mod.rs`) with the `SetDisks::delete_all` borrow service unit
//! (relocated from `set_disk/list.rs`). Method bodies are moved verbatim and
//! runtime behavior is unchanged.
use super::super::ctx::SetDisksCtx;
use super::super::*;
impl SetDisks {
#[tracing::instrument(skip(self))]
@@ -91,3 +100,64 @@ impl<'a> ListOperations<'a> {
Ok(())
}
}
#[async_trait::async_trait]
impl crate::storage_api_contracts::list::ListOperations for SetDisks {
type Error = Error;
type ListObjectsV2Info = ListObjectsV2Info;
type ListObjectVersionsInfo = ListObjectVersionsInfo;
type ObjectInfoOrErr = ObjectInfoOrErr;
type WalkOptions = WalkOptions;
type WalkCancellation = CancellationToken;
type WalkResultSender = Sender<ObjectInfoOrErr>;
#[tracing::instrument(skip(self))]
async fn list_objects_v2(
self: Arc<Self>,
bucket: &str,
prefix: &str,
continuation_token: Option<String>,
delimiter: Option<String>,
max_keys: i32,
fetch_owner: bool,
start_after: Option<String>,
incl_deleted: bool,
) -> Result<ListObjectsV2Info> {
self.inner_list_objects_v2(
bucket,
prefix,
continuation_token,
delimiter,
max_keys,
fetch_owner,
start_after,
incl_deleted,
)
.await
}
#[tracing::instrument(skip(self))]
async fn list_object_versions(
self: Arc<Self>,
bucket: &str,
prefix: &str,
marker: Option<String>,
version_marker: Option<String>,
delimiter: Option<String>,
max_keys: i32,
) -> Result<ListObjectVersionsInfo> {
self.inner_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys)
.await
}
async fn walk(
self: Arc<Self>,
rx: CancellationToken,
bucket: &str,
prefix: &str,
result: Sender<ObjectInfoOrErr>,
opts: WalkOptions,
) -> Result<()> {
self.walk_internal(rx, bucket, prefix, result, opts).await
}
}
crates/ecstore/src/set_disk/ops/mod.rs
+2
View File
@@ -17,5 +17,7 @@
//! borrows shared state through [`super::ctx::SetDisksCtx`]; the storage-api
//! contract impls stay `for SetDisks`, so contract bounds are unchanged.
pub(crate) mod bucket;
pub(crate) mod heal;
pub(crate) mod list;
pub(crate) mod multipart;