mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
refactor(server): dispatch S3 requests through a per-server context slot (#4615)
This commit is contained in:
@@ -20,11 +20,13 @@ mod global;
|
||||
mod handles;
|
||||
mod interfaces;
|
||||
mod runtime_sources;
|
||||
mod server_slot;
|
||||
mod startup;
|
||||
|
||||
pub use global::*;
|
||||
pub use handles::*;
|
||||
pub use interfaces::*;
|
||||
pub use server_slot::ServerContextSlot;
|
||||
|
||||
use super::storage_api::context::bucket::metadata_sys::BucketMetadataSys;
|
||||
use super::storage_api::context::runtime::{
|
||||
@@ -1216,6 +1218,7 @@ mod tests {
|
||||
assert_eq!(context_server_config_published.load(Ordering::SeqCst), 1);
|
||||
assert!(publish_storage_class_config_with(Some(context.clone()), StorageClassConfig::default()));
|
||||
assert_eq!(context_storage_class_published.load(Ordering::SeqCst), 1);
|
||||
let slot_context = context.clone();
|
||||
assert_eq!(
|
||||
resolve_buffer_config_with(Some(context))
|
||||
.expect("context buffer config")
|
||||
@@ -1258,5 +1261,19 @@ mod tests {
|
||||
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());
|
||||
|
||||
// backlog#1052 S2: a per-server context slot resolves its *installed*
|
||||
// context — not whatever another server installed — and the first
|
||||
// installation wins.
|
||||
let slot = ServerContextSlot::new();
|
||||
assert!(slot.install(slot_context.clone()), "first install must succeed");
|
||||
assert!(!slot.install(slot_context.clone()), "the slot is install-once");
|
||||
let resolved = slot.app_context().expect("installed slot must resolve its context");
|
||||
assert!(Arc::ptr_eq(&resolved, &slot_context), "the slot must hand back the installed context");
|
||||
let resolved_store = slot.object_store().expect("installed slot must resolve the store");
|
||||
assert!(
|
||||
Arc::ptr_eq(&resolved_store, &slot_context.object_store()),
|
||||
"the slot must dispatch to the installed context's store"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
//! Per-server application-context slot (backlog#1052 S2).
|
||||
//!
|
||||
//! The S3 service starts serving before storage and IAM finish initializing,
|
||||
//! so a server cannot hold its `AppContext` at construction time. This slot is
|
||||
//! the late-bound carrier: startup creates one per server, hands it to the
|
||||
//! request-path owners (the `FS` service), and installs the `AppContext` into
|
||||
//! it once IAM bootstrap completes.
|
||||
//!
|
||||
//! Until every app subsystem is per-server (backlog#1052 S3), resolution falls
|
||||
//! back to the process-global `AppContext` singleton when the slot has not
|
||||
//! been installed — byte-for-byte the ambient resolution the request path used
|
||||
//! before this seam existed. The fallback is removed when multi-instance flips
|
||||
//! on (backlog#1052 S5).
|
||||
|
||||
use super::global::{AppContext, get_global_app_context};
|
||||
use crate::app::storage_api::context::ECStore;
|
||||
use std::sync::{Arc, OnceLock};
|
||||
|
||||
/// Late-bound, per-server handle to the application context.
|
||||
#[derive(Default)]
|
||||
pub struct ServerContextSlot {
|
||||
app_context: OnceLock<Arc<AppContext>>,
|
||||
}
|
||||
|
||||
// Manual Debug: AppContext is not Debug, so summarize installation state.
|
||||
impl std::fmt::Debug for ServerContextSlot {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("ServerContextSlot")
|
||||
.field("app_context_installed", &self.app_context.get().is_some())
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
impl ServerContextSlot {
|
||||
pub fn new() -> Arc<Self> {
|
||||
Arc::new(Self {
|
||||
app_context: OnceLock::new(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Install this server's application context (once). Returns `false` if
|
||||
/// the slot was already installed; the first installation wins, matching
|
||||
/// the process-global singleton's `get_or_init` semantics.
|
||||
pub fn install(&self, context: Arc<AppContext>) -> bool {
|
||||
self.app_context.set(context).is_ok()
|
||||
}
|
||||
|
||||
/// This server's application context: the installed one, or the
|
||||
/// process-global singleton as the single-instance legacy default.
|
||||
pub fn app_context(&self) -> Option<Arc<AppContext>> {
|
||||
self.app_context.get().cloned().or_else(get_global_app_context)
|
||||
}
|
||||
|
||||
/// This server's object store, resolved through [`Self::app_context`].
|
||||
pub fn object_store(&self) -> Option<Arc<ECStore>> {
|
||||
self.app_context().map(|context| context.object_store())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{ServerContextSlot, get_global_app_context};
|
||||
|
||||
// Before anything is installed — and with no global AppContext in this
|
||||
// process (nextest runs each test in its own process) — resolution yields
|
||||
// nothing: the same "not ready yet" answer the ambient path gives before
|
||||
// IAM bootstrap completes.
|
||||
#[test]
|
||||
fn empty_slot_resolves_like_the_ambient_path() {
|
||||
let slot = ServerContextSlot::new();
|
||||
assert_eq!(slot.app_context().is_some(), get_global_app_context().is_some());
|
||||
assert_eq!(slot.object_store().is_some(), get_global_app_context().is_some());
|
||||
}
|
||||
}
|
||||
@@ -15,6 +15,7 @@
|
||||
use super::super::storage_api::context::ECStore;
|
||||
use super::global::{AppContext, get_global_app_context, init_global_app_context};
|
||||
use super::runtime_sources;
|
||||
use super::server_slot::ServerContextSlot;
|
||||
use rustfs_kms::KmsServiceManager;
|
||||
use std::io::{Error, Result};
|
||||
use std::sync::Arc;
|
||||
@@ -30,7 +31,11 @@ impl AppContext {
|
||||
ensure_startup_kms_interface_with(runtime_sources::kms_service_manager, runtime_sources::init_kms_service_manager)
|
||||
}
|
||||
|
||||
pub(crate) fn ensure_startup_after_iam(store: Arc<ECStore>, kms_interface: Arc<KmsServiceManager>) -> Result<()> {
|
||||
pub(crate) fn ensure_startup_after_iam(
|
||||
store: Arc<ECStore>,
|
||||
kms_interface: Arc<KmsServiceManager>,
|
||||
server_ctx: &ServerContextSlot,
|
||||
) -> Result<()> {
|
||||
ensure_startup_app_context_after_iam_with(
|
||||
|| get_global_app_context().is_some(),
|
||||
|| {
|
||||
@@ -40,6 +45,12 @@ impl AppContext {
|
||||
Ok(())
|
||||
},
|
||||
)?;
|
||||
// Install this server's context slot (backlog#1052 S2). Today the
|
||||
// context is still the process singleton, so the slot mirrors it;
|
||||
// once contexts become per-server (S3) this hands each server its own.
|
||||
if let Some(context) = get_global_app_context() {
|
||||
let _ = server_ctx.install(context);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,7 +16,7 @@ use crate::app::context;
|
||||
use std::sync::Arc;
|
||||
|
||||
pub(crate) use context::{
|
||||
AppContext, NotifyInterface, default_notify_interface as fallback_notify_interface,
|
||||
AppContext, NotifyInterface, ServerContextSlot, default_notify_interface as fallback_notify_interface,
|
||||
default_object_data_cache_handle as fallback_object_data_cache_handle,
|
||||
default_outbound_tls_runtime_interface as fallback_outbound_tls_runtime_interface,
|
||||
default_s3select_db_interface as fallback_s3select_db_interface,
|
||||
|
||||
@@ -34,7 +34,7 @@ use crate::storage_api::server::http as storage;
|
||||
use crate::storage_api::server::http::request_context::{RequestContext, extract_request_id_from_headers};
|
||||
use crate::storage_api::server::http::rpc::InternodeRpcService;
|
||||
use crate::storage_api::server::http::tonic_service::make_server;
|
||||
use crate::storage_api::server::http::{TONIC_RPC_PREFIX, verify_rpc_signature};
|
||||
use crate::storage_api::server::http::{ServerContextSlot, TONIC_RPC_PREFIX, verify_rpc_signature};
|
||||
use bytes::Bytes;
|
||||
use http::{HeaderMap, Method, Request as HttpRequest, Response};
|
||||
use hyper::body::Incoming;
|
||||
@@ -331,7 +331,11 @@ fn trace_on_response<ResBody>(response: &Response<ResBody>, latency: Duration, s
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalReadiness>) -> Result<(ShutdownHandle, SocketAddr)> {
|
||||
pub async fn start_http_server(
|
||||
config: &config::Config,
|
||||
readiness: Arc<GlobalReadiness>,
|
||||
server_ctx: Arc<ServerContextSlot>,
|
||||
) -> Result<(ShutdownHandle, SocketAddr)> {
|
||||
let server_addr = parse_and_resolve_address(config.address.as_str()).map_err(Error::other)?;
|
||||
|
||||
// The listening address and port are obtained from the parameters
|
||||
@@ -598,7 +602,9 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
|
||||
// Setup S3 service
|
||||
// This project uses the S3S library to implement S3 services
|
||||
let s3_service = {
|
||||
let store = storage::ecfs::FS::new();
|
||||
// Bind the S3 service to this server's context slot (backlog#1052 S2)
|
||||
// so request dispatch resolves the server's own store.
|
||||
let store = storage::ecfs::FS::with_server_ctx(server_ctx);
|
||||
let mut b = S3ServiceBuilder::new(store.clone());
|
||||
|
||||
let access_key = config.access_key.clone();
|
||||
|
||||
@@ -24,6 +24,7 @@ use crate::{
|
||||
startup_services::init_embedded_startup_runtime_services,
|
||||
startup_shutdown::signal_embedded_startup_shutdown,
|
||||
startup_storage::{init_embedded_startup_storage_foundation, init_embedded_startup_storage_runtime},
|
||||
storage_api::server::http::ServerContextSlot,
|
||||
storage_api::startup::storage::bootstrap_instance_ctx,
|
||||
};
|
||||
use std::{io, net::SocketAddr, path::PathBuf};
|
||||
@@ -114,6 +115,8 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result<Em
|
||||
// are per-server too (backlog#1052 S2–S4), each server constructs its own
|
||||
// context here instead.
|
||||
let instance_ctx = bootstrap_instance_ctx();
|
||||
// This server's request-path context slot (backlog#1052 S2).
|
||||
let server_ctx = ServerContextSlot::new();
|
||||
|
||||
let EmbeddedStartupConfig {
|
||||
config,
|
||||
@@ -139,7 +142,7 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result<Em
|
||||
.await
|
||||
.map_err(init_error)?;
|
||||
|
||||
let http_server = start_embedded_http_server(&config, listen_context.readiness.clone()).await?;
|
||||
let http_server = start_embedded_http_server(&config, listen_context.readiness.clone(), server_ctx.clone()).await?;
|
||||
let shutdown_handle = http_server.shutdown_handle;
|
||||
let bound_addr = http_server.bound_addr;
|
||||
let cancel_token = CancellationToken::new();
|
||||
@@ -166,6 +169,7 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result<Em
|
||||
storage_runtime.store,
|
||||
cancel_token.clone(),
|
||||
listen_context.readiness.clone(),
|
||||
server_ctx,
|
||||
)
|
||||
.await
|
||||
.map_err(|err| {
|
||||
|
||||
@@ -19,6 +19,7 @@ use crate::{
|
||||
startup_server::{StartupHttpServers, StartupListenContext, init_startup_http_servers, init_startup_listen_context},
|
||||
startup_services::init_startup_runtime_services,
|
||||
startup_storage::{StartupStorageRuntime, init_startup_storage_foundation, init_startup_storage_runtime},
|
||||
storage_api::server::http::ServerContextSlot,
|
||||
storage_api::startup::storage::bootstrap_instance_ctx,
|
||||
};
|
||||
use std::io::{Error, Result};
|
||||
@@ -108,6 +109,9 @@ async fn run(config: Config) -> Result<()> {
|
||||
// the storage path explicitly (Phase 5 follow-up, backlog#1052); a future
|
||||
// multi-instance server constructs its own context here instead.
|
||||
let instance_ctx = bootstrap_instance_ctx();
|
||||
// This server's request-path context slot (backlog#1052 S2): handed to the
|
||||
// HTTP service now, installed once IAM bootstrap completes.
|
||||
let server_ctx = ServerContextSlot::new();
|
||||
|
||||
let StartupListenContext {
|
||||
readiness,
|
||||
@@ -120,7 +124,7 @@ async fn run(config: Config) -> Result<()> {
|
||||
state_manager,
|
||||
s3_shutdown_tx,
|
||||
console_shutdown_tx,
|
||||
} = init_startup_http_servers(&config, readiness.clone()).await?;
|
||||
} = init_startup_http_servers(&config, readiness.clone(), server_ctx.clone()).await?;
|
||||
|
||||
let StartupStorageRuntime {
|
||||
store,
|
||||
@@ -134,6 +138,7 @@ async fn run(config: Config) -> Result<()> {
|
||||
ctx.clone(),
|
||||
readiness.clone(),
|
||||
state_manager.clone(),
|
||||
server_ctx,
|
||||
)
|
||||
.await?;
|
||||
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::runtime_sources::AppContext;
|
||||
use crate::runtime_sources::{AppContext, ServerContextSlot};
|
||||
use crate::server::{ServiceStateManager, publish_ready_when_runtime_ready};
|
||||
use crate::storage_api::startup::iam::ECStore;
|
||||
use rustfs_common::{GlobalReadiness, SystemStage};
|
||||
@@ -76,8 +76,9 @@ async fn finalize_iam_recovery(
|
||||
kms_interface: Arc<KmsServiceManager>,
|
||||
readiness: Arc<GlobalReadiness>,
|
||||
state_manager: Option<Arc<ServiceStateManager>>,
|
||||
server_ctx: Arc<ServerContextSlot>,
|
||||
) -> Result<()> {
|
||||
AppContext::ensure_startup_after_iam(store, kms_interface)?;
|
||||
AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx)?;
|
||||
|
||||
readiness.mark_stage(SystemStage::IamReady);
|
||||
publish_ready_when_runtime_ready(readiness.as_ref(), state_manager.as_deref()).await
|
||||
@@ -90,6 +91,7 @@ fn compute_backoff_interval(attempt: u64, initial: Duration, max: Duration) -> D
|
||||
if backoff > max { max } else { backoff }
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn spawn_iam_recovery_task(
|
||||
initial_interval: Duration,
|
||||
max_interval: Duration,
|
||||
@@ -98,12 +100,14 @@ fn spawn_iam_recovery_task(
|
||||
kms_interface: Arc<KmsServiceManager>,
|
||||
readiness: Arc<GlobalReadiness>,
|
||||
state_manager: Option<Arc<ServiceStateManager>>,
|
||||
server_ctx: Arc<ServerContextSlot>,
|
||||
) {
|
||||
let init_store = store.clone();
|
||||
let finalize_store = store;
|
||||
let finalize_kms_interface = kms_interface;
|
||||
let finalize_readiness = readiness;
|
||||
let finalize_state_manager = state_manager;
|
||||
let finalize_server_ctx = server_ctx;
|
||||
tokio::spawn(async move {
|
||||
run_iam_recovery_loop(
|
||||
initial_interval,
|
||||
@@ -118,7 +122,8 @@ fn spawn_iam_recovery_task(
|
||||
let kms_interface = finalize_kms_interface.clone();
|
||||
let readiness = finalize_readiness.clone();
|
||||
let state_manager = finalize_state_manager.clone();
|
||||
Box::pin(async move { finalize_iam_recovery(store, kms_interface, readiness, state_manager).await })
|
||||
let server_ctx = finalize_server_ctx.clone();
|
||||
Box::pin(async move { finalize_iam_recovery(store, kms_interface, readiness, state_manager, server_ctx).await })
|
||||
},
|
||||
)
|
||||
.await;
|
||||
@@ -333,10 +338,11 @@ pub(crate) async fn bootstrap_or_defer_iam_init(
|
||||
readiness: Arc<GlobalReadiness>,
|
||||
state_manager: Option<Arc<ServiceStateManager>>,
|
||||
shutdown_token: Option<tokio_util::sync::CancellationToken>,
|
||||
server_ctx: Arc<ServerContextSlot>,
|
||||
) -> Result<IamBootstrapDisposition> {
|
||||
match attempt_init_iam_sys(store.clone()).await {
|
||||
Ok(()) => {
|
||||
AppContext::ensure_startup_after_iam(store, kms_interface)?;
|
||||
AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx)?;
|
||||
readiness.mark_stage(SystemStage::IamReady);
|
||||
return Ok(IamBootstrapDisposition::ReadyInline);
|
||||
}
|
||||
@@ -362,6 +368,7 @@ pub(crate) async fn bootstrap_or_defer_iam_init(
|
||||
kms_interface,
|
||||
readiness,
|
||||
state_manager,
|
||||
server_ctx,
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -374,17 +381,19 @@ pub(crate) async fn bootstrap_or_defer_iam_init_with_startup_kms(
|
||||
readiness: Arc<GlobalReadiness>,
|
||||
state_manager: Option<Arc<ServiceStateManager>>,
|
||||
shutdown_token: Option<tokio_util::sync::CancellationToken>,
|
||||
server_ctx: Arc<ServerContextSlot>,
|
||||
) -> Result<IamBootstrapDisposition> {
|
||||
let kms_interface = AppContext::ensure_startup_kms_interface();
|
||||
bootstrap_or_defer_iam_init(store, kms_interface, readiness, state_manager, shutdown_token).await
|
||||
bootstrap_or_defer_iam_init(store, kms_interface, readiness, state_manager, shutdown_token, server_ctx).await
|
||||
}
|
||||
|
||||
pub(crate) async fn init_embedded_iam_runtime(
|
||||
store: Arc<ECStore>,
|
||||
ctx: tokio_util::sync::CancellationToken,
|
||||
readiness: Arc<GlobalReadiness>,
|
||||
server_ctx: Arc<ServerContextSlot>,
|
||||
) -> Result<IamBootstrapDisposition> {
|
||||
bootstrap_or_defer_iam_init_with_startup_kms(store, readiness, None, Some(ctx)).await
|
||||
bootstrap_or_defer_iam_init_with_startup_kms(store, readiness, None, Some(ctx), server_ctx).await
|
||||
}
|
||||
|
||||
pub(crate) async fn init_iam_runtime(
|
||||
@@ -392,8 +401,9 @@ pub(crate) async fn init_iam_runtime(
|
||||
ctx: tokio_util::sync::CancellationToken,
|
||||
readiness: Arc<GlobalReadiness>,
|
||||
state_manager: Arc<ServiceStateManager>,
|
||||
server_ctx: Arc<ServerContextSlot>,
|
||||
) -> Result<IamBootstrapDisposition> {
|
||||
bootstrap_or_defer_iam_init_with_startup_kms(store, readiness, Some(state_manager), Some(ctx)).await
|
||||
bootstrap_or_defer_iam_init_with_startup_kms(store, readiness, Some(state_manager), Some(ctx), server_ctx).await
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -17,6 +17,7 @@ use crate::{
|
||||
config::Config,
|
||||
server::{ServiceState, ServiceStateManager, ShutdownHandle, start_http_server},
|
||||
startup_runtime_sources,
|
||||
storage_api::server::http::ServerContextSlot,
|
||||
storage_api::startup::runtime_sources::InstanceContext,
|
||||
};
|
||||
use rustfs_common::GlobalReadiness;
|
||||
@@ -231,9 +232,13 @@ pub(crate) async fn init_embedded_startup_listen_context(
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) async fn start_embedded_http_server(config: &Config, readiness: Arc<GlobalReadiness>) -> Result<EmbeddedHttpServer> {
|
||||
pub(crate) async fn start_embedded_http_server(
|
||||
config: &Config,
|
||||
readiness: Arc<GlobalReadiness>,
|
||||
server_ctx: Arc<ServerContextSlot>,
|
||||
) -> Result<EmbeddedHttpServer> {
|
||||
let s3_config = s3_http_server_config(config);
|
||||
let (shutdown_handle, bound_addr) = start_http_server(&s3_config, readiness).await?;
|
||||
let (shutdown_handle, bound_addr) = start_http_server(&s3_config, readiness, server_ctx).await?;
|
||||
|
||||
Ok(EmbeddedHttpServer {
|
||||
shutdown_handle,
|
||||
@@ -241,16 +246,22 @@ pub(crate) async fn start_embedded_http_server(config: &Config, readiness: Arc<G
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) async fn init_startup_http_servers(config: &Config, readiness: Arc<GlobalReadiness>) -> Result<StartupHttpServers> {
|
||||
pub(crate) async fn init_startup_http_servers(
|
||||
config: &Config,
|
||||
readiness: Arc<GlobalReadiness>,
|
||||
server_ctx: Arc<ServerContextSlot>,
|
||||
) -> Result<StartupHttpServers> {
|
||||
init_capacity_management().await;
|
||||
let state_manager = Arc::new(ServiceStateManager::new());
|
||||
state_manager.update(ServiceState::Starting);
|
||||
|
||||
let s3_config = s3_http_server_config(config);
|
||||
let (s3_shutdown_tx, _) = start_http_server(&s3_config, readiness.clone()).await?;
|
||||
let (s3_shutdown_tx, _) = start_http_server(&s3_config, readiness.clone(), server_ctx.clone()).await?;
|
||||
|
||||
let console_shutdown_tx = match console_http_server_config(config) {
|
||||
Some(console_config) => Some(start_http_server(&console_config, readiness).await?.0),
|
||||
// The console shares the S3 server's context slot: it is the same
|
||||
// logical server on a second listener.
|
||||
Some(console_config) => Some(start_http_server(&console_config, readiness, server_ctx).await?.0),
|
||||
None => None,
|
||||
};
|
||||
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::storage_api::startup::services::{ECStore, EndpointServerPools};
|
||||
use crate::storage_api::startup::services::{ECStore, EndpointServerPools, ServerContextSlot};
|
||||
use crate::{
|
||||
config::Config,
|
||||
init::{init_buffer_profile_system, init_kms_system},
|
||||
@@ -48,10 +48,11 @@ pub(crate) async fn init_embedded_startup_runtime_services(
|
||||
store: Arc<ECStore>,
|
||||
ctx: CancellationToken,
|
||||
readiness: Arc<GlobalReadiness>,
|
||||
server_ctx: Arc<ServerContextSlot>,
|
||||
) -> Result<EmbeddedStartupServiceRuntime> {
|
||||
init_embedded_optional_service_runtime(config).await;
|
||||
let buckets = init_embedded_bucket_metadata_runtime(store.clone()).await?;
|
||||
let iam_bootstrap = init_embedded_iam_runtime(store, ctx, readiness)
|
||||
let iam_bootstrap = init_embedded_iam_runtime(store, ctx, readiness, server_ctx)
|
||||
.await
|
||||
.map_err(|err| std::io::Error::other(format!("IAM bootstrap setup: {err}")))?;
|
||||
init_embedded_notification_runtime(endpoint_pools, buckets).await;
|
||||
@@ -66,6 +67,7 @@ pub(crate) async fn init_startup_runtime_services(
|
||||
ctx: CancellationToken,
|
||||
readiness: Arc<GlobalReadiness>,
|
||||
state_manager: Arc<ServiceStateManager>,
|
||||
server_ctx: Arc<ServerContextSlot>,
|
||||
) -> Result<StartupServiceRuntime> {
|
||||
init_kms_system(config).await?;
|
||||
|
||||
@@ -76,7 +78,7 @@ pub(crate) async fn init_startup_runtime_services(
|
||||
init_deadlock_detector_runtime();
|
||||
|
||||
let buckets = init_bucket_metadata_runtime(store.clone(), ctx.clone()).await?;
|
||||
let iam_bootstrap = init_iam_runtime(store.clone(), ctx.clone(), readiness, state_manager).await?;
|
||||
let iam_bootstrap = init_iam_runtime(store.clone(), ctx.clone(), readiness, state_manager, server_ctx).await?;
|
||||
init_auth_integrations().await?;
|
||||
init_notification_runtime(endpoint_pools, buckets).await?;
|
||||
let enable_scanner = init_background_service_runtime(store.clone()).await?;
|
||||
|
||||
+38
-25
@@ -60,7 +60,13 @@ use crate::storage::storage_api::ecfs_consumer::StorageObjectOptions as ObjectOp
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct FS {
|
||||
// pub store: ECStore,
|
||||
/// This server's late-bound application-context slot (backlog#1052 S2).
|
||||
///
|
||||
/// Handlers resolve the object store through this slot instead of the
|
||||
/// ambient process singleton, so each server dispatches requests to its
|
||||
/// own store once contexts become per-server. An uninstalled slot falls
|
||||
/// back to the global `AppContext` — the single-instance legacy default.
|
||||
server_ctx: std::sync::Arc<runtime_sources::ServerContextSlot>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, serde::Deserialize)]
|
||||
@@ -77,9 +83,16 @@ impl Default for FS {
|
||||
|
||||
impl FS {
|
||||
pub fn new() -> Self {
|
||||
Self::with_server_ctx(runtime_sources::ServerContextSlot::new())
|
||||
}
|
||||
|
||||
/// Build the service bound to an explicit per-server context slot
|
||||
/// (backlog#1052 S2). [`FS::new`] hands out a fresh, never-installed slot,
|
||||
/// which resolves through the global fallback — the legacy behavior.
|
||||
pub(crate) fn with_server_ctx(server_ctx: std::sync::Arc<runtime_sources::ServerContextSlot>) -> Self {
|
||||
rustfs_io_metrics::init_s3_metrics();
|
||||
rustfs_io_metrics::init_list_objects_metrics();
|
||||
Self {}
|
||||
Self { server_ctx }
|
||||
}
|
||||
|
||||
async fn replication_tagging_enabled(bucket: &str, object: &str) -> bool {
|
||||
@@ -102,7 +115,7 @@ impl FS {
|
||||
object: &str,
|
||||
version_id: Option<&str>,
|
||||
) -> S3Result<std::collections::HashMap<String, Vec<String>>> {
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Ok(std::collections::HashMap::new());
|
||||
};
|
||||
let opts = ObjectOptions {
|
||||
@@ -348,7 +361,7 @@ impl S3 for FS {
|
||||
&self,
|
||||
req: S3Request<DeleteBucketWebsiteInput>,
|
||||
) -> S3Result<S3Response<DeleteBucketWebsiteOutput>> {
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(s3_error!(InternalError, "Not init"));
|
||||
};
|
||||
|
||||
@@ -398,7 +411,7 @@ impl S3 for FS {
|
||||
|
||||
validate_table_catalog_object_mutation(&bucket, &object).await?;
|
||||
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
error!(
|
||||
component = LOG_COMPONENT_STORAGE,
|
||||
subsystem = LOG_SUBSYSTEM_TAGGING,
|
||||
@@ -480,7 +493,7 @@ impl S3 for FS {
|
||||
record_s3_op(S3Operation::GetBucketAcl);
|
||||
let GetBucketAclInput { bucket, .. } = req.input;
|
||||
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
||||
};
|
||||
|
||||
@@ -496,7 +509,7 @@ impl S3 for FS {
|
||||
&self,
|
||||
req: S3Request<GetBucketAccelerateConfigurationInput>,
|
||||
) -> S3Result<S3Response<GetBucketAccelerateConfigurationOutput>> {
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(s3_error!(InternalError, "Not init"));
|
||||
};
|
||||
|
||||
@@ -586,7 +599,7 @@ impl S3 for FS {
|
||||
&self,
|
||||
req: S3Request<GetBucketRequestPaymentInput>,
|
||||
) -> S3Result<S3Response<GetBucketRequestPaymentOutput>> {
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(s3_error!(InternalError, "Not init"));
|
||||
};
|
||||
|
||||
@@ -634,7 +647,7 @@ impl S3 for FS {
|
||||
}
|
||||
|
||||
async fn get_bucket_website(&self, req: S3Request<GetBucketWebsiteInput>) -> S3Result<S3Response<GetBucketWebsiteOutput>> {
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(s3_error!(InternalError, "Not init"));
|
||||
};
|
||||
|
||||
@@ -673,7 +686,7 @@ impl S3 for FS {
|
||||
bucket, key, version_id, ..
|
||||
} = req.input;
|
||||
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
||||
};
|
||||
|
||||
@@ -703,7 +716,7 @@ impl S3 for FS {
|
||||
bucket, key, version_id, ..
|
||||
} = req.input.clone();
|
||||
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
||||
};
|
||||
|
||||
@@ -764,7 +777,7 @@ impl S3 for FS {
|
||||
record_s3_op(S3Operation::GetObjectLockConfiguration);
|
||||
let GetObjectLockConfigurationInput { bucket, .. } = req.input;
|
||||
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
||||
};
|
||||
|
||||
@@ -812,7 +825,7 @@ impl S3 for FS {
|
||||
bucket, key, version_id, ..
|
||||
} = req.input.clone();
|
||||
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
||||
};
|
||||
|
||||
@@ -864,7 +877,7 @@ impl S3 for FS {
|
||||
let bucket = req.input.bucket.as_str();
|
||||
let object = req.input.key.as_str();
|
||||
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
error!(
|
||||
component = LOG_COMPONENT_STORAGE,
|
||||
subsystem = LOG_SUBSYSTEM_TAGGING,
|
||||
@@ -1008,7 +1021,7 @@ impl S3 for FS {
|
||||
} = req.input;
|
||||
record_s3_op(S3Operation::PutBucketAcl);
|
||||
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
||||
};
|
||||
|
||||
@@ -1031,7 +1044,7 @@ impl S3 for FS {
|
||||
&self,
|
||||
req: S3Request<PutBucketAccelerateConfigurationInput>,
|
||||
) -> S3Result<S3Response<PutBucketAccelerateConfigurationOutput>> {
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(s3_error!(InternalError, "Not init"));
|
||||
};
|
||||
store
|
||||
@@ -1056,7 +1069,7 @@ impl S3 for FS {
|
||||
|
||||
async fn get_bucket_logging(&self, req: S3Request<GetBucketLoggingInput>) -> S3Result<S3Response<GetBucketLoggingOutput>> {
|
||||
record_s3_op(S3Operation::GetBucketLogging);
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(s3_error!(InternalError, "Not init"));
|
||||
};
|
||||
store
|
||||
@@ -1075,7 +1088,7 @@ impl S3 for FS {
|
||||
|
||||
async fn put_bucket_logging(&self, req: S3Request<PutBucketLoggingInput>) -> S3Result<S3Response<PutBucketLoggingOutput>> {
|
||||
record_s3_op(S3Operation::PutBucketLogging);
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(s3_error!(InternalError, "Not init"));
|
||||
};
|
||||
store
|
||||
@@ -1134,7 +1147,7 @@ impl S3 for FS {
|
||||
&self,
|
||||
req: S3Request<PutBucketRequestPaymentInput>,
|
||||
) -> S3Result<S3Response<PutBucketRequestPaymentOutput>> {
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(s3_error!(InternalError, "Not init"));
|
||||
};
|
||||
store
|
||||
@@ -1176,7 +1189,7 @@ impl S3 for FS {
|
||||
}
|
||||
|
||||
async fn put_bucket_website(&self, req: S3Request<PutBucketWebsiteInput>) -> S3Result<S3Response<PutBucketWebsiteOutput>> {
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(s3_error!(InternalError, "Not init"));
|
||||
};
|
||||
store
|
||||
@@ -1207,7 +1220,7 @@ impl S3 for FS {
|
||||
let key = &req.input.key;
|
||||
let version_id = req.input.version_id.clone();
|
||||
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
||||
};
|
||||
|
||||
@@ -1249,7 +1262,7 @@ impl S3 for FS {
|
||||
|
||||
validate_table_catalog_object_mutation(&bucket, &key).await?;
|
||||
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
||||
};
|
||||
|
||||
@@ -1303,7 +1316,7 @@ impl S3 for FS {
|
||||
|
||||
let Some(input_cfg) = object_lock_configuration else { return Err(s3_error!(InvalidArgument)) };
|
||||
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
||||
};
|
||||
|
||||
@@ -1398,7 +1411,7 @@ impl S3 for FS {
|
||||
|
||||
validate_table_catalog_object_mutation(&bucket, &key).await?;
|
||||
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
||||
};
|
||||
|
||||
@@ -1477,7 +1490,7 @@ impl S3 for FS {
|
||||
|
||||
crate::storage::s3_api::tagging::validate_object_tag_set(&tagging.tag_set)?;
|
||||
|
||||
let Some(store) = runtime_sources::current_object_store_handle() else {
|
||||
let Some(store) = self.server_ctx.object_store() else {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
||||
};
|
||||
|
||||
|
||||
@@ -22,7 +22,7 @@ use rustfs_kms::ObjectEncryptionService;
|
||||
use rustfs_lock::LockClient;
|
||||
use std::sync::Arc;
|
||||
|
||||
pub(crate) use crate::runtime_sources::AppContext;
|
||||
pub(crate) use crate::runtime_sources::{AppContext, ServerContextSlot};
|
||||
|
||||
pub(crate) fn current_app_context() -> Option<Arc<AppContext>> {
|
||||
root_runtime_sources::current_app_context()
|
||||
|
||||
@@ -518,6 +518,7 @@ pub(crate) type FileWriter = ecstore_disk::FileWriter;
|
||||
pub(crate) type FS = super::ecfs::FS;
|
||||
pub(crate) type HashReader = ecstore_rio::HashReader;
|
||||
pub(crate) type InstanceContext = ecstore_runtime::InstanceContext;
|
||||
pub(crate) type ServerContextSlot = crate::storage::runtime_sources::ServerContextSlot;
|
||||
pub(crate) type LocalPeerS3Client = ecstore_rpc::LocalPeerS3Client;
|
||||
pub(crate) type MetricType = ecstore_metrics::MetricType;
|
||||
pub(crate) type ObjectPartInfo = rustfs_filemeta::ObjectPartInfo;
|
||||
|
||||
@@ -93,7 +93,7 @@ pub(crate) mod server {
|
||||
}
|
||||
|
||||
pub(crate) mod http {
|
||||
pub(crate) use crate::storage::storage_api::{TONIC_RPC_PREFIX, verify_rpc_signature};
|
||||
pub(crate) use crate::storage::storage_api::{ServerContextSlot, TONIC_RPC_PREFIX, verify_rpc_signature};
|
||||
|
||||
pub(crate) mod ecfs {
|
||||
pub(crate) type FS = crate::storage::storage_api::FS;
|
||||
@@ -215,7 +215,7 @@ pub(crate) mod startup {
|
||||
}
|
||||
|
||||
pub(crate) mod services {
|
||||
pub(crate) use crate::storage::storage_api::{ECStore, EndpointServerPools};
|
||||
pub(crate) use crate::storage::storage_api::{ECStore, EndpointServerPools, ServerContextSlot};
|
||||
}
|
||||
|
||||
pub(crate) mod shutdown {
|
||||
|
||||
Reference in New Issue
Block a user