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

refactor: centralize startup lifecycle runtime (#3468)

Browse Source
This commit is contained in:
安正超
2026-06-15 14:07:24 +08:00
committed by GitHub
parent e26eea7f36
commit 218f473fb5
4 changed files with 379 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/architecture/migration-progress.md
+38 -20
View File
@@ -5,18 +5,16 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
## Current Context ## Current Context
- Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660) - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660)
- Branch: `overtrue/arch-startup-runtime-services` - Branch: `overtrue/arch-startup-ready-shutdown-runtime`
- Baseline: `origin/main` at `3e723a2476692051cac1eab2cc31984c62063130` - Baseline: `origin/main` at `e26eea7f3a5ef6c272a322a47f58f5a10cb2342b0`
- PR type for this branch: `pure-move` - PR type for this branch: `pure-move`
- Runtime behavior changes: no external behavior change expected; KMS, optional - Runtime behavior changes: no external behavior change expected; server-ready
protocols, diagnostics services, bucket metadata migration, IAM bootstrap, publication, global init time, scanner start, shutdown wait, and shutdown
auth integrations, notification runtime, heal/scanner setup, and metrics ordering still run in the same relative order after runtime services.
initialization still run in the same relative order after storage runtime and - Rust code changes: add `startup_services::run_startup_runtime_lifecycle` and
before the server-ready log.
- Rust code changes: add `startup_services::init_startup_runtime_services` and
use it from binary startup. use it from binary startup.
- CI/script changes: none. - CI/script changes: none.
- Docs changes: record `R-017` startup runtime service bootstrap progress and - Docs changes: record `R-018` startup runtime lifecycle progress and
verification. verification.
## Phase 0 Tasks ## Phase 0 Tasks
@@ -704,10 +702,29 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
formatting, migration guards, Rust risk scan, branch freshness check, and formatting, migration guards, Rust risk scan, branch freshness check, and
pre-commit quality gate. pre-commit quality gate.
- [x] `R-018` Centralize startup ready, scanner, and shutdown lifecycle.
- Do: move server-ready logging, IAM readiness publication, global init time,
scanner start, shutdown signal wait, background shutdown ordering, protocol
shutdown, notifier/audit/profiling shutdown, HTTP shutdown, and final stopped
state logging behind the `startup_services` boundary.
- Acceptance: `main.rs` still initializes listen/storage/runtime services in
the same order, then delegates lifecycle completion; `startup_services`
owns the shutdown handles, runtime token, readiness handle, store, and
service runtime needed for ready/scanner/shutdown orchestration.
- Must preserve: server-ready log fields, inline/deferred IAM readiness
behavior, global init time timing, scanner start timing, shutdown signal log,
runtime token cancellation before service-specific shutdown, scanner before
AHM shutdown order, protocol shutdown order, notifier/audit/profiling
shutdown order, HTTP shutdown order, stopped service state, and final stopped
logs.
- Verification: focused startup services tests, binary/lib compile checks,
formatting, migration guards, Rust risk scan, branch freshness check, and
pre-commit quality gate.
## Next PRs ## Next PRs
1. `pure-move`: continue extracting startup ready/scanner/shutdown wrappers while 1. `pure-move`: continue collapsing remaining binary startup glue around command
preserving startup order, readiness ownership, and shutdown ownership. dispatch and server bootstrap without changing fatal stderr behavior.
2. `ci-gate`: finish `G-006` public re-export and storage trait coverage checks 2. `ci-gate`: finish `G-006` public re-export and storage trait coverage checks
before the remaining cleanup slices. before the remaining cleanup slices.
@@ -715,13 +732,13 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
| Expert | Status | Notes | | Expert | Status | Notes |
|---|---|---| |---|---|---|
| Quality/architecture | passed | Pure-move slice removes runtime service details from binary startup behind the existing startup services boundary. | | Quality/architecture | passed | Pure-move slice removes ready/scanner/shutdown lifecycle details from binary startup behind the existing startup services boundary. |
| Migration preservation | passed | Storage runtime, KMS, optional protocols, diagnostics, bucket metadata, IAM, auth, notification, heal/scanner, metrics, and server-ready order are preserved. | | Migration preservation | passed | Server-ready publication, scanner start, shutdown wait, background/protocol/notifier/audit/profiling/HTTP shutdown order, and stopped state are preserved. |
| Testing/verification | passed | Focused startup services tests, compile checks, formatting, migration/layer guards, Rust risk scan, branch freshness check, and full `make pre-commit` passed. | | Testing/verification | passed | Focused startup services tests, compile checks, formatting, migration/layer guards, Rust risk scan, branch freshness check, and full `make pre-commit` passed. |
## Verification Notes ## Verification Notes
Passed on `3e723a2476692051cac1eab2cc31984c62063130`: Passed on `e26eea7f3a5ef6c272a322a47f58f5a10cb2342b0`:
- `cargo test -p rustfs startup_services --no-fail-fast`: passed. - `cargo test -p rustfs startup_services --no-fail-fast`: passed.
- `cargo check -p rustfs --lib`: passed. - `cargo check -p rustfs --lib`: passed.
@@ -740,14 +757,15 @@ Passed on `3e723a2476692051cac1eab2cc31984c62063130`:
Notes: Notes:
- This slice centralizes startup runtime services without changing startup - This slice centralizes startup lifecycle completion without changing startup
ordering, shutdown token ownership, readiness ownership, or endpoint pool ordering, shutdown token ownership, readiness ownership, or shutdown handle
ownership. ownership.
- Runtime services remain after storage runtime and before the server-ready log. - Ready publication and scanner start remain after runtime services; shutdown
wait remains after scanner start.
## Handoff Notes ## Handoff Notes
- R-017 is implemented, locally verified, and current with `origin/main`. - R-018 is implemented, locally verified, and current with `origin/main`.
- Next startup slices can keep using larger pure moves, but must keep startup - Next startup slices can keep using larger pure moves, but must keep startup
ordering, fatal/non-fatal boundaries, shutdown ownership, and readiness ordering, fatal/non-fatal boundaries, shutdown ownership, readiness ownership,
ownership explicit in tests and review notes. and fatal stderr behavior explicit in tests and review notes.
docs/architecture/startup-timeline.md
+8 -8
View File
@@ -44,9 +44,9 @@ new startup semantics.
| `RUN-019` | `rustfs/src/startup_services.rs` | Add bucket notification config and initialize notification system. | Adds bucket notification configuration and publishes the global notification system. | Notification config add is non-fatal in this path; global notification init is fatal on error. | None | | `RUN-019` | `rustfs/src/startup_services.rs` | Add bucket notification config and initialize notification system. | Adds bucket notification configuration and publishes the global notification system. | Notification config add is non-fatal in this path; global notification init is fatal on error. | None |
| `RUN-020` | `rustfs/src/startup_services.rs` | Create AHM cancellation token and initialize heal manager when scanner or heal is enabled. | Creates AHM cancellation token and starts heal manager for heal/scanner workflows. | Heal manager init is fatal when enabled. | None | | `RUN-020` | `rustfs/src/startup_services.rs` | Create AHM cancellation token and initialize heal manager when scanner or heal is enabled. | Creates AHM cancellation token and starts heal manager for heal/scanner workflows. | Heal manager init is fatal when enabled. | None |
| `RUN-021` | `rustfs/src/startup_services.rs` | Print server info, init update check, allocator reclaim, metrics, memory observability, and auto-tuner. | Starts informational/update/memory/metrics background tasks when enabled. | Non-fatal in this path. | None | | `RUN-021` | `rustfs/src/startup_services.rs` | Print server info, init update check, allocator reclaim, metrics, memory observability, and auto-tuner. | Starts informational/update/memory/metrics background tasks when enabled. | Non-fatal in this path. | None |
| `RUN-022` | `rustfs/src/main.rs:599` | Log successful startup and publish full readiness for inline IAM. | Logs version/address, checks runtime readiness, marks `FullReady`, and sets service state to `Ready` when IAM was ready inline. | Fatal if runtime readiness is not reached within the startup wait. | Marks `FullReady` only for inline IAM here | | `RUN-022` | `rustfs/src/startup_services.rs` and `rustfs/src/startup_iam.rs` | Log successful startup and publish full readiness for inline IAM. | Logs version/address, checks runtime readiness, marks `FullReady`, and sets service state to `Ready` when IAM was ready inline. | Fatal if runtime readiness is not reached within the startup wait. | Marks `FullReady` only for inline IAM here |
| `RUN-023` | `rustfs/src/main.rs:609` | Publish global init time and start data scanner when enabled. | Sets global init time and starts scanner after the successful-startup log. | Scanner start is non-fatal in this path. | Full readiness may already be published or may await deferred IAM recovery | | `RUN-023` | `rustfs/src/startup_services.rs` | Publish global init time and start data scanner when enabled. | Sets global init time and starts scanner after the successful-startup log. | Scanner start is non-fatal in this path. | Full readiness may already be published or may await deferred IAM recovery |
| `RUN-024` | `rustfs/src/main.rs:616` | Wait for shutdown signal. | Blocks the main task until a shutdown signal is received. | Non-fatal. | Runtime remains in its current readiness state | | `RUN-024` | `rustfs/src/startup_services.rs` | Wait for shutdown signal. | Blocks the main task until a shutdown signal is received. | Non-fatal. | Runtime remains in its current readiness state |
## Deferred IAM Readiness ## Deferred IAM Readiness
@@ -68,11 +68,11 @@ new startup semantics.
| Step | Source | Current action | Side effects | Fatal boundary | Ready stage | | Step | Source | Current action | Side effects | Fatal boundary | Ready stage |
|---|---|---|---|---|---| |---|---|---|---|---|---|
| `STOP-001` | `rustfs/src/main.rs:647` | Cancel runtime token and move service state to `Stopping`. | Notifies cancellation-aware background tasks. | Non-fatal. | Service state moves to `Stopping`; readiness stages are not cleared here | | `STOP-001` | `rustfs/src/startup_services.rs` | Cancel runtime token and move service state to `Stopping`. | Notifies cancellation-aware background tasks. | Non-fatal. | Service state moves to `Stopping`; readiness stages are not cleared here |
| `STOP-002` | `rustfs/src/main.rs:675` | Stop scanner/background services and AHM services according to enable flags. | Calls ECStore background shutdown and heal/scanner shutdown helpers. | Non-fatal in this path. | No readiness-stage change | | `STOP-002` | `rustfs/src/startup_services.rs` | Stop scanner/background services and AHM services according to enable flags. | Calls ECStore background shutdown and heal/scanner shutdown helpers. | Non-fatal in this path. | No readiness-stage change |
| `STOP-003` | `rustfs/src/main.rs:699` | Signal optional FTP/FTPS/WebDAV/SFTP protocol servers. | Collects protocol shutdown futures. | Non-fatal in this path. | No readiness-stage change | | `STOP-003` | `rustfs/src/startup_services.rs` | Signal optional FTP/FTPS/WebDAV/SFTP protocol servers. | Collects protocol shutdown futures. | Non-fatal in this path. | No readiness-stage change |
| `STOP-004` | `rustfs/src/main.rs:735` | Stop event notifier, audit system, and profiling tasks. | Stops notifier and profiling tasks; audit stop failures are logged. | Non-fatal in this path. | No readiness-stage change | | `STOP-004` | `rustfs/src/startup_services.rs` | Stop event notifier, audit system, and profiling tasks. | Stops notifier and profiling tasks; audit stop failures are logged. | Non-fatal in this path. | No readiness-stage change |
| `STOP-005` | `rustfs/src/main.rs:763` | Stop S3 and console HTTP servers, wait for protocol shutdowns, then mark service state `Stopped`. | HTTP shutdown happens after notifier/audit/profiling shutdown in current order. | Join failures are logged by shutdown handles; this path does not return errors. | Service state moves to `Stopped`; readiness stages are not cleared here | | `STOP-005` | `rustfs/src/startup_services.rs` | Stop S3 and console HTTP servers, wait for protocol shutdowns, then mark service state `Stopped`. | HTTP shutdown happens after notifier/audit/profiling shutdown in current order. | Join failures are logged by shutdown handles; this path does not return errors. | Service state moves to `Stopped`; readiness stages are not cleared here |
## Migration Rules ## Migration Rules
rustfs/src/main.rs
+12 -312
View File
@@ -13,40 +13,16 @@
// limitations under the License. // limitations under the License.
// Ensure the correct path for parse_license is imported // Ensure the correct path for parse_license is imported
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::publish_ready_for_iam_bootstrap;
use rustfs::startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight}; use rustfs::startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight};
use rustfs::startup_protocols::ProtocolShutdownSenders;
use rustfs::startup_server::{StartupHttpServers, StartupListenContext, init_startup_http_servers, init_startup_listen_context}; 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_services::{StartupRuntimeLifecycle, init_startup_runtime_services, run_startup_runtime_lifecycle};
use rustfs::startup_storage::{StartupStorageRuntime, init_startup_storage_foundation, init_startup_storage_runtime}; use rustfs::startup_storage::{StartupStorageRuntime, init_startup_storage_foundation, init_startup_storage_runtime};
use rustfs_ecstore::global::shutdown_background_services;
use rustfs_heal::shutdown_ahm_services;
use rustfs_scanner::init_data_scanner;
use rustfs_utils::get_env_bool_with_aliases;
use std::io::{Error, Result}; use std::io::{Error, Result};
use tokio_util::sync::CancellationToken; use tracing::{error, instrument};
use tracing::{error, info, instrument};
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_MAIN: &str = "main";
const LOG_SUBSYSTEM_STARTUP: &str = "startup"; const LOG_SUBSYSTEM_STARTUP: &str = "startup";
const EVENT_SERVER_RUNTIME_FAILED: &str = "server_runtime_failed"; 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_SERVER_READY: &str = "server_ready";
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";
const EVENT_PROFILING_SHUTDOWN: &str = "profiling_shutdown";
const EVENT_SERVER_SHUTDOWN_STATE: &str = "server_shutdown_state";
const OBSERVABILITY_INIT_FATAL_ALREADY_REPORTED: &str = "observability initialization failure already reported"; const OBSERVABILITY_INIT_FATAL_ALREADY_REPORTED: &str = "observability initialization failure already reported";
#[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))] #[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
#[global_allocator] #[global_allocator]
@@ -147,11 +123,7 @@ async fn run(config: rustfs::config::Config) -> Result<()> {
shutdown_token: ctx, shutdown_token: ctx,
} = init_startup_storage_runtime(server_addr, &endpoint_pools, readiness.clone()).await?; } = init_startup_storage_runtime(server_addr, &endpoint_pools, readiness.clone()).await?;
let StartupServiceRuntime { let service_runtime = init_startup_runtime_services(
protocol_shutdowns,
iam_bootstrap,
enable_scanner,
} = init_startup_runtime_services(
&config, &config,
endpoint_pools, endpoint_pools,
store.clone(), store.clone(),
@@ -161,275 +133,17 @@ async fn run(config: rustfs::config::Config) -> Result<()> {
) )
.await?; .await?;
info!( run_startup_runtime_lifecycle(StartupRuntimeLifecycle {
target: "rustfs::main::run", server_address,
event = EVENT_SERVER_READY, state_manager,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
version = %rustfs::version::get_version(),
server_address = %server_address,
started_at = %jiff::Zoned::now(),
iam_bootstrap = ?iam_bootstrap,
"RustFS server ready"
);
publish_ready_for_iam_bootstrap(iam_bootstrap, readiness.as_ref(), Some(state_manager.as_ref())).await?;
// Set the global RustFS initialization time to now
rustfs_common::set_global_init_time_now().await;
if enable_scanner {
init_data_scanner(ctx.clone(), store.clone()).await;
}
// listen to the shutdown signal
let shutdown_signal = wait_for_shutdown().await;
handle_shutdown(
&state_manager,
shutdown_signal,
s3_shutdown_tx, s3_shutdown_tx,
console_shutdown_tx, console_shutdown_tx,
protocol_shutdowns, service_runtime,
ctx.clone(), store,
) shutdown_token: ctx,
.await; readiness,
})
info!( .await
target: "rustfs::main::run",
event = EVENT_SERVER_SHUTDOWN_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = ?state_manager.current_state(),
result = "stopped",
"RustFS server stopped"
);
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum BackgroundShutdownStep {
DataScanner,
Ahm,
}
fn background_shutdown_steps(enable_scanner: bool, enable_heal: bool) -> Vec<BackgroundShutdownStep> {
let mut steps = Vec::with_capacity(2);
if enable_scanner {
steps.push(BackgroundShutdownStep::DataScanner);
}
if enable_heal || enable_scanner {
steps.push(BackgroundShutdownStep::Ahm);
}
steps
}
/// Handles the shutdown process of the server
async fn handle_shutdown(
state_manager: &ServiceStateManager,
shutdown_signal: ShutdownSignal,
s3_shutdown_handle: Option<ShutdownHandle>,
console_shutdown_handle: Option<ShutdownHandle>,
protocols: ProtocolShutdownSenders,
ctx: CancellationToken,
) {
let ProtocolShutdownSenders {
ftp: ftp_shutdown_tx,
ftps: ftps_shutdown_tx,
webdav: webdav_shutdown_tx,
sftp: sftp_shutdown_tx,
} = protocols;
ctx.cancel();
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_SHUTDOWN_SIGNAL_RECEIVED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
signal = shutdown_signal.log_label(),
"Shutdown signal received"
);
// update the status to stopping first
state_manager.update(ServiceState::Stopping);
// Check environment variables to determine what services need to be stopped
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);
// Stop background services based on what was enabled.
let background_steps = background_shutdown_steps(enable_scanner, enable_heal);
for step in &background_steps {
match step {
BackgroundShutdownStep::DataScanner => {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_BACKGROUND_SERVICE_SHUTDOWN,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
service = "data_scanner",
state = "stopping",
"Background service shutdown started"
);
shutdown_background_services();
}
BackgroundShutdownStep::Ahm => {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_BACKGROUND_SERVICE_SHUTDOWN,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
service = "ahm",
state = "stopping",
"Background service shutdown started"
);
shutdown_ahm_services();
}
}
}
if background_steps.is_empty() {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_BACKGROUND_SERVICE_SHUTDOWN,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
service = "ahm",
state = "skipped",
reason = "disabled",
"Background service shutdown skipped"
);
}
// Shutdown FTP and FTPS servers
let mut protocol_shutdowns = Vec::new();
if let Some(ftp_shutdown_tx) = ftp_shutdown_tx {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_PROTOCOL_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
protocol = "ftp",
state = "stopping",
"Protocol runtime stopping"
);
protocol_shutdowns.push(ftp_shutdown_tx.shutdown());
}
if let Some(ftps_shutdown_tx) = ftps_shutdown_tx {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_PROTOCOL_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
protocol = "ftps",
state = "stopping",
"Protocol runtime stopping"
);
protocol_shutdowns.push(ftps_shutdown_tx.shutdown());
}
// Shutdown WebDAV server
if let Some(webdav_shutdown_tx) = webdav_shutdown_tx {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_PROTOCOL_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
protocol = "webdav",
state = "stopping",
"Protocol runtime stopping"
);
protocol_shutdowns.push(webdav_shutdown_tx.shutdown());
}
// Shutdown SFTP server
if let Some(sftp_shutdown_tx) = sftp_shutdown_tx {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_PROTOCOL_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
protocol = "sftp",
state = "stopping",
"Protocol runtime stopping"
);
protocol_shutdowns.push(sftp_shutdown_tx.shutdown());
}
// Stop the notification system
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_EVENT_NOTIFIER_SHUTDOWN,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stopping",
"Event notifier shutdown started"
);
shutdown_event_notifier().await;
// Stop the audit system
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_AUDIT_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stopping",
"Audit runtime stopping"
);
match stop_audit_system().await {
Ok(_) => info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_AUDIT_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stopped",
"Audit runtime stopped"
),
Err(e) => error!(
target: "rustfs::main::handle_shutdown",
event = EVENT_AUDIT_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stop_failed",
error = %e,
"Audit runtime failed to stop"
),
}
// Stop profiling tasks
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_PROFILING_SHUTDOWN,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stopping",
"Profiling shutdown started"
);
rustfs::profiling::shutdown_profiling();
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_SERVER_SHUTDOWN_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stopping",
"RustFS server stopping"
);
if let Some(s3_shutdown_handle) = s3_shutdown_handle {
s3_shutdown_handle.shutdown().await;
}
if let Some(console_shutdown_handle) = console_shutdown_handle {
console_shutdown_handle.shutdown().await;
}
join_all(protocol_shutdowns).await;
// the last updated status is stopped
state_manager.update(ServiceState::Stopped);
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_SERVER_SHUTDOWN_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stopped",
"RustFS server stopped"
);
} }
#[cfg(test)] #[cfg(test)]
@@ -443,18 +157,4 @@ mod tests {
"[FATAL] Observability initialization failed: collector unavailable" "[FATAL] Observability initialization failed: collector unavailable"
); );
} }
#[test]
fn background_shutdown_plan_keeps_scanner_before_ahm() {
assert_eq!(
background_shutdown_steps(true, true),
vec![BackgroundShutdownStep::DataScanner, BackgroundShutdownStep::Ahm]
);
assert_eq!(
background_shutdown_steps(true, false),
vec![BackgroundShutdownStep::DataScanner, BackgroundShutdownStep::Ahm]
);
assert_eq!(background_shutdown_steps(false, true), vec![BackgroundShutdownStep::Ahm]);
assert!(background_shutdown_steps(false, false).is_empty());
}
} }
rustfs/src/startup_services.rs
+321 -4
View File
@@ -18,10 +18,14 @@ use crate::{
add_bucket_notification_configuration, init_auto_tuner, init_buffer_profile_system, init_kms_system, init_update_check, add_bucket_notification_configuration, init_auto_tuner, init_buffer_profile_system, init_kms_system, init_update_check,
print_server_info, print_server_info,
}, },
server::{ServiceStateManager, init_event_notifier, start_audit_system}, server::{
startup_iam::{IamBootstrapDisposition, bootstrap_or_defer_iam_init}, ServiceState, ServiceStateManager, ShutdownHandle, ShutdownSignal, init_event_notifier, shutdown_event_notifier,
start_audit_system, stop_audit_system, wait_for_shutdown,
},
startup_iam::{IamBootstrapDisposition, bootstrap_or_defer_iam_init, publish_ready_for_iam_bootstrap},
startup_protocols::{ProtocolShutdownSenders, init_protocol_shutdown_senders}, startup_protocols::{ProtocolShutdownSenders, init_protocol_shutdown_senders},
}; };
use futures_util::future::join_all;
use rustfs_audit::AuditResult; use rustfs_audit::AuditResult;
use rustfs_common::GlobalReadiness; use rustfs_common::GlobalReadiness;
use rustfs_ecstore::{ use rustfs_ecstore::{
@@ -31,13 +35,17 @@ use rustfs_ecstore::{
replication::get_global_replication_pool, replication::get_global_replication_pool,
}, },
endpoints::EndpointServerPools, endpoints::EndpointServerPools,
global::shutdown_background_services,
notification_sys::new_global_notification_sys, notification_sys::new_global_notification_sys,
store::ECStore, store::ECStore,
store_api::BucketOperations, store_api::BucketOperations,
}; };
use rustfs_heal::{create_ahm_services_cancel_token, heal::storage::ECStoreHealStorage, init_heal_manager}; 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_iam::init_oidc_sys;
use rustfs_obs::init_metrics_runtime; use rustfs_obs::init_metrics_runtime;
use rustfs_scanner::init_data_scanner;
use rustfs_storage_api::BucketOptions; use rustfs_storage_api::BucketOptions;
use rustfs_utils::get_env_bool_with_aliases; use rustfs_utils::get_env_bool_with_aliases;
use std::{ use std::{
@@ -56,12 +64,19 @@ const LOG_COMPONENT_MAIN: &str = "main";
const LOG_SUBSYSTEM_STARTUP: &str = "startup"; const LOG_SUBSYSTEM_STARTUP: &str = "startup";
const LOG_SUBSYSTEM_AUTH: &str = "auth"; const LOG_SUBSYSTEM_AUTH: &str = "auth";
const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state"; const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state";
const EVENT_PROTOCOL_SYSTEM_STATE: &str = "protocol_system_state";
const EVENT_DEADLOCK_DETECTOR_STATE: &str = "deadlock_detector_state"; const EVENT_DEADLOCK_DETECTOR_STATE: &str = "deadlock_detector_state";
const EVENT_KEYSTONE_AUTH_INITIALIZED: &str = "keystone_auth_initialized"; const EVENT_KEYSTONE_AUTH_INITIALIZED: &str = "keystone_auth_initialized";
const EVENT_KEYSTONE_AUTH_INITIALIZATION_FAILED: &str = "keystone_auth_initialization_failed"; const EVENT_KEYSTONE_AUTH_INITIALIZATION_FAILED: &str = "keystone_auth_initialization_failed";
const EVENT_OIDC_INITIALIZATION_FAILED: &str = "oidc_initialization_failed"; const EVENT_OIDC_INITIALIZATION_FAILED: &str = "oidc_initialization_failed";
const EVENT_NOTIFICATION_SYSTEM_INITIALIZATION_FAILED: &str = "notification_system_initialization_failed"; const EVENT_NOTIFICATION_SYSTEM_INITIALIZATION_FAILED: &str = "notification_system_initialization_failed";
const EVENT_BACKGROUND_SERVICES_CONFIGURED: &str = "background_services_configured"; 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";
const EVENT_EVENT_NOTIFIER_SHUTDOWN: &str = "event_notifier_shutdown";
const EVENT_PROFILING_SHUTDOWN: &str = "profiling_shutdown";
const EVENT_SERVER_SHUTDOWN_STATE: &str = "server_shutdown_state";
pub struct StartupServiceRuntime { pub struct StartupServiceRuntime {
pub protocol_shutdowns: ProtocolShutdownSenders, pub protocol_shutdowns: ProtocolShutdownSenders,
@@ -69,6 +84,17 @@ pub struct StartupServiceRuntime {
pub enable_scanner: bool, pub enable_scanner: bool,
} }
pub struct StartupRuntimeLifecycle {
pub server_address: String,
pub state_manager: Arc<ServiceStateManager>,
pub s3_shutdown_tx: Option<ShutdownHandle>,
pub console_shutdown_tx: Option<ShutdownHandle>,
pub service_runtime: StartupServiceRuntime,
pub store: Arc<ECStore>,
pub shutdown_token: CancellationToken,
pub readiness: Arc<GlobalReadiness>,
}
pub async fn init_startup_runtime_services( pub async fn init_startup_runtime_services(
config: &Config, config: &Config,
endpoint_pools: EndpointServerPools, endpoint_pools: EndpointServerPools,
@@ -99,6 +125,64 @@ pub async fn init_startup_runtime_services(
}) })
} }
pub async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifecycle) -> Result<()> {
let StartupRuntimeLifecycle {
server_address,
state_manager,
s3_shutdown_tx,
console_shutdown_tx,
service_runtime,
store,
shutdown_token,
readiness,
} = lifecycle;
let StartupServiceRuntime {
protocol_shutdowns,
iam_bootstrap,
enable_scanner,
} = service_runtime;
info!(
target: "rustfs::main::run",
event = EVENT_SERVER_READY,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
version = %crate::version::get_version(),
server_address = %server_address,
started_at = %jiff::Zoned::now(),
iam_bootstrap = ?iam_bootstrap,
"RustFS server ready"
);
publish_ready_for_iam_bootstrap(iam_bootstrap, readiness.as_ref(), Some(state_manager.as_ref())).await?;
rustfs_common::set_global_init_time_now().await;
if enable_scanner {
init_data_scanner(shutdown_token.clone(), store).await;
}
let shutdown_signal = wait_for_shutdown().await;
handle_shutdown(
&state_manager,
shutdown_signal,
s3_shutdown_tx,
console_shutdown_tx,
protocol_shutdowns,
shutdown_token,
)
.await;
info!(
target: "rustfs::main::run",
event = EVENT_SERVER_SHUTDOWN_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = ?state_manager.current_state(),
result = "stopped",
"RustFS server stopped"
);
Ok(())
}
async fn init_audit_runtime() { async fn init_audit_runtime() {
match init_event_notifier_and_audit().await { match init_event_notifier_and_audit().await {
Ok(()) => info!( Ok(()) => info!(
@@ -278,6 +362,223 @@ async fn init_observability_runtime(ctx: CancellationToken) {
} }
} }
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum BackgroundShutdownStep {
DataScanner,
Ahm,
}
fn background_shutdown_steps(enable_scanner: bool, enable_heal: bool) -> Vec<BackgroundShutdownStep> {
let mut steps = Vec::with_capacity(2);
if enable_scanner {
steps.push(BackgroundShutdownStep::DataScanner);
}
if enable_heal || enable_scanner {
steps.push(BackgroundShutdownStep::Ahm);
}
steps
}
async fn handle_shutdown(
state_manager: &ServiceStateManager,
shutdown_signal: ShutdownSignal,
s3_shutdown_handle: Option<ShutdownHandle>,
console_shutdown_handle: Option<ShutdownHandle>,
protocols: ProtocolShutdownSenders,
ctx: CancellationToken,
) {
let ProtocolShutdownSenders {
ftp: ftp_shutdown_tx,
ftps: ftps_shutdown_tx,
webdav: webdav_shutdown_tx,
sftp: sftp_shutdown_tx,
} = protocols;
ctx.cancel();
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_SHUTDOWN_SIGNAL_RECEIVED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
signal = shutdown_signal.log_label(),
"Shutdown signal received"
);
state_manager.update(ServiceState::Stopping);
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);
let background_steps = background_shutdown_steps(enable_scanner, enable_heal);
for step in &background_steps {
match step {
BackgroundShutdownStep::DataScanner => {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_BACKGROUND_SERVICE_SHUTDOWN,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
service = "data_scanner",
state = "stopping",
"Background service shutdown started"
);
shutdown_background_services();
}
BackgroundShutdownStep::Ahm => {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_BACKGROUND_SERVICE_SHUTDOWN,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
service = "ahm",
state = "stopping",
"Background service shutdown started"
);
shutdown_ahm_services();
}
}
}
if background_steps.is_empty() {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_BACKGROUND_SERVICE_SHUTDOWN,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
service = "ahm",
state = "skipped",
reason = "disabled",
"Background service shutdown skipped"
);
}
let mut protocol_shutdowns = Vec::new();
if let Some(ftp_shutdown_tx) = ftp_shutdown_tx {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_PROTOCOL_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
protocol = "ftp",
state = "stopping",
"Protocol runtime stopping"
);
protocol_shutdowns.push(ftp_shutdown_tx.shutdown());
}
if let Some(ftps_shutdown_tx) = ftps_shutdown_tx {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_PROTOCOL_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
protocol = "ftps",
state = "stopping",
"Protocol runtime stopping"
);
protocol_shutdowns.push(ftps_shutdown_tx.shutdown());
}
if let Some(webdav_shutdown_tx) = webdav_shutdown_tx {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_PROTOCOL_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
protocol = "webdav",
state = "stopping",
"Protocol runtime stopping"
);
protocol_shutdowns.push(webdav_shutdown_tx.shutdown());
}
if let Some(sftp_shutdown_tx) = sftp_shutdown_tx {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_PROTOCOL_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
protocol = "sftp",
state = "stopping",
"Protocol runtime stopping"
);
protocol_shutdowns.push(sftp_shutdown_tx.shutdown());
}
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_EVENT_NOTIFIER_SHUTDOWN,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stopping",
"Event notifier shutdown started"
);
shutdown_event_notifier().await;
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_AUDIT_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stopping",
"Audit runtime stopping"
);
match stop_audit_system().await {
Ok(_) => info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_AUDIT_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stopped",
"Audit runtime stopped"
),
Err(e) => error!(
target: "rustfs::main::handle_shutdown",
event = EVENT_AUDIT_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stop_failed",
error = %e,
"Audit runtime failed to stop"
),
}
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_PROFILING_SHUTDOWN,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stopping",
"Profiling shutdown started"
);
crate::profiling::shutdown_profiling();
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_SERVER_SHUTDOWN_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stopping",
"RustFS server stopping"
);
if let Some(s3_shutdown_handle) = s3_shutdown_handle {
s3_shutdown_handle.shutdown().await;
}
if let Some(console_shutdown_handle) = console_shutdown_handle {
console_shutdown_handle.shutdown().await;
}
join_all(protocol_shutdowns).await;
state_manager.update(ServiceState::Stopped);
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_SERVER_SHUTDOWN_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
state = "stopped",
"RustFS server stopped"
);
}
pub async fn init_event_notifier_and_audit() -> AuditResult<()> { pub async fn init_event_notifier_and_audit() -> AuditResult<()> {
init_event_notifier_and_audit_with(init_event_notifier, start_audit_system).await init_event_notifier_and_audit_with(init_event_notifier, start_audit_system).await
} }
@@ -310,7 +611,9 @@ where
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::{init_event_notifier_and_audit_with, init_notification_system_with}; use super::{
BackgroundShutdownStep, background_shutdown_steps, init_event_notifier_and_audit_with, init_notification_system_with,
};
use rustfs_audit::AuditError; use rustfs_audit::AuditError;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
@@ -363,4 +666,18 @@ mod tests {
assert!(result.is_err()); assert!(result.is_err());
} }
#[test]
fn background_shutdown_plan_keeps_scanner_before_ahm() {
assert_eq!(
background_shutdown_steps(true, true),
vec![BackgroundShutdownStep::DataScanner, BackgroundShutdownStep::Ahm]
);
assert_eq!(
background_shutdown_steps(true, false),
vec![BackgroundShutdownStep::DataScanner, BackgroundShutdownStep::Ahm]
);
assert_eq!(background_shutdown_steps(false, true), vec![BackgroundShutdownStep::Ahm]);
assert!(background_shutdown_steps(false, false).is_empty());
}
} }