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

refactor(app): route metadata/endpoints access through AppContext (#1949)

Browse Source
This commit is contained in:
安正超
2026-02-25 15:07:09 +08:00
committed by GitHub
parent 672c255567
commit d774d6821b
4 changed files with 94 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/app/admin_usecase.rs
+11 -3
View File
@@ -20,9 +20,10 @@ use crate::error::ApiError;
use rustfs_common::data_usage::DataUsageInfo;
use rustfs_ecstore::admin_server_info::get_server_info;
use rustfs_ecstore::data_usage::load_data_usage_from_backend;
use rustfs_ecstore::endpoints::EndpointServerPools;
use rustfs_ecstore::new_object_layer_fn;
use rustfs_ecstore::pools::{PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free};
use rustfs_ecstore::store_api::StorageAPI;
use rustfs_ecstore::{GLOBAL_Endpoints, new_object_layer_fn};
use rustfs_madmin::{InfoMessage, StorageInfo};
use s3s::S3ErrorCode;
use std::sync::Arc;
@@ -96,6 +97,13 @@ impl DefaultAdminUsecase {
self.context.clone()
}
fn endpoints(&self) -> Option<EndpointServerPools> {
self.context
.as_ref()
.and_then(|context| context.endpoints().handle())
.or_else(|| rustfs_ecstore::GLOBAL_Endpoints.get().cloned())
}
fn app_error(code: S3ErrorCode, message: impl Into<String>) -> ApiError {
ApiError {
code,
@@ -221,7 +229,7 @@ impl DefaultAdminUsecase {
return Err(Self::app_error(S3ErrorCode::InternalError, "Not init"));
};
let Some(endpoints) = GLOBAL_Endpoints.get() else {
let Some(endpoints) = self.endpoints() else {
return Err(Self::app_error_default(S3ErrorCode::NotImplemented));
};
@@ -243,7 +251,7 @@ impl DefaultAdminUsecase {
let _ = context.object_store();
}
let Some(endpoints) = GLOBAL_Endpoints.get() else {
let Some(endpoints) = self.endpoints() else {
return Err(Self::app_error_default(S3ErrorCode::NotImplemented));
};
rustfs/src/app/context.rs
+54
View File
@@ -18,12 +18,16 @@
#![allow(dead_code)]
use async_trait::async_trait;
use rustfs_ecstore::GLOBAL_Endpoints;
use rustfs_ecstore::bucket::metadata_sys::{BucketMetadataSys, GLOBAL_BucketMetadataSys};
use rustfs_ecstore::endpoints::EndpointServerPools;
use rustfs_ecstore::store::ECStore;
use rustfs_iam::{store::object::ObjectStore, sys::IamSys};
use rustfs_kms::KmsServiceManager;
use rustfs_notify::{EventArgs, NotificationError, notifier_global};
use rustfs_targets::{EventName, arn::TargetID};
use std::sync::{Arc, OnceLock};
use tokio::sync::RwLock;
/// IAM interface for application-layer use-cases.
pub trait IamInterface: Send + Sync {
@@ -51,6 +55,16 @@ pub trait NotifyInterface: Send + Sync {
async fn clear_bucket_notification_rules(&self, bucket_name: &str) -> Result<(), NotificationError>;
}
/// Bucket metadata interface for application-layer use-cases.
pub trait BucketMetadataInterface: Send + Sync {
fn handle(&self) -> Option<Arc<RwLock<BucketMetadataSys>>>;
}
/// Endpoints interface for application-layer use-cases.
pub trait EndpointsInterface: Send + Sync {
fn handle(&self) -> Option<EndpointServerPools>;
}
/// Default IAM interface adapter.
pub struct IamHandle {
iam: Arc<IamSys<ObjectStore>>,
@@ -113,6 +127,26 @@ impl NotifyInterface for NotifyHandle {
}
}
/// Default bucket metadata interface adapter.
#[derive(Default)]
pub struct BucketMetadataHandle;
impl BucketMetadataInterface for BucketMetadataHandle {
fn handle(&self) -> Option<Arc<RwLock<BucketMetadataSys>>> {
GLOBAL_BucketMetadataSys.get().cloned()
}
}
/// Default endpoints interface adapter.
#[derive(Default)]
pub struct EndpointsHandle;
impl EndpointsInterface for EndpointsHandle {
fn handle(&self) -> Option<EndpointServerPools> {
GLOBAL_Endpoints.get().cloned()
}
}
/// Application-layer context with explicit dependencies.
#[derive(Clone)]
pub struct AppContext {
@@ -120,6 +154,8 @@ pub struct AppContext {
iam: Arc<dyn IamInterface>,
kms: Arc<dyn KmsInterface>,
notify: Arc<dyn NotifyInterface>,
bucket_metadata: Arc<dyn BucketMetadataInterface>,
endpoints: Arc<dyn EndpointsInterface>,
}
impl AppContext {
@@ -129,6 +165,8 @@ impl AppContext {
iam,
kms,
notify: default_notify_interface(),
bucket_metadata: default_bucket_metadata_interface(),
endpoints: default_endpoints_interface(),
}
}
@@ -155,12 +193,28 @@ impl AppContext {
pub fn notify(&self) -> Arc<dyn NotifyInterface> {
self.notify.clone()
}
pub fn bucket_metadata(&self) -> Arc<dyn BucketMetadataInterface> {
self.bucket_metadata.clone()
}
pub fn endpoints(&self) -> Arc<dyn EndpointsInterface> {
self.endpoints.clone()
}
}
pub fn default_notify_interface() -> Arc<dyn NotifyInterface> {
Arc::new(NotifyHandle)
}
pub fn default_bucket_metadata_interface() -> Arc<dyn BucketMetadataInterface> {
Arc::new(BucketMetadataHandle)
}
pub fn default_endpoints_interface() -> Arc<dyn EndpointsInterface> {
Arc::new(EndpointsHandle)
}
static GLOBAL_APP_CONTEXT: OnceLock<Arc<AppContext>> = OnceLock::new();
/// Initialize global application context once and return the canonical instance.
rustfs/src/app/multipart_usecase.rs
+12 -5
View File
@@ -30,6 +30,7 @@ use futures::StreamExt;
use rustfs_ecstore::StorageAPI;
use rustfs_ecstore::bucket::quota::checker::QuotaChecker;
use rustfs_ecstore::bucket::{
metadata_sys,
quota::QuotaOperation,
replication::{get_must_replicate_options, must_replicate, schedule_replication},
};
@@ -52,6 +53,7 @@ use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::Arc;
use tokio::sync::RwLock;
use tokio_util::io::StreamReader;
use tracing::{info, instrument, warn};
@@ -158,6 +160,13 @@ impl DefaultMultipartUsecase {
self.context.clone()
}
fn bucket_metadata_sys(&self) -> Option<Arc<RwLock<metadata_sys::BucketMetadataSys>>> {
self.context
.as_ref()
.and_then(|context| context.bucket_metadata().handle())
.or_else(|| rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get().cloned())
}
#[instrument(level = "debug", skip(self))]
pub async fn execute_abort_multipart_upload(
&self,
@@ -337,8 +346,8 @@ impl DefaultMultipartUsecase {
.map_err(ApiError::from)?;
// check quota after completing multipart upload
if let Some(metadata_sys) = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get() {
let quota_checker = QuotaChecker::new(metadata_sys.clone());
if let Some(metadata_sys) = self.bucket_metadata_sys() {
let quota_checker = QuotaChecker::new(metadata_sys);
match quota_checker
.check_quota(&bucket, QuotaOperation::PutObject, obj_info.size as u64)
@@ -358,9 +367,7 @@ impl DefaultMultipartUsecase {
));
}
// Update quota tracking after successful multipart upload
if rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get().is_some() {
rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, obj_info.size as u64).await;
}
rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, obj_info.size as u64).await;
}
Err(e) => {
warn!("Quota check failed for bucket {}: {}, allowing operation", bucket, e);
rustfs/src/app/object_usecase.rs
+17 -6
View File
@@ -101,6 +101,7 @@ use std::ops::Add;
use std::str::FromStr;
use std::sync::Arc;
use time::{OffsetDateTime, format_description::well_known::Rfc3339};
use tokio::sync::RwLock;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tokio_util::io::{ReaderStream, StreamReader};
@@ -199,6 +200,13 @@ impl DefaultObjectUsecase {
self.context.clone()
}
fn bucket_metadata_sys(&self) -> Option<Arc<RwLock<metadata_sys::BucketMetadataSys>>> {
self.context
.as_ref()
.and_then(|context| context.bucket_metadata().handle())
.or_else(|| rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get().cloned())
}
#[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 {
@@ -258,9 +266,9 @@ impl DefaultObjectUsecase {
// check quota for put operation
if let Some(size) = content_length
&& let Some(metadata_sys) = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get()
&& let Some(metadata_sys) = self.bucket_metadata_sys()
{
let quota_checker = QuotaChecker::new(metadata_sys.clone());
let quota_checker = QuotaChecker::new(metadata_sys);
match quota_checker
.check_quota(&bucket, QuotaOperation::PutObject, size as u64)
@@ -2091,8 +2099,8 @@ impl DefaultObjectUsecase {
}
// check quota for copy operation
if let Some(metadata_sys) = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get() {
let quota_checker = QuotaChecker::new(metadata_sys.clone());
let has_bucket_metadata = if let Some(metadata_sys) = self.bucket_metadata_sys() {
let quota_checker = QuotaChecker::new(metadata_sys);
match quota_checker
.check_quota(&bucket, QuotaOperation::CopyObject, src_info.size as u64)
@@ -2114,7 +2122,10 @@ impl DefaultObjectUsecase {
warn!("Quota check failed for bucket {}: {}, allowing operation", bucket, e);
}
}
}
true
} else {
false
};
let oi = store
.copy_object(&src_bucket, &src_key, &bucket, &key, &mut src_info, &src_opts, &dst_opts)
@@ -2122,7 +2133,7 @@ impl DefaultObjectUsecase {
.map_err(ApiError::from)?;
// Update quota tracking after successful copy
if rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get().is_some() {
if has_bucket_metadata {
rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, oi.size as u64).await;
}