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

refactor: extract embedded startup server helpers (#3660)

Browse Source
This commit is contained in:
安正超
2026-06-20 18:38:02 +08:00
committed by GitHub
parent 3edff6a436
commit 84e292c6da
3 changed files with 167 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/embedded.rs
+17 -33
View File
@@ -46,16 +46,17 @@
//! storage engine uses process-global singletons (`OnceLock`). Attempting to
//! start a second server will return an error.
use crate::config::Config;
use crate::server::{ShutdownHandle, start_http_server};
use crate::server::ShutdownHandle;
use crate::startup_lifecycle::{log_embedded_server_ready, publish_embedded_startup_ready};
use crate::startup_runtime_hooks::init_embedded_runtime_hooks;
use crate::startup_server::init_embedded_startup_listen_context;
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;
use crate::startup_storage::{init_embedded_startup_storage_foundation, init_embedded_startup_storage_runtime};
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::path::{Path, PathBuf};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use tokio_util::sync::CancellationToken;
@@ -237,31 +238,15 @@ impl RustFSServerBuilder {
Ok(())
};
// Keep a TempDir guard alive so that if build fails the directory is
// cleaned up automatically. We disarm (keep) on success.
let mut temp_dir_guard: Option<tempfile::TempDir> = None;
if self.volumes.is_empty() {
let dir = tempfile::tempdir().map_err(|e| ServerError::Init(format!("failed to create temp dir: {e}")))?;
self.volumes.push(dir.path().display().to_string());
temp_dir_guard = Some(dir);
}
// Ensure volume directories exist.
for v in &self.volumes {
let p = Path::new(v);
if !p.exists() {
tokio::fs::create_dir_all(p)
.await
.map_err(|e| ServerError::Init(format!("failed to create volume dir {v}: {e}")))?;
}
}
// Build Config.
let mut config = Config::new(&self.address, self.volumes.clone());
config.access_key = self.access_key.clone();
config.secret_key = self.secret_key.clone();
config.region = Some(self.region.clone());
config.console_enable = false;
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) ---
@@ -279,10 +264,9 @@ impl RustFSServerBuilder {
.await
.map_err(|e| ServerError::Init(e.to_string()))?;
// Start HTTP server.
let mut s3_config = config.clone();
s3_config.console_enable = false;
let (shutdown_handle, bound_addr) = start_http_server(&s3_config, listen_context.readiness.clone()).await?;
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 shutdown_embedded_server = || {
shutdown_handle.signal();
rustfs/src/startup_server.rs
+99 -1
View File
@@ -24,8 +24,10 @@ use rustfs_utils::net::parse_and_resolve_address;
use std::{
io::{Error, Result},
net::SocketAddr,
path::Path,
sync::Arc,
};
use tempfile::TempDir;
use tracing::{debug, error, info, warn};
const LOG_COMPONENT_MAIN: &str = "main";
@@ -50,6 +52,16 @@ pub struct EmbeddedStartupListenContext {
pub server_address: String,
}
pub(crate) struct EmbeddedStartupConfig {
pub config: Config,
pub(crate) temp_dir_guard: Option<TempDir>,
}
pub(crate) struct EmbeddedHttpServer {
pub shutdown_handle: ShutdownHandle,
pub bound_addr: SocketAddr,
}
pub struct StartupHttpServers {
pub state_manager: Arc<ServiceStateManager>,
pub s3_shutdown_tx: Option<ShutdownHandle>,
@@ -105,6 +117,38 @@ pub async fn init_startup_listen_context(config: &Config) -> Result<StartupListe
})
}
pub(crate) async fn prepare_embedded_startup_config(
address: String,
access_key: String,
secret_key: String,
mut volumes: Vec<String>,
region: String,
) -> Result<EmbeddedStartupConfig> {
let mut temp_dir_guard = None;
if volumes.is_empty() {
let dir = tempfile::tempdir().map_err(|err| Error::other(format!("failed to create temp dir: {err}")))?;
volumes.push(dir.path().display().to_string());
temp_dir_guard = Some(dir);
}
for volume in &volumes {
let path = Path::new(volume);
if !path.exists() {
tokio::fs::create_dir_all(path)
.await
.map_err(|err| Error::other(format!("failed to create volume dir {volume}: {err}")))?;
}
}
let mut config = Config::new(&address, volumes);
config.access_key = access_key;
config.secret_key = secret_key;
config.region = Some(region);
config.console_enable = false;
Ok(EmbeddedStartupConfig { config, temp_dir_guard })
}
pub async fn init_embedded_startup_listen_context(config: &Config) -> Result<EmbeddedStartupListenContext> {
let readiness = Arc::new(GlobalReadiness::new());
@@ -138,6 +182,16 @@ pub async fn init_embedded_startup_listen_context(config: &Config) -> Result<Emb
})
}
pub(crate) async fn start_embedded_http_server(config: &Config, readiness: Arc<GlobalReadiness>) -> Result<EmbeddedHttpServer> {
let s3_config = s3_http_server_config(config);
let (shutdown_handle, bound_addr) = start_http_server(&s3_config, readiness).await?;
Ok(EmbeddedHttpServer {
shutdown_handle,
bound_addr,
})
}
pub async fn init_startup_http_servers(config: &Config, readiness: Arc<GlobalReadiness>) -> Result<StartupHttpServers> {
init_capacity_management().await;
let state_manager = Arc::new(ServiceStateManager::new());
@@ -224,7 +278,9 @@ fn console_http_server_config(config: &Config) -> Option<Config> {
#[cfg(test)]
mod tests {
use super::{DEFAULT_CREDENTIALS_WARNING_MESSAGE, console_http_server_config, s3_http_server_config};
use super::{
DEFAULT_CREDENTIALS_WARNING_MESSAGE, console_http_server_config, prepare_embedded_startup_config, s3_http_server_config,
};
use crate::config::Config;
#[test]
@@ -292,4 +348,46 @@ mod tests {
assert!(!DEFAULT_CREDENTIALS_WARNING_MESSAGE.contains(rustfs_credentials::DEFAULT_ACCESS_KEY));
assert!(!DEFAULT_CREDENTIALS_WARNING_MESSAGE.contains(rustfs_credentials::DEFAULT_SECRET_KEY));
}
#[tokio::test]
async fn prepare_embedded_startup_config_creates_temp_volume_when_missing() {
let prepared = prepare_embedded_startup_config(
"127.0.0.1:9000".to_string(),
"access".to_string(),
"secret".to_string(),
Vec::new(),
"us-west-2".to_string(),
)
.await
.expect("embedded startup config should be prepared");
assert_eq!(prepared.config.address, "127.0.0.1:9000");
assert_eq!(prepared.config.access_key, "access");
assert_eq!(prepared.config.secret_key, "secret");
assert_eq!(prepared.config.region.as_deref(), Some("us-west-2"));
assert!(!prepared.config.console_enable);
assert_eq!(prepared.config.volumes.len(), 1);
assert!(std::path::Path::new(&prepared.config.volumes[0]).exists());
assert!(prepared.temp_dir_guard.is_some());
}
#[tokio::test]
async fn prepare_embedded_startup_config_creates_missing_custom_volume() {
let parent = tempfile::tempdir().expect("temp parent");
let volume = parent.path().join("data");
let prepared = prepare_embedded_startup_config(
"127.0.0.1:9000".to_string(),
"access".to_string(),
"secret".to_string(),
vec![volume.display().to_string()],
"us-east-1".to_string(),
)
.await
.expect("embedded startup config should create custom volume");
assert_eq!(prepared.config.volumes, vec![volume.display().to_string()]);
assert!(volume.exists());
assert!(prepared.temp_dir_guard.is_none());
}
}