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 storage runtime (#3465)

Browse Source
This commit is contained in:
安正超
2026-06-15 12:12:27 +08:00
committed by GitHub
parent 46fb4bdc2f
commit 3e723a2476
4 changed files with 146 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/main.rs
+6 -61
View File
@@ -26,15 +26,12 @@ use rustfs::startup_iam::{bootstrap_or_defer_iam_init, publish_ready_for_iam_boo
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_server::{StartupHttpServers, StartupListenContext, init_startup_http_servers, init_startup_listen_context};
use rustfs::startup_storage::init_startup_storage_foundation;
use rustfs_common::SystemStage;
use rustfs::startup_storage::{StartupStorageRuntime, init_startup_storage_foundation, init_startup_storage_runtime};
use rustfs_ecstore::{
bucket::metadata_sys::init_bucket_metadata_sys,
bucket::migration::{try_migrate_bucket_metadata, try_migrate_iam_config},
bucket::replication::{get_global_replication_pool, init_background_replication},
config as ecconfig,
bucket::replication::get_global_replication_pool,
global::shutdown_background_services,
store::ECStore,
store_api::BucketOperations,
};
use rustfs_heal::{
@@ -57,9 +54,7 @@ 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 LOG_SUBSYSTEM_STORAGE: &str = "storage";
const EVENT_SERVER_RUNTIME_FAILED: &str = "server_runtime_failed";
const EVENT_STARTUP_STORAGE_STAGE: &str = "startup_storage_stage";
const EVENT_PROTOCOL_SYSTEM_STATE: &str = "protocol_system_state";
const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state";
const EVENT_DEADLOCK_DETECTOR_STATE: &str = "deadlock_detector_state";
@@ -169,61 +164,11 @@ async fn run(config: rustfs::config::Config) -> Result<()> {
console_shutdown_tx,
} = init_startup_http_servers(&config, readiness.clone()).await?;
let ctx = CancellationToken::new();
let StartupStorageRuntime {
store,
shutdown_token: ctx,
} = init_startup_storage_runtime(server_addr, &endpoint_pools, readiness.clone()).await?;
// init store
// 2. Start Storage Engine (ECStore)
debug!(
target: "rustfs::main::run",
event = EVENT_STARTUP_STORAGE_STAGE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STORAGE,
stage = "ecstore_initialization",
state = "starting",
"starting ECStore initialization"
);
let store = ECStore::new(server_addr, endpoint_pools.clone(), ctx.clone())
.await
.inspect_err(|err| {
error!(
target: "rustfs::main::run",
event = EVENT_STARTUP_STORAGE_STAGE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STORAGE,
stage = "ecstore_initialization",
state = "failed",
error = ?err,
"ECStore initialization failed"
);
})?;
ecconfig::init();
ecconfig::try_migrate_server_config(store.clone()).await;
// // Initialize global configuration system
let mut retry_count = 0;
while let Err(e) = ecconfig::init_global_config_sys(store.clone()).await {
error!(
target: "rustfs::main::run",
event = EVENT_STARTUP_STORAGE_STAGE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STORAGE,
stage = "global_config_initialization",
state = "retrying",
retry_count = retry_count + 1,
error = ?e,
"Global config initialization retry failed"
);
// TODO: check error type
retry_count += 1;
if retry_count > 15 {
return Err(Error::other("ecconfig::init_global_config_sys failed"));
}
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
}
readiness.mark_stage(SystemStage::StorageReady);
// init replication_pool
init_background_replication(store.clone()).await;
// Initialize KMS system if enabled
init_kms_system(&config).await?;
rustfs/src/startup_storage.rs
+97 -3
View File
@@ -13,22 +13,36 @@
// limitations under the License.
use crate::startup_fs_guard::enforce_unsupported_fs_policy;
use rustfs_common::{GlobalReadiness, SystemStage};
use rustfs_ecstore::{
bucket::replication::init_background_replication,
config as ecconfig,
endpoints::EndpointServerPools,
set_global_endpoints,
store::{init_local_disks, init_lock_clients, prewarm_local_disk_id_map},
store::{ECStore, init_local_disks, init_lock_clients, prewarm_local_disk_id_map},
update_erasure_type,
};
use std::io::{Error, Result};
use std::{
io::{Error, Result},
net::SocketAddr,
sync::Arc,
};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
const LOG_COMPONENT_MAIN: &str = "main";
const LOG_SUBSYSTEM_STORAGE: &str = "storage";
const GLOBAL_CONFIG_INIT_MAX_RETRIES: usize = 15;
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";
const EVENT_STORAGE_POOL_HOST_RISK: &str = "storage_pool_host_risk";
pub struct StartupStorageRuntime {
pub store: Arc<ECStore>,
pub shutdown_token: CancellationToken,
}
pub async fn init_startup_storage_foundation(server_address: &str, volumes: &[String]) -> Result<EndpointServerPools> {
info!(
target: "rustfs::main::run",
@@ -91,6 +105,76 @@ pub async fn init_startup_storage_foundation(server_address: &str, volumes: &[St
Ok(endpoint_pools)
}
pub async fn init_startup_storage_runtime(
server_addr: SocketAddr,
endpoint_pools: &EndpointServerPools,
readiness: Arc<GlobalReadiness>,
) -> Result<StartupStorageRuntime> {
let ctx = CancellationToken::new();
debug!(
target: "rustfs::main::run",
event = EVENT_STARTUP_STORAGE_STAGE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STORAGE,
stage = "ecstore_initialization",
state = "starting",
"starting ECStore initialization"
);
let store = ECStore::new(server_addr, endpoint_pools.clone(), ctx.clone())
.await
.inspect_err(|err| {
error!(
target: "rustfs::main::run",
event = EVENT_STARTUP_STORAGE_STAGE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STORAGE,
stage = "ecstore_initialization",
state = "failed",
error = ?err,
"ECStore initialization failed"
);
})?;
init_startup_storage_global_config(store.clone()).await?;
readiness.mark_stage(SystemStage::StorageReady);
init_background_replication(store.clone()).await;
Ok(StartupStorageRuntime {
store,
shutdown_token: ctx,
})
}
async fn init_startup_storage_global_config(store: Arc<ECStore>) -> Result<()> {
ecconfig::init();
ecconfig::try_migrate_server_config(store.clone()).await;
let mut retry_count = 0;
while let Err(e) = ecconfig::init_global_config_sys(store.clone()).await {
let next_retry_count = retry_count + 1;
error!(
target: "rustfs::main::run",
event = EVENT_STARTUP_STORAGE_STAGE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STORAGE,
stage = "global_config_initialization",
state = "retrying",
retry_count = next_retry_count,
error = ?e,
"Global config initialization retry failed"
);
// TODO: check error type
retry_count = next_retry_count;
if global_config_retry_exhausted(retry_count) {
return Err(Error::other("ecconfig::init_global_config_sys failed"));
}
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
}
Ok(())
}
fn log_storage_pool_layout(endpoint_pools: &EndpointServerPools) {
for (i, eps) in endpoint_pools.as_ref().iter().enumerate() {
info!(
@@ -142,9 +226,13 @@ fn storage_pool_has_host_failure_risk(drives_per_set: usize) -> bool {
drives_per_set > 1
}
fn global_config_retry_exhausted(retry_count: usize) -> bool {
retry_count > GLOBAL_CONFIG_INIT_MAX_RETRIES
}
#[cfg(test)]
mod tests {
use super::storage_pool_has_host_failure_risk;
use super::{global_config_retry_exhausted, storage_pool_has_host_failure_risk};
#[test]
fn reports_host_failure_risk_only_for_multi_drive_sets() {
@@ -152,4 +240,10 @@ mod tests {
assert!(!storage_pool_has_host_failure_risk(1));
assert!(storage_pool_has_host_failure_risk(2));
}
#[test]
fn global_config_retry_limit_matches_startup_policy() {
assert!(!global_config_retry_exhausted(15));
assert!(global_config_retry_exhausted(16));
}
}