diff --git a/rustfs/src/admin/handlers/pools.rs b/rustfs/src/admin/handlers/pools.rs index a2a5e42dd..5f6839f90 100644 --- a/rustfs/src/admin/handlers/pools.rs +++ b/rustfs/src/admin/handlers/pools.rs @@ -27,12 +27,17 @@ use crate::{ router::{AdminOperation, Operation, S3Router}, }, app::admin_usecase::{DefaultAdminUsecase, QueryPoolStatusRequest}, + app::context::get_global_app_context, auth::{check_key_valid, get_session_token}, error::ApiError, server::{ADMIN_PREFIX, RemoteAddr}, }; use hyper::Method; -use rustfs_ecstore::{GLOBAL_Endpoints, new_object_layer_fn}; +use rustfs_ecstore::new_object_layer_fn; + +fn endpoints_from_context() -> Option { + get_global_app_context().and_then(|context| context.endpoints().handle()) +} pub fn register_pool_route(r: &mut S3Router) -> std::io::Result<()> { r.insert( @@ -196,7 +201,7 @@ impl Operation for StartDecommission { ) .await?; - let Some(endpoints) = GLOBAL_Endpoints.get() else { + let Some(endpoints) = endpoints_from_context() else { return Err(s3_error!(NotImplemented)); }; @@ -295,7 +300,7 @@ impl Operation for CancelDecommission { ) .await?; - let Some(endpoints) = GLOBAL_Endpoints.get() else { + let Some(endpoints) = endpoints_from_context() else { return Err(s3_error!(NotImplemented)); }; diff --git a/rustfs/src/admin/handlers/quota.rs b/rustfs/src/admin/handlers/quota.rs index 544f20c4b..623bae8a7 100644 --- a/rustfs/src/admin/handlers/quota.rs +++ b/rustfs/src/admin/handlers/quota.rs @@ -16,16 +16,20 @@ use crate::admin::auth::{validate_admin_request, validate_admin_request_with_bucket}; use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::app::context::get_global_app_context; use crate::auth::{check_key_valid, get_session_token}; use crate::server::ADMIN_PREFIX; use hyper::{Method, StatusCode}; use matchit::Params; +use rustfs_ecstore::bucket::metadata_sys::BucketMetadataSys; use rustfs_ecstore::bucket::quota::checker::QuotaChecker; use rustfs_ecstore::bucket::quota::{BucketQuota, QuotaError, QuotaOperation}; use rustfs_policy::policy::action::{Action, AdminAction, S3Action}; use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; use serde_json; +use std::sync::Arc; +use tokio::sync::RwLock; use tracing::{debug, info, warn}; #[derive(Debug, Deserialize)] @@ -81,6 +85,25 @@ pub struct ClearBucketQuotaHandler; pub struct GetBucketQuotaStatsHandler; pub struct CheckBucketQuotaHandler; +fn bucket_metadata_from_context() -> Option>> { + get_global_app_context().and_then(|context| context.bucket_metadata().handle()) +} + +async fn current_usage_from_context(bucket: &str) -> u64 { + let Some(store) = get_global_app_context().map(|context| context.object_store()) else { + return 0; + }; + + match rustfs_ecstore::data_usage::load_data_usage_from_backend(store).await { + Ok(data_usage_info) => data_usage_info + .buckets_usage + .get(bucket) + .map(|bucket_usage| bucket_usage.size) + .unwrap_or(0), + Err(_) => 0, + } +} + pub fn register_quota_route(r: &mut S3Router) -> std::io::Result<()> { r.insert( Method::PUT, @@ -164,8 +187,7 @@ impl Operation for SetBucketQuotaHandler { let quota = BucketQuota::new(request.quota); - let metadata_sys_lock = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys - .get() + let metadata_sys_lock = bucket_metadata_from_context() .ok_or_else(|| s3_error!(InternalError, "{}", rustfs_config::QUOTA_METADATA_SYSTEM_ERROR_MSG))?; let mut quota_checker = QuotaChecker::new(metadata_sys_lock.clone()); @@ -175,18 +197,7 @@ impl Operation for SetBucketQuotaHandler { .map_err(|e| s3_error!(InternalError, "Failed to set quota: {}", e))?; // Get real-time usage from data usage system - let current_usage = if let Some(store) = rustfs_ecstore::global::GLOBAL_OBJECT_API.get() { - match rustfs_ecstore::data_usage::load_data_usage_from_backend(store.clone()).await { - Ok(data_usage_info) => data_usage_info - .buckets_usage - .get(&bucket) - .map(|bucket_usage| bucket_usage.size) - .unwrap_or(0), - Err(_) => 0, - } - } else { - 0 - }; + let current_usage = current_usage_from_context(&bucket).await; let response = BucketQuotaResponse { bucket, @@ -231,9 +242,8 @@ impl Operation for GetBucketQuotaHandler { ) .await?; - let metadata_sys_lock = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys - .get() - .ok_or_else(|| s3_error!(InternalError, "Bucket metadata system not initialized"))?; + let metadata_sys_lock = + bucket_metadata_from_context().ok_or_else(|| s3_error!(InternalError, "Bucket metadata system not initialized"))?; let quota_checker = QuotaChecker::new(metadata_sys_lock.clone()); @@ -288,9 +298,8 @@ impl Operation for ClearBucketQuotaHandler { info!("Clearing quota for bucket: {}", bucket); - let metadata_sys_lock = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys - .get() - .ok_or_else(|| s3_error!(InternalError, "Bucket metadata system not initialized"))?; + let metadata_sys_lock = + bucket_metadata_from_context().ok_or_else(|| s3_error!(InternalError, "Bucket metadata system not initialized"))?; let mut quota_checker = QuotaChecker::new(metadata_sys_lock.clone()); @@ -304,18 +313,7 @@ impl Operation for ClearBucketQuotaHandler { info!("Successfully cleared quota for bucket: {}", bucket); // Get real-time usage from data usage system - let current_usage = if let Some(store) = rustfs_ecstore::global::GLOBAL_OBJECT_API.get() { - match rustfs_ecstore::data_usage::load_data_usage_from_backend(store.clone()).await { - Ok(data_usage_info) => data_usage_info - .buckets_usage - .get(&bucket) - .map(|bucket_usage| bucket_usage.size) - .unwrap_or(0), - Err(_) => 0, - } - } else { - 0 - }; + let current_usage = current_usage_from_context(&bucket).await; let response = BucketQuotaResponse { bucket, @@ -360,9 +358,8 @@ impl Operation for GetBucketQuotaStatsHandler { ) .await?; - let metadata_sys_lock = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys - .get() - .ok_or_else(|| s3_error!(InternalError, "Bucket metadata system not initialized"))?; + let metadata_sys_lock = + bucket_metadata_from_context().ok_or_else(|| s3_error!(InternalError, "Bucket metadata system not initialized"))?; let quota_checker = QuotaChecker::new(metadata_sys_lock.clone()); @@ -445,9 +442,8 @@ impl Operation for CheckBucketQuotaHandler { bucket, request.operation_type, request.operation_size ); - let metadata_sys_lock = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys - .get() - .ok_or_else(|| s3_error!(InternalError, "Bucket metadata system not initialized"))?; + let metadata_sys_lock = + bucket_metadata_from_context().ok_or_else(|| s3_error!(InternalError, "Bucket metadata system not initialized"))?; let quota_checker = QuotaChecker::new(metadata_sys_lock.clone()); diff --git a/rustfs/src/admin/handlers/trace.rs b/rustfs/src/admin/handlers/trace.rs index 1b9577a12..777330782 100644 --- a/rustfs/src/admin/handlers/trace.rs +++ b/rustfs/src/admin/handlers/trace.rs @@ -13,10 +13,11 @@ // limitations under the License. use crate::admin::router::Operation; +use crate::app::context::get_global_app_context; use http::StatusCode; use hyper::Uri; use matchit::Params; -use rustfs_ecstore::{GLOBAL_Endpoints, rpc::PeerRestClient}; +use rustfs_ecstore::rpc::PeerRestClient; use rustfs_madmin::service_commands::ServiceTraceOpts; use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use tracing::warn; @@ -42,7 +43,7 @@ impl Operation for Trace { let _trace_opts = extract_trace_options(&req.uri)?; // let (tx, rx) = mpsc::channel(10000); - let _peers = match GLOBAL_Endpoints.get() { + let _peers = match get_global_app_context().and_then(|context| context.endpoints().handle()) { Some(ep) => PeerRestClient::new_clients(ep.clone()).await, None => (Vec::new(), Vec::new()), }; diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 085555b02..b02c799ea 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -165,6 +165,10 @@ impl DefaultBucketUsecase { self.context.clone() } + fn global_region(&self) -> Option { + self.context.as_ref().and_then(|context| context.region().get()) + } + #[instrument( level = "debug", skip(self, req), @@ -425,7 +429,7 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; - if let Some(region) = rustfs_ecstore::global::get_global_region() { + if let Some(region) = self.global_region() { return Ok(S3Response::new(GetBucketLocationOutput { location_constraint: Some(BucketLocationConstraint::from(region)), })); @@ -1187,7 +1191,7 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; - let region = resolve_notification_region(rustfs_ecstore::global::get_global_region(), request_region); + let region = resolve_notification_region(self.global_region(), request_region); let notify = self .context .as_ref() diff --git a/rustfs/src/app/context.rs b/rustfs/src/app/context.rs index 74d59e6fc..5c3d0aa55 100644 --- a/rustfs/src/app/context.rs +++ b/rustfs/src/app/context.rs @@ -21,6 +21,7 @@ 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::global::get_global_region; use rustfs_ecstore::store::ECStore; use rustfs_iam::{store::object::ObjectStore, sys::IamSys}; use rustfs_kms::KmsServiceManager; @@ -65,6 +66,11 @@ pub trait EndpointsInterface: Send + Sync { fn handle(&self) -> Option; } +/// Region interface for application-layer use-cases. +pub trait RegionInterface: Send + Sync { + fn get(&self) -> Option; +} + /// Default IAM interface adapter. pub struct IamHandle { iam: Arc>, @@ -147,6 +153,16 @@ impl EndpointsInterface for EndpointsHandle { } } +/// Default region interface adapter. +#[derive(Default)] +pub struct RegionHandle; + +impl RegionInterface for RegionHandle { + fn get(&self) -> Option { + get_global_region() + } +} + /// Application-layer context with explicit dependencies. #[derive(Clone)] pub struct AppContext { @@ -156,6 +172,7 @@ pub struct AppContext { notify: Arc, bucket_metadata: Arc, endpoints: Arc, + region: Arc, } impl AppContext { @@ -167,6 +184,7 @@ impl AppContext { notify: default_notify_interface(), bucket_metadata: default_bucket_metadata_interface(), endpoints: default_endpoints_interface(), + region: default_region_interface(), } } @@ -201,6 +219,10 @@ impl AppContext { pub fn endpoints(&self) -> Arc { self.endpoints.clone() } + + pub fn region(&self) -> Arc { + self.region.clone() + } } pub fn default_notify_interface() -> Arc { @@ -215,6 +237,10 @@ pub fn default_endpoints_interface() -> Arc { Arc::new(EndpointsHandle) } +pub fn default_region_interface() -> Arc { + Arc::new(RegionHandle) +} + static GLOBAL_APP_CONTEXT: OnceLock> = OnceLock::new(); /// Initialize global application context once and return the canonical instance. diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 0690d3deb..24cc209dd 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -164,6 +164,10 @@ impl DefaultMultipartUsecase { self.context.as_ref().and_then(|context| context.bucket_metadata().handle()) } + fn global_region(&self) -> Option { + self.context.as_ref().and_then(|context| context.region().get()) + } + #[instrument(level = "debug", skip(self))] pub async fn execute_abort_multipart_upload( &self, @@ -418,7 +422,7 @@ impl DefaultMultipartUsecase { } } - let region = rustfs_ecstore::global::get_global_region().unwrap_or_else(|| "us-east-1".to_string()); + let region = self.global_region().unwrap_or_else(|| "us-east-1".to_string()); let output = CompleteMultipartUploadOutput { bucket: Some(bucket.clone()), key: Some(key.clone()),