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

refactor: route admin topology reads through app context (#3774)

Browse Source 
* refactor: route admin topology reads through app context

* chore: accept app context layer baseline entries
This commit is contained in:
Zhengchao An
2026-06-23 11:22:20 +08:00
committed by GitHub
parent 0acc8f8b43
commit 2fe059811a
12 changed files with 215 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/replication.rs
+2 -3
View File
@@ -14,7 +14,6 @@
use super::super::StorageError;
use super::super::bucket_target_sys::{BucketTargetError, BucketTargetSys};
use super::super::global_rustfs_port;
use super::super::metadata::BUCKET_TARGETS_FILE;
use super::super::metadata_sys;
use super::super::metadata_sys::get_replication_config;
@@ -25,7 +24,7 @@ use crate::admin::auth::validate_admin_request;
use crate::admin::handlers::site_replication::site_replication_peer_deployment_id_for_endpoint;
use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::admin::utils::read_compatible_admin_body;
use crate::app::context::resolve_object_store_handle;
use crate::app::context::{resolve_object_store_handle, resolve_runtime_port};
use crate::auth::{check_key_valid, get_session_token};
use crate::error::ApiError;
use crate::server::{ADMIN_PREFIX, RemoteAddr};
@@ -223,7 +222,7 @@ impl Operation for SetRemoteTargetHandler {
let same_target = rustfs_utils::net::is_local_host(
target_url.host().unwrap_or(Host::Domain("localhost")),
target_url.port().unwrap_or(80),
global_rustfs_port(),
resolve_runtime_port(),
)
.unwrap_or_default();
rustfs/src/admin/handlers/site_replication.rs
+11 -9
View File
@@ -25,7 +25,6 @@ use super::super::replication::{ResyncOpts, get_global_replication_pool};
use super::super::target::{ARN, BucketTarget, BucketTargetType, BucketTargets, Credentials};
use super::super::{AdminReplicationConfigExt as _, AdminVersioningConfigExt as _};
use super::super::{delete_admin_config, read_admin_config, save_admin_config};
use super::super::{get_global_deployment_id, get_global_endpoints_opt, global_rustfs_port};
use crate::admin::auth::validate_admin_request;
use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::admin::site_replication_identity::{
@@ -33,7 +32,10 @@ use crate::admin::site_replication_identity::{
site_identity_key,
};
use crate::admin::utils::{encode_compatible_admin_payload, read_compatible_admin_body};
use crate::app::context::{resolve_object_store_handle, resolve_region, resolve_server_config};
use crate::app::context::{
resolve_deployment_id, resolve_endpoints_handle, resolve_object_store_handle, resolve_region, resolve_runtime_port,
resolve_server_config,
};
use crate::auth::{check_key_valid, get_session_token};
use crate::config::get_config_snapshot;
use crate::error::ApiError;
@@ -671,7 +673,7 @@ fn runtime_tls_enabled_with(endpoints: Option<&super::super::EndpointServerPools
}
fn runtime_tls_enabled() -> bool {
let endpoints = get_global_endpoints_opt();
let endpoints = resolve_endpoints_handle();
runtime_tls_enabled_with(endpoints.as_ref())
}
@@ -822,7 +824,7 @@ fn request_endpoint(uri: &Uri, headers: &HeaderMap) -> String {
.map(str::to_string)
.or_else(|| uri.authority().map(|value| value.as_str().to_string()))
.or_else(|| {
get_global_endpoints_opt().and_then(|endpoints| {
resolve_endpoints_handle().and_then(|endpoints| {
endpoints
.as_ref()
.iter()
@@ -831,7 +833,7 @@ fn request_endpoint(uri: &Uri, headers: &HeaderMap) -> String {
.map(|endpoint| endpoint.host_port())
})
})
.unwrap_or_else(|| format!("127.0.0.1:{}", global_rustfs_port()));
.unwrap_or_else(|| format!("127.0.0.1:{}", resolve_runtime_port()));
format!("{scheme}://{host}")
}
@@ -859,7 +861,7 @@ fn site_replication_local_endpoint(uri: &Uri, headers: &HeaderMap) -> String {
if !matches!(parsed.scheme(), "http" | "https") || parsed.host_str().is_none() {
return request_endpoint(&Uri::from_static("/"), &HeaderMap::new());
}
if parsed.port_or_known_default() == runtime_console_port() && parsed.set_port(Some(global_rustfs_port())).is_ok() {
if parsed.port_or_known_default() == runtime_console_port() && parsed.set_port(Some(resolve_runtime_port())).is_ok() {
parsed.to_string().trim_end_matches('/').to_string()
} else {
endpoint
@@ -899,7 +901,7 @@ fn non_negative_u64(value: i64) -> u64 {
fn current_local_peer(req: &S3Request<Body>, state: &SiteReplicationState) -> PeerInfo {
let endpoint = site_replication_local_endpoint(&req.uri, &req.headers);
let deployment_id = get_global_deployment_id().unwrap_or_else(|| deployment_id_for_endpoint(&endpoint));
let deployment_id = resolve_deployment_id().unwrap_or_else(|| deployment_id_for_endpoint(&endpoint));
let stored_peer = state.peers.get(&deployment_id);
PeerInfo {
@@ -923,7 +925,7 @@ fn current_local_peer(req: &S3Request<Body>, state: &SiteReplicationState) -> Pe
fn current_local_runtime_peer(state: &SiteReplicationState) -> PeerInfo {
let endpoint = current_local_runtime_endpoint();
let deployment_id = get_global_deployment_id().unwrap_or_else(|| deployment_id_for_endpoint(&endpoint));
let deployment_id = resolve_deployment_id().unwrap_or_else(|| deployment_id_for_endpoint(&endpoint));
let stored_peer = state.peers.get(&deployment_id);
PeerInfo {
@@ -4287,7 +4289,7 @@ impl Operation for SiteReplicationEditHandler {
};
for target in current_state.peers.values() {
let local_target = get_global_deployment_id()
let local_target = resolve_deployment_id()
.as_ref()
.is_some_and(|deployment_id| deployment_id == &target.deployment_id);
if local_target {
rustfs/src/admin/mod.rs
+1 -15
View File
@@ -136,9 +136,7 @@ mod ecstore_error {
}
mod ecstore_global {
pub(crate) use crate::storage::ecstore_global::{
GLOBAL_BOOT_TIME, get_global_bucket_monitor, get_global_deployment_id, get_global_endpoints_opt, global_rustfs_port,
};
pub(crate) use crate::storage::ecstore_global::{GLOBAL_BOOT_TIME, get_global_bucket_monitor};
}
#[allow(unused_imports)]
@@ -516,18 +514,6 @@ pub(crate) fn get_global_bucket_monitor() -> Option<Arc<bandwidth::monitor::Moni
ecstore_global::get_global_bucket_monitor()
}
pub(crate) fn get_global_deployment_id() -> Option<String> {
ecstore_global::get_global_deployment_id()
}
pub(crate) fn get_global_endpoints_opt() -> Option<EndpointServerPools> {
ecstore_global::get_global_endpoints_opt()
}
pub(crate) fn global_rustfs_port() -> u16 {
ecstore_global::global_rustfs_port()
}
pub(crate) async fn collect_local_metrics(
types: MetricType,
opts: &CollectMetricsOpts,
rustfs/src/admin/router.rs
+3 -3
View File
@@ -16,6 +16,7 @@ use super::GLOBAL_BOOT_TIME;
use super::PeerRestClient;
use super::bandwidth::monitor::BandwidthDetails;
use super::bucket_target_sys::{BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, TargetClient};
use super::get_global_bucket_monitor;
use super::get_global_notification_sys;
use super::metadata::BUCKET_TARGETS_FILE;
use super::metadata_sys;
@@ -26,10 +27,9 @@ use super::replication::{
use super::target::{BucketTarget, BucketTargetType, BucketTargets};
use super::versioning_sys::BucketVersioningSys;
use super::{AdminReplicationConfigExt as _, AdminVersioningConfigExt as _};
use super::{get_global_bucket_monitor, get_global_deployment_id};
use crate::admin::console::{is_console_path, make_console_server};
use crate::admin::handlers::oidc::is_oidc_path;
use crate::app::context::{resolve_object_store_handle, resolve_region, resolve_server_config};
use crate::app::context::{resolve_deployment_id, resolve_object_store_handle, resolve_region, resolve_server_config};
use crate::app::object_usecase::DefaultObjectUsecase;
use crate::auth::{check_key_valid, get_session_token};
use crate::error::ApiError;
@@ -1829,7 +1829,7 @@ async fn check_replication_target(bucket: &str, target: &BucketTarget) -> Replic
if target.target_bucket == bucket
&& !target.deployment_id.is_empty()
&& get_global_deployment_id().as_deref() == Some(target.deployment_id.as_str())
&& resolve_deployment_id().as_deref() == Some(target.deployment_id.as_str())
{
result.status = "FAILED".to_string();
result.error = Some("target bucket must not match source bucket on the same deployment".to_string());
rustfs/src/app/context.rs
+65 -3
View File
@@ -84,6 +84,16 @@ pub fn resolve_endpoints_handle() -> Option<EndpointServerPools> {
resolve_endpoints_handle_with(get_global_app_context(), || default_endpoints_interface().handle())
}
/// Resolve deployment identity using AppContext-first precedence.
pub fn resolve_deployment_id() -> Option<String> {
resolve_deployment_id_with(get_global_app_context(), || default_deployment_id_interface().get())
}
/// Resolve runtime port using AppContext-first precedence.
pub fn resolve_runtime_port() -> u16 {
resolve_runtime_port_with(get_global_app_context(), || default_runtime_port_interface().get())
}
/// Resolve lock client using AppContext-first precedence.
pub fn resolve_lock_client() -> Option<Arc<dyn LockClient>> {
resolve_lock_client_with(get_global_app_context(), || default_lock_client_interface().handle())
@@ -163,6 +173,14 @@ fn resolve_endpoints_handle_with(
context.and_then(|context| context.endpoints().handle()).or_else(fallback)
}
fn resolve_deployment_id_with(context: Option<Arc<AppContext>>, fallback: impl FnOnce() -> Option<String>) -> Option<String> {
context.and_then(|context| context.deployment_id().get()).or_else(fallback)
}
fn resolve_runtime_port_with(context: Option<Arc<AppContext>>, fallback: impl FnOnce() -> u16) -> u16 {
context.map_or_else(fallback, |context| context.runtime_port().get())
}
fn resolve_lock_client_with(
context: Option<Arc<AppContext>>,
fallback: impl FnOnce() -> Option<Arc<dyn LockClient>>,
@@ -226,9 +244,9 @@ mod tests {
use crate::app::context::global::AppContextTestInterfaces;
use crate::app::context::handles::default_notify_interface;
use crate::app::context::interfaces::{
ActionCredentialInterface, BucketMetadataInterface, BufferConfigInterface, EndpointsInterface, IamInterface,
KmsInterface, KmsRuntimeInterface, LocalNodeNameInterface, LockClientInterface, RegionInterface, ServerConfigInterface,
TierConfigInterface,
ActionCredentialInterface, BucketMetadataInterface, BufferConfigInterface, DeploymentIdInterface, EndpointsInterface,
IamInterface, KmsInterface, KmsRuntimeInterface, LocalNodeNameInterface, LockClientInterface, RegionInterface,
RuntimePortInterface, ServerConfigInterface, TierConfigInterface,
};
use crate::config::{RustFSBufferConfig, WorkloadProfile};
use async_trait::async_trait;
@@ -292,6 +310,26 @@ mod tests {
}
}
struct TestDeploymentIdInterface {
id: Option<String>,
}
impl DeploymentIdInterface for TestDeploymentIdInterface {
fn get(&self) -> Option<String> {
self.id.clone()
}
}
struct TestRuntimePortInterface {
port: u16,
}
impl RuntimePortInterface for TestRuntimePortInterface {
fn get(&self) -> u16 {
self.port
}
}
struct TestLockClientInterface {
client: Option<Arc<dyn LockClient>>,
}
@@ -426,6 +464,10 @@ mod tests {
let fallback_lock_client: Arc<dyn LockClient> = Arc::new(LocalClient::new());
let context_node_name = "context-node".to_string();
let fallback_node_name = "fallback-node".to_string();
let context_deployment_id = "context-deployment".to_string();
let fallback_deployment_id = "fallback-deployment".to_string();
let context_runtime_port = 19000;
let fallback_runtime_port = 29000;
let context_credentials = Credentials {
access_key: "context-access-key".to_string(),
..Default::default()
@@ -454,6 +496,12 @@ mod tests {
endpoints: Arc::new(TestEndpointsInterface {
endpoints: Some(endpoints.clone()),
}),
deployment_id: Arc::new(TestDeploymentIdInterface {
id: Some(context_deployment_id.clone()),
}),
runtime_port: Arc::new(TestRuntimePortInterface {
port: context_runtime_port,
}),
lock_client: Arc::new(TestLockClientInterface {
client: Some(context_lock_client.clone()),
}),
@@ -497,6 +545,15 @@ mod tests {
.drives_per_set,
endpoints.as_ref()[0].drives_per_set
);
assert_eq!(
resolve_deployment_id_with(Some(context.clone()), || Some(fallback_deployment_id.clone()))
.expect("context deployment id"),
context_deployment_id
);
assert_eq!(
resolve_runtime_port_with(Some(context.clone()), || fallback_runtime_port),
context_runtime_port
);
assert!(Arc::ptr_eq(
&resolve_lock_client_with(Some(context.clone()), || None).expect("context lock client"),
&context_lock_client
@@ -549,6 +606,11 @@ mod tests {
.drives_per_set,
endpoints.as_ref()[0].drives_per_set
);
assert_eq!(
resolve_deployment_id_with(None, || Some(fallback_deployment_id.clone())).expect("fallback deployment id"),
fallback_deployment_id
);
assert_eq!(resolve_runtime_port_with(None, || fallback_runtime_port), fallback_runtime_port);
assert!(Arc::ptr_eq(
&resolve_lock_client_with(None, || Some(fallback_lock_client.clone())).expect("fallback lock client"),
&fallback_lock_client
rustfs/src/app/context/global.rs
+21 -5
View File
@@ -15,14 +15,14 @@
use super::super::{ECStore, set_object_store_resolver};
use super::handles::{
IamHandle, KmsHandle, default_action_credential_interface, default_bucket_metadata_interface,
default_buffer_config_interface, default_endpoints_interface, default_kms_runtime_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_notify_interface, default_region_interface,
default_server_config_interface, default_tier_config_interface,
default_runtime_port_interface, default_server_config_interface, default_tier_config_interface,
};
use super::interfaces::{
ActionCredentialInterface, BucketMetadataInterface, BufferConfigInterface, EndpointsInterface, IamInterface, KmsInterface,
KmsRuntimeInterface, LocalNodeNameInterface, LockClientInterface, NotifyInterface, RegionInterface, ServerConfigInterface,
TierConfigInterface,
ActionCredentialInterface, BucketMetadataInterface, BufferConfigInterface, DeploymentIdInterface, EndpointsInterface,
IamInterface, KmsInterface, KmsRuntimeInterface, LocalNodeNameInterface, LockClientInterface, NotifyInterface,
RegionInterface, RuntimePortInterface, ServerConfigInterface, TierConfigInterface,
};
use rustfs_iam::{store::object::ObjectStore, sys::IamSys};
use rustfs_kms::KmsServiceManager;
@@ -39,6 +39,8 @@ pub struct AppContext {
notify: Arc<dyn NotifyInterface>,
bucket_metadata: Arc<dyn BucketMetadataInterface>,
endpoints: Arc<dyn EndpointsInterface>,
deployment_id: Arc<dyn DeploymentIdInterface>,
runtime_port: Arc<dyn RuntimePortInterface>,
lock_client: Arc<dyn LockClientInterface>,
local_node_name: Arc<dyn LocalNodeNameInterface>,
action_credentials: Arc<dyn ActionCredentialInterface>,
@@ -58,6 +60,8 @@ impl AppContext {
notify: default_notify_interface(),
bucket_metadata: default_bucket_metadata_interface(),
endpoints: default_endpoints_interface(),
deployment_id: default_deployment_id_interface(),
runtime_port: default_runtime_port_interface(),
lock_client: default_lock_client_interface(),
local_node_name: default_local_node_name_interface(),
action_credentials: default_action_credential_interface(),
@@ -105,6 +109,14 @@ impl AppContext {
self.endpoints.clone()
}
pub fn deployment_id(&self) -> Arc<dyn DeploymentIdInterface> {
self.deployment_id.clone()
}
pub fn runtime_port(&self) -> Arc<dyn RuntimePortInterface> {
self.runtime_port.clone()
}
pub fn lock_client(&self) -> Arc<dyn LockClientInterface> {
self.lock_client.clone()
}
@@ -142,6 +154,8 @@ pub(super) struct AppContextTestInterfaces {
pub(super) notify: Arc<dyn NotifyInterface>,
pub(super) bucket_metadata: Arc<dyn BucketMetadataInterface>,
pub(super) endpoints: Arc<dyn EndpointsInterface>,
pub(super) deployment_id: Arc<dyn DeploymentIdInterface>,
pub(super) runtime_port: Arc<dyn RuntimePortInterface>,
pub(super) lock_client: Arc<dyn LockClientInterface>,
pub(super) local_node_name: Arc<dyn LocalNodeNameInterface>,
pub(super) action_credentials: Arc<dyn ActionCredentialInterface>,
@@ -162,6 +176,8 @@ impl AppContext {
notify: interfaces.notify,
bucket_metadata: interfaces.bucket_metadata,
endpoints: interfaces.endpoints,
deployment_id: interfaces.deployment_id,
runtime_port: interfaces.runtime_port,
lock_client: interfaces.lock_client,
local_node_name: interfaces.local_node_name,
action_credentials: interfaces.action_credentials,
rustfs/src/app/context/handles.rs
+35 -4
View File
@@ -15,11 +15,14 @@
use super::super::EndpointServerPools;
use super::super::TierConfigMgr;
use super::super::metadata_sys::{BucketMetadataSys, get_global_bucket_metadata_sys};
use super::super::{get_global_endpoints_opt, get_global_lock_client, get_global_region, get_global_tier_config_mgr};
use super::super::{
get_global_deployment_id, get_global_endpoints_opt, get_global_lock_client, get_global_region, get_global_tier_config_mgr,
global_rustfs_port,
};
use super::interfaces::{
ActionCredentialInterface, BucketMetadataInterface, BufferConfigInterface, EndpointsInterface, IamInterface, KmsInterface,
KmsRuntimeInterface, LocalNodeNameInterface, LockClientInterface, NotifyInterface, RegionInterface, ServerConfigInterface,
TierConfigInterface,
ActionCredentialInterface, BucketMetadataInterface, BufferConfigInterface, DeploymentIdInterface, EndpointsInterface,
IamInterface, KmsInterface, KmsRuntimeInterface, LocalNodeNameInterface, LockClientInterface, NotifyInterface,
RegionInterface, RuntimePortInterface, ServerConfigInterface, TierConfigInterface,
};
use crate::config::{RustFSBufferConfig, get_global_buffer_config};
use async_trait::async_trait;
@@ -129,6 +132,26 @@ impl EndpointsInterface for EndpointsHandle {
}
}
/// Default deployment identity interface adapter.
#[derive(Default)]
pub struct DeploymentIdHandle;
impl DeploymentIdInterface for DeploymentIdHandle {
fn get(&self) -> Option<String> {
get_global_deployment_id()
}
}
/// Default runtime port interface adapter.
#[derive(Default)]
pub struct RuntimePortHandle;
impl RuntimePortInterface for RuntimePortHandle {
fn get(&self) -> u16 {
global_rustfs_port()
}
}
/// Default lock client interface adapter.
#[derive(Default)]
pub struct LockClientHandle;
@@ -216,6 +239,14 @@ pub fn default_endpoints_interface() -> Arc<dyn EndpointsInterface> {
Arc::new(EndpointsHandle)
}
pub fn default_deployment_id_interface() -> Arc<dyn DeploymentIdInterface> {
Arc::new(DeploymentIdHandle)
}
pub fn default_runtime_port_interface() -> Arc<dyn RuntimePortInterface> {
Arc::new(RuntimePortHandle)
}
pub fn default_lock_client_interface() -> Arc<dyn LockClientInterface> {
Arc::new(LockClientHandle)
}
rustfs/src/app/context/interfaces.rs
+10
View File
@@ -70,6 +70,16 @@ pub trait EndpointsInterface: Send + Sync {
fn handle(&self) -> Option<EndpointServerPools>;
}
/// Deployment identity interface for admin topology integration.
pub trait DeploymentIdInterface: Send + Sync {
fn get(&self) -> Option<String>;
}
/// Runtime port interface for admin topology integration.
pub trait RuntimePortInterface: Send + Sync {
fn get(&self) -> u16;
}
/// Lock client interface for application-layer use-cases.
pub trait LockClientInterface: Send + Sync {
fn handle(&self) -> Option<Arc<dyn LockClient>>;
rustfs/src/app/mod.rs
+8
View File
@@ -630,6 +630,10 @@ pub(crate) fn get_global_endpoints_opt() -> Option<EndpointServerPools> {
crate::storage::get_global_endpoints_opt()
}
pub(crate) fn get_global_deployment_id() -> Option<String> {
crate::storage::get_global_deployment_id()
}
pub(crate) fn get_global_lock_client() -> Option<Arc<dyn rustfs_lock::client::LockClient>> {
crate::storage::get_global_lock_client()
}
@@ -638,6 +642,10 @@ pub(crate) fn get_global_region() -> Option<s3s::region::Region> {
crate::storage::get_global_region()
}
pub(crate) fn global_rustfs_port() -> u16 {
crate::storage::global_rustfs_port()
}
pub(crate) fn get_global_tier_config_mgr() -> Arc<tokio::sync::RwLock<TierConfigMgr>> {
crate::storage::get_global_tier_config_mgr()
}
rustfs/src/storage/mod.rs
+8
View File
@@ -813,10 +813,18 @@ pub(crate) fn get_global_endpoints_opt() -> Option<EndpointServerPools> {
ecstore_global::get_global_endpoints_opt()
}
pub(crate) fn get_global_deployment_id() -> Option<String> {
ecstore_global::get_global_deployment_id()
}
pub(crate) fn get_global_region() -> Option<s3s::region::Region> {
ecstore_global::get_global_region()
}
pub(crate) fn global_rustfs_port() -> u16 {
ecstore_global::global_rustfs_port()
}
pub(crate) fn get_global_tier_config_mgr() -> Arc<tokio::sync::RwLock<TierConfigMgr>> {
ecstore_global::get_global_tier_config_mgr()
}