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

refactor: centralize startup server preflight (#3459)

Browse Source
This commit is contained in:
安正超
2026-06-15 08:32:44 +08:00
committed by GitHub
parent 2288b1bb63
commit 513f06f268
5 changed files with 229 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
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_preflight;
pub mod startup_protocols;
pub mod startup_runtime;
pub mod startup_services;
rustfs/src/main.rs
+8 -110
View File
@@ -19,15 +19,14 @@ use rustfs::init::{
use futures_util::future::join_all;
use rustfs::capacity::capacity_integration::init_capacity_management;
use rustfs::license::init_license;
use rustfs::server::{
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};
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_runtime::init_startup_runtime_foundation;
use rustfs_common::{GlobalReadiness, SystemStage, set_global_addr};
use rustfs_credentials::init_global_action_credentials;
use rustfs_ecstore::store::init_lock_clients;
@@ -49,12 +48,10 @@ 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, init_obs, set_global_guard};
use rustfs_obs::init_metrics_runtime;
use rustfs_scanner::init_data_scanner;
use rustfs_storage_api::BucketOptions;
use rustfs_utils::{
ExternalEnvCompatReport, apply_external_env_compat, get_env_bool_with_aliases, net::parse_and_resolve_address,
};
use rustfs_utils::{get_env_bool_with_aliases, net::parse_and_resolve_address};
use std::io::{Error, Result};
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
@@ -68,16 +65,12 @@ const LOG_COMPONENT_MAIN: &str = "main";
const LOG_SUBSYSTEM_STARTUP: &str = "startup";
const LOG_SUBSYSTEM_AUTH: &str = "auth";
const LOG_SUBSYSTEM_STORAGE: &str = "storage";
const EVENT_EXTERNAL_ENV_COMPAT_CONFLICT: &str = "external_env_compat_conflict";
const EVENT_EXTERNAL_ENV_COMPAT_APPLIED: &str = "external_env_compat_applied";
const EVENT_OBSERVABILITY_GUARD_SET_FAILED: &str = "observability_guard_set_failed";
const EVENT_DEFAULT_CREDENTIALS_DETECTED: &str = "default_credentials_detected";
const EVENT_SERVER_CONFIG_SANITIZED: &str = "server_config_sanitized";
const EVENT_SERVER_STARTING: &str = "server_starting";
const EVENT_SERVER_RUNTIME_FAILED: &str = "server_runtime_failed";
const EVENT_ACTION_CREDENTIALS_INITIALIZED: &str = "action_credentials_initialized";
const EVENT_ACTION_CREDENTIALS_INITIALIZATION_FAILED: &str = "action_credentials_initialization_failed";
const EVENT_OBSERVABILITY_GUARD_SET: &str = "observability_guard_set";
const EVENT_ENDPOINT_PARSING_STARTED: &str = "endpoint_parsing_started";
const EVENT_STARTUP_STORAGE_STAGE: &str = "startup_storage_stage";
const EVENT_STORAGE_POOL_FORMATTING: &str = "storage_pool_formatting";
@@ -126,20 +119,6 @@ fn main() {
}
}
fn bootstrap_external_prefix_compat() -> Result<ExternalEnvCompatReport> {
let env_compat_report = apply_external_env_compat();
Ok(env_compat_report)
}
fn format_external_prefix_mappings(report: &ExternalEnvCompatReport) -> String {
report
.mapped_pairs
.iter()
.map(|(source_key, rustfs_key)| format!("{source_key}->{rustfs_key}"))
.collect::<Vec<_>>()
.join(", ")
}
fn is_using_default_credentials(config: &rustfs::config::Config) -> bool {
config.is_using_default_credentials()
}
@@ -169,50 +148,16 @@ async fn async_main() -> Result<()> {
rustfs::config::CommandResult::Server(config) => config,
};
// Initialize the global config snapshot for info command
rustfs::config::init_config_snapshot(&config);
// Initialize the configuration
init_license(config.license.clone());
// Initialize Observability
let guard = match init_obs(Some(config.clone().obs_endpoint)).await {
Ok(g) => g,
Err(e) => {
match init_startup_server_preflight(&config, &env_compat_report).await {
Ok(()) => {}
Err(StartupServerPreflightError::ObservabilityInit(err)) => {
// Structured logging is unavailable until observability initializes.
emit_fatal_stderr("Observability initialization failed", &e);
emit_fatal_stderr("Observability initialization failed", err);
return Err(Error::other(OBSERVABILITY_INIT_FATAL_ALREADY_REPORTED));
}
};
// Store in global storage
match set_global_guard(guard).map_err(Error::other) {
Ok(_) => {
debug!(
target: "rustfs::main",
event = EVENT_OBSERVABILITY_GUARD_SET,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
result = "ok",
"Stored global observability guard"
);
}
Err(e) => {
error!(
target: "rustfs::main",
event = EVENT_OBSERVABILITY_GUARD_SET_FAILED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
error = %e,
"Failed to store global observability guard"
);
return Err(e);
}
Err(StartupServerPreflightError::Other(err)) => return Err(err),
}
log_external_prefix_compat_report(&env_compat_report);
init_startup_runtime_foundation(&config).await?;
// Run parameters
match run(*config).await {
Ok(_) => Ok(()),
@@ -724,32 +669,6 @@ async fn run(config: rustfs::config::Config) -> Result<()> {
Ok(())
}
fn log_external_prefix_compat_report(report: &ExternalEnvCompatReport) {
if report.conflict_count() > 0 {
warn!(
target: "rustfs::main",
event = EVENT_EXTERNAL_ENV_COMPAT_CONFLICT,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
conflict_count = report.conflict_count(),
conflict_keys = %report.conflict_keys.join(", "),
"Detected external-prefix compatibility conflicts; keeping RUSTFS_ values"
);
}
if report.mapped_count() > 0 {
info!(
target: "rustfs::main",
event = EVENT_EXTERNAL_ENV_COMPAT_APPLIED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
mapped_count = report.mapped_count(),
mapped_pairs = %format_external_prefix_mappings(report),
"Applied external-prefix compatibility mappings"
);
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum BackgroundShutdownStep {
DataScanner,
@@ -982,27 +901,6 @@ async fn handle_shutdown(
mod tests {
use super::*;
#[test]
fn format_external_prefix_mappings_lists_mapped_pairs() {
let report = ExternalEnvCompatReport {
mapped_pairs: vec![
("MINIO_ROOT_USER".to_string(), "RUSTFS_ROOT_USER".to_string()),
(
"MINIO_NOTIFY_WEBHOOK_ENABLE_PRIMARY".to_string(),
"RUSTFS_NOTIFY_WEBHOOK_ENABLE_PRIMARY".to_string(),
),
],
conflict_keys: Vec::new(),
};
let formatted = format_external_prefix_mappings(&report);
assert_eq!(
formatted,
"MINIO_ROOT_USER->RUSTFS_ROOT_USER, MINIO_NOTIFY_WEBHOOK_ENABLE_PRIMARY->RUSTFS_NOTIFY_WEBHOOK_ENABLE_PRIMARY"
);
}
#[test]
fn fatal_stderr_message_uses_consistent_prefix_and_context() {
assert_eq!(
rustfs/src/startup_preflight.rs
+174
View File
@@ -0,0 +1,174 @@
// 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::{config::Config, license::init_license, startup_runtime::init_startup_runtime_foundation};
use rustfs_obs::{init_obs, set_global_guard};
use rustfs_utils::{ExternalEnvCompatReport, apply_external_env_compat};
use std::{
fmt,
io::{Error, Result},
};
use tracing::{debug, error, info, warn};
const LOG_COMPONENT_MAIN: &str = "main";
const LOG_SUBSYSTEM_STARTUP: &str = "startup";
const EVENT_EXTERNAL_ENV_COMPAT_CONFLICT: &str = "external_env_compat_conflict";
const EVENT_EXTERNAL_ENV_COMPAT_APPLIED: &str = "external_env_compat_applied";
const EVENT_OBSERVABILITY_GUARD_SET: &str = "observability_guard_set";
const EVENT_OBSERVABILITY_GUARD_SET_FAILED: &str = "observability_guard_set_failed";
#[derive(Debug)]
pub enum StartupServerPreflightError {
ObservabilityInit(Error),
Other(Error),
}
impl StartupServerPreflightError {
fn as_io_error(&self) -> &Error {
match self {
Self::ObservabilityInit(err) | Self::Other(err) => err,
}
}
}
impl fmt::Display for StartupServerPreflightError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
self.as_io_error().fmt(formatter)
}
}
impl std::error::Error for StartupServerPreflightError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
Some(self.as_io_error())
}
}
pub fn bootstrap_external_prefix_compat() -> Result<ExternalEnvCompatReport> {
let env_compat_report = apply_external_env_compat();
Ok(env_compat_report)
}
pub async fn init_startup_server_preflight(
config: &Config,
env_compat_report: &ExternalEnvCompatReport,
) -> std::result::Result<(), StartupServerPreflightError> {
crate::config::init_config_snapshot(config);
init_license(config.license.clone());
init_startup_observability(config.obs_endpoint.clone()).await?;
log_external_prefix_compat_report(env_compat_report);
init_startup_runtime_foundation(config)
.await
.map_err(StartupServerPreflightError::Other)
}
async fn init_startup_observability(obs_endpoint: String) -> std::result::Result<(), StartupServerPreflightError> {
let guard = init_obs(Some(obs_endpoint))
.await
.map_err(|err| StartupServerPreflightError::ObservabilityInit(Error::other(err)))?;
match set_global_guard(guard).map_err(Error::other) {
Ok(_) => {
debug!(
target: "rustfs::main",
event = EVENT_OBSERVABILITY_GUARD_SET,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
result = "ok",
"Stored global observability guard"
);
Ok(())
}
Err(err) => {
error!(
target: "rustfs::main",
event = EVENT_OBSERVABILITY_GUARD_SET_FAILED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
error = %err,
"Failed to store global observability guard"
);
Err(StartupServerPreflightError::Other(err))
}
}
}
fn format_external_prefix_mappings(report: &ExternalEnvCompatReport) -> String {
report
.mapped_pairs
.iter()
.map(|(source_key, rustfs_key)| format!("{source_key}->{rustfs_key}"))
.collect::<Vec<_>>()
.join(", ")
}
fn log_external_prefix_compat_report(report: &ExternalEnvCompatReport) {
if report.conflict_count() > 0 {
warn!(
target: "rustfs::main",
event = EVENT_EXTERNAL_ENV_COMPAT_CONFLICT,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
conflict_count = report.conflict_count(),
conflict_keys = %report.conflict_keys.join(", "),
"Detected external-prefix compatibility conflicts; keeping RUSTFS_ values"
);
}
if report.mapped_count() > 0 {
info!(
target: "rustfs::main",
event = EVENT_EXTERNAL_ENV_COMPAT_APPLIED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
mapped_count = report.mapped_count(),
mapped_pairs = %format_external_prefix_mappings(report),
"Applied external-prefix compatibility mappings"
);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn format_external_prefix_mappings_lists_mapped_pairs() {
let report = ExternalEnvCompatReport {
mapped_pairs: vec![
("MINIO_ROOT_USER".to_string(), "RUSTFS_ROOT_USER".to_string()),
(
"MINIO_NOTIFY_WEBHOOK_ENABLE_PRIMARY".to_string(),
"RUSTFS_NOTIFY_WEBHOOK_ENABLE_PRIMARY".to_string(),
),
],
conflict_keys: Vec::new(),
};
let formatted = format_external_prefix_mappings(&report);
assert_eq!(
formatted,
"MINIO_ROOT_USER->RUSTFS_ROOT_USER, MINIO_NOTIFY_WEBHOOK_ENABLE_PRIMARY->RUSTFS_NOTIFY_WEBHOOK_ENABLE_PRIMARY"
);
}
#[test]
fn preflight_error_preserves_observability_init_classification() {
let err = StartupServerPreflightError::ObservabilityInit(Error::other("collector unavailable"));
assert!(matches!(&err, StartupServerPreflightError::ObservabilityInit(_)));
assert_eq!(err.to_string(), "collector unavailable");
assert!(std::error::Error::source(&err).is_some());
}
}