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

refactor: centralize startup service bootstrap (#3448)

Browse Source
This commit is contained in:
安正超
2026-06-14 21:59:17 +08:00
committed by GitHub
parent b2e6cc520b
commit 9499391370
5 changed files with 165 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/embedded.rs
+3 -11
View File
@@ -48,9 +48,7 @@
use crate::config::Config;
use crate::init::{add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system};
use crate::server::{
ShutdownHandle, init_event_notifier, shutdown_event_notifier, start_audit_system, start_http_server, stop_audit_system,
};
use crate::server::{ShutdownHandle, shutdown_event_notifier, start_http_server, stop_audit_system};
use crate::startup_fs_guard::enforce_unsupported_fs_policy;
use crate::startup_iam::{bootstrap_or_defer_iam_init, publish_ready_for_iam_bootstrap};
use rustfs_common::{GlobalReadiness, SystemStage, set_global_addr};
@@ -65,7 +63,6 @@ use rustfs_ecstore::{
config as ecconfig,
endpoints::EndpointServerPools,
global::set_global_rustfs_port,
notification_sys::new_global_notification_sys,
set_global_endpoints,
store::ECStore,
store::init_local_disks,
@@ -435,11 +432,7 @@ impl RustFSServerBuilder {
// Buffer profiles.
init_buffer_profile_system(&config);
// Event notifier.
init_event_notifier().await;
// Audit (non-fatal).
if let Err(e) = start_audit_system().await {
if let Err(e) = crate::startup_services::init_event_notifier_and_audit().await {
warn!(
component = LOG_COMPONENT_EMBEDDED,
subsystem = LOG_SUBSYSTEM_EMBEDDED,
@@ -482,8 +475,7 @@ impl RustFSServerBuilder {
// Bucket notifications.
add_bucket_notification_configuration(buckets.clone()).await;
// Notification system.
if let Err(e) = new_global_notification_sys(endpoint_pools.clone()).await {
if let Err(e) = crate::startup_services::init_notification_system(endpoint_pools.clone()).await {
warn!(
component = LOG_COMPONENT_EMBEDDED,
subsystem = LOG_SUBSYSTEM_EMBEDDED,
rustfs/src/lib.rs
+1
View File
@@ -69,6 +69,7 @@ pub mod protocols;
pub mod server;
pub mod startup_fs_guard;
pub mod startup_iam;
pub mod startup_services;
pub mod storage;
pub(crate) mod table_catalog;
pub mod tls;
rustfs/src/main.rs
+16 -20
View File
@@ -30,8 +30,8 @@ use futures_util::future::join_all;
use rustfs::capacity::capacity_integration::init_capacity_management;
use rustfs::license::{init_license, license_status};
use rustfs::server::{
ServiceState, ServiceStateManager, ShutdownHandle, ShutdownSignal, init_event_notifier, shutdown_event_notifier,
start_audit_system, start_http_server, stop_audit_system, wait_for_shutdown,
ServiceState, ServiceStateManager, ShutdownHandle, ShutdownSignal, shutdown_event_notifier, start_http_server,
stop_audit_system, wait_for_shutdown,
};
use rustfs::startup_fs_guard::enforce_unsupported_fs_policy;
use rustfs::startup_iam::{bootstrap_or_defer_iam_init, publish_ready_for_iam_bootstrap};
@@ -45,7 +45,6 @@ use rustfs_ecstore::{
config as ecconfig,
endpoints::EndpointServerPools,
global::{set_global_rustfs_port, shutdown_background_services},
notification_sys::new_global_notification_sys,
set_global_endpoints,
store::ECStore,
store::init_local_disks,
@@ -778,12 +777,8 @@ async fn run(config: rustfs::config::Config) -> Result<()> {
// Initialize buffer profiling system
init_buffer_profile_system(&config);
// Initialize event notifier
init_event_notifier().await;
// Start the audit system
match start_audit_system().await {
Ok(_) => info!(
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,
@@ -893,17 +888,18 @@ async fn run(config: rustfs::config::Config) -> Result<()> {
add_bucket_notification_configuration(buckets.clone()).await;
// Initialize the global notification system
new_global_notification_sys(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)
})?;
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();
rustfs/src/startup_services.rs
+106
View File
@@ -0,0 +1,106 @@
// 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.
use crate::server::{init_event_notifier, start_audit_system};
use rustfs_audit::AuditResult;
use rustfs_ecstore::endpoints::EndpointServerPools;
use rustfs_ecstore::notification_sys::new_global_notification_sys;
use std::future::Future;
pub async fn init_event_notifier_and_audit() -> AuditResult<()> {
init_event_notifier_and_audit_with(init_event_notifier, start_audit_system).await
}
async fn init_event_notifier_and_audit_with<NotifyFn, NotifyFuture, AuditFn, AuditFuture>(
notify: NotifyFn,
start_audit: AuditFn,
) -> AuditResult<()>
where
NotifyFn: FnOnce() -> NotifyFuture,
NotifyFuture: Future<Output = ()>,
AuditFn: FnOnce() -> AuditFuture,
AuditFuture: Future<Output = AuditResult<()>>,
{
notify().await;
start_audit().await
}
pub async fn init_notification_system(endpoint_pools: EndpointServerPools) -> rustfs_ecstore::error::Result<()> {
init_notification_system_with(|| new_global_notification_sys(endpoint_pools)).await
}
async fn init_notification_system_with<InitFn, InitFuture>(init_notification: InitFn) -> rustfs_ecstore::error::Result<()>
where
InitFn: FnOnce() -> InitFuture,
InitFuture: Future<Output = rustfs_ecstore::error::Result<()>>,
{
init_notification().await
}
#[cfg(test)]
mod tests {
use super::{init_event_notifier_and_audit_with, init_notification_system_with};
use rustfs_audit::AuditError;
use std::sync::{Arc, Mutex};
#[tokio::test]
async fn event_notifier_runs_before_successful_audit_start() {
let events = Arc::new(Mutex::new(Vec::new()));
let notify_events = events.clone();
let audit_events = events.clone();
let result = init_event_notifier_and_audit_with(
move || async move {
notify_events.lock().unwrap_or_else(|err| err.into_inner()).push("notify");
},
move || async move {
audit_events.lock().unwrap_or_else(|err| err.into_inner()).push("audit");
Ok(())
},
)
.await;
assert!(result.is_ok());
let events = events.lock().unwrap_or_else(|err| err.into_inner()).clone();
assert_eq!(events, ["notify", "audit"]);
}
#[tokio::test]
async fn event_notifier_runs_before_failed_audit_result() {
let events = Arc::new(Mutex::new(Vec::new()));
let notify_events = events.clone();
let audit_events = events.clone();
let result = init_event_notifier_and_audit_with(
move || async move {
notify_events.lock().unwrap_or_else(|err| err.into_inner()).push("notify");
},
move || async move {
audit_events.lock().unwrap_or_else(|err| err.into_inner()).push("audit");
Err(AuditError::ConfigNotLoaded)
},
)
.await;
assert!(result.is_err());
let events = events.lock().unwrap_or_else(|err| err.into_inner()).clone();
assert_eq!(events, ["notify", "audit"]);
}
#[tokio::test]
async fn notification_system_returns_source_error() {
let result = init_notification_system_with(|| async { Err(rustfs_ecstore::error::Error::FaultyDisk) }).await;
assert!(result.is_err());
}
}