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

refactor: extract embedded build orchestration (#3663)

Browse Source
This commit is contained in:
安正超
2026-06-20 22:10:40 +08:00
committed by GitHub
parent 0cf7e5cf03
commit ebd4882ec1
4 changed files with 210 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/embedded.rs
+25 -81
View File
@@ -47,14 +47,9 @@
//! start a second server will return an error.
use crate::server::ShutdownHandle;
use crate::startup_lifecycle::{EmbeddedStartupGuard, log_embedded_server_ready, publish_embedded_startup_ready};
use crate::startup_runtime_hooks::init_embedded_runtime_hooks;
use crate::startup_server::{
EmbeddedStartupConfig, init_embedded_startup_listen_context, prepare_embedded_startup_config, start_embedded_http_server,
};
use crate::startup_services::init_embedded_startup_runtime_services;
use crate::startup_shutdown::{run_embedded_server_shutdown, signal_embedded_startup_shutdown};
use crate::startup_storage::{init_embedded_startup_storage_foundation, init_embedded_startup_storage_runtime};
use crate::startup_embedded::{EmbeddedStartupArgs, EmbeddedStartupError, run_embedded_startup};
use crate::startup_lifecycle::log_embedded_server_ready;
use crate::startup_shutdown::run_embedded_server_shutdown;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::path::PathBuf;
use tokio_util::sync::CancellationToken;
@@ -99,6 +94,16 @@ impl From<std::io::Error> for ServerError {
}
}
impl From<EmbeddedStartupError> for ServerError {
fn from(err: EmbeddedStartupError) -> Self {
match err {
EmbeddedStartupError::AlreadyStarted => ServerError::AlreadyStarted,
EmbeddedStartupError::Init(message) => ServerError::Init(message),
EmbeddedStartupError::Io(err) => ServerError::Io(err),
}
}
}
/// Builder for configuring and starting an embedded RustFS server.
///
/// # Examples
@@ -213,85 +218,24 @@ impl RustFSServerBuilder {
self.do_build().await
}
/// Inner build implementation. Separated from [`build`] so the outer
/// method can enforce the one-shot process-global startup guard.
async fn do_build(&mut self) -> Result<RustFSServer, ServerError> {
// Build is allowed to fail before irreversible global initialization
// (for example on temporary I/O or directory setup errors), and in that
// case callers can retry.
let mut startup_guard = EmbeddedStartupGuard::new();
let EmbeddedStartupConfig { config, temp_dir_guard } = prepare_embedded_startup_config(
self.address.clone(),
self.access_key.clone(),
self.secret_key.clone(),
self.volumes.clone(),
self.region.clone(),
)
.await
.map_err(|e| ServerError::Init(e.to_string()))?;
// --- Initialization sequence (mirrors main.rs::run) ---
init_embedded_runtime_hooks(config.obs_endpoint.clone())
.await
.map_err(|e| ServerError::Init(e.to_string()))?;
let listen_context = init_embedded_startup_listen_context(&config)
.await
.map_err(|e| ServerError::Init(e.to_string()))?;
startup_guard
.mark_global_init_started()
.map_err(|_| ServerError::AlreadyStarted)?;
let endpoint_pools = init_embedded_startup_storage_foundation(&listen_context.server_address, &config.volumes)
.await
.map_err(|e| ServerError::Init(e.to_string()))?;
let http_server = start_embedded_http_server(&config, listen_context.readiness.clone()).await?;
let shutdown_handle = http_server.shutdown_handle;
let bound_addr = http_server.bound_addr;
let ctx = CancellationToken::new();
let storage_runtime = match init_embedded_startup_storage_runtime(
listen_context.server_addr,
&endpoint_pools,
listen_context.readiness.clone(),
ctx.clone(),
)
.await
{
Ok(runtime) => runtime,
Err(e) => {
signal_embedded_startup_shutdown(&shutdown_handle, &ctx);
return Err(ServerError::Init(e.to_string()));
}
};
let store = storage_runtime.store;
let service_runtime =
init_embedded_startup_runtime_services(&config, endpoint_pools, store, ctx.clone(), listen_context.readiness.clone())
.await
.map_err(|e| {
signal_embedded_startup_shutdown(&shutdown_handle, &ctx);
ServerError::Init(e.to_string())
})?;
publish_embedded_startup_ready(service_runtime.iam_bootstrap, listen_context.readiness.as_ref())
.await
.map_err(|e| {
signal_embedded_startup_shutdown(&shutdown_handle, &ctx);
ServerError::Init(format!("runtime readiness: {e}"))
})?;
let started = run_embedded_startup(EmbeddedStartupArgs {
address: self.address.clone(),
access_key: self.access_key.clone(),
secret_key: self.secret_key.clone(),
volumes: self.volumes.clone(),
region: self.region.clone(),
})
.await?;
let server = RustFSServer {
address: bound_addr,
address: started.bound_addr,
access_key: self.access_key.clone(),
secret_key: self.secret_key.clone(),
region: self.region.clone(),
shutdown_handle: Some(shutdown_handle),
cancel_token: ctx,
temp_dir: temp_dir_guard.map(|g| g.keep()),
shutdown_handle: Some(started.shutdown_handle),
cancel_token: started.cancel_token,
temp_dir: started.temp_dir,
};
log_embedded_server_ready(server.endpoint_address());
rustfs/src/lib.rs
+1
View File
@@ -68,6 +68,7 @@ pub mod profiling;
pub mod protocols;
pub mod runtime_capabilities;
pub mod server;
pub(crate) mod startup_embedded;
pub mod startup_entrypoint;
pub mod startup_fs_guard;
pub mod startup_iam;
rustfs/src/startup_embedded.rs
+132
View File
@@ -0,0 +1,132 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::{
server::ShutdownHandle,
startup_lifecycle::{EmbeddedStartupGuard, publish_embedded_startup_ready},
startup_runtime_hooks::init_embedded_runtime_hooks,
startup_server::{
EmbeddedStartupConfig, init_embedded_startup_listen_context, prepare_embedded_startup_config, start_embedded_http_server,
},
startup_services::init_embedded_startup_runtime_services,
startup_shutdown::signal_embedded_startup_shutdown,
startup_storage::{init_embedded_startup_storage_foundation, init_embedded_startup_storage_runtime},
};
use std::{io, net::SocketAddr, path::PathBuf};
use tokio_util::sync::CancellationToken;
pub(crate) struct EmbeddedStartupArgs {
pub address: String,
pub access_key: String,
pub secret_key: String,
pub volumes: Vec<String>,
pub region: String,
}
pub(crate) struct EmbeddedStartedServer {
pub bound_addr: SocketAddr,
pub shutdown_handle: ShutdownHandle,
pub cancel_token: CancellationToken,
pub temp_dir: Option<PathBuf>,
}
#[derive(Debug)]
pub(crate) enum EmbeddedStartupError {
AlreadyStarted,
Init(String),
Io(io::Error),
}
impl From<io::Error> for EmbeddedStartupError {
fn from(err: io::Error) -> Self {
Self::Io(err)
}
}
pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result<EmbeddedStartedServer, EmbeddedStartupError> {
// Build is allowed to fail before irreversible global initialization
// (for example on temporary I/O or directory setup errors), and in that
// case callers can retry.
let mut startup_guard = EmbeddedStartupGuard::new();
let EmbeddedStartupConfig { config, temp_dir_guard } =
prepare_embedded_startup_config(args.address, args.access_key, args.secret_key, args.volumes, args.region)
.await
.map_err(init_error)?;
init_embedded_runtime_hooks(config.obs_endpoint.clone())
.await
.map_err(init_error)?;
let listen_context = init_embedded_startup_listen_context(&config).await.map_err(init_error)?;
startup_guard
.mark_global_init_started()
.map_err(|_| EmbeddedStartupError::AlreadyStarted)?;
let endpoint_pools = init_embedded_startup_storage_foundation(&listen_context.server_address, &config.volumes)
.await
.map_err(init_error)?;
let http_server = start_embedded_http_server(&config, listen_context.readiness.clone()).await?;
let shutdown_handle = http_server.shutdown_handle;
let bound_addr = http_server.bound_addr;
let cancel_token = CancellationToken::new();
let storage_runtime = match init_embedded_startup_storage_runtime(
listen_context.server_addr,
&endpoint_pools,
listen_context.readiness.clone(),
cancel_token.clone(),
)
.await
{
Ok(runtime) => runtime,
Err(err) => {
signal_embedded_startup_shutdown(&shutdown_handle, &cancel_token);
return Err(init_error(err));
}
};
let service_runtime = init_embedded_startup_runtime_services(
&config,
endpoint_pools,
storage_runtime.store,
cancel_token.clone(),
listen_context.readiness.clone(),
)
.await
.map_err(|err| {
signal_embedded_startup_shutdown(&shutdown_handle, &cancel_token);
init_error(err)
})?;
publish_embedded_startup_ready(service_runtime.iam_bootstrap, listen_context.readiness.as_ref())
.await
.map_err(|err| {
signal_embedded_startup_shutdown(&shutdown_handle, &cancel_token);
EmbeddedStartupError::Init(format!("runtime readiness: {err}"))
})?;
Ok(EmbeddedStartedServer {
bound_addr,
shutdown_handle,
cancel_token,
temp_dir: temp_dir_guard.map(|guard| guard.keep()),
})
}
fn init_error(err: impl std::fmt::Display) -> EmbeddedStartupError {
EmbeddedStartupError::Init(err.to_string())
}