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

refactor: route site replication tls through app context (#3778)

Browse Source
This commit is contained in:
Zhengchao An
2026-06-23 12:56:54 +08:00
committed by GitHub
parent a6878e8fce
commit 08d092562c
7 changed files with 199 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/site_replication.rs
+6 -5
View File
@@ -32,8 +32,9 @@ use crate::admin::site_replication_identity::{
};
use crate::admin::utils::{encode_compatible_admin_payload, read_compatible_admin_body};
use crate::app::context::{
resolve_deployment_id, resolve_endpoints_handle, resolve_iam_handle, resolve_object_store_handle, resolve_region,
resolve_replication_pool_handle, resolve_runtime_port, resolve_server_config,
resolve_deployment_id, resolve_endpoints_handle, resolve_iam_handle, resolve_object_store_handle,
resolve_outbound_tls_generation, resolve_outbound_tls_state, resolve_region, resolve_replication_pool_handle,
resolve_runtime_port, resolve_server_config,
};
use crate::auth::{check_key_valid, get_session_token};
use crate::config::get_config_snapshot;
@@ -71,7 +72,7 @@ use rustfs_policy::policy::{
use rustfs_signer::constants::UNSIGNED_PAYLOAD;
use rustfs_signer::sign_v4;
use rustfs_storage_api::{BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions, SRBucketDeleteOp};
use rustfs_tls_runtime::{GlobalPublishedOutboundTlsState, load_global_outbound_tls_generation, load_global_outbound_tls_state};
use rustfs_tls_runtime::GlobalPublishedOutboundTlsState;
use rustfs_utils::http::get_source_scheme;
use rustls_pki_types::pem::PemObject;
use s3s::dto::{
@@ -627,14 +628,14 @@ fn build_site_replication_peer_client(outbound_tls: &GlobalPublishedOutboundTlsS
}
async fn site_replication_peer_client() -> S3Result<reqwest::Client> {
let generation = load_global_outbound_tls_generation().0;
let generation = resolve_outbound_tls_generation().0;
let cache = SITE_REPLICATION_PEER_CLIENT.lock().await;
if let Some(hit) = site_replication_peer_client_cache_hit(&cache, generation) {
return hit;
}
drop(cache);
let outbound_tls = load_global_outbound_tls_state().await;
let outbound_tls = resolve_outbound_tls_state().await;
let built = build_site_replication_peer_client(&outbound_tls);
let cache_entry = match &built {
Ok(client) => SiteReplicationPeerClientCacheEntry::Ready(client.clone()),
rustfs/src/admin/handlers/tls_debug.rs
+18 -17
View File
@@ -14,15 +14,13 @@
use super::profile::authorize_profile_request;
use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::app::context::resolve_outbound_tls_state;
use crate::server::ADMIN_PREFIX;
use http::StatusCode;
use http::{HeaderMap, HeaderValue};
use hyper::Method;
use matchit::Params;
use rustfs_tls_runtime::{
OutboundOnlySnapshotArgs, TlsConsumerStatusItem, TlsDebugStatusResponse, TlsRuntimeStatusSnapshot,
summarize_global_outbound_tls_state,
};
use rustfs_tls_runtime::{OutboundOnlySnapshotArgs, TlsConsumerStatusItem, TlsDebugStatusResponse, TlsRuntimeStatusSnapshot};
use s3s::header::CONTENT_TYPE;
use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
@@ -47,10 +45,13 @@ impl Operation for TlsStatusHandler {
authorize_profile_request(&req).await?;
let tls_path = rustfs_utils::get_env_opt_str(rustfs_config::ENV_RUSTFS_TLS_PATH).unwrap_or_default();
let outbound = summarize_global_outbound_tls_state().await;
let outbound = resolve_outbound_tls_state().await;
let generation = outbound.generation.0;
let has_root_ca = outbound.root_ca_pem.as_ref().is_some_and(|pem| !pem.is_empty());
let has_mtls_identity = outbound.mtls_identity.is_some();
let status = TlsRuntimeStatusSnapshot::from_outbound_only(OutboundOnlySnapshotArgs {
source_path: tls_path,
generation: outbound.generation.0,
generation,
reload_enabled: rustfs_utils::get_env_bool(
rustfs_config::ENV_TLS_RELOAD_ENABLE,
rustfs_config::DEFAULT_TLS_RELOAD_ENABLE,
@@ -59,28 +60,28 @@ impl Operation for TlsStatusHandler {
last_attempt_time: None,
last_success_time: None,
last_error: None,
has_roots: outbound.has_root_ca,
has_mtls_identity: outbound.has_mtls_identity,
has_roots: has_root_ca,
has_mtls_identity,
});
let payload = TlsDebugStatusResponse::builder(status)
.push_consumers([
TlsConsumerStatusItem {
consumer: PROTOS_GRPC_CHANNEL_CONSUMER,
generation: outbound.generation.0,
has_root_ca: outbound.has_root_ca,
has_mtls_identity: outbound.has_mtls_identity,
generation,
has_root_ca,
has_mtls_identity,
},
TlsConsumerStatusItem {
consumer: RIO_HTTP_READER_CONSUMER,
generation: outbound.generation.0,
has_root_ca: outbound.has_root_ca,
has_mtls_identity: outbound.has_mtls_identity,
generation,
has_root_ca,
has_mtls_identity,
},
TlsConsumerStatusItem {
consumer: ECSTORE_TRANSITION_CLIENT_CONSUMER,
generation: outbound.generation.0,
has_root_ca: outbound.has_root_ca,
has_mtls_identity: outbound.has_mtls_identity,
generation,
has_root_ca,
has_mtls_identity,
},
])
.build();
rustfs/src/app/context.rs
+60 -2
View File
@@ -37,6 +37,7 @@ use rustfs_credentials::Credentials;
use rustfs_iam::{store::object::ObjectStore, sys::IamSys};
use rustfs_kms::KmsServiceManager;
use rustfs_lock::LockClient;
use rustfs_tls_runtime::{GlobalPublishedOutboundTlsState, TlsGeneration};
use std::{future::Future, sync::Arc};
use tokio::sync::RwLock;
@@ -45,6 +46,16 @@ pub fn resolve_kms_runtime_service_manager() -> Option<Arc<KmsServiceManager>> {
resolve_kms_runtime_service_manager_with(get_global_app_context(), || default_kms_runtime_interface().service_manager())
}
/// Resolve outbound TLS generation using AppContext-first precedence.
pub fn resolve_outbound_tls_generation() -> TlsGeneration {
resolve_outbound_tls_generation_with(get_global_app_context(), || default_outbound_tls_runtime_interface().generation())
}
/// Resolve outbound TLS state using AppContext-first precedence.
pub async fn resolve_outbound_tls_state() -> GlobalPublishedOutboundTlsState {
resolve_outbound_tls_state_with(get_global_app_context()).await
}
/// Resolve IAM readiness using AppContext-first precedence.
pub fn resolve_iam_ready() -> bool {
resolve_iam_ready_with(get_global_app_context(), || {
@@ -154,6 +165,21 @@ fn resolve_kms_runtime_service_manager_with(
.or_else(fallback)
}
fn resolve_outbound_tls_generation_with(
context: Option<Arc<AppContext>>,
fallback: impl FnOnce() -> TlsGeneration,
) -> TlsGeneration {
context.map_or_else(fallback, |context| context.outbound_tls_runtime().generation())
}
async fn resolve_outbound_tls_state_with(context: Option<Arc<AppContext>>) -> GlobalPublishedOutboundTlsState {
if let Some(context) = context {
return context.outbound_tls_runtime().state().await;
}
default_outbound_tls_runtime_interface().state().await
}
fn resolve_iam_ready_with(context: Option<Arc<AppContext>>, fallback: impl FnOnce() -> bool) -> bool {
context.map_or_else(fallback, |context| context.iam().is_ready())
}
@@ -291,8 +317,8 @@ mod tests {
};
use crate::app::context::interfaces::{
ActionCredentialInterface, BucketMetadataInterface, BufferConfigInterface, DeploymentIdInterface, EndpointsInterface,
IamInterface, KmsInterface, KmsRuntimeInterface, LocalNodeNameInterface, LockClientInterface, RegionInterface,
RuntimePortInterface, ServerConfigInterface, TierConfigInterface,
IamInterface, KmsInterface, KmsRuntimeInterface, LocalNodeNameInterface, LockClientInterface,
OutboundTlsRuntimeInterface, RegionInterface, RuntimePortInterface, ServerConfigInterface, TierConfigInterface,
};
use crate::config::{RustFSBufferConfig, WorkloadProfile};
use async_trait::async_trait;
@@ -336,6 +362,21 @@ mod tests {
}
}
struct TestOutboundTlsRuntimeInterface {
state: GlobalPublishedOutboundTlsState,
}
#[async_trait]
impl OutboundTlsRuntimeInterface for TestOutboundTlsRuntimeInterface {
fn generation(&self) -> TlsGeneration {
self.state.generation
}
async fn state(&self) -> GlobalPublishedOutboundTlsState {
self.state.clone()
}
}
struct TestBucketMetadataInterface {
metadata: Option<Arc<RwLock<BucketMetadataSys>>>,
}
@@ -514,6 +555,11 @@ mod tests {
let fallback_deployment_id = "fallback-deployment".to_string();
let context_runtime_port = 19000;
let fallback_runtime_port = 29000;
let context_outbound_tls_state = GlobalPublishedOutboundTlsState {
generation: TlsGeneration(41),
root_ca_pem: Some(b"context-root-ca".to_vec()),
mtls_identity: None,
};
let context_credentials = Credentials {
access_key: "context-access-key".to_string(),
..Default::default()
@@ -535,6 +581,9 @@ mod tests {
kms_runtime: Arc::new(TestKmsRuntimeInterface {
kms: Some(context_kms.clone()),
}),
outbound_tls_runtime: Arc::new(TestOutboundTlsRuntimeInterface {
state: context_outbound_tls_state.clone(),
}),
notify: default_notify_interface(),
notification_system: default_notification_system_interface(),
bucket_metadata: Arc::new(TestBucketMetadataInterface {
@@ -578,6 +627,14 @@ mod tests {
.expect("context KMS runtime"),
&context_kms
));
assert_eq!(
resolve_outbound_tls_generation_with(Some(context.clone()), || TlsGeneration(99)),
context_outbound_tls_state.generation
);
assert_eq!(
resolve_outbound_tls_state_with(Some(context.clone())).await.generation,
context_outbound_tls_state.generation
);
assert!(resolve_iam_ready_with(Some(context.clone()), || false));
assert!(Arc::ptr_eq(
&resolve_bucket_metadata_handle_with(Some(context.clone()), || None).expect("context bucket metadata"),
@@ -638,6 +695,7 @@ mod tests {
&resolve_kms_runtime_service_manager_with(None, || Some(fallback_kms.clone())).expect("fallback KMS runtime"),
&fallback_kms
));
assert_eq!(resolve_outbound_tls_generation_with(None, || TlsGeneration(99)), TlsGeneration(99));
assert!(!resolve_iam_ready_with(None, || false));
assert!(resolve_iam_handle_with(None, || None).is_none());
assert!(Arc::ptr_eq(
rustfs/src/app/context/global.rs
+13 -5
View File
@@ -17,15 +17,15 @@ use super::handles::{
IamHandle, KmsHandle, default_action_credential_interface, default_bucket_metadata_interface,
default_bucket_monitor_interface, default_buffer_config_interface, default_deployment_id_interface,
default_endpoints_interface, default_kms_runtime_interface, default_local_node_name_interface, default_lock_client_interface,
default_notification_system_interface, default_notify_interface, default_region_interface,
default_replication_pool_interface, default_runtime_port_interface, default_server_config_interface,
default_tier_config_interface,
default_notification_system_interface, default_notify_interface, default_outbound_tls_runtime_interface,
default_region_interface, default_replication_pool_interface, default_runtime_port_interface,
default_server_config_interface, default_tier_config_interface,
};
use super::interfaces::{
ActionCredentialInterface, BucketMetadataInterface, BucketMonitorInterface, BufferConfigInterface, DeploymentIdInterface,
EndpointsInterface, IamInterface, KmsInterface, KmsRuntimeInterface, LocalNodeNameInterface, LockClientInterface,
NotificationSystemInterface, NotifyInterface, RegionInterface, ReplicationPoolInterface, RuntimePortInterface,
ServerConfigInterface, TierConfigInterface,
NotificationSystemInterface, NotifyInterface, OutboundTlsRuntimeInterface, RegionInterface, ReplicationPoolInterface,
RuntimePortInterface, ServerConfigInterface, TierConfigInterface,
};
use rustfs_iam::{store::object::ObjectStore, sys::IamSys};
use rustfs_kms::KmsServiceManager;
@@ -39,6 +39,7 @@ pub struct AppContext {
#[allow(dead_code)]
kms: Arc<dyn KmsInterface>,
kms_runtime: Arc<dyn KmsRuntimeInterface>,
outbound_tls_runtime: Arc<dyn OutboundTlsRuntimeInterface>,
notify: Arc<dyn NotifyInterface>,
notification_system: Arc<dyn NotificationSystemInterface>,
bucket_metadata: Arc<dyn BucketMetadataInterface>,
@@ -63,6 +64,7 @@ impl AppContext {
iam,
kms,
kms_runtime: default_kms_runtime_interface(),
outbound_tls_runtime: default_outbound_tls_runtime_interface(),
notify: default_notify_interface(),
notification_system: default_notification_system_interface(),
bucket_metadata: default_bucket_metadata_interface(),
@@ -106,6 +108,10 @@ impl AppContext {
self.kms_runtime.clone()
}
pub fn outbound_tls_runtime(&self) -> Arc<dyn OutboundTlsRuntimeInterface> {
self.outbound_tls_runtime.clone()
}
pub fn notify(&self) -> Arc<dyn NotifyInterface> {
self.notify.clone()
}
@@ -172,6 +178,7 @@ pub(super) struct AppContextTestInterfaces {
pub(super) iam: Arc<dyn IamInterface>,
pub(super) kms: Arc<dyn KmsInterface>,
pub(super) kms_runtime: Arc<dyn KmsRuntimeInterface>,
pub(super) outbound_tls_runtime: Arc<dyn OutboundTlsRuntimeInterface>,
pub(super) notify: Arc<dyn NotifyInterface>,
pub(super) notification_system: Arc<dyn NotificationSystemInterface>,
pub(super) bucket_metadata: Arc<dyn BucketMetadataInterface>,
@@ -197,6 +204,7 @@ impl AppContext {
iam: interfaces.iam,
kms: interfaces.kms,
kms_runtime: interfaces.kms_runtime,
outbound_tls_runtime: interfaces.outbound_tls_runtime,
notify: interfaces.notify,
notification_system: interfaces.notification_system,
bucket_metadata: interfaces.bucket_metadata,
rustfs/src/app/context/handles.rs
+24 -2
View File
@@ -22,8 +22,8 @@ use super::super::{
use super::interfaces::{
ActionCredentialInterface, BucketMetadataInterface, BucketMonitorInterface, BufferConfigInterface, DeploymentIdInterface,
EndpointsInterface, IamInterface, KmsInterface, KmsRuntimeInterface, LocalNodeNameInterface, LockClientInterface,
NotificationSystemInterface, NotifyInterface, RegionInterface, ReplicationPoolInterface, RuntimePortInterface,
ServerConfigInterface, TierConfigInterface,
NotificationSystemInterface, NotifyInterface, OutboundTlsRuntimeInterface, RegionInterface, ReplicationPoolInterface,
RuntimePortInterface, ServerConfigInterface, TierConfigInterface,
};
use crate::config::{RustFSBufferConfig, get_global_buffer_config};
use async_trait::async_trait;
@@ -36,6 +36,9 @@ use rustfs_kms::{KmsServiceManager, get_global_kms_service_manager};
use rustfs_lock::LockClient;
use rustfs_notify::{EventArgs, NotificationError, notifier_global};
use rustfs_targets::{EventName, arn::TargetID};
use rustfs_tls_runtime::{
GlobalPublishedOutboundTlsState, TlsGeneration, load_global_outbound_tls_generation, load_global_outbound_tls_state,
};
use std::sync::Arc;
use tokio::sync::RwLock;
@@ -89,6 +92,21 @@ impl KmsRuntimeInterface for KmsRuntimeHandle {
}
}
/// Default outbound TLS runtime interface adapter.
#[derive(Default)]
pub struct OutboundTlsRuntimeHandle;
#[async_trait]
impl OutboundTlsRuntimeInterface for OutboundTlsRuntimeHandle {
fn generation(&self) -> TlsGeneration {
load_global_outbound_tls_generation()
}
async fn state(&self) -> GlobalPublishedOutboundTlsState {
load_global_outbound_tls_state().await
}
}
/// Default notify interface adapter.
#[derive(Default)]
pub struct NotifyHandle;
@@ -266,6 +284,10 @@ pub fn default_kms_runtime_interface() -> Arc<dyn KmsRuntimeInterface> {
Arc::new(KmsRuntimeHandle)
}
pub fn default_outbound_tls_runtime_interface() -> Arc<dyn OutboundTlsRuntimeInterface> {
Arc::new(OutboundTlsRuntimeHandle)
}
pub fn default_bucket_metadata_interface() -> Arc<dyn BucketMetadataInterface> {
Arc::new(BucketMetadataHandle)
}
rustfs/src/app/context/interfaces.rs
+8
View File
@@ -25,6 +25,7 @@ use rustfs_kms::KmsServiceManager;
use rustfs_lock::LockClient;
use rustfs_notify::{EventArgs, NotificationError};
use rustfs_targets::{EventName, arn::TargetID};
use rustfs_tls_runtime::{GlobalPublishedOutboundTlsState, TlsGeneration};
use std::sync::Arc;
use tokio::sync::RwLock;
@@ -46,6 +47,13 @@ pub trait KmsRuntimeInterface: Send + Sync {
fn service_manager(&self) -> Option<Arc<KmsServiceManager>>;
}
/// Outbound TLS runtime interface for client material consumers.
#[async_trait]
pub trait OutboundTlsRuntimeInterface: Send + Sync {
fn generation(&self) -> TlsGeneration;
async fn state(&self) -> GlobalPublishedOutboundTlsState;
}
/// Notify interface for application-layer use-cases.
#[async_trait]
pub trait NotifyInterface: Send + Sync {