mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 08:18:18 +00:00
refactor(app): converge lower-priority global reads via AppContext (#1958)
This commit is contained in:
@@ -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<rustfs_ecstore::endpoints::EndpointServerPools> {
|
||||
get_global_app_context().and_then(|context| context.endpoints().handle())
|
||||
}
|
||||
|
||||
pub fn register_pool_route(r: &mut S3Router<AdminOperation>) -> 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));
|
||||
};
|
||||
|
||||
|
||||
@@ -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<Arc<RwLock<BucketMetadataSys>>> {
|
||||
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<AdminOperation>) -> 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());
|
||||
|
||||
|
||||
@@ -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()),
|
||||
};
|
||||
|
||||
@@ -165,6 +165,10 @@ impl DefaultBucketUsecase {
|
||||
self.context.clone()
|
||||
}
|
||||
|
||||
fn global_region(&self) -> Option<String> {
|
||||
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()
|
||||
|
||||
@@ -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<EndpointServerPools>;
|
||||
}
|
||||
|
||||
/// Region interface for application-layer use-cases.
|
||||
pub trait RegionInterface: Send + Sync {
|
||||
fn get(&self) -> Option<String>;
|
||||
}
|
||||
|
||||
/// Default IAM interface adapter.
|
||||
pub struct IamHandle {
|
||||
iam: Arc<IamSys<ObjectStore>>,
|
||||
@@ -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<String> {
|
||||
get_global_region()
|
||||
}
|
||||
}
|
||||
|
||||
/// Application-layer context with explicit dependencies.
|
||||
#[derive(Clone)]
|
||||
pub struct AppContext {
|
||||
@@ -156,6 +172,7 @@ pub struct AppContext {
|
||||
notify: Arc<dyn NotifyInterface>,
|
||||
bucket_metadata: Arc<dyn BucketMetadataInterface>,
|
||||
endpoints: Arc<dyn EndpointsInterface>,
|
||||
region: Arc<dyn RegionInterface>,
|
||||
}
|
||||
|
||||
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<dyn EndpointsInterface> {
|
||||
self.endpoints.clone()
|
||||
}
|
||||
|
||||
pub fn region(&self) -> Arc<dyn RegionInterface> {
|
||||
self.region.clone()
|
||||
}
|
||||
}
|
||||
|
||||
pub fn default_notify_interface() -> Arc<dyn NotifyInterface> {
|
||||
@@ -215,6 +237,10 @@ pub fn default_endpoints_interface() -> Arc<dyn EndpointsInterface> {
|
||||
Arc::new(EndpointsHandle)
|
||||
}
|
||||
|
||||
pub fn default_region_interface() -> Arc<dyn RegionInterface> {
|
||||
Arc::new(RegionHandle)
|
||||
}
|
||||
|
||||
static GLOBAL_APP_CONTEXT: OnceLock<Arc<AppContext>> = OnceLock::new();
|
||||
|
||||
/// Initialize global application context once and return the canonical instance.
|
||||
|
||||
@@ -164,6 +164,10 @@ impl DefaultMultipartUsecase {
|
||||
self.context.as_ref().and_then(|context| context.bucket_metadata().handle())
|
||||
}
|
||||
|
||||
fn global_region(&self) -> Option<String> {
|
||||
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()),
|
||||
|
||||
Reference in New Issue
Block a user