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(storage): trim thin s3 forwarding layers (#2474)

Browse Source
This commit is contained in:
安正超
2026-04-10 23:17:18 +08:00
committed by GitHub
parent b8c45fc9e3
commit 4472324125
8 changed files with 7 additions and 464 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/app/admin_usecase.rs
+1 -28
View File
@@ -93,19 +93,11 @@ impl DefaultAdminUsecase {
}
pub async fn execute_query_server_info(&self, req: QueryServerInfoRequest) -> AdminUsecaseResult<QueryServerInfoResponse> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let info = get_server_info(req.include_pools).await;
Ok(QueryServerInfoResponse { info })
}
pub async fn execute_query_storage_info(&self) -> AdminUsecaseResult<StorageInfo> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let Some(store) = new_object_layer_fn() else {
return Err(Self::app_error(S3ErrorCode::InternalError, "Not init"));
};
@@ -114,10 +106,6 @@ impl DefaultAdminUsecase {
}
pub async fn execute_query_data_usage_info(&self) -> AdminUsecaseResult<DataUsageInfo> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let Some(store) = new_object_layer_fn() else {
return Err(Self::app_error(S3ErrorCode::InternalError, "Not init"));
};
@@ -196,10 +184,6 @@ impl DefaultAdminUsecase {
}
pub async fn execute_list_pool_statuses(&self) -> AdminUsecaseResult<Vec<PoolStatus>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let Some(store) = new_object_layer_fn() else {
return Err(Self::app_error(S3ErrorCode::InternalError, "Not init"));
};
@@ -222,10 +206,6 @@ impl DefaultAdminUsecase {
}
pub async fn execute_query_pool_status(&self, req: QueryPoolStatusRequest) -> AdminUsecaseResult<PoolStatus> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let Some(endpoints) = self.endpoints() else {
return Err(Self::app_error_default(S3ErrorCode::NotImplemented));
};
@@ -254,14 +234,7 @@ impl DefaultAdminUsecase {
}
pub fn execute_collect_dependency_readiness(&self) -> DependencyReadiness {
let iam_ready = self
.context
.as_ref()
.map(|context| {
let _ = context.object_store();
context.iam().is_ready()
})
.unwrap_or(false);
let iam_ready = self.context.as_ref().map(|context| context.iam().is_ready()).unwrap_or(false);
DependencyReadiness {
storage_ready: new_object_layer_fn().is_some(),
rustfs/src/app/bucket_usecase.rs
+2 -200
View File
@@ -26,7 +26,7 @@ use crate::storage::helper::{OperationHelper, spawn_background_with_context};
use crate::storage::s3_api::acl;
use crate::storage::s3_api::bucket::{
ListObjectVersionsParams, ListObjectsV2Params, build_list_buckets_output, build_list_object_versions_output,
build_list_objects_v2_output, parse_list_object_versions_params, parse_list_objects_v2_params,
build_list_objects_output, build_list_objects_v2_output, parse_list_object_versions_params, parse_list_objects_v2_params,
};
use crate::storage::s3_api::common::rustfs_owner;
use crate::storage::*;
@@ -539,10 +539,6 @@ impl DefaultBucketUsecase {
fields(start_time=?time::OffsetDateTime::now_utc())
)]
pub async fn execute_create_bucket(&self, req: S3Request<CreateBucketInput>) -> S3Result<S3Response<CreateBucketOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let helper = OperationHelper::new(&req, EventName::BucketCreated, S3Operation::CreateBucket);
let requester_is_owner = match req_info_ref(&req) {
Ok(r) => r.is_owner,
@@ -596,10 +592,6 @@ impl DefaultBucketUsecase {
}
pub async fn execute_put_bucket_acl(&self, req: S3Request<PutBucketAclInput>) -> S3Result<S3Response<PutBucketAclOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let PutBucketAclInput {
bucket,
access_control_policy,
@@ -627,10 +619,6 @@ impl DefaultBucketUsecase {
#[instrument(level = "debug", skip(self, req))]
pub async fn execute_delete_bucket(&self, mut req: S3Request<DeleteBucketInput>) -> S3Result<S3Response<DeleteBucketOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let helper = OperationHelper::new(&req, EventName::BucketRemoved, S3Operation::DeleteBucket);
let input = req.input.clone();
@@ -670,10 +658,6 @@ impl DefaultBucketUsecase {
#[instrument(level = "debug", skip(self, req))]
pub async fn execute_head_bucket(&self, req: S3Request<HeadBucketInput>) -> S3Result<S3Response<HeadBucketOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let input = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -689,10 +673,6 @@ impl DefaultBucketUsecase {
}
pub async fn execute_get_bucket_acl(&self, req: S3Request<GetBucketAclInput>) -> S3Result<S3Response<GetBucketAclOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let GetBucketAclInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -712,10 +692,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<GetBucketLocationInput>,
) -> S3Result<S3Response<GetBucketLocationOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let input = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -738,10 +714,6 @@ impl DefaultBucketUsecase {
#[instrument(level = "debug", skip(self))]
pub async fn execute_list_buckets(&self, req: S3Request<ListBucketsInput>) -> S3Result<S3Response<ListBucketsOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let Some(store) = new_object_layer_fn() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
@@ -798,10 +770,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<DeleteBucketEncryptionInput>,
) -> S3Result<S3Response<DeleteBucketEncryptionOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let DeleteBucketEncryptionInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -830,10 +798,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<DeleteBucketCorsInput>,
) -> S3Result<S3Response<DeleteBucketCorsOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let DeleteBucketCorsInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -862,10 +826,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<DeleteBucketLifecycleInput>,
) -> S3Result<S3Response<DeleteBucketLifecycleOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let DeleteBucketLifecycleInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -893,10 +853,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<DeleteBucketPolicyInput>,
) -> S3Result<S3Response<DeleteBucketPolicyOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let DeleteBucketPolicyInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -924,10 +880,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<DeleteBucketReplicationInput>,
) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let DeleteBucketReplicationInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -958,10 +910,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<DeleteBucketTaggingInput>,
) -> S3Result<S3Response<DeleteBucketTaggingOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let DeleteBucketTaggingInput { bucket, .. } = req.input;
metadata_sys::delete(&bucket, BUCKET_TAGGING_CONFIG)
@@ -981,10 +929,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<DeletePublicAccessBlockInput>,
) -> S3Result<S3Response<DeletePublicAccessBlockOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let DeletePublicAccessBlockInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -1007,10 +951,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<GetBucketEncryptionInput>,
) -> S3Result<S3Response<GetBucketEncryptionOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let GetBucketEncryptionInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -1040,10 +980,6 @@ impl DefaultBucketUsecase {
#[instrument(level = "debug", skip(self))]
pub async fn execute_get_bucket_cors(&self, req: S3Request<GetBucketCorsInput>) -> S3Result<S3Response<GetBucketCorsOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let GetBucketCorsInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -1079,10 +1015,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<GetBucketLifecycleConfigurationInput>,
) -> S3Result<S3Response<GetBucketLifecycleConfigurationOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let GetBucketLifecycleConfigurationInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -1111,10 +1043,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<GetBucketNotificationConfigurationInput>,
) -> S3Result<S3Response<GetBucketNotificationConfigurationOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let GetBucketNotificationConfigurationInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -1153,10 +1081,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<GetBucketPolicyInput>,
) -> S3Result<S3Response<GetBucketPolicyOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let GetBucketPolicyInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -1187,10 +1111,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<GetBucketPolicyStatusInput>,
) -> S3Result<S3Response<GetBucketPolicyStatusOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let GetBucketPolicyStatusInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -1268,10 +1188,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<GetBucketReplicationInput>,
) -> S3Result<S3Response<GetBucketReplicationOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let GetBucketReplicationInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -1307,10 +1223,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<GetBucketTaggingInput>,
) -> S3Result<S3Response<GetBucketTaggingOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let GetBucketTaggingInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -1341,10 +1253,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<GetPublicAccessBlockInput>,
) -> S3Result<S3Response<GetPublicAccessBlockOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let GetPublicAccessBlockInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -1379,10 +1287,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<GetBucketVersioningInput>,
) -> S3Result<S3Response<GetBucketVersioningOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let GetBucketVersioningInput { bucket, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
@@ -1405,10 +1309,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<PutBucketEncryptionInput>,
) -> S3Result<S3Response<PutBucketEncryptionOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let PutBucketEncryptionInput {
bucket,
server_side_encryption_configuration,
@@ -1447,10 +1347,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<PutBucketLifecycleConfigurationInput>,
) -> S3Result<S3Response<PutBucketLifecycleConfigurationOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let PutBucketLifecycleConfigurationInput {
bucket,
lifecycle_configuration,
@@ -1512,9 +1408,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<PutBucketNotificationConfigurationInput>,
) -> S3Result<S3Response<PutBucketNotificationConfigurationOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let request_region = req.region.clone();
let PutBucketNotificationConfigurationInput {
@@ -1588,10 +1481,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<PutBucketPolicyInput>,
) -> S3Result<S3Response<PutBucketPolicyOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let PutBucketPolicyInput { bucket, policy, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -1649,10 +1538,6 @@ impl DefaultBucketUsecase {
#[instrument(level = "debug", skip(self))]
pub async fn execute_put_bucket_cors(&self, req: S3Request<PutBucketCorsInput>) -> S3Result<S3Response<PutBucketCorsOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let PutBucketCorsInput {
bucket,
cors_configuration,
@@ -1687,10 +1572,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<PutBucketReplicationInput>,
) -> S3Result<S3Response<PutBucketReplicationOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let PutBucketReplicationInput {
bucket,
replication_configuration,
@@ -1729,10 +1610,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<PutPublicAccessBlockInput>,
) -> S3Result<S3Response<PutPublicAccessBlockOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let PutPublicAccessBlockInput {
bucket,
public_access_block_configuration,
@@ -1761,10 +1638,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<PutBucketTaggingInput>,
) -> S3Result<S3Response<PutBucketTaggingOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let PutBucketTaggingInput { bucket, tagging, .. } = req.input;
let Some(store) = new_object_layer_fn() else {
@@ -1796,10 +1669,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<PutBucketVersioningInput>,
) -> S3Result<S3Response<PutBucketVersioningOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let PutBucketVersioningInput {
bucket,
versioning_configuration,
@@ -1827,10 +1696,6 @@ impl DefaultBucketUsecase {
#[instrument(level = "debug", skip(self, req))]
pub async fn execute_list_objects_v2(&self, req: S3Request<ListObjectsV2Input>) -> S3Result<S3Response<ListObjectsV2Output>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
// warn!("list_objects_v2 req {:?}", &req.input);
let ListObjectsV2Input {
bucket,
@@ -1895,10 +1760,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<ListObjectsV2Input>,
) -> S3Result<S3Response<ListObjectsV2MOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let input = req.input.clone();
let ListObjectsV2Input {
bucket,
@@ -1963,10 +1824,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<ListObjectVersionsInput>,
) -> S3Result<S3Response<ListObjectVersionsOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let ListObjectVersionsInput {
bucket,
delimiter,
@@ -2001,10 +1858,6 @@ impl DefaultBucketUsecase {
&self,
req: S3Request<ListObjectVersionsInput>,
) -> S3Result<S3Response<ListObjectVersionsMOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let input = req.input.clone();
let ListObjectVersionsInput {
bucket,
@@ -2055,61 +1908,10 @@ impl DefaultBucketUsecase {
#[instrument(level = "debug", skip(self, req))]
pub async fn execute_list_objects(&self, req: S3Request<ListObjectsInput>) -> S3Result<S3Response<ListObjectsOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let request_marker = req.input.marker.clone();
let v2_resp = self.execute_list_objects_v2(req.map_input(Into::into)).await?;
Ok(v2_resp.map_output(|v2| {
let next_marker = if v2.is_truncated.unwrap_or(false) {
let last_key = v2
.contents
.as_ref()
.and_then(|contents| contents.last())
.and_then(|obj| obj.key.as_ref())
.cloned();
let last_prefix = v2
.common_prefixes
.as_ref()
.and_then(|prefixes| prefixes.last())
.and_then(|prefix| prefix.prefix.as_ref())
.cloned();
match (last_key, last_prefix) {
(Some(k), Some(p)) => {
if k > p {
Some(k)
} else {
Some(p)
}
}
(Some(k), None) => Some(k),
(None, Some(p)) => Some(p),
(None, None) => None,
}
} else {
None
};
let marker = Some(request_marker.unwrap_or_default());
ListObjectsOutput {
contents: v2.contents,
delimiter: v2.delimiter,
encoding_type: v2.encoding_type,
name: v2.name,
prefix: v2.prefix,
max_keys: v2.max_keys,
common_prefixes: v2.common_prefixes,
is_truncated: v2.is_truncated,
marker,
next_marker,
..Default::default()
}
}))
Ok(v2_resp.map_output(|v2| build_list_objects_output(v2, request_marker)))
}
}
rustfs/src/app/multipart_usecase.rs
+3 -47
View File
@@ -19,7 +19,6 @@ use crate::app::object_usecase::{build_put_like_object_lock_metadata, validate_e
use crate::capacity::record_capacity_write;
use crate::error::ApiError;
use crate::storage::access::has_bypass_governance_header;
use crate::storage::entity;
use crate::storage::helper::OperationHelper;
use crate::storage::options::{
copy_src_opts, extract_metadata, get_complete_multipart_upload_opts, get_content_sha256_with_query, get_opts,
@@ -185,10 +184,6 @@ impl DefaultMultipartUsecase {
&self,
req: S3Request<AbortMultipartUploadInput>,
) -> S3Result<S3Response<AbortMultipartUploadOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let AbortMultipartUploadInput {
bucket, key, upload_id, ..
} = req.input;
@@ -225,10 +220,6 @@ impl DefaultMultipartUsecase {
&self,
req: S3Request<CompleteMultipartUploadInput>,
) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let mut helper = OperationHelper::new(
&req,
EventName::ObjectCreatedCompleteMultipartUpload,
@@ -432,21 +423,6 @@ impl DefaultMultipartUsecase {
version_id: mpu_version,
..Default::default()
};
let helper_output = entity::CompleteMultipartUploadOutput {
bucket: Some(bucket.clone()),
key: Some(key.clone()),
e_tag: obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)),
location: Some(location),
server_side_encryption,
ssekms_key_id,
checksum_crc32,
checksum_crc32c,
checksum_sha1,
checksum_sha256,
checksum_crc64nvme,
checksum_type,
..Default::default()
};
let mt2 = HashMap::new();
let replicate_options =
get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts.clone());
@@ -464,9 +440,9 @@ impl DefaultMultipartUsecase {
helper = helper.version_id(version_id.clone());
}
let helper_result = Ok(S3Response::new(helper_output));
let _ = helper.complete(&helper_result);
Ok(S3Response::new(output))
let result = Ok(S3Response::new(output));
let _ = helper.complete(&result);
result
}
#[instrument(level = "debug", skip(self, req))]
@@ -474,10 +450,6 @@ impl DefaultMultipartUsecase {
&self,
req: S3Request<CreateMultipartUploadInput>,
) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let helper =
OperationHelper::new(&req, EventName::ObjectCreatedCreateMultipartUpload, S3Operation::CreateMultipartUpload)
.suppress_event();
@@ -622,10 +594,6 @@ impl DefaultMultipartUsecase {
#[instrument(level = "debug", skip(self, req))]
pub async fn execute_upload_part(&self, req: S3Request<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let input = req.input;
let UploadPartInput {
body,
@@ -868,10 +836,6 @@ impl DefaultMultipartUsecase {
&self,
req: S3Request<ListMultipartUploadsInput>,
) -> S3Result<S3Response<ListMultipartUploadsOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let ListMultipartUploadsInput {
bucket,
prefix,
@@ -901,10 +865,6 @@ impl DefaultMultipartUsecase {
}
pub async fn execute_list_parts(&self, req: S3Request<ListPartsInput>) -> S3Result<S3Response<ListPartsOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let ListPartsInput {
bucket,
key,
@@ -940,10 +900,6 @@ impl DefaultMultipartUsecase {
&self,
req: S3Request<UploadPartCopyInput>,
) -> S3Result<S3Response<UploadPartCopyOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let UploadPartCopyInput {
bucket,
key,
rustfs/src/app/object_usecase.rs
-80
View File
@@ -693,10 +693,6 @@ impl DefaultObjectUsecase {
#[instrument(level = "debug", skip(self, _fs, req))]
pub async fn execute_put_object(&self, _fs: &FS, req: S3Request<PutObjectInput>) -> S3Result<S3Response<PutObjectOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let request_context = PutObjectRequestContext {
headers: req.headers.clone(),
trailing_headers: req.trailing_headers.clone(),
@@ -747,10 +743,6 @@ impl DefaultObjectUsecase {
}
pub async fn execute_put_object_acl(&self, req: S3Request<PutObjectAclInput>) -> S3Result<S3Response<PutObjectAclOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let mut helper = OperationHelper::new(&req, EventName::ObjectAclPut, S3Operation::PutObjectAcl);
let PutObjectAclInput {
bucket,
@@ -790,10 +782,6 @@ impl DefaultObjectUsecase {
&self,
req: S3Request<PutObjectLegalHoldInput>,
) -> S3Result<S3Response<PutObjectLegalHoldOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let mut helper =
OperationHelper::new(&req, EventName::ObjectCreatedPutLegalHold, S3Operation::PutObjectLegalHold).suppress_event();
let PutObjectLegalHoldInput {
@@ -849,10 +837,6 @@ impl DefaultObjectUsecase {
&self,
req: S3Request<PutObjectLockConfigurationInput>,
) -> S3Result<S3Response<PutObjectLockConfigurationOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let PutObjectLockConfigurationInput {
bucket,
object_lock_configuration,
@@ -922,10 +906,6 @@ impl DefaultObjectUsecase {
&self,
req: S3Request<PutObjectRetentionInput>,
) -> S3Result<S3Response<PutObjectRetentionOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let mut helper =
OperationHelper::new(&req, EventName::ObjectCreatedPutRetention, S3Operation::PutObjectRetention).suppress_event();
let PutObjectRetentionInput {
@@ -1015,10 +995,6 @@ impl DefaultObjectUsecase {
&self,
req: S3Request<PutObjectTaggingInput>,
) -> S3Result<S3Response<PutObjectTaggingOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let start_time = std::time::Instant::now();
let mut helper = OperationHelper::new(&req, EventName::ObjectTaggingPut, S3Operation::PutObjectTagging);
let PutObjectTaggingInput {
@@ -1097,10 +1073,6 @@ impl DefaultObjectUsecase {
fields(start_time=?time::OffsetDateTime::now_utc())
)]
pub async fn execute_get_object(&self, req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let request_id = req
.extensions
.get::<request_context::RequestContext>()
@@ -1183,10 +1155,6 @@ impl DefaultObjectUsecase {
}
pub async fn execute_get_object_acl(&self, req: S3Request<GetObjectAclInput>) -> S3Result<S3Response<GetObjectAclOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let GetObjectAclInput {
bucket, key, version_id, ..
} = req.input;
@@ -1207,10 +1175,6 @@ impl DefaultObjectUsecase {
&self,
req: S3Request<GetObjectAttributesInput>,
) -> S3Result<S3Response<GetObjectAttributesOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let mut helper =
OperationHelper::new(&req, EventName::ObjectAccessedAttributes, S3Operation::GetObjectAttributes).suppress_event();
let GetObjectAttributesInput {
@@ -1440,10 +1404,6 @@ impl DefaultObjectUsecase {
&self,
req: S3Request<GetObjectLegalHoldInput>,
) -> S3Result<S3Response<GetObjectLegalHoldOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let mut helper =
OperationHelper::new(&req, EventName::ObjectAccessedGetLegalHold, S3Operation::GetObjectLegalHold).suppress_event();
let GetObjectLegalHoldInput {
@@ -1500,10 +1460,6 @@ impl DefaultObjectUsecase {
&self,
req: S3Request<GetObjectLockConfigurationInput>,
) -> S3Result<S3Response<GetObjectLockConfigurationOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let GetObjectLockConfigurationInput { bucket, .. } = req.input;
let object_lock_configuration = match metadata_sys::get_object_lock_config(&bucket).await {
@@ -1532,10 +1488,6 @@ impl DefaultObjectUsecase {
&self,
req: S3Request<GetObjectRetentionInput>,
) -> S3Result<S3Response<GetObjectRetentionOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let mut helper =
OperationHelper::new(&req, EventName::ObjectAccessedGetRetention, S3Operation::GetObjectRetention).suppress_event();
let GetObjectRetentionInput {
@@ -1584,10 +1536,6 @@ impl DefaultObjectUsecase {
&self,
req: S3Request<GetObjectTaggingInput>,
) -> S3Result<S3Response<GetObjectTaggingOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let start_time = std::time::Instant::now();
let GetObjectTaggingInput { bucket, key: object, .. } = req.input;
@@ -1627,10 +1575,6 @@ impl DefaultObjectUsecase {
#[instrument(level = "debug", skip(self, req))]
pub async fn execute_copy_object(&self, req: S3Request<CopyObjectInput>) -> S3Result<S3Response<CopyObjectOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let mut helper = OperationHelper::new(&req, EventName::ObjectCreatedCopy, S3Operation::CopyObject);
let CopyObjectInput {
copy_source,
@@ -1939,10 +1883,6 @@ impl DefaultObjectUsecase {
&self,
mut req: S3Request<DeleteObjectsInput>,
) -> S3Result<S3Response<DeleteObjectsOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let helper = OperationHelper::new(&req, EventName::ObjectRemovedDelete, S3Operation::DeleteObjects).suppress_event();
let (bucket, delete) = {
let bucket = req.input.bucket.clone();
@@ -2244,10 +2184,6 @@ impl DefaultObjectUsecase {
#[instrument(level = "debug", skip(self, req))]
pub async fn execute_delete_object(&self, mut req: S3Request<DeleteObjectInput>) -> S3Result<S3Response<DeleteObjectOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let mut helper = OperationHelper::new(&req, EventName::ObjectRemovedDelete, S3Operation::DeleteObject);
let DeleteObjectInput {
bucket, key, version_id, ..
@@ -2444,10 +2380,6 @@ impl DefaultObjectUsecase {
&self,
req: S3Request<DeleteObjectTaggingInput>,
) -> S3Result<S3Response<DeleteObjectTaggingOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let start_time = std::time::Instant::now();
let mut helper = OperationHelper::new(&req, EventName::ObjectTaggingDelete, S3Operation::DeleteObjectTagging);
let DeleteObjectTaggingInput {
@@ -2513,10 +2445,6 @@ impl DefaultObjectUsecase {
#[instrument(level = "debug", skip(self, req))]
pub async fn execute_head_object(&self, req: S3Request<HeadObjectInput>) -> S3Result<S3Response<HeadObjectOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let mut helper = OperationHelper::new(&req, EventName::ObjectAccessedHead, S3Operation::HeadObject).suppress_event();
// mc get 2
let HeadObjectInput {
@@ -2839,10 +2767,6 @@ impl DefaultObjectUsecase {
#[instrument(level = "debug", skip(self, req))]
pub async fn execute_restore_object(&self, req: S3Request<RestoreObjectInput>) -> S3Result<S3Response<RestoreObjectOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
let mut helper = OperationHelper::new(&req, EventName::ObjectRestorePost, S3Operation::RestoreObject);
let RestoreObjectInput {
bucket,
@@ -3032,10 +2956,6 @@ impl DefaultObjectUsecase {
&self,
req: S3Request<SelectObjectContentInput>,
) -> S3Result<S3Response<SelectObjectContentOutput>> {
if let Some(context) = &self.context {
let _ = context.object_store();
}
info!("handle select_object_content");
let input = Arc::new(req.input);
rustfs/src/storage/ecfs_test.rs
-45
View File
@@ -364,51 +364,6 @@ mod tests {
set_buffer_profile_enabled(false);
}
#[test]
fn test_phase5_s3_entrypoints_delegate_to_usecases() {
fn assert_delegates_within_method(src: &str, signature: &str, delegation_call: &str, error_msg: &str) {
let sig_pos = src
.find(signature)
.unwrap_or_else(|| panic!("Expected to find method signature: {signature}"));
let after_sig = &src[sig_pos + signature.len()..];
let method_body_end_rel = after_sig.find("async fn ").unwrap_or(after_sig.len());
let method_body = &after_sig[..method_body_end_rel];
assert!(method_body.contains(delegation_call), "{error_msg}");
}
let src = include_str!("ecfs.rs");
assert_delegates_within_method(
src,
"async fn put_object(&self, req: S3Request<PutObjectInput>)",
"usecase.execute_put_object(self, req).await",
"put_object must delegate to DefaultObjectUsecase::execute_put_object",
);
assert_delegates_within_method(
src,
"async fn get_object(&self, req: S3Request<GetObjectInput>)",
"usecase.execute_get_object(req).await",
"get_object must delegate to DefaultObjectUsecase::execute_get_object",
);
assert_delegates_within_method(
src,
"async fn list_objects_v2(&self, req: S3Request<ListObjectsV2Input>)",
"usecase.execute_list_objects_v2(req).await",
"list_objects_v2 must delegate to DefaultBucketUsecase::execute_list_objects_v2",
);
assert_delegates_within_method(
src,
"async fn list_objects_v2m(&self, req: S3Request<ListObjectsV2Input>)",
"usecase.execute_list_objects_v2m(req).await",
"list_objects_v2m must delegate to DefaultBucketUsecase::execute_list_objects_v2m",
);
}
#[test]
fn test_validate_list_object_unordered_with_delimiter() {
// [1] Normal case: No delimiter specified.
rustfs/src/storage/entity.rs
-62
View File
@@ -1,62 +0,0 @@
// 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.
use s3s::dto::{
BucketKeyEnabled, BucketName, ChecksumCRC32, ChecksumCRC32C, ChecksumCRC64NVME, ChecksumSHA1, ChecksumSHA256, ChecksumType,
ETag, Expiration, Location, ObjectKey, ObjectVersionId, RequestCharged, SSEKMSKeyId, ServerSideEncryption,
};
#[allow(dead_code)]
#[derive(Debug, Clone, Default)]
pub struct CompleteMultipartUploadOutput {
pub bucket: Option<BucketName>,
pub bucket_key_enabled: Option<BucketKeyEnabled>,
pub checksum_crc32: Option<ChecksumCRC32>,
pub checksum_crc32c: Option<ChecksumCRC32C>,
pub checksum_crc64nvme: Option<ChecksumCRC64NVME>,
pub checksum_sha1: Option<ChecksumSHA1>,
pub checksum_sha256: Option<ChecksumSHA256>,
pub checksum_type: Option<ChecksumType>,
pub e_tag: Option<ETag>,
pub expiration: Option<Expiration>,
pub key: Option<ObjectKey>,
pub location: Option<Location>,
pub request_charged: Option<RequestCharged>,
pub ssekms_key_id: Option<SSEKMSKeyId>,
pub server_side_encryption: Option<ServerSideEncryption>,
pub version_id: Option<ObjectVersionId>,
}
impl From<s3s::dto::CompleteMultipartUploadOutput> for CompleteMultipartUploadOutput {
fn from(output: s3s::dto::CompleteMultipartUploadOutput) -> Self {
Self {
bucket: output.bucket,
bucket_key_enabled: output.bucket_key_enabled,
checksum_crc32: output.checksum_crc32,
checksum_crc32c: output.checksum_crc32c,
checksum_crc64nvme: output.checksum_crc64nvme,
checksum_sha1: output.checksum_sha1,
checksum_sha256: output.checksum_sha256,
checksum_type: output.checksum_type,
e_tag: output.e_tag,
expiration: output.expiration,
key: output.key,
location: output.location,
request_charged: output.request_charged,
ssekms_key_id: output.ssekms_key_id,
server_side_encryption: output.server_side_encryption,
version_id: output.version_id,
}
}
}
rustfs/src/storage/helper.rs
+1 -1
View File
@@ -213,7 +213,7 @@ impl OperationHelper {
/// Complete operational details from S3 results.
/// This method should be called immediately before the function returns.
/// It consumes and prepares auxiliary structures for use during `drop`.
pub fn complete(mut self, result: &S3Result<S3Response<impl Send + Sync>>) -> Self {
pub fn complete<T>(mut self, result: &S3Result<S3Response<T>>) -> Self {
// Complete audit log
if let Some(builder) = self.audit_builder.take() {
let (status, status_code, error_msg) = match result {
rustfs/src/storage/mod.rs
-1
View File
@@ -17,7 +17,6 @@ pub mod backpressure;
pub mod concurrency;
pub mod deadlock_detector;
pub mod ecfs;
pub(crate) mod entity;
pub(crate) mod helper;
pub mod lock_optimizer;
pub mod options;