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

refactor: extract optional runtime sidecar boundary (#3637)

Browse Source
This commit is contained in:
安正超
2026-06-20 04:28:33 +08:00
committed by GitHub
parent aa5de1c908
commit 5e96151d9c
8 changed files with 242 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/lib.rs
+1
View File
@@ -72,6 +72,7 @@ pub mod startup_entrypoint;
pub mod startup_fs_guard;
pub mod startup_iam;
pub mod startup_lifecycle;
pub mod startup_optional_runtime_sidecars;
pub mod startup_optional_runtimes;
pub mod startup_preflight;
pub mod startup_profiling;
rustfs/src/startup_optional_runtime_sidecars.rs
+183
View File
@@ -0,0 +1,183 @@
// 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_protocols::{ProtocolShutdownSenders, init_protocol_shutdown_senders},
};
use futures_util::future::join_all;
use std::io::Result;
use tracing::info;
const LOG_COMPONENT_MAIN: &str = "main";
const LOG_SUBSYSTEM_STARTUP: &str = "startup";
const EVENT_PROTOCOL_SYSTEM_STATE: &str = "protocol_system_state";
pub struct OptionalRuntimeServices {
pub protocols: ProtocolShutdownSenders,
}
impl OptionalRuntimeServices {
pub fn new(protocols: ProtocolShutdownSenders) -> Self {
Self { protocols }
}
}
pub async fn init_optional_runtime_services() -> Result<OptionalRuntimeServices> {
let protocols = init_protocol_shutdown_senders().await?;
Ok(OptionalRuntimeServices::new(protocols))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum OptionalRuntimeShutdownStep {
Ftp,
Ftps,
Webdav,
Sftp,
}
impl OptionalRuntimeShutdownStep {
const fn protocol(self) -> &'static str {
match self {
Self::Ftp => "ftp",
Self::Ftps => "ftps",
Self::Webdav => "webdav",
Self::Sftp => "sftp",
}
}
}
#[cfg(test)]
fn optional_runtime_shutdown_steps(protocols: &ProtocolShutdownSenders) -> Vec<OptionalRuntimeShutdownStep> {
let mut steps = Vec::with_capacity(4);
if protocols.ftp.is_some() {
steps.push(OptionalRuntimeShutdownStep::Ftp);
}
if protocols.ftps.is_some() {
steps.push(OptionalRuntimeShutdownStep::Ftps);
}
if protocols.webdav.is_some() {
steps.push(OptionalRuntimeShutdownStep::Webdav);
}
if protocols.sftp.is_some() {
steps.push(OptionalRuntimeShutdownStep::Sftp);
}
steps
}
pub fn prepare_optional_runtime_shutdowns(optional_runtimes: OptionalRuntimeServices) -> Vec<ShutdownHandle> {
let ProtocolShutdownSenders { ftp, ftps, webdav, sftp } = optional_runtimes.protocols;
let mut protocol_shutdowns = Vec::new();
if let Some(handle) = ftp {
log_protocol_stopping(OptionalRuntimeShutdownStep::Ftp);
protocol_shutdowns.push(handle);
}
if let Some(handle) = ftps {
log_protocol_stopping(OptionalRuntimeShutdownStep::Ftps);
protocol_shutdowns.push(handle);
}
if let Some(handle) = webdav {
log_protocol_stopping(OptionalRuntimeShutdownStep::Webdav);
protocol_shutdowns.push(handle);
}
if let Some(handle) = sftp {
log_protocol_stopping(OptionalRuntimeShutdownStep::Sftp);
protocol_shutdowns.push(handle);
}
protocol_shutdowns
}
pub async fn shutdown_optional_runtime_services(protocol_shutdowns: Vec<ShutdownHandle>) {
join_all(protocol_shutdowns.into_iter().map(ShutdownHandle::shutdown)).await;
}
fn log_protocol_stopping(step: OptionalRuntimeShutdownStep) {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_PROTOCOL_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
protocol = step.protocol(),
state = "stopping",
"Protocol runtime stopping"
);
}
#[cfg(test)]
mod tests {
use super::{
OptionalRuntimeServices, OptionalRuntimeShutdownStep, optional_runtime_shutdown_steps,
prepare_optional_runtime_shutdowns, shutdown_optional_runtime_services,
};
use crate::{server::ShutdownHandle, startup_protocols::ProtocolShutdownSenders};
use std::sync::{Arc, Mutex};
use tokio::sync::broadcast;
fn completed_shutdown_handle() -> ShutdownHandle {
let (tx, _rx) = broadcast::channel(1);
ShutdownHandle::new(tx, tokio::spawn(async {}))
}
fn observed_shutdown_handle(label: &'static str, events: Arc<Mutex<Vec<&'static str>>>) -> ShutdownHandle {
let (tx, mut rx) = broadcast::channel(1);
let task_handle = tokio::spawn(async move {
let _ = rx.recv().await;
events.lock().unwrap_or_else(|err| err.into_inner()).push(label);
});
ShutdownHandle::new(tx, task_handle)
}
#[tokio::test]
async fn optional_runtime_shutdown_plan_preserves_protocol_order() {
let protocols = ProtocolShutdownSenders {
ftp: Some(completed_shutdown_handle()),
ftps: Some(completed_shutdown_handle()),
webdav: Some(completed_shutdown_handle()),
sftp: Some(completed_shutdown_handle()),
};
assert_eq!(
optional_runtime_shutdown_steps(&protocols),
vec![
OptionalRuntimeShutdownStep::Ftp,
OptionalRuntimeShutdownStep::Ftps,
OptionalRuntimeShutdownStep::Webdav,
OptionalRuntimeShutdownStep::Sftp,
]
);
}
#[tokio::test]
async fn optional_runtime_shutdown_signals_registered_protocols() {
let events = Arc::new(Mutex::new(Vec::new()));
let protocols = ProtocolShutdownSenders {
ftp: Some(observed_shutdown_handle("ftp", events.clone())),
ftps: None,
webdav: Some(observed_shutdown_handle("webdav", events.clone())),
sftp: None,
};
let protocol_shutdowns = prepare_optional_runtime_shutdowns(OptionalRuntimeServices::new(protocols));
tokio::task::yield_now().await;
assert!(events.lock().unwrap_or_else(|err| err.into_inner()).is_empty());
shutdown_optional_runtime_services(protocol_shutdowns).await;
let mut events = events.lock().unwrap_or_else(|err| err.into_inner()).clone();
events.sort_unstable();
assert_eq!(events, ["ftp", "webdav"]);
}
}
rustfs/src/startup_optional_runtimes.rs
+3 -168
View File
@@ -12,172 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::{
server::ShutdownHandle,
startup_protocols::{ProtocolShutdownSenders, init_protocol_shutdown_senders},
pub use crate::startup_optional_runtime_sidecars::{
OptionalRuntimeServices, init_optional_runtime_services, prepare_optional_runtime_shutdowns,
shutdown_optional_runtime_services,
};
use futures_util::future::join_all;
use std::io::Result;
use tracing::info;
const LOG_COMPONENT_MAIN: &str = "main";
const LOG_SUBSYSTEM_STARTUP: &str = "startup";
const EVENT_PROTOCOL_SYSTEM_STATE: &str = "protocol_system_state";
pub struct OptionalRuntimeServices {
pub protocols: ProtocolShutdownSenders,
}
impl OptionalRuntimeServices {
pub fn new(protocols: ProtocolShutdownSenders) -> Self {
Self { protocols }
}
}
pub async fn init_optional_runtime_services() -> Result<OptionalRuntimeServices> {
let protocols = init_protocol_shutdown_senders().await?;
Ok(OptionalRuntimeServices::new(protocols))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum OptionalRuntimeShutdownStep {
Ftp,
Ftps,
Webdav,
Sftp,
}
impl OptionalRuntimeShutdownStep {
const fn protocol(self) -> &'static str {
match self {
Self::Ftp => "ftp",
Self::Ftps => "ftps",
Self::Webdav => "webdav",
Self::Sftp => "sftp",
}
}
}
#[cfg(test)]
fn optional_runtime_shutdown_steps(protocols: &ProtocolShutdownSenders) -> Vec<OptionalRuntimeShutdownStep> {
let mut steps = Vec::with_capacity(4);
if protocols.ftp.is_some() {
steps.push(OptionalRuntimeShutdownStep::Ftp);
}
if protocols.ftps.is_some() {
steps.push(OptionalRuntimeShutdownStep::Ftps);
}
if protocols.webdav.is_some() {
steps.push(OptionalRuntimeShutdownStep::Webdav);
}
if protocols.sftp.is_some() {
steps.push(OptionalRuntimeShutdownStep::Sftp);
}
steps
}
pub fn prepare_optional_runtime_shutdowns(optional_runtimes: OptionalRuntimeServices) -> Vec<ShutdownHandle> {
let ProtocolShutdownSenders { ftp, ftps, webdav, sftp } = optional_runtimes.protocols;
let mut protocol_shutdowns = Vec::new();
if let Some(handle) = ftp {
log_protocol_stopping(OptionalRuntimeShutdownStep::Ftp);
protocol_shutdowns.push(handle);
}
if let Some(handle) = ftps {
log_protocol_stopping(OptionalRuntimeShutdownStep::Ftps);
protocol_shutdowns.push(handle);
}
if let Some(handle) = webdav {
log_protocol_stopping(OptionalRuntimeShutdownStep::Webdav);
protocol_shutdowns.push(handle);
}
if let Some(handle) = sftp {
log_protocol_stopping(OptionalRuntimeShutdownStep::Sftp);
protocol_shutdowns.push(handle);
}
protocol_shutdowns
}
pub async fn shutdown_optional_runtime_services(protocol_shutdowns: Vec<ShutdownHandle>) {
join_all(protocol_shutdowns.into_iter().map(ShutdownHandle::shutdown)).await;
}
fn log_protocol_stopping(step: OptionalRuntimeShutdownStep) {
info!(
target: "rustfs::main::handle_shutdown",
event = EVENT_PROTOCOL_SYSTEM_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
protocol = step.protocol(),
state = "stopping",
"Protocol runtime stopping"
);
}
#[cfg(test)]
mod tests {
use super::{
OptionalRuntimeServices, OptionalRuntimeShutdownStep, optional_runtime_shutdown_steps,
prepare_optional_runtime_shutdowns, shutdown_optional_runtime_services,
};
use crate::{server::ShutdownHandle, startup_protocols::ProtocolShutdownSenders};
use std::sync::{Arc, Mutex};
use tokio::sync::broadcast;
fn completed_shutdown_handle() -> ShutdownHandle {
let (tx, _rx) = broadcast::channel(1);
ShutdownHandle::new(tx, tokio::spawn(async {}))
}
fn observed_shutdown_handle(label: &'static str, events: Arc<Mutex<Vec<&'static str>>>) -> ShutdownHandle {
let (tx, mut rx) = broadcast::channel(1);
let task_handle = tokio::spawn(async move {
let _ = rx.recv().await;
events.lock().unwrap_or_else(|err| err.into_inner()).push(label);
});
ShutdownHandle::new(tx, task_handle)
}
#[tokio::test]
async fn optional_runtime_shutdown_plan_preserves_protocol_order() {
let protocols = ProtocolShutdownSenders {
ftp: Some(completed_shutdown_handle()),
ftps: Some(completed_shutdown_handle()),
webdav: Some(completed_shutdown_handle()),
sftp: Some(completed_shutdown_handle()),
};
assert_eq!(
optional_runtime_shutdown_steps(&protocols),
vec![
OptionalRuntimeShutdownStep::Ftp,
OptionalRuntimeShutdownStep::Ftps,
OptionalRuntimeShutdownStep::Webdav,
OptionalRuntimeShutdownStep::Sftp,
]
);
}
#[tokio::test]
async fn optional_runtime_shutdown_signals_registered_protocols() {
let events = Arc::new(Mutex::new(Vec::new()));
let protocols = ProtocolShutdownSenders {
ftp: Some(observed_shutdown_handle("ftp", events.clone())),
ftps: None,
webdav: Some(observed_shutdown_handle("webdav", events.clone())),
sftp: None,
};
let protocol_shutdowns = prepare_optional_runtime_shutdowns(OptionalRuntimeServices::new(protocols));
tokio::task::yield_now().await;
assert!(events.lock().unwrap_or_else(|err| err.into_inner()).is_empty());
shutdown_optional_runtime_services(protocol_shutdowns).await;
let mut events = events.lock().unwrap_or_else(|err| err.into_inner()).clone();
events.sort_unstable();
assert_eq!(events, ["ftp", "webdav"]);
}
}
rustfs/src/startup_services.rs
+1 -1
View File
@@ -18,7 +18,7 @@ use crate::{
init::{init_buffer_profile_system, init_kms_system},
server::ServiceStateManager,
startup_iam::IamBootstrapDisposition,
startup_optional_runtimes::{OptionalRuntimeServices, init_optional_runtime_services},
startup_optional_runtime_sidecars::{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,
rustfs/src/startup_shutdown.rs
+1 -1
View File
@@ -14,7 +14,7 @@
use crate::{
server::{ServiceState, ServiceStateManager, ShutdownHandle, ShutdownSignal, shutdown_event_notifier, stop_audit_system},
startup_optional_runtimes::{
startup_optional_runtime_sidecars::{
OptionalRuntimeServices, prepare_optional_runtime_shutdowns, shutdown_optional_runtime_services,
},
storage_compat::shutdown_background_services,