gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-11 23:56:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: isolate specialized runtime fallbacks (#3958)

Browse Source
This commit is contained in:
Zhengchao An
2026-06-27 21:35:47 +08:00
committed by GitHub
parent 672f6e9ea9
commit 68d5d1d41d
3 changed files with 144 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/app/context.rs
+36 -60
View File
@@ -51,9 +51,8 @@ pub fn resolve_kms_runtime_service_manager() -> Option<Arc<KmsServiceManager>> {
}
/// Resolve or initialize the KMS runtime service manager using AppContext-first precedence.
pub fn resolve_or_init_kms_runtime_service_manager() -> Arc<KmsServiceManager> {
pub fn resolve_or_init_kms_runtime_service_manager() -> Option<Arc<KmsServiceManager>> {
resolve_or_init_kms_runtime_service_manager_with(get_global_app_context())
.unwrap_or_else(runtime_sources::init_kms_service_manager)
}
/// Resolve KMS encryption service using AppContext-first precedence.
@@ -72,7 +71,7 @@ pub(crate) fn set_test_outbound_tls_generation(generation: u64) {
}
/// Resolve outbound TLS state using AppContext-first precedence.
pub async fn resolve_outbound_tls_state() -> GlobalPublishedOutboundTlsState {
pub async fn resolve_outbound_tls_state() -> Option<GlobalPublishedOutboundTlsState> {
resolve_outbound_tls_state_with(get_global_app_context()).await
}
@@ -127,16 +126,14 @@ pub fn resolve_object_store_handle_for_context(context: Option<&AppContext>) ->
}
/// Resolve notify interface using AppContext-first precedence.
pub fn resolve_notify_interface() -> Arc<dyn NotifyInterface> {
pub fn resolve_notify_interface() -> Option<Arc<dyn NotifyInterface>> {
let context = get_global_app_context();
resolve_notify_interface_for_context(context.as_deref())
}
/// Resolve notify interface using an explicit AppContext, falling back to the legacy global notifier.
pub fn resolve_notify_interface_for_context(context: Option<&AppContext>) -> Arc<dyn NotifyInterface> {
context
.map(|context| context.notify())
.unwrap_or_else(default_notify_interface)
/// Resolve notify interface using an explicit AppContext.
pub fn resolve_notify_interface_for_context(context: Option<&AppContext>) -> Option<Arc<dyn NotifyInterface>> {
context.map(|context| context.notify())
}
/// Resolve notification system handle using AppContext-first precedence.
@@ -180,12 +177,8 @@ pub fn resolve_daily_tier_stats() -> Option<DailyAllTierStats> {
}
/// Resolve scanner metrics report using AppContext-first precedence.
pub async fn resolve_scanner_metrics_report() -> ScannerMetricsReport {
if let Some(report) = resolve_scanner_metrics_report_with(get_global_app_context()).await {
return report;
}
default_scanner_metrics_interface().report().await
pub async fn resolve_scanner_metrics_report() -> Option<ScannerMetricsReport> {
resolve_scanner_metrics_report_with(get_global_app_context()).await
}
/// Resolve deployment identity using AppContext-first precedence.
@@ -222,12 +215,12 @@ pub fn resolve_internode_metrics() -> Option<Arc<InternodeMetrics>> {
pub async fn resolve_s3select_db(
input: SelectObjectContentInput,
enable_debug: bool,
) -> QueryResult<Arc<dyn DatabaseManagerSystem + Send + Sync>> {
) -> Option<QueryResult<Arc<dyn DatabaseManagerSystem + Send + Sync>>> {
if let Some(context) = get_global_app_context() {
return resolve_s3select_db_with(context, input, enable_debug).await;
return Some(resolve_s3select_db_with(context, input, enable_debug).await);
}
default_s3select_db_interface().get(input, enable_debug).await
None
}
/// Resolve local node name using AppContext-first precedence.
@@ -266,15 +259,13 @@ pub fn resolve_server_config_for_context(context: Option<&AppContext>) -> Option
}
/// Publish server config using AppContext-first precedence.
pub fn publish_server_config(config: Config) {
publish_server_config_with(get_global_app_context(), config, |config| default_server_config_interface().set(config));
pub fn publish_server_config(config: Config) -> bool {
publish_server_config_with(get_global_app_context(), config)
}
/// Publish storage class config using AppContext-first precedence.
pub fn publish_storage_class_config(config: StorageClassConfig) {
publish_storage_class_config_with(get_global_app_context(), config, |config| {
default_storage_class_interface().set(config);
});
pub fn publish_storage_class_config(config: StorageClassConfig) -> bool {
publish_storage_class_config_with(get_global_app_context(), config)
}
/// Resolve buffer profile config using AppContext-first precedence.
@@ -299,12 +290,12 @@ fn resolve_outbound_tls_generation_with(context: Option<Arc<AppContext>>) -> Opt
context.map(|context| context.outbound_tls_runtime().generation())
}
async fn resolve_outbound_tls_state_with(context: Option<Arc<AppContext>>) -> GlobalPublishedOutboundTlsState {
async fn resolve_outbound_tls_state_with(context: Option<Arc<AppContext>>) -> Option<GlobalPublishedOutboundTlsState> {
if let Some(context) = context {
return context.outbound_tls_runtime().state().await;
return Some(context.outbound_tls_runtime().state().await);
}
default_outbound_tls_runtime_interface().state().await
None
}
fn resolve_iam_ready_with(context: Option<Arc<AppContext>>) -> Option<bool> {
@@ -440,24 +431,22 @@ fn resolve_server_config_with(context: Option<Arc<AppContext>>) -> Option<Config
context.and_then(|context| context.server_config().get())
}
fn publish_server_config_with(context: Option<Arc<AppContext>>, config: Config, fallback: impl FnOnce(Config)) {
fn publish_server_config_with(context: Option<Arc<AppContext>>, config: Config) -> bool {
if let Some(context) = context {
context.server_config().set(config);
} else {
fallback(config);
return true;
}
false
}
fn publish_storage_class_config_with(
context: Option<Arc<AppContext>>,
config: StorageClassConfig,
fallback: impl FnOnce(StorageClassConfig),
) {
fn publish_storage_class_config_with(context: Option<Arc<AppContext>>, config: StorageClassConfig) -> bool {
if let Some(context) = context {
context.storage_class().set(config);
} else {
fallback(config);
return true;
}
false
}
fn resolve_buffer_config_with(context: Option<Arc<AppContext>>) -> Option<RustFSBufferConfig> {
@@ -904,9 +893,7 @@ mod tests {
let context_expiry_state = ExpiryState::new();
let server_config = Config::new();
let context_server_config_published = Arc::new(AtomicUsize::new(0));
let fallback_server_config_published = Arc::new(AtomicUsize::new(0));
let context_storage_class_published = Arc::new(AtomicUsize::new(0));
let fallback_storage_class_published = Arc::new(AtomicUsize::new(0));
let buffer_config = RustFSBufferConfig::new(WorkloadProfile::AiTraining);
let context_lock_client: Arc<dyn LockClient> = Arc::new(LocalClient::new());
let context_node_name = "context-node".to_string();
@@ -1038,7 +1025,10 @@ mod tests {
context_outbound_tls_state.generation
);
assert_eq!(
resolve_outbound_tls_state_with(Some(context.clone())).await.generation,
resolve_outbound_tls_state_with(Some(context.clone()))
.await
.expect("context outbound TLS state")
.generation,
context_outbound_tls_state.generation
);
assert!(resolve_iam_ready_with(Some(context.clone())).expect("context IAM ready"));
@@ -1144,18 +1134,10 @@ mod tests {
resolve_server_config_with(Some(context.clone())).expect("context server config"),
server_config
);
publish_server_config_with(Some(context.clone()), Config::new(), |config| {
drop(config);
fallback_server_config_published.fetch_add(1, Ordering::SeqCst);
});
assert!(publish_server_config_with(Some(context.clone()), Config::new()));
assert_eq!(context_server_config_published.load(Ordering::SeqCst), 1);
assert_eq!(fallback_server_config_published.load(Ordering::SeqCst), 0);
publish_storage_class_config_with(Some(context.clone()), StorageClassConfig::default(), |config| {
drop(config);
fallback_storage_class_published.fetch_add(1, Ordering::SeqCst);
});
assert!(publish_storage_class_config_with(Some(context.clone()), StorageClassConfig::default()));
assert_eq!(context_storage_class_published.load(Ordering::SeqCst), 1);
assert_eq!(fallback_storage_class_published.load(Ordering::SeqCst), 0);
assert_eq!(
resolve_buffer_config_with(Some(context))
.expect("context buffer config")
@@ -1166,10 +1148,12 @@ mod tests {
assert!(resolve_kms_runtime_service_manager_with(None).is_none());
assert!(resolve_or_init_kms_runtime_service_manager_with(None).is_none());
assert!(resolve_outbound_tls_generation_with(None).is_none());
assert!(resolve_outbound_tls_state_with(None).await.is_none());
assert!(resolve_iam_ready_with(None).is_none());
assert!(resolve_iam_handle_with(None).is_none());
assert!(resolve_oidc_handle_with(None).is_none());
assert!(resolve_token_signing_key_with(None).is_none());
assert!(resolve_notify_interface_for_context(None).is_none());
assert!(!publish_oidc_handle_with(None, context_oidc));
assert!(resolve_bucket_metadata_handle_with(None).is_none());
assert!(resolve_bucket_monitor_handle_with(None).is_none());
@@ -1192,16 +1176,8 @@ mod tests {
assert!(resolve_tier_config_handle_with(None).is_none());
assert!(resolve_expiry_state_handle_with(None).is_none());
assert!(resolve_server_config_with(None).is_none());
publish_server_config_with(None, Config::new(), |config| {
drop(config);
fallback_server_config_published.fetch_add(1, Ordering::SeqCst);
});
assert_eq!(fallback_server_config_published.load(Ordering::SeqCst), 1);
publish_storage_class_config_with(None, StorageClassConfig::default(), |config| {
drop(config);
fallback_storage_class_published.fetch_add(1, Ordering::SeqCst);
});
assert_eq!(fallback_storage_class_published.load(Ordering::SeqCst), 1);
assert!(!publish_server_config_with(None, Config::new()));
assert!(!publish_storage_class_config_with(None, StorageClassConfig::default()));
assert!(resolve_buffer_config_with(None).is_none());
}
}
rustfs/src/runtime_sources.rs
+60 -10
View File
@@ -13,35 +13,34 @@
// limitations under the License.
use crate::app::context;
use crate::app::storage_api::runtime::{ScannerMetricsReport, StorageClassConfig};
use crate::config::RustFSBufferConfig;
use crate::storage_api::server::runtime_sources::{DailyAllTierStats, ExpiryState, TierConfigMgr};
use rustfs_config::server_config::Config;
use rustfs_io_metrics::{PerformanceMetrics, internode_metrics::InternodeMetrics};
use rustfs_kms::KmsServiceManager;
use rustfs_s3select_api::{QueryResult, server::dbms::DatabaseManagerSystem};
use rustfs_tls_runtime::TlsGeneration;
use s3s::dto::SelectObjectContentInput;
use std::sync::Arc;
#[cfg(test)]
use std::sync::atomic::{AtomicU64, Ordering};
use tokio::sync::RwLock;
pub(crate) use context::{
AppContext, NotifyInterface, publish_oidc_handle, publish_server_config, publish_storage_class_config,
resolve_action_credentials as current_action_credentials, resolve_boot_time as current_boot_time,
resolve_bucket_metadata_handle as current_bucket_metadata_handle,
AppContext, NotifyInterface, publish_oidc_handle, resolve_action_credentials as current_action_credentials,
resolve_boot_time as current_boot_time, resolve_bucket_metadata_handle as current_bucket_metadata_handle,
resolve_bucket_monitor_handle as current_bucket_monitor_handle, resolve_deployment_id as current_deployment_id,
resolve_encryption_service as current_encryption_service, resolve_endpoints_handle as current_endpoints_handle,
resolve_iam_handle as current_iam_handle, resolve_iam_ready as current_iam_ready,
resolve_kms_runtime_service_manager as current_kms_runtime_service_manager, resolve_lock_client as current_lock_client,
resolve_lock_clients_handle as current_lock_clients_handle, resolve_notification_system as current_notification_system,
resolve_notification_system_for_context as current_notification_system_for_context,
resolve_notify_interface as current_notify_interface,
resolve_notify_interface_for_context as current_notify_interface_for_context,
resolve_object_store_handle as current_object_store_handle,
resolve_object_store_handle_for_context as current_object_store_handle_for_context,
resolve_oidc_handle as current_oidc_handle,
resolve_or_init_kms_runtime_service_manager as current_or_init_kms_runtime_service_manager,
resolve_outbound_tls_state as current_outbound_tls_state, resolve_ready_iam_handle as current_ready_iam_handle,
resolve_oidc_handle as current_oidc_handle, resolve_ready_iam_handle as current_ready_iam_handle,
resolve_region as current_region, resolve_replication_pool_handle as current_replication_pool_handle,
resolve_replication_stats_handle as current_replication_stats_handle, resolve_s3select_db as current_s3select_db,
resolve_scanner_metrics_report as current_scanner_metrics_report, resolve_server_config as current_server_config,
resolve_replication_stats_handle as current_replication_stats_handle, resolve_server_config as current_server_config,
resolve_server_config_for_context as current_server_config_for_context,
resolve_token_signing_key as current_token_signing_key,
};
@@ -59,10 +58,30 @@ pub(crate) fn current_app_context() -> Option<Arc<AppContext>> {
context::get_global_app_context()
}
pub(crate) fn current_or_init_kms_runtime_service_manager() -> Arc<KmsServiceManager> {
context::resolve_or_init_kms_runtime_service_manager().unwrap_or_else(rustfs_kms::init_global_kms_service_manager)
}
pub(crate) fn current_notify_interface() -> Arc<dyn NotifyInterface> {
context::resolve_notify_interface().unwrap_or_else(context::default_notify_interface)
}
pub(crate) fn current_notify_interface_for_context(context: Option<&AppContext>) -> Arc<dyn NotifyInterface> {
context::resolve_notify_interface_for_context(context).unwrap_or_else(context::default_notify_interface)
}
pub(crate) fn current_outbound_tls_generation() -> TlsGeneration {
context::resolve_outbound_tls_generation().unwrap_or_else(empty_outbound_tls_generation)
}
pub(crate) async fn current_outbound_tls_state() -> rustfs_tls_runtime::GlobalPublishedOutboundTlsState {
if let Some(state) = context::resolve_outbound_tls_state().await {
return state;
}
context::default_outbound_tls_runtime_interface().state().await
}
#[cfg(test)]
fn empty_outbound_tls_generation() -> TlsGeneration {
TlsGeneration(TEST_OUTBOUND_TLS_GENERATION.load(Ordering::Relaxed))
@@ -89,6 +108,25 @@ pub(crate) fn current_internode_metrics() -> Arc<InternodeMetrics> {
context::resolve_internode_metrics().unwrap_or_else(|| Arc::new(InternodeMetrics::default()))
}
pub(crate) async fn current_scanner_metrics_report() -> ScannerMetricsReport {
if let Some(report) = context::resolve_scanner_metrics_report().await {
return report;
}
context::default_scanner_metrics_interface().report().await
}
pub(crate) async fn current_s3select_db(
input: SelectObjectContentInput,
enable_debug: bool,
) -> QueryResult<Arc<dyn DatabaseManagerSystem + Send + Sync>> {
if let Some(result) = context::resolve_s3select_db(input.clone(), enable_debug).await {
return result;
}
context::default_s3select_db_interface().get(input, enable_debug).await
}
pub(crate) async fn current_local_node_name() -> String {
context::resolve_local_node_name().await.unwrap_or_default()
}
@@ -104,3 +142,15 @@ pub(crate) fn current_expiry_state_handle() -> Arc<RwLock<ExpiryState>> {
pub(crate) fn current_buffer_config() -> RustFSBufferConfig {
context::resolve_buffer_config().unwrap_or_default()
}
pub(crate) fn publish_server_config(config: Config) {
if !context::publish_server_config(config.clone()) {
context::default_server_config_interface().set(config);
}
}
pub(crate) fn publish_storage_class_config(config: StorageClassConfig) {
if !context::publish_storage_class_config(config.clone()) {
context::default_storage_class_interface().set(config);
}
}