From aa5de1c908d6ffb778533c895c200f3c60ca1ed5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Sat, 20 Jun 2026 03:26:09 +0800 Subject: [PATCH] refactor: extract startup service component boundary (#3636) --- docs/architecture/migration-progress.md | 49 +++- docs/architecture/runtime-lifecycle.md | 8 + docs/architecture/startup-timeline.md | 14 +- rustfs/src/embedded.rs | 4 +- rustfs/src/lib.rs | 1 + rustfs/src/startup_service_components.rs | 319 +++++++++++++++++++++++ rustfs/src/startup_services.rs | 313 +--------------------- 7 files changed, 386 insertions(+), 322 deletions(-) create mode 100644 rustfs/src/startup_service_components.rs diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 0879cb8ad..cdef3dee8 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,15 +5,16 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block ## Current Context - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660) -- Branch: `overtrue/arch-startup-ready-lifecycle-boundary` -- Baseline: `origin/main` (`e9037f9eb09fcbb31d980efac8b8e6b8f12bcaef`). +- Branch: `overtrue/arch-startup-service-components-boundary` +- Baseline: `overtrue/arch-startup-ready-lifecycle-boundary` + (`a518f40489515c1ec37c3cbc5738e04955a942f2`). +- Stacked on: rustfs/rustfs#3635. - PR type for this branch: `pure-move` - Runtime behavior changes: none. -- Rust code changes: move ready publication, scanner startup, shutdown-signal - wait, and stopped-state final logging from `startup_services` into a - dedicated startup lifecycle boundary. +- Rust code changes: move startup service component initialization helpers from + `startup_services` into a dedicated startup service component boundary. - CI/script changes: none. -- Docs changes: record the R-024 startup lifecycle boundary slice. +- Docs changes: record the R-025 startup service component boundary slice. ## Phase 0 Tasks @@ -2077,6 +2078,22 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block migration/layer guards, formatting, diff hygiene, Rust risk scan, branch freshness check, pre-commit quality gate, and three-expert review. +- [x] `R-025` Extract startup service component boundary. + - Do: add `startup_service_components` and move audit/deadlock, bucket + metadata, IAM bootstrap, auth integration, notification, background service, + and observability component helpers out of `startup_services`. + - Acceptance: `startup_services` keeps the same runtime service orchestration + order while component helpers own the individual service startup side + effects. + - Must preserve: KMS before optional runtime startup, buffer profiling before + audit, event notifier before audit, bucket metadata before IAM, IAM before + auth and notification, notification before background services, and + observability startup after background service setup. + - Verification: focused startup service component/service/lifecycle tests, + RustFS lib check, migration/layer guards, formatting, diff hygiene, Rust + risk scan, branch freshness check, pre-commit quality gate, and + three-expert review. + ## Next PRs 1. `pure-move`: continue larger lifecycle hook slices for optional runtime @@ -2088,9 +2105,9 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | R-024 keeps runtime service initialization in `startup_services` and moves ready/scanner/shutdown-wait orchestration into `startup_lifecycle`. | -| Migration preservation | passed | Ready publication, global init time, scanner startup, shutdown wait/delegation, and final stopped logging remain ordered. | -| Testing/verification | passed | Focused lifecycle, shutdown, and startup service tests plus final checks are tracked below before push. | +| Quality/architecture | passed | R-025 keeps `startup_services` as the orchestrator and moves component helpers into `startup_service_components`. | +| Migration preservation | passed | Runtime service startup order and embedded helper reuse remain unchanged. | +| Testing/verification | passed | Focused component, service, lifecycle, final hygiene, and full pre-commit checks passed. | ## Verification Notes @@ -2172,6 +2189,20 @@ Passed before push: - `make pre-commit`: passed. - Three-expert review: passed. +- Issue #660 R-025 current slice: + - `cargo test -p rustfs --lib startup_service_components -- --nocapture`: + passed. + - `cargo test -p rustfs --lib startup_services -- --nocapture`: passed. + - `cargo test -p rustfs --lib startup_lifecycle -- --nocapture`: passed. + - `cargo check -p rustfs --lib`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_layer_dependencies.sh`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - Rust risk scan on changed Rust files: passed. + - `make pre-commit`: passed. + - Three-expert review: passed. + - Issue #660 R-020 current slice: - `cargo test -p rustfs --lib startup_profiling -- --nocapture`: passed. - `cargo check -p rustfs --lib`: passed. diff --git a/docs/architecture/runtime-lifecycle.md b/docs/architecture/runtime-lifecycle.md index 2e68bc2b9..bce617cd9 100644 --- a/docs/architecture/runtime-lifecycle.md +++ b/docs/architecture/runtime-lifecycle.md @@ -47,6 +47,14 @@ boundary, but they must not reorder ready publication, global init-time publication, scanner startup, shutdown-signal wait, shutdown delegation, or the final stopped-state log. +## Startup Service Component Boundary + +`startup_service_components` owns individual runtime service startup component +helpers while `startup_services` preserves their orchestration order. Migration +PRs must not change KMS, optional runtime, audit, metadata, IAM, auth, +notification, background service, or observability startup ordering while moving +these helpers. + ## Optional Runtime Boundary `startup_optional_runtimes` owns startup and shutdown handoff for optional diff --git a/docs/architecture/startup-timeline.md b/docs/architecture/startup-timeline.md index 100b474a6..9b3e93e98 100644 --- a/docs/architecture/startup-timeline.md +++ b/docs/architecture/startup-timeline.md @@ -38,13 +38,13 @@ new startup semantics. | `RUN-012` | `rustfs/src/startup_storage.rs` | Initialize ECStore config and global config system. | Initializes ECStore config, attempts server-config migration, then retries global config init up to 15 times. | Migration attempt is non-fatal in this path; global config init becomes fatal after retries. | Marks the `GlobalReadiness` `StorageReady` stage after global config init succeeds; later runtime readiness still rechecks storage, IAM, and lock quorum before `FullReady` | | `RUN-013` | `rustfs/src/startup_storage.rs` and `rustfs/src/startup_services.rs` | Start replication and KMS systems. | Starts background replication pool, then initializes KMS from startup services. | Replication init is non-fatal in this path; KMS init is fatal on error. | `StorageReady` stage is already marked; dynamic runtime storage readiness is still checked before `FullReady` | | `RUN-014` | `rustfs/src/startup_optional_runtimes.rs` and `rustfs/src/startup_protocols.rs` | Initialize optional protocol servers. | Starts FTP/FTPS/WebDAV/SFTP when feature-enabled and configured, collecting shutdown handles. | Feature-enabled protocol init is fatal on error; disabled protocols are non-fatal. | None | -| `RUN-015` | `rustfs/src/startup_services.rs` | Initialize buffer profiling, event notifier, audit, and deadlock detector. | Starts buffer profile system, event notifier, audit system, and optional deadlock detector. | Audit startup failure is logged and non-fatal; the others are non-fatal in this path. | None | -| `RUN-016` | `rustfs/src/startup_services.rs` | List buckets and run bucket/replication/IAM metadata migrations. | Reads bucket names, migrates bucket metadata, initializes replication resync, migrates IAM config, and initializes bucket metadata system. | Bucket list and replication resync are fatal on error; metadata migration calls are non-fatal in this path. | Storage remains ready; IAM not yet ready | -| `RUN-017` | `rustfs/src/startup_services.rs` and `rustfs/src/startup_iam.rs` | Bootstrap IAM inline or defer recovery. | Initializes IAM when possible; otherwise starts the deferred IAM recovery path through `startup_iam`. | Fatal only when `bootstrap_or_defer_iam_init` returns an unrecoverable error. | Inline success marks `IamReady`; deferred mode publishes `IamReady` later from the recovery task | -| `RUN-018` | `rustfs/src/startup_services.rs` | Initialize Keystone and OIDC auth integrations. | Loads Keystone env config and initializes OIDC providers. | Keystone config parse is fatal; Keystone runtime init failure is non-fatal; OIDC init failure is non-fatal. | 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-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-015` | `rustfs/src/startup_services.rs` and `rustfs/src/startup_service_components.rs` | Initialize buffer profiling, event notifier, audit, and deadlock detector. | Starts buffer profile system, event notifier, audit system, and optional deadlock detector. | Audit startup failure is logged and non-fatal; the others are non-fatal in this path. | None | +| `RUN-016` | `rustfs/src/startup_service_components.rs` | List buckets and run bucket/replication/IAM metadata migrations. | Reads bucket names, migrates bucket metadata, initializes replication resync, migrates IAM config, and initializes bucket metadata system. | Bucket list and replication resync are fatal on error; metadata migration calls are non-fatal in this path. | Storage remains ready; IAM not yet ready | +| `RUN-017` | `rustfs/src/startup_service_components.rs` and `rustfs/src/startup_iam.rs` | Bootstrap IAM inline or defer recovery. | Initializes IAM when possible; otherwise starts the deferred IAM recovery path through `startup_iam`. | Fatal only when `bootstrap_or_defer_iam_init` returns an unrecoverable error. | Inline success marks `IamReady`; deferred mode publishes `IamReady` later from the recovery task | +| `RUN-018` | `rustfs/src/startup_service_components.rs` | Initialize Keystone and OIDC auth integrations. | Loads Keystone env config and initializes OIDC providers. | Keystone config parse is fatal; Keystone runtime init failure is non-fatal; OIDC init failure is non-fatal. | None | +| `RUN-019` | `rustfs/src/startup_service_components.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_service_components.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_service_components.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/startup_lifecycle.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/startup_lifecycle.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/startup_lifecycle.rs` | Wait for shutdown signal. | Blocks the main task until a shutdown signal is received. | Non-fatal. | Runtime remains in its current readiness state | diff --git a/rustfs/src/embedded.rs b/rustfs/src/embedded.rs index 61a436fb9..b4f1953eb 100644 --- a/rustfs/src/embedded.rs +++ b/rustfs/src/embedded.rs @@ -422,7 +422,7 @@ impl RustFSServerBuilder { // Buffer profiles. init_buffer_profile_system(&config); - if let Err(e) = crate::startup_services::init_event_notifier_and_audit().await { + if let Err(e) = crate::startup_service_components::init_event_notifier_and_audit().await { warn!( component = LOG_COMPONENT_EMBEDDED, subsystem = LOG_SUBSYSTEM_EMBEDDED, @@ -465,7 +465,7 @@ impl RustFSServerBuilder { // Bucket notifications. add_bucket_notification_configuration(buckets.clone()).await; - if let Err(e) = crate::startup_services::init_notification_system(endpoint_pools.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, diff --git a/rustfs/src/lib.rs b/rustfs/src/lib.rs index e01743c44..9af84638e 100644 --- a/rustfs/src/lib.rs +++ b/rustfs/src/lib.rs @@ -78,6 +78,7 @@ pub mod startup_profiling; pub mod startup_protocols; pub mod startup_runtime; pub mod startup_server; +pub mod startup_service_components; pub mod startup_services; pub mod startup_shutdown; pub mod startup_storage; diff --git a/rustfs/src/startup_service_components.rs b/rustfs/src/startup_service_components.rs new file mode 100644 index 000000000..bab83f623 --- /dev/null +++ b/rustfs/src/startup_service_components.rs @@ -0,0 +1,319 @@ +// 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::storage_compat::{ + ECStore, EndpointServerPools, get_global_replication_pool, init_bucket_metadata_sys, new_global_notification_sys, + try_migrate_bucket_metadata, try_migrate_iam_config, +}; +use crate::{ + init::{add_bucket_notification_configuration, init_auto_tuner, init_update_check, print_server_info}, + server::{ServiceStateManager, init_event_notifier, start_audit_system}, + startup_iam::{IamBootstrapDisposition, bootstrap_or_defer_iam_init}, +}; +use rustfs_audit::AuditResult; +use rustfs_common::GlobalReadiness; +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::{BucketOperations, 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(crate) 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" + ), + } +} + +pub(crate) 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" + ); + } +} + +pub(crate) async fn init_bucket_metadata_runtime(store: Arc, ctx: CancellationToken) -> Result> { + let buckets_list = store + .list_bucket(&BucketOptions { + no_metadata: true, + ..Default::default() + }) + .await + .map_err(Error::other)?; + + let buckets: Vec = 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) +} + +pub(crate) async fn init_iam_runtime( + store: Arc, + ctx: CancellationToken, + readiness: Arc, + state_manager: Arc, +) -> Result { + 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 +} + +pub(crate) 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(()) +} + +pub(crate) async fn init_notification_runtime(endpoint_pools: EndpointServerPools, buckets: Vec) -> 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) + }) +} + +pub(crate) async fn init_background_service_runtime(store: Arc) -> Result { + 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) +} + +pub(crate) 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() { + rustfs_io_metrics::set_put_stage_metrics_enabled(true); + init_metrics_runtime(ctx.clone()); + crate::memory_observability::init_memory_observability(ctx.clone()); + init_auto_tuner(ctx).await; + } +} + +pub(crate) 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( + notify: NotifyFn, + start_audit: AuditFn, +) -> AuditResult<()> +where + NotifyFn: FnOnce() -> NotifyFuture, + NotifyFuture: Future, + AuditFn: FnOnce() -> AuditFuture, + AuditFuture: Future>, +{ + notify().await; + start_audit().await +} + +pub(crate) async fn init_notification_system(endpoint_pools: EndpointServerPools) -> crate::storage_compat::EcstoreResult<()> { + init_notification_system_with(|| new_global_notification_sys(endpoint_pools)).await +} + +async fn init_notification_system_with(init_notification: InitFn) -> crate::storage_compat::EcstoreResult<()> +where + InitFn: FnOnce() -> InitFuture, + InitFuture: Future>, +{ + 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(crate::storage_compat::EcstoreError::FaultyDisk) }).await; + + assert!(result.is_err()); + } +} diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index 9d54eb61d..b93c1a91b 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -12,49 +12,21 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::storage_compat::{ - ECStore, EndpointServerPools, get_global_replication_pool, init_bucket_metadata_sys, new_global_notification_sys, - try_migrate_bucket_metadata, try_migrate_iam_config, -}; +use crate::storage_compat::{ECStore, EndpointServerPools}; 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}, + init::{init_buffer_profile_system, init_kms_system}, + server::ServiceStateManager, + startup_iam::IamBootstrapDisposition, startup_optional_runtimes::{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, + }, }; -use rustfs_audit::AuditResult; use rustfs_common::GlobalReadiness; -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::{BucketOperations, BucketOptions}; -use rustfs_utils::get_env_bool_with_aliases; -use std::{ - future::Future, - io::{Error, Result}, - sync::Arc, -}; +use std::{io::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 optional_runtimes: OptionalRuntimeServices, @@ -91,270 +63,3 @@ pub async fn init_startup_runtime_services( 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, ctx: CancellationToken) -> Result> { - let buckets_list = store - .list_bucket(&BucketOptions { - no_metadata: true, - ..Default::default() - }) - .await - .map_err(Error::other)?; - - let buckets: Vec = 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, - ctx: CancellationToken, - readiness: Arc, - state_manager: Arc, -) -> Result { - 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) -> 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) -> Result { - 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() { - rustfs_io_metrics::set_put_stage_metrics_enabled(true); - 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 -} - -async fn init_event_notifier_and_audit_with( - notify: NotifyFn, - start_audit: AuditFn, -) -> AuditResult<()> -where - NotifyFn: FnOnce() -> NotifyFuture, - NotifyFuture: Future, - AuditFn: FnOnce() -> AuditFuture, - AuditFuture: Future>, -{ - notify().await; - start_audit().await -} - -pub async fn init_notification_system(endpoint_pools: EndpointServerPools) -> crate::storage_compat::EcstoreResult<()> { - init_notification_system_with(|| new_global_notification_sys(endpoint_pools)).await -} - -async fn init_notification_system_with(init_notification: InitFn) -> crate::storage_compat::EcstoreResult<()> -where - InitFn: FnOnce() -> InitFuture, - InitFuture: Future>, -{ - 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(crate::storage_compat::EcstoreError::FaultyDisk) }).await; - - assert!(result.is_err()); - } -}