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

Centralize lifecycle state updates and fix systemd running status (#2567)

Browse Source
This commit is contained in:
houseme
2026-04-17 11:20:11 +08:00
committed by GitHub
parent 6b4172998b
commit 478720d2ee
28 changed files with 230 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
-3
View File
@@ -3202,13 +3202,10 @@ dependencies = [
"base64 0.22.1",
"bytes",
"chrono",
"clap",
"flatbuffers",
"flate2",
"futures",
"http 1.4.0",
"http-body 1.0.1",
"http-body-util",
"md5",
"rand 0.10.1",
"rcgen",
crates/common/src/data_usage.rs
+1 -1
View File
@@ -673,7 +673,7 @@ impl DataUsageCache {
let mut leaves = Vec::new();
let mut remove = total - limit;
add(self, path, &mut leaves);
leaves.sort_by(|a, b| a.objects.cmp(&b.objects));
leaves.sort_by_key(|a| a.objects);
while remove > 0 && !leaves.is_empty() {
let Some(e) = leaves.first() else {
crates/common/src/internode_metrics.rs
+1 -5
View File
@@ -101,11 +101,7 @@ impl InternodeMetrics {
pub fn snapshot(&self) -> InternodeMetricsSnapshot {
let dial_samples_total = self.dial_samples_total.load(Ordering::Relaxed);
let dial_total_time_nanos = self.dial_total_time_nanos.load(Ordering::Relaxed);
let dial_avg_time_nanos = if dial_samples_total == 0 {
0
} else {
dial_total_time_nanos / dial_samples_total
};
let dial_avg_time_nanos = dial_total_time_nanos.checked_div(dial_samples_total).unwrap_or(0);
InternodeMetricsSnapshot {
sent_bytes_total: self.sent_bytes_total.load(Ordering::Relaxed),
crates/e2e_test/Cargo.toml
-3
View File
@@ -41,7 +41,6 @@ serde_json.workspace = true
tonic = { workspace = true }
tokio = { workspace = true }
tokio-stream = { workspace = true }
clap = { workspace = true }
rustfs-madmin.workspace = true
rustfs-filemeta.workspace = true
bytes.workspace = true
@@ -53,8 +52,6 @@ async-compression = { workspace = true, features = ["tokio", "bzip2", "xz"] }
async-trait = { workspace = true }
flate2.workspace = true
http.workspace = true
http-body.workspace = true
http-body-util.workspace = true
reqwest = { workspace = true }
rustfs-signer.workspace = true
tracing = { workspace = true }
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+2 -4
View File
@@ -1687,10 +1687,8 @@ pub async fn eval_action_from_lifecycle(
let lock_enabled = if let Some(lr) = lr { lr.mode.is_some() } else { false };
match event.action {
IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction => {
if lock_enabled {
return lifecycle::Event::default();
}
IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction if lock_enabled => {
return lifecycle::Event::default();
}
IlmAction::DeleteVersionAction | IlmAction::DeleteRestoredVersionAction => {
if oi.version_id.is_none() {
crates/ecstore/src/bucket/replication/replication_state.rs
+46 -52
View File
@@ -832,61 +832,55 @@ impl ReplicationStats {
let mut rs = ReplStat::new();
match status {
ReplicationStatusType::Pending => {
if ri.op_type.is_data_replication() && prev_status != status {
rs.set(
ri.arn.clone(),
ri.size,
Duration::default(),
status,
ri.op_type,
ri.endpoint.clone(),
ri.secure,
ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())),
);
}
ReplicationStatusType::Pending if ri.op_type.is_data_replication() && prev_status != status => {
rs.set(
ri.arn.clone(),
ri.size,
Duration::default(),
status,
ri.op_type,
ri.endpoint.clone(),
ri.secure,
ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())),
);
}
ReplicationStatusType::Completed => {
if ri.op_type.is_data_replication() {
rs.set(
ri.arn.clone(),
ri.size,
ri.duration,
status,
ri.op_type,
ri.endpoint.clone(),
ri.secure,
ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())),
);
}
ReplicationStatusType::Completed if ri.op_type.is_data_replication() => {
rs.set(
ri.arn.clone(),
ri.size,
ri.duration,
status,
ri.op_type,
ri.endpoint.clone(),
ri.secure,
ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())),
);
}
ReplicationStatusType::Failed => {
if ri.op_type.is_data_replication() && prev_status == ReplicationStatusType::Pending {
rs.set(
ri.arn.clone(),
ri.size,
ri.duration,
status,
ri.op_type,
ri.endpoint.clone(),
ri.secure,
ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())),
);
}
ReplicationStatusType::Failed
if ri.op_type.is_data_replication() && prev_status == ReplicationStatusType::Pending =>
{
rs.set(
ri.arn.clone(),
ri.size,
ri.duration,
status,
ri.op_type,
ri.endpoint.clone(),
ri.secure,
ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())),
);
}
ReplicationStatusType::Replica => {
if ri.op_type == ReplicationType::Object {
rs.set(
ri.arn.clone(),
ri.size,
Duration::default(),
status,
ri.op_type,
String::new(),
false,
ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())),
);
}
ReplicationStatusType::Replica if ri.op_type == ReplicationType::Object => {
rs.set(
ri.arn.clone(),
ri.size,
Duration::default(),
status,
ri.op_type,
String::new(),
false,
ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())),
);
}
_ => {}
}
crates/ecstore/src/config/com.rs
+3 -3
View File
@@ -537,7 +537,7 @@ fn build_oidc_object(cfg: &Config) -> Map<String, Value> {
};
let mut providers = subsystem.iter().collect::<Vec<_>>();
providers.sort_by(|(lhs, _), (rhs, _)| lhs.cmp(rhs));
providers.sort_by_key(|(lhs, _)| *lhs);
let mut oidc_obj = Map::new();
for (instance_key, kvs) in providers {
@@ -572,7 +572,7 @@ fn build_semantic_oidc_object(cfg: &Config) -> Map<String, Value> {
};
let mut providers = subsystem.iter().collect::<Vec<_>>();
providers.sort_by(|(lhs, _), (rhs, _)| lhs.cmp(rhs));
providers.sort_by_key(|(lhs, _)| *lhs);
let mut oidc_obj = Map::new();
for (instance_key, kvs) in providers {
@@ -691,7 +691,7 @@ fn build_notify_subsystem_object(
.iter()
.filter(|(instance_key, _)| instance_key.as_str() != DEFAULT_DELIMITER)
.collect::<Vec<_>>();
instances.sort_by(|(lhs, _), (rhs, _)| lhs.cmp(rhs));
instances.sort_by_key(|(lhs, _)| *lhs);
for (instance_key, kvs) in instances {
let instance_obj = build_notify_instance_diff_object(kvs, &effective_default, valid_keys, default_kvs);
crates/ecstore/src/pools.rs
+2 -1
View File
@@ -1405,7 +1405,8 @@ impl ECStore {
let mut fivs = load_decommission_entry_versions(&entry, &bucket, "file_info_versions")?;
fivs.versions.sort_by(|a, b| b.mod_time.cmp(&a.mod_time));
fivs.versions
.sort_by_key(|v| (v.mod_time.is_none(), std::cmp::Reverse(v.mod_time)));
let mut decommissioned: usize = 0;
let mut expired: usize = 0;
crates/ecstore/src/rebalance.rs
+2 -1
View File
@@ -1530,7 +1530,8 @@ impl ECStore {
let mut fivs =
resolve_rebalance_file_info_versions_result(entry.file_info_versions(&bucket), bucket.as_str(), entry.name.as_str())?;
fivs.versions.sort_by(|a, b| b.mod_time.cmp(&a.mod_time));
fivs.versions
.sort_by_key(|v| (v.mod_time.is_none(), std::cmp::Reverse(v.mod_time)));
let mut rebalanced: usize = 0;
let mut expired: usize = 0;
crates/ecstore/src/set_disk.rs
+3 -3
View File
@@ -1627,7 +1627,7 @@ impl ObjectOperations for SetDisks {
let mut vers = Vec::with_capacity(vers_map.len());
for (_, mut fi_vers) in vers_map {
fi_vers.versions.sort_by(|a, b| a.deleted.cmp(&b.deleted));
fi_vers.versions.sort_by_key(|a| a.deleted);
if let Some(index) = fi_vers.versions.iter().position(|fi| fi.deleted) {
fi_vers.versions.truncate(index + 1);
@@ -2961,7 +2961,7 @@ impl MultipartOperations for SetDisks {
populated_upload_ids.insert(upload_id);
}
uploads.sort_by(|a, b| a.initiated.cmp(&b.initiated));
uploads.sort_by_key(|a| a.initiated);
let mut upload_idx = 0;
if let Some(upload_id_marker) = &upload_id_marker {
@@ -4212,7 +4212,7 @@ async fn get_storage_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> rust
// },
// }
let mut disks = get_disks_info(disks, eps).await;
disks.sort_by(|a, b| a.total_space.cmp(&b.total_space));
disks.sort_by_key(|a| a.total_space);
// Provide minimal backend shape for callers. Do NOT guess parity here since it belongs to higher-level config.
// Missing/empty standard_sc_data will be handled by capacity fallback logic.
crates/ecstore/src/set_disk/lock.rs
+1 -1
View File
@@ -168,7 +168,7 @@ impl SetDisks {
let mut new_disks = Vec::new();
let mut new_infos = Vec::new();
for (disk, info) in disks.into_iter().zip(infos.into_iter()) {
for (disk, info) in disks.into_iter().zip(infos) {
let Some(info) = info else {
continue;
};
crates/ecstore/src/store_list_objects.rs
+2 -4
View File
@@ -162,10 +162,8 @@ impl ListPathOptions {
}
match kv[0] {
"rustfs_cache" => {
if kv[1] != MARKER_TAG_VERSION {
continue;
}
"rustfs_cache" if kv[1] != MARKER_TAG_VERSION => {
continue;
}
"id" => self.id = Some(kv[1].to_owned()),
"return" => {
crates/filemeta/examples/dump_versions.rs
+14
View File
@@ -1,3 +1,17 @@
// 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 rustfs_filemeta::FileMeta;
use std::{env, fs, path::PathBuf};
crates/filemeta/src/fileinfo.rs
+6 -8
View File
@@ -349,7 +349,7 @@ impl FileInfo {
self.parts.push(part);
self.parts.sort_by(|a, b| a.number.cmp(&b.number));
self.parts.sort_by_key(|a| a.number);
}
// to_part_offset gets the part index where offset is located, returns part index and offset
@@ -643,13 +643,11 @@ pub fn parse_restore_obj_status(restore_hdr: &str) -> Result<RestoreStatus> {
}
match progress_tokens[1] {
"true" | "\"true\"" => {
if tokens.len() == 1 {
return Ok(RestoreStatus {
is_restore_in_progress: Some(true),
..Default::default()
});
}
"true" | "\"true\"" if tokens.len() == 1 => {
return Ok(RestoreStatus {
is_restore_in_progress: Some(true),
..Default::default()
});
}
"false" | "\"false\"" => {
if tokens.len() != 2 {
crates/filemeta/src/test_data.rs
+9 -7
View File
@@ -104,7 +104,8 @@ pub fn create_real_xlmeta() -> Result<Vec<u8>> {
fm.versions.push(legacy_shallow);
// Sort by modification time (newest first)
fm.versions.sort_by(|a, b| b.header.mod_time.cmp(&a.header.mod_time));
fm.versions
.sort_by_key(|v| (v.header.mod_time.is_none(), std::cmp::Reverse(v.header.mod_time)));
fm.marshal_msg()
}
@@ -263,11 +264,11 @@ pub fn create_legacy_v1_object_xlmeta() -> Result<Vec<u8>> {
wr.extend_from_slice(&[0xc6, 0, 0, 0, 0]);
let offset = wr.len();
rmp::encode::write_uint(&mut wr, 1).unwrap();
rmp::encode::write_uint(&mut wr, 1).unwrap();
rmp::encode::write_sint(&mut wr, 1).unwrap();
rmp::encode::write_bin(&mut wr, &header).unwrap();
rmp::encode::write_bin(&mut wr, &body).unwrap();
rmp::encode::write_uint(&mut wr, 1)?;
rmp::encode::write_uint(&mut wr, 1)?;
rmp::encode::write_sint(&mut wr, 1)?;
rmp::encode::write_bin(&mut wr, &header)?;
rmp::encode::write_bin(&mut wr, &body)?;
let data_len = (wr.len() - offset) as u32;
wr[offset - 4..offset].copy_from_slice(&data_len.to_be_bytes());
@@ -350,7 +351,8 @@ pub fn create_complex_xlmeta() -> Result<Vec<u8>> {
}
// Sort by modification time (newest first)
fm.versions.sort_by(|a, b| b.header.mod_time.cmp(&a.header.mod_time));
fm.versions
.sort_by_key(|v| (v.header.mod_time.is_none(), std::cmp::Reverse(v.header.mod_time)));
fm.marshal_msg()
}
crates/io-core/src/lock_optimizer.rs
+1 -5
View File
@@ -124,11 +124,7 @@ impl LockStats {
pub fn avg_hold_time(&self) -> Duration {
let total = self.total_hold_time_ns.load(Ordering::Relaxed);
let count = self.locks_acquired.load(Ordering::Relaxed);
if count == 0 {
Duration::ZERO
} else {
Duration::from_nanos(total / count)
}
total.checked_div(count).map(Duration::from_nanos).unwrap_or(Duration::ZERO)
}
/// Get maximum hold time.
crates/io-core/src/timeout_wrapper.rs
+1 -5
View File
@@ -342,11 +342,7 @@ impl TimeoutStats {
pub fn avg_wait_time(&self) -> Duration {
let total = self.total_wait_time_ns.load(Ordering::Relaxed);
let count = self.total_operations.load(Ordering::Relaxed);
if count == 0 {
Duration::ZERO
} else {
Duration::from_nanos(total / count)
}
total.checked_div(count).map(Duration::from_nanos).unwrap_or(Duration::ZERO)
}
/// Reset statistics.
crates/io-metrics/src/adaptive_ttl.rs
+1 -1
View File
@@ -255,7 +255,7 @@ impl AccessTracker {
/// Get keys sorted by access count (descending).
pub fn top_keys(&self, n: usize) -> Vec<(&String, &AccessRecord)> {
let mut entries: Vec<_> = self.records.iter().collect();
entries.sort_by(|a, b| b.1.count.cmp(&a.1.count));
entries.sort_by_key(|entry| std::cmp::Reverse(entry.1.count));
entries.into_iter().take(n).collect()
}
crates/lock/src/fast_lock/metrics.rs
+4 -5
View File
@@ -170,11 +170,10 @@ impl MetricsSnapshot {
}
pub fn avg_wait_time(&self) -> Duration {
if self.slow_path_success == 0 {
Duration::ZERO
} else {
Duration::from_nanos(self.total_wait_time_ns / self.slow_path_success)
}
self.total_wait_time_ns
.checked_div(self.slow_path_success)
.map(Duration::from_nanos)
.unwrap_or(Duration::ZERO)
}
pub fn max_wait_time(&self) -> Duration {
crates/object-capacity/src/scan.rs
+1 -5
View File
@@ -603,11 +603,7 @@ async fn get_dir_size_async(path: &Path) -> Result<CapacityScanResult, std::io::
} else if file_count > max_files_threshold {
let overflow_count = file_count - max_files_threshold;
let exact_prefix_count = file_count.min(max_files_threshold) as u64;
let avg_prefix_size = if exact_prefix_count > 0 {
exact_prefix_bytes / exact_prefix_count
} else {
0
};
let avg_prefix_size = exact_prefix_bytes.checked_div(exact_prefix_count).unwrap_or(0);
let estimated_overflow = avg_prefix_size.saturating_mul(overflow_count as u64);
let estimated_size = exact_prefix_bytes.saturating_add(estimated_overflow);
info!(
crates/scanner/src/data_usage_define.rs
+1 -1
View File
@@ -765,7 +765,7 @@ impl DataUsageCache {
let mut leaves = Vec::new();
let mut remove = total - limit;
add(self, path, &mut leaves);
leaves.sort_by(|a, b| a.objects.cmp(&b.objects));
leaves.sort_by_key(|a| a.objects);
while remove > 0 && !leaves.is_empty() {
let e = leaves.first().unwrap();
rustfs/src/admin/handlers/policies.rs
+2 -2
View File
@@ -641,8 +641,8 @@ fn build_policy_mappings(
}
let mut results: Vec<PolicyEntities> = policy_map
.into_iter()
.filter_map(|(_, mut mapping)| {
.into_values()
.filter_map(|mut mapping| {
if !requested_policies.is_empty() && !requested_policies.iter().any(|policy| policy == &mapping.policy) {
return None;
}
rustfs/src/embedded.rs
+19 -24
View File
@@ -28,8 +28,8 @@
//! let port = find_available_port()?;
//! let server = RustFSServerBuilder::new()
//! .address(format!("127.0.0.1:{port}"))
//! .access_key("minioadmin")
//! .secret_key("minioadmin")
//! .access_key("rustfsadmin")
//! .secret_key("rustfsadmin")
//! .build()
//! .await?;
//!
@@ -49,10 +49,7 @@
use crate::app::context::{AppContext, init_global_app_context};
use crate::config::Config;
use crate::init::{add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system};
use crate::server::{
ServiceState, ServiceStateManager, init_event_notifier, shutdown_event_notifier, start_audit_system, start_http_server,
stop_audit_system,
};
use crate::server::{init_event_notifier, shutdown_event_notifier, start_audit_system, start_http_server, stop_audit_system};
use rustfs_common::{GlobalReadiness, SystemStage, set_global_addr};
use rustfs_credentials::init_global_action_credentials;
use rustfs_ecstore::store::init_lock_clients;
@@ -166,7 +163,7 @@ impl RustFSServerBuilder {
///
/// Defaults:
/// - address: `"127.0.0.1:9000"`
/// - access_key / secret_key: `"minioadmin"`
/// - access_key / secret_key: `"rustfsadmin"`
/// - region: `"us-east-1"`
/// - A temporary directory is created automatically for data storage
///
@@ -175,10 +172,10 @@ impl RustFSServerBuilder {
pub fn new() -> Self {
Self {
address: "127.0.0.1:9000".to_string(),
access_key: "minioadmin".to_string(),
secret_key: "minioadmin".to_string(),
access_key: rustfs_credentials::DEFAULT_ACCESS_KEY.to_string(),
secret_key: rustfs_credentials::DEFAULT_SECRET_KEY.to_string(),
volumes: Vec::new(),
region: "us-east-1".to_string(),
region: rustfs_config::RUSTFS_REGION.to_string(),
}
}
@@ -196,13 +193,13 @@ impl RustFSServerBuilder {
self
}
/// Set the S3 access key (default: `"minioadmin"`).
/// Set the S3 access key (default: `"rustfsadmin"`).
pub fn access_key(mut self, key: impl Into<String>) -> Self {
self.access_key = key.into();
self
}
/// Set the S3 secret key (default: `"minioadmin"`).
/// Set the S3 secret key (default: `"rustfsadmin"`).
pub fn secret_key(mut self, key: impl Into<String>) -> Self {
self.secret_key = key.into();
self
@@ -355,13 +352,11 @@ impl RustFSServerBuilder {
// Service state.
let readiness = Arc::new(GlobalReadiness::new());
let state_manager = ServiceStateManager::new();
state_manager.update(ServiceState::Starting);
// Start HTTP server.
let mut s3_config = config.clone();
s3_config.console_enable = false;
let (shutdown_tx, bound_addr) = start_http_server(&s3_config, state_manager.clone(), readiness.clone()).await?;
let (shutdown_tx, bound_addr) = start_http_server(&s3_config, readiness.clone()).await?;
let ctx = CancellationToken::new();
let shutdown_embedded_server = || {
let _ = shutdown_tx.send(());
@@ -599,15 +594,15 @@ impl Drop for RustFSServer {
/// ```rust,no_run
/// use rustfs::embedded::{find_available_port, RustFSServerBuilder};
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
/// let port = find_available_port()?;
/// let server = RustFSServerBuilder::new()
/// .address(format!("127.0.0.1:{port}"))
/// .build()
/// .await?;
/// println!("Listening on port {port}");
/// # Ok(())
/// # }
/// async fn example() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
/// let port = find_available_port()?;
/// let server = RustFSServerBuilder::new()
/// .address(format!("127.0.0.1:{port}"))
/// .build()
/// .await?;
/// println!("Listening on port {port}");
/// Ok(())
/// }
/// ```
pub fn find_available_port() -> Result<u16, std::io::Error> {
let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
rustfs/src/main.rs
+4 -4
View File
@@ -340,14 +340,14 @@ async fn run(config: rustfs::config::Config) -> Result<()> {
let s3_shutdown_tx = {
let mut s3_config = config.clone();
s3_config.console_enable = false;
let (s3_shutdown_tx, _) = start_http_server(&s3_config, state_manager.clone(), readiness.clone()).await?;
let (s3_shutdown_tx, _) = start_http_server(&s3_config, readiness.clone()).await?;
Some(s3_shutdown_tx)
};
let console_shutdown_tx = if config.console_enable && !config.console_address.is_empty() {
let mut console_config = config.clone();
console_config.address = console_config.console_address.clone();
let (console_shutdown_tx, _) = start_http_server(&console_config, state_manager.clone(), readiness.clone()).await?;
let (console_shutdown_tx, _) = start_http_server(&console_config, readiness.clone()).await?;
Some(console_shutdown_tx)
} else {
None
@@ -576,11 +576,11 @@ async fn run(config: rustfs::config::Config) -> Result<()> {
);
// 4. Mark as Full Ready now that critical components are warm
readiness.mark_stage(SystemStage::FullReady);
// Update service status to Ready
state_manager.update(ServiceState::Ready);
// Set the global RustFS initialization time to now
rustfs_common::set_global_init_time_now().await;
// Publish ready only after all critical bootstrap metadata is in place
state_manager.update(ServiceState::Ready);
// Perform hibernation for 1 second
tokio::time::sleep(SHUTDOWN_TIMEOUT).await;
rustfs/src/server/http.rs
+4 -11
View File
@@ -18,7 +18,7 @@ use crate::auth::IAMAuth;
use crate::auth_keystone;
use crate::config;
use crate::server::{
ReadinessGateLayer, RemoteAddr, ServiceState, ServiceStateManager,
ReadinessGateLayer, RemoteAddr,
compress::{CompressionConfig, PathAwareCompressionPredicate, PathCategoryInjectionLayer},
hybrid::hybrid,
layer::{
@@ -68,7 +68,6 @@ use tracing_opentelemetry::OpenTelemetrySpanExt;
pub async fn start_http_server(
config: &config::Config,
worker_state_manager: ServiceStateManager,
readiness: Arc<GlobalReadiness>,
) -> Result<(tokio::sync::broadcast::Sender<()>, SocketAddr)> {
let server_addr = parse_and_resolve_address(config.address.as_str()).map_err(Error::other)?;
@@ -161,12 +160,12 @@ pub async fn start_http_server(
// Note: outbound material (root CAs, mTLS identity) is already applied in main.rs.
let tls_snapshot = TlsMaterialSnapshot::load(tls_path)
.await
.map_err(|e| std::io::Error::other(e.to_string()))?;
.map_err(|e| Error::other(e.to_string()))?;
let tls_acceptor = tls_snapshot
.build_tls_acceptor(tls_path)
.await
.map_err(|e| std::io::Error::other(e.to_string()))?;
.map_err(|e| Error::other(e.to_string()))?;
let tls_enabled = tls_acceptor.is_some();
let protocol = if tls_enabled { "https" } else { "http" };
@@ -357,10 +356,6 @@ pub async fn start_http_server(
let graceful = Arc::new(GracefulShutdown::new());
debug!("graceful initiated");
// service ready
worker_state_manager.update(ServiceState::Ready);
// tls_acceptor is already Option<Arc<TlsAcceptorHolder>>, clone for the loop
loop {
debug!("Waiting for new connection...");
let (socket, _) = {
@@ -458,7 +453,6 @@ pub async fn start_http_server(
process_connection(socket, tls_acceptor.clone(), connection_ctx, graceful.clone());
}
worker_state_manager.update(ServiceState::Stopping);
match Arc::try_unwrap(graceful) {
Ok(g) => {
tokio::select! {
@@ -476,7 +470,6 @@ pub async fn start_http_server(
debug!("Timeout reached, forcing shutdown");
}
}
worker_state_manager.update(ServiceState::Stopped);
});
Ok((shutdown_tx, local_addr))
@@ -490,7 +483,7 @@ struct ConnectionContext {
is_console: bool,
readiness: Arc<GlobalReadiness>,
/// Pre-computed Keystone auth provider (avoids per-connection OnceLock read).
keystone_auth: Option<std::sync::Arc<rustfs_keystone::KeystoneAuthProvider>>,
keystone_auth: Option<Arc<rustfs_keystone::KeystoneAuthProvider>>,
/// Pre-computed trusted proxy layer (avoids per-connection is_enabled() check).
trusted_proxy_layer: Option<rustfs_trusted_proxies::TrustedProxyLayer>,
}
rustfs/src/server/service_state.rs
+98 -44
View File
@@ -13,39 +13,18 @@
// limitations under the License.
use atomic_enum::atomic_enum;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tracing::info;
use tracing::{info, warn};
// a configurable shutdown timeout
pub const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(1);
#[cfg(target_os = "linux")]
fn notify_systemd(state: &str) {
use libsystemd::daemon::{NotifyState, notify};
use tracing::{debug, error};
let notify_state = match state {
"ready" => NotifyState::Ready,
"stopping" => NotifyState::Stopping,
_ => {
info!("Unsupported state passed to notify_systemd: {}", state);
return;
}
};
if let Err(e) = notify(false, &[notify_state]) {
error!("Failed to notify systemd: {}", e);
} else {
debug!("Successfully notified systemd: {}", state);
}
info!("Systemd notifications are enabled on linux (state: {})", state);
}
#[cfg(not(target_os = "linux"))]
fn notify_systemd(state: &str) {
info!("Systemd notifications are not available on this platform not linux (state: {})", state);
}
const SERVICE_STATUS_STARTING: &str = "Starting";
const SERVICE_STATUS_RUNNING: &str = "Running";
const SERVICE_STATUS_STOPPING: &str = "Stopping";
const SERVICE_STATUS_STOPPED: &str = "Stopped";
#[derive(Debug)]
pub enum ShutdownSignal {
@@ -100,56 +79,112 @@ pub async fn wait_for_shutdown() -> ShutdownSignal {
#[derive(Clone)]
pub struct ServiceStateManager {
state: Arc<AtomicServiceState>,
published_state: Arc<Mutex<Option<ServiceState>>>,
}
impl ServiceStateManager {
pub fn new() -> Self {
Self {
state: Arc::new(AtomicServiceState::new(ServiceState::Starting)),
published_state: Arc::new(Mutex::new(None)),
}
}
pub fn update(&self, new_state: ServiceState) {
// Serialize transition check + state write + publish dedupe + notify as one
// critical section to keep notification order monotonic under concurrency.
let mut published_state = self.published_state.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let current_state = self.current_state();
if service_state_rank(new_state) < service_state_rank(current_state) {
warn!(
current = ?current_state,
attempted = ?new_state,
"Ignoring regressive service state transition"
);
return;
}
self.state.store(new_state, Ordering::SeqCst);
self.notify_systemd(&new_state);
if *published_state != Some(new_state) {
*published_state = Some(new_state);
self.notify_systemd(new_state);
}
}
pub fn current_state(&self) -> ServiceState {
self.state.load(Ordering::SeqCst)
}
fn notify_systemd(&self, state: &ServiceState) {
fn notify_systemd(&self, state: ServiceState) {
match state {
ServiceState::Starting => {
info!("RustFS Service is starting...");
#[cfg(target_os = "linux")]
if let Err(e) =
libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Starting...".to_string())])
{
tracing::error!("Failed to notify systemd of starting state: {}", e);
}
notify_systemd_daemon(state);
}
ServiceState::Ready => {
info!("RustFS Service is ready");
notify_systemd("ready");
info!("RustFS Service is running");
notify_systemd_daemon(state);
}
ServiceState::Stopping => {
info!("RustFS Service is stopping...");
notify_systemd("stopping");
notify_systemd_daemon(state);
}
ServiceState::Stopped => {
info!("RustFS Service has stopped");
#[cfg(target_os = "linux")]
if let Err(e) =
libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Stopped".to_string())])
{
tracing::error!("Failed to notify systemd of stopped state: {}", e);
}
notify_systemd_daemon(state);
}
}
}
}
fn service_state_rank(state: ServiceState) -> u8 {
match state {
ServiceState::Starting => 0,
ServiceState::Ready => 1,
ServiceState::Stopping => 2,
ServiceState::Stopped => 3,
}
}
fn systemd_status_text(state: ServiceState) -> &'static str {
match state {
ServiceState::Starting => SERVICE_STATUS_STARTING,
ServiceState::Ready => SERVICE_STATUS_RUNNING,
ServiceState::Stopping => SERVICE_STATUS_STOPPING,
ServiceState::Stopped => SERVICE_STATUS_STOPPED,
}
}
#[cfg(target_os = "linux")]
fn notify_systemd_daemon(state: ServiceState) {
use libsystemd::daemon::{NotifyState, notify};
use tracing::{debug, error};
let status = systemd_status_text(state);
let result = match state {
ServiceState::Starting => notify(false, &[NotifyState::Status(status.to_string())]),
ServiceState::Ready => notify(false, &[NotifyState::Ready, NotifyState::Status(status.to_string())]),
ServiceState::Stopping => notify(false, &[NotifyState::Stopping, NotifyState::Status(status.to_string())]),
ServiceState::Stopped => notify(false, &[NotifyState::Status(status.to_string())]),
};
if let Err(e) = result {
error!(%status, ?state, "Failed to notify systemd: {}", e);
} else {
debug!(%status, ?state, "Successfully notified systemd");
}
}
#[cfg(not(target_os = "linux"))]
fn notify_systemd_daemon(state: ServiceState) {
info!(
status = systemd_status_text(state),
?state,
"Systemd notifications are not available on this platform"
);
}
impl Default for ServiceStateManager {
fn default() -> Self {
Self::new()
@@ -180,4 +215,23 @@ mod tests {
manager.update(ServiceState::Stopped);
assert_eq!(manager.current_state(), ServiceState::Stopped);
}
#[test]
fn test_service_state_manager_ignores_regression() {
let manager = ServiceStateManager::new();
manager.update(ServiceState::Starting);
manager.update(ServiceState::Ready);
manager.update(ServiceState::Starting);
assert_eq!(manager.current_state(), ServiceState::Ready);
manager.update(ServiceState::Stopping);
manager.update(ServiceState::Ready);
assert_eq!(manager.current_state(), ServiceState::Stopping);
}
#[test]
fn test_ready_maps_to_running_status() {
assert_eq!(systemd_status_text(ServiceState::Ready), SERVICE_STATUS_RUNNING);
}
}
rustfs/src/storage/concurrency/io_schedule.rs
+1 -5
View File
@@ -1263,11 +1263,7 @@ impl IoLoadMetrics {
pub(crate) fn lifetime_average_wait(&self) -> Duration {
let total = self.total_wait_ns.load(Ordering::Relaxed);
let count = self.observation_count.load(Ordering::Relaxed);
if count == 0 {
Duration::ZERO
} else {
Duration::from_nanos(total / count)
}
total.checked_div(count).map(Duration::from_nanos).unwrap_or(Duration::ZERO)
}
/// Get the total observation count
rustfs/src/storage/lock_optimizer.rs
+1 -5
View File
@@ -147,11 +147,7 @@ impl LockStats {
pub fn avg_hold_time(&self) -> Duration {
let total = self.total_hold_time_us.load(Ordering::Relaxed);
let count = self.locks_released_early.load(Ordering::Relaxed);
if count > 0 {
Duration::from_micros(total / count)
} else {
Duration::ZERO
}
total.checked_div(count).map(Duration::from_micros).unwrap_or(Duration::ZERO)
}
/// Get maximum hold time.