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

refactor: reuse embedded runtime service boundaries (#3641)

Browse Source
This commit is contained in:
安正超
2026-06-20 08:15:44 +08:00
committed by GitHub
parent 131e9dc804
commit 77d842ae34
6 changed files with 159 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/embedded.rs
+9 -85
View File
@@ -47,14 +47,13 @@
//! start a second server will return an error.
use crate::config::Config;
use crate::init::{add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system};
use crate::server::{ShutdownHandle, shutdown_event_notifier, start_http_server, stop_audit_system};
use crate::startup_iam::{bootstrap_or_defer_iam_init, publish_ready_for_iam_bootstrap};
use crate::server::{ShutdownHandle, start_http_server};
use crate::startup_iam::publish_ready_for_iam_bootstrap;
use crate::startup_server::init_embedded_startup_listen_context;
use crate::startup_services::init_embedded_startup_runtime_services;
use crate::startup_shutdown::run_embedded_shutdown_cleanup;
use crate::startup_storage::{init_embedded_startup_storage_foundation, init_embedded_startup_storage_runtime};
use crate::storage_compat::{init_bucket_metadata_sys, try_migrate_bucket_metadata, try_migrate_iam_config};
use rustfs_obs::{init_obs, set_global_guard};
use rustfs_storage_api::{BucketOperations, BucketOptions};
use rustls::crypto::aws_lc_rs::default_provider;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::path::{Path, PathBuf};
@@ -328,77 +327,15 @@ impl RustFSServerBuilder {
};
let store = storage_runtime.store;
// KMS (optional, non-fatal for embedded).
if let Err(e) = init_kms_system(&config).await {
warn!(
component = LOG_COMPONENT_EMBEDDED,
subsystem = LOG_SUBSYSTEM_EMBEDDED,
event = "embedded_optional_service_skipped",
service = "kms",
error = %e,
"Embedded optional service initialization skipped"
);
}
// Buffer profiles.
init_buffer_profile_system(&config);
if let Err(e) = crate::startup_service_components::init_event_notifier_and_audit().await {
warn!(
component = LOG_COMPONENT_EMBEDDED,
subsystem = LOG_SUBSYSTEM_EMBEDDED,
event = "embedded_optional_service_skipped",
service = "audit",
error = %e,
"Embedded optional service initialization skipped"
);
}
// Bucket listing for metadata + notification init.
let buckets: Vec<String> = store
.list_bucket(&BucketOptions {
no_metadata: true,
..Default::default()
})
.await
.map_err(|e| {
shutdown_embedded_server();
ServerError::Init(format!("list_bucket: {e}"))
})?
.into_iter()
.map(|v| v.name)
.collect();
try_migrate_bucket_metadata(store.clone()).await;
init_bucket_metadata_sys(store.clone(), buckets.clone()).await;
try_migrate_iam_config(store.clone()).await;
// IAM.
let kms_interface =
rustfs_kms::get_global_kms_service_manager().unwrap_or_else(rustfs_kms::init_global_kms_service_manager);
let iam_bootstrap =
bootstrap_or_defer_iam_init(store.clone(), kms_interface, listen_context.readiness.clone(), None, Some(ctx.clone()))
let service_runtime =
init_embedded_startup_runtime_services(&config, endpoint_pools, store, ctx.clone(), listen_context.readiness.clone())
.await
.map_err(|e| {
shutdown_embedded_server();
ServerError::Init(format!("IAM bootstrap setup: {e}"))
ServerError::Init(e.to_string())
})?;
// Bucket notifications.
add_bucket_notification_configuration(buckets.clone()).await;
if let Err(e) = crate::startup_service_components::init_notification_system(endpoint_pools.clone()).await {
warn!(
component = LOG_COMPONENT_EMBEDDED,
subsystem = LOG_SUBSYSTEM_EMBEDDED,
event = "embedded_optional_service_skipped",
service = "notification",
error = %e,
"Embedded optional service initialization skipped"
);
}
publish_ready_for_iam_bootstrap(iam_bootstrap, listen_context.readiness.as_ref(), None)
publish_ready_for_iam_bootstrap(service_runtime.iam_bootstrap, listen_context.readiness.as_ref(), None)
.await
.map_err(|e| {
shutdown_embedded_server();
@@ -505,20 +442,7 @@ impl RustFSServer {
// Cancel background services.
self.cancel_token.cancel();
// Shutdown event notifier.
shutdown_event_notifier().await;
// Stop the audit system.
if let Err(e) = stop_audit_system().await {
warn!(
component = LOG_COMPONENT_EMBEDDED,
subsystem = LOG_SUBSYSTEM_EMBEDDED,
event = "embedded_shutdown_cleanup_failed",
service = "audit",
error = %e,
"Embedded shutdown cleanup failed"
);
}
run_embedded_shutdown_cleanup().await;
// Signal HTTP server to stop.
if let Some(shutdown_handle) = self.shutdown_handle.take() {
rustfs/src/startup_service_components.rs
+66 -1
View File
@@ -17,7 +17,11 @@ use crate::storage_compat::{
try_migrate_bucket_metadata, try_migrate_iam_config,
};
use crate::{
init::{add_bucket_notification_configuration, init_auto_tuner, init_update_check, print_server_info},
config::Config,
init::{
add_bucket_notification_configuration, init_auto_tuner, init_buffer_profile_system, init_kms_system, init_update_check,
print_server_info,
},
server::{ServiceStateManager, init_event_notifier, start_audit_system},
startup_iam::{IamBootstrapDisposition, bootstrap_or_defer_iam_init},
};
@@ -41,10 +45,13 @@ const ENV_SCANNER_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_SCANNER";
const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED";
const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL";
const LOG_COMPONENT_MAIN: &str = "main";
const LOG_COMPONENT_EMBEDDED: &str = "embedded";
const LOG_SUBSYSTEM_STARTUP: &str = "startup";
const LOG_SUBSYSTEM_AUTH: &str = "auth";
const LOG_SUBSYSTEM_EMBEDDED: &str = "embedded";
const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state";
const EVENT_DEADLOCK_DETECTOR_STATE: &str = "deadlock_detector_state";
const EVENT_EMBEDDED_OPTIONAL_SERVICE_SKIPPED: &str = "embedded_optional_service_skipped";
const EVENT_KEYSTONE_AUTH_INITIALIZED: &str = "keystone_auth_initialized";
const EVENT_KEYSTONE_AUTH_INITIALIZATION_FAILED: &str = "keystone_auth_initialization_failed";
const EVENT_OIDC_INITIALIZATION_FAILED: &str = "oidc_initialization_failed";
@@ -73,6 +80,18 @@ pub(crate) async fn init_audit_runtime() {
}
}
pub(crate) async fn init_embedded_optional_service_runtime(config: &Config) {
if let Err(err) = init_kms_system(config).await {
log_embedded_optional_service_skipped("kms", err);
}
init_buffer_profile_system(config);
if let Err(err) = init_event_notifier_and_audit().await {
log_embedded_optional_service_skipped("audit", err);
}
}
pub(crate) fn init_deadlock_detector_runtime() {
let detector = crate::storage::deadlock_detector::get_deadlock_detector();
if detector.is_enabled() {
@@ -97,6 +116,24 @@ pub(crate) fn init_deadlock_detector_runtime() {
}
}
pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc<ECStore>) -> Result<Vec<String>> {
let buckets_list = store
.list_bucket(&BucketOptions {
no_metadata: true,
..Default::default()
})
.await
.map_err(|err| Error::other(format!("list_bucket: {err}")))?;
let buckets: Vec<String> = buckets_list.into_iter().map(|v| v.name).collect();
try_migrate_bucket_metadata(store.clone()).await;
init_bucket_metadata_sys(store.clone(), buckets.clone()).await;
try_migrate_iam_config(store).await;
Ok(buckets)
}
pub(crate) async fn init_bucket_metadata_runtime(store: Arc<ECStore>, ctx: CancellationToken) -> Result<Vec<String>> {
let buckets_list = store
.list_bucket(&BucketOptions {
@@ -120,6 +157,15 @@ pub(crate) async fn init_bucket_metadata_runtime(store: Arc<ECStore>, ctx: Cance
Ok(buckets)
}
pub(crate) async fn init_embedded_iam_runtime(
store: Arc<ECStore>,
ctx: CancellationToken,
readiness: Arc<GlobalReadiness>,
) -> Result<IamBootstrapDisposition> {
let kms_interface = rustfs_kms::get_global_kms_service_manager().unwrap_or_else(rustfs_kms::init_global_kms_service_manager);
bootstrap_or_defer_iam_init(store, kms_interface, readiness, None, Some(ctx)).await
}
pub(crate) async fn init_iam_runtime(
store: Arc<ECStore>,
ctx: CancellationToken,
@@ -130,6 +176,14 @@ pub(crate) async fn init_iam_runtime(
bootstrap_or_defer_iam_init(store, kms_interface, readiness, Some(state_manager), Some(ctx)).await
}
pub(crate) async fn init_embedded_notification_runtime(endpoint_pools: EndpointServerPools, buckets: Vec<String>) {
add_bucket_notification_configuration(buckets).await;
if let Err(err) = init_notification_system(endpoint_pools).await {
log_embedded_optional_service_skipped("notification", err);
}
}
pub(crate) async fn init_auth_integrations() -> Result<()> {
let keystone_config = rustfs_keystone::KeystoneConfig::from_env().map_err(Error::other)?;
if keystone_config.enable {
@@ -261,6 +315,17 @@ where
init_notification().await
}
fn log_embedded_optional_service_skipped(service: &str, err: impl std::fmt::Display) {
warn!(
component = LOG_COMPONENT_EMBEDDED,
subsystem = LOG_SUBSYSTEM_EMBEDDED,
event = EVENT_EMBEDDED_OPTIONAL_SERVICE_SKIPPED,
service,
error = %err,
"Embedded optional service initialization skipped"
);
}
#[cfg(test)]
mod tests {
use super::{init_event_notifier_and_audit_with, init_notification_system_with};
rustfs/src/startup_services.rs
+24 -1
View File
@@ -21,7 +21,9 @@ use crate::{
startup_optional_runtime_sidecars::{OptionalRuntimeServices, init_optional_runtime_services},
startup_service_components::{
init_audit_runtime, init_auth_integrations, init_background_service_runtime, init_bucket_metadata_runtime,
init_deadlock_detector_runtime, init_iam_runtime, init_notification_runtime, init_observability_runtime,
init_deadlock_detector_runtime, init_embedded_bucket_metadata_runtime, init_embedded_iam_runtime,
init_embedded_notification_runtime, init_embedded_optional_service_runtime, init_iam_runtime, init_notification_runtime,
init_observability_runtime,
},
};
use rustfs_common::GlobalReadiness;
@@ -34,6 +36,27 @@ pub struct StartupServiceRuntime {
pub enable_scanner: bool,
}
pub struct EmbeddedStartupServiceRuntime {
pub iam_bootstrap: IamBootstrapDisposition,
}
pub async fn init_embedded_startup_runtime_services(
config: &Config,
endpoint_pools: EndpointServerPools,
store: Arc<ECStore>,
ctx: CancellationToken,
readiness: Arc<GlobalReadiness>,
) -> 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)
.await
.map_err(|err| std::io::Error::other(format!("IAM bootstrap setup: {err}")))?;
init_embedded_notification_runtime(endpoint_pools, buckets).await;
Ok(EmbeddedStartupServiceRuntime { iam_bootstrap })
}
pub async fn init_startup_runtime_services(
config: &Config,
endpoint_pools: EndpointServerPools,
rustfs/src/startup_shutdown.rs
+19 -1
View File
@@ -22,15 +22,18 @@ use crate::{
use rustfs_heal::shutdown_ahm_services;
use rustfs_utils::get_env_bool_with_aliases;
use tokio_util::sync::CancellationToken;
use tracing::{error, info};
use tracing::{error, info, warn};
const ENV_SCANNER_ENABLED: &str = "RUSTFS_SCANNER_ENABLED";
const ENV_SCANNER_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_SCANNER";
const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED";
const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL";
const LOG_COMPONENT_MAIN: &str = "main";
const LOG_COMPONENT_EMBEDDED: &str = "embedded";
const LOG_SUBSYSTEM_STARTUP: &str = "startup";
const LOG_SUBSYSTEM_EMBEDDED: &str = "embedded";
const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state";
const EVENT_EMBEDDED_SHUTDOWN_CLEANUP_FAILED: &str = "embedded_shutdown_cleanup_failed";
const EVENT_SHUTDOWN_SIGNAL_RECEIVED: &str = "shutdown_signal_received";
const EVENT_BACKGROUND_SERVICE_SHUTDOWN: &str = "background_service_shutdown";
const EVENT_EVENT_NOTIFIER_SHUTDOWN: &str = "event_notifier_shutdown";
@@ -196,6 +199,21 @@ pub async fn run_startup_shutdown_sequence(
);
}
pub async fn run_embedded_shutdown_cleanup() {
shutdown_event_notifier().await;
if let Err(err) = stop_audit_system().await {
warn!(
component = LOG_COMPONENT_EMBEDDED,
subsystem = LOG_SUBSYSTEM_EMBEDDED,
event = EVENT_EMBEDDED_SHUTDOWN_CLEANUP_FAILED,
service = "audit",
error = %err,
"Embedded shutdown cleanup failed"
);
}
}
#[cfg(test)]
mod tests {
use super::{BackgroundShutdownStep, background_shutdown_steps};