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

refactor: centralize startup runtime services (#3467)

Browse Source
This commit is contained in:
安正超
2026-06-15 13:24:17 +08:00
committed by GitHub
parent 3e723a2476
commit e26eea7f36
4 changed files with 340 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/main.rs
+15 -207
View File
@@ -13,39 +13,24 @@
// limitations under the License.
// Ensure the correct path for parse_license is imported
use rustfs::init::{
add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system, init_update_check, print_server_info,
};
use futures_util::future::join_all;
use rustfs::server::{
ServiceState, ServiceStateManager, ShutdownHandle, ShutdownSignal, shutdown_event_notifier, stop_audit_system,
wait_for_shutdown,
};
use rustfs::startup_iam::{bootstrap_or_defer_iam_init, publish_ready_for_iam_bootstrap};
use rustfs::startup_iam::publish_ready_for_iam_bootstrap;
use rustfs::startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight};
use rustfs::startup_protocols::{ProtocolShutdownSenders, init_protocol_shutdown_senders};
use rustfs::startup_protocols::ProtocolShutdownSenders;
use rustfs::startup_server::{StartupHttpServers, StartupListenContext, init_startup_http_servers, init_startup_listen_context};
use rustfs::startup_services::{StartupServiceRuntime, init_startup_runtime_services};
use rustfs::startup_storage::{StartupStorageRuntime, init_startup_storage_foundation, init_startup_storage_runtime};
use rustfs_ecstore::{
bucket::metadata_sys::init_bucket_metadata_sys,
bucket::migration::{try_migrate_bucket_metadata, try_migrate_iam_config},
bucket::replication::get_global_replication_pool,
global::shutdown_background_services,
store_api::BucketOperations,
};
use rustfs_heal::{
create_ahm_services_cancel_token, heal::storage::ECStoreHealStorage, init_heal_manager, shutdown_ahm_services,
};
use rustfs_iam::init_oidc_sys;
use rustfs_obs::init_metrics_runtime;
use rustfs_ecstore::global::shutdown_background_services;
use rustfs_heal::shutdown_ahm_services;
use rustfs_scanner::init_data_scanner;
use rustfs_storage_api::BucketOptions;
use rustfs_utils::get_env_bool_with_aliases;
use std::io::{Error, Result};
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, instrument, warn};
use tracing::{error, info, instrument};
const ENV_SCANNER_ENABLED: &str = "RUSTFS_SCANNER_ENABLED";
const ENV_SCANNER_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_SCANNER";
@@ -53,16 +38,9 @@ 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_SUBSYSTEM_STARTUP: &str = "startup";
const LOG_SUBSYSTEM_AUTH: &str = "auth";
const EVENT_SERVER_RUNTIME_FAILED: &str = "server_runtime_failed";
const EVENT_PROTOCOL_SYSTEM_STATE: &str = "protocol_system_state";
const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state";
const EVENT_DEADLOCK_DETECTOR_STATE: &str = "deadlock_detector_state";
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";
const EVENT_NOTIFICATION_SYSTEM_INITIALIZATION_FAILED: &str = "notification_system_initialization_failed";
const EVENT_BACKGROUND_SERVICES_CONFIGURED: &str = "background_services_configured";
const EVENT_SERVER_READY: &str = "server_ready";
const EVENT_SHUTDOWN_SIGNAL_RECEIVED: &str = "shutdown_signal_received";
const EVENT_BACKGROUND_SERVICE_SHUTDOWN: &str = "background_service_shutdown";
@@ -169,190 +147,20 @@ async fn run(config: rustfs::config::Config) -> Result<()> {
shutdown_token: ctx,
} = init_startup_storage_runtime(server_addr, &endpoint_pools, readiness.clone()).await?;
// Initialize KMS system if enabled
init_kms_system(&config).await?;
let protocol_shutdowns = init_protocol_shutdown_senders().await?;
// Initialize buffer profiling system
init_buffer_profile_system(&config);
match rustfs::startup_services::init_event_notifier_and_audit().await {
Ok(()) => info!(
target: "rustfs::main::run",
event = EVENT_AUDIT_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "started",
"Audit runtime started"
),
Err(e) => error!(
target: "rustfs::main::run",
event = EVENT_AUDIT_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "start_failed",
error = %e,
"Audit runtime failed to start"
),
}
// Initialize deadlock detector if enabled
let detector = rustfs::storage::deadlock_detector::get_deadlock_detector();
if detector.is_enabled() {
detector.start();
info!(
target: "rustfs::main::run",
event = EVENT_DEADLOCK_DETECTOR_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "started",
"Deadlock detector started"
);
} else {
info!(
target: "rustfs::main::run",
event = EVENT_DEADLOCK_DETECTOR_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "disabled",
"Deadlock detector disabled"
);
}
let buckets_list = store
.list_bucket(&BucketOptions {
no_metadata: true,
..Default::default()
})
.await
.map_err(Error::other)?;
// Collect bucket names into a vector
let buckets: Vec<String> = buckets_list.into_iter().map(|v| v.name).collect();
try_migrate_bucket_metadata(store.clone()).await;
if let Some(pool) = get_global_replication_pool() {
pool.init_resync(ctx.clone(), buckets.clone()).await?;
}
try_migrate_iam_config(store.clone()).await;
init_bucket_metadata_sys(store.clone(), buckets.clone()).await;
// 3. Initialize IAM System (Blocking load)
// This ensures data is in memory before moving forward
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(
let StartupServiceRuntime {
protocol_shutdowns,
iam_bootstrap,
enable_scanner,
} = init_startup_runtime_services(
&config,
endpoint_pools,
store.clone(),
kms_interface,
ctx.clone(),
readiness.clone(),
Some(state_manager.clone()),
Some(ctx.clone()),
state_manager.clone(),
)
.await?;
// 3a. Initialize Keystone authentication if enabled
let keystone_config = rustfs_keystone::KeystoneConfig::from_env().map_err(Error::other)?;
if keystone_config.enable {
match rustfs::auth_keystone::init_keystone_auth(keystone_config).await {
Ok(_) => info!(
event = EVENT_KEYSTONE_AUTH_INITIALIZED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_AUTH,
"Initialized Keystone authentication"
),
Err(e) => {
error!(
event = EVENT_KEYSTONE_AUTH_INITIALIZATION_FAILED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_AUTH,
error = %e,
"Failed to initialize Keystone authentication"
);
// Continue without Keystone - fall back to standard auth
}
}
}
// 3b. Initialize OIDC System (non-fatal if no providers configured)
if let Err(e) = init_oidc_sys().await {
warn!(
event = EVENT_OIDC_INITIALIZATION_FAILED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_AUTH,
error = %e,
"OIDC initialization failed; continuing without OIDC providers"
);
}
add_bucket_notification_configuration(buckets.clone()).await;
rustfs::startup_services::init_notification_system(endpoint_pools.clone())
.await
.map_err(|err| {
error!(
event = EVENT_NOTIFICATION_SYSTEM_INITIALIZATION_FAILED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
error = ?err,
"Failed to initialize notification system"
);
Error::other(err)
})?;
// Create a cancellation token for AHM services
let _ = create_ahm_services_cancel_token();
// Check environment variables to determine if scanner and heal should be enabled
let enable_scanner = get_env_bool_with_aliases(ENV_SCANNER_ENABLED, &[ENV_SCANNER_ENABLED_DEPRECATED], true);
let enable_heal = get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true);
info!(
target: "rustfs::main::run",
event = EVENT_BACKGROUND_SERVICES_CONFIGURED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
enable_scanner = enable_scanner,
enable_heal = enable_heal,
"Background services configured"
);
// Scanner depends on the heal channel/manager, so scanner implies heal.
if enable_heal || enable_scanner {
let heal_storage = Arc::new(ECStoreHealStorage::new(store.clone()));
init_heal_manager(heal_storage, None).await?;
}
if !enable_heal && !enable_scanner {
debug!(
target: "rustfs::main::run",
event = EVENT_BACKGROUND_SERVICES_CONFIGURED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
enable_scanner = false,
enable_heal = false,
ahm_state = "skipped",
reason = "disabled",
"Background services disabled"
);
}
// print server info
print_server_info();
init_update_check();
rustfs::allocator_reclaim::init_allocator_reclaim(ctx.clone());
if rustfs_obs::observability_metric_enabled() {
// Initialize metrics system
init_metrics_runtime(ctx.clone());
rustfs::memory_observability::init_memory_observability(ctx.clone());
// Initialize auto-tuner for performance optimization (optional)
rustfs::init::init_auto_tuner(ctx.clone()).await;
}
info!(
target: "rustfs::main::run",
event = EVENT_SERVER_READY,
rustfs/src/startup_services.rs
+264 -4
View File
@@ -12,11 +12,271 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::server::{init_event_notifier, start_audit_system};
use crate::{
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},
startup_protocols::{ProtocolShutdownSenders, init_protocol_shutdown_senders},
};
use rustfs_audit::AuditResult;
use rustfs_ecstore::endpoints::EndpointServerPools;
use rustfs_ecstore::notification_sys::new_global_notification_sys;
use std::future::Future;
use rustfs_common::GlobalReadiness;
use rustfs_ecstore::{
bucket::{
metadata_sys::init_bucket_metadata_sys,
migration::{try_migrate_bucket_metadata, try_migrate_iam_config},
replication::get_global_replication_pool,
},
endpoints::EndpointServerPools,
notification_sys::new_global_notification_sys,
store::ECStore,
store_api::BucketOperations,
};
use rustfs_heal::{create_ahm_services_cancel_token, heal::storage::ECStoreHealStorage, init_heal_manager};
use rustfs_iam::init_oidc_sys;
use rustfs_obs::init_metrics_runtime;
use rustfs_storage_api::BucketOptions;
use rustfs_utils::get_env_bool_with_aliases;
use std::{
future::Future,
io::{Error, Result},
sync::Arc,
};
use tokio_util::sync::CancellationToken;
use tracing::{debug, 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_SUBSYSTEM_STARTUP: &str = "startup";
const LOG_SUBSYSTEM_AUTH: &str = "auth";
const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state";
const EVENT_DEADLOCK_DETECTOR_STATE: &str = "deadlock_detector_state";
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";
const EVENT_NOTIFICATION_SYSTEM_INITIALIZATION_FAILED: &str = "notification_system_initialization_failed";
const EVENT_BACKGROUND_SERVICES_CONFIGURED: &str = "background_services_configured";
pub struct StartupServiceRuntime {
pub protocol_shutdowns: ProtocolShutdownSenders,
pub iam_bootstrap: IamBootstrapDisposition,
pub enable_scanner: bool,
}
pub async fn init_startup_runtime_services(
config: &Config,
endpoint_pools: EndpointServerPools,
store: Arc<ECStore>,
ctx: CancellationToken,
readiness: Arc<GlobalReadiness>,
state_manager: Arc<ServiceStateManager>,
) -> Result<StartupServiceRuntime> {
init_kms_system(config).await?;
let protocol_shutdowns = init_protocol_shutdown_senders().await?;
init_buffer_profile_system(config);
init_audit_runtime().await;
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?;
init_auth_integrations().await?;
init_notification_runtime(endpoint_pools, buckets).await?;
let enable_scanner = init_background_service_runtime(store.clone()).await?;
init_observability_runtime(ctx.clone()).await;
Ok(StartupServiceRuntime {
protocol_shutdowns,
iam_bootstrap,
enable_scanner,
})
}
async fn init_audit_runtime() {
match init_event_notifier_and_audit().await {
Ok(()) => info!(
target: "rustfs::main::run",
event = EVENT_AUDIT_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "started",
"Audit runtime started"
),
Err(e) => error!(
target: "rustfs::main::run",
event = EVENT_AUDIT_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "start_failed",
error = %e,
"Audit runtime failed to start"
),
}
}
fn init_deadlock_detector_runtime() {
let detector = crate::storage::deadlock_detector::get_deadlock_detector();
if detector.is_enabled() {
detector.start();
info!(
target: "rustfs::main::run",
event = EVENT_DEADLOCK_DETECTOR_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "started",
"Deadlock detector started"
);
} else {
info!(
target: "rustfs::main::run",
event = EVENT_DEADLOCK_DETECTOR_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "disabled",
"Deadlock detector disabled"
);
}
}
async fn init_bucket_metadata_runtime(store: Arc<ECStore>, ctx: CancellationToken) -> Result<Vec<String>> {
let buckets_list = store
.list_bucket(&BucketOptions {
no_metadata: true,
..Default::default()
})
.await
.map_err(Error::other)?;
let buckets: Vec<String> = buckets_list.into_iter().map(|v| v.name).collect();
try_migrate_bucket_metadata(store.clone()).await;
if let Some(pool) = get_global_replication_pool() {
pool.init_resync(ctx, buckets.clone()).await?;
}
try_migrate_iam_config(store.clone()).await;
init_bucket_metadata_sys(store, buckets.clone()).await;
Ok(buckets)
}
async fn init_iam_runtime(
store: Arc<ECStore>,
ctx: CancellationToken,
readiness: Arc<GlobalReadiness>,
state_manager: Arc<ServiceStateManager>,
) -> 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, Some(state_manager), Some(ctx)).await
}
async fn init_auth_integrations() -> Result<()> {
let keystone_config = rustfs_keystone::KeystoneConfig::from_env().map_err(Error::other)?;
if keystone_config.enable {
match crate::auth_keystone::init_keystone_auth(keystone_config).await {
Ok(_) => info!(
event = EVENT_KEYSTONE_AUTH_INITIALIZED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_AUTH,
"Initialized Keystone authentication"
),
Err(e) => {
error!(
event = EVENT_KEYSTONE_AUTH_INITIALIZATION_FAILED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_AUTH,
error = %e,
"Failed to initialize Keystone authentication"
);
}
}
}
if let Err(e) = init_oidc_sys().await {
warn!(
event = EVENT_OIDC_INITIALIZATION_FAILED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_AUTH,
error = %e,
"OIDC initialization failed; continuing without OIDC providers"
);
}
Ok(())
}
async fn init_notification_runtime(endpoint_pools: EndpointServerPools, buckets: Vec<String>) -> Result<()> {
add_bucket_notification_configuration(buckets).await;
init_notification_system(endpoint_pools).await.map_err(|err| {
error!(
event = EVENT_NOTIFICATION_SYSTEM_INITIALIZATION_FAILED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
error = ?err,
"Failed to initialize notification system"
);
Error::other(err)
})
}
async fn init_background_service_runtime(store: Arc<ECStore>) -> Result<bool> {
let _ = create_ahm_services_cancel_token();
let enable_scanner = get_env_bool_with_aliases(ENV_SCANNER_ENABLED, &[ENV_SCANNER_ENABLED_DEPRECATED], true);
let enable_heal = get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true);
info!(
target: "rustfs::main::run",
event = EVENT_BACKGROUND_SERVICES_CONFIGURED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
enable_scanner = enable_scanner,
enable_heal = enable_heal,
"Background services configured"
);
if enable_heal || enable_scanner {
let heal_storage = Arc::new(ECStoreHealStorage::new(store));
init_heal_manager(heal_storage, None).await?;
}
if !enable_heal && !enable_scanner {
debug!(
target: "rustfs::main::run",
event = EVENT_BACKGROUND_SERVICES_CONFIGURED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
enable_scanner = false,
enable_heal = false,
ahm_state = "skipped",
reason = "disabled",
"Background services disabled"
);
}
Ok(enable_scanner)
}
async fn init_observability_runtime(ctx: CancellationToken) {
print_server_info();
init_update_check();
crate::allocator_reclaim::init_allocator_reclaim(ctx.clone());
if rustfs_obs::observability_metric_enabled() {
init_metrics_runtime(ctx.clone());
crate::memory_observability::init_memory_observability(ctx.clone());
init_auto_tuner(ctx).await;
}
}
pub async fn init_event_notifier_and_audit() -> AuditResult<()> {
init_event_notifier_and_audit_with(init_event_notifier, start_audit_system).await