improve signal watch

Cargo.lock

@@ -669,6 +669,17 @@ version = "1.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
 
+[[package]]
+name = "atomic_enum"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "99e1aca718ea7b89985790c94aad72d77533063fe00bc497bb79a7c2dae6a661"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.100",
+]
+
 [[package]]
 name = "autocfg"
 version = "1.4.0"
@@ -7133,6 +7144,7 @@ dependencies = [
  "appauth",
  "async-trait",
  "atoi",
+ "atomic_enum",
  "axum",
  "axum-extra",
  "axum-server",
@@ -7279,9 +7291,9 @@ dependencies = [
 
 [[package]]
 name = "rustls"
-version = "0.23.25"
+version = "0.23.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "822ee9188ac4ec04a2f0531e55d035fb2de73f18b41a63c70c2712503b6fb13c"
+checksum = "df51b5869f3a441595eac5e8ff14d486ff285f7b8c0df8770e49c3b56351f0f0"
 dependencies = [
  "aws-lc-rs",
  "log",

Cargo.toml

@@ -36,6 +36,7 @@ madmin = { path = "./madmin" }
 atoi = "2.0.0"
 async-recursion = "1.0.5"
 async-trait = "0.1.87"
+atomic_enum = "0.3.0"
 axum = "0.8.3"
 axum-extra = "0.10.1"
 axum-server = { version = "0.7.2", features = ["tls-rustls"] }
@@ -98,7 +99,7 @@ rmp = "0.8.14"
 rmp-serde = "1.3.0"
 rustfs-obs = { path = "crates/obs", version = "0.0.1" }
 rust-embed = "8.6.0"
-rustls = { version = "0.23" }
+rustls = { version = "0.23.26" }
 rustls-pki-types = "1.11.0"
 rustls-pemfile = "2.2.0"
 s3s = { git = "https://github.com/Nugine/s3s.git", rev = "ab139f72fe768fb9d8cecfe36269451da1ca9779", default-features = true, features = [

crates/obs/src/config.rs

@@ -1,5 +1,5 @@
-use crate::global::{ENVIRONMENT, LOGGER_LEVEL, METER_INTERVAL, SAMPLE_RATIO, SERVICE_NAME, SERVICE_VERSION};
+use crate::global::{ENVIRONMENT, LOGGER_LEVEL, METER_INTERVAL, SAMPLE_RATIO, SERVICE_NAME, SERVICE_VERSION, USE_STDOUT};
-use config::{Config, File, FileFormat};
+use config::{Config, Environment, File, FileFormat};
 use serde::Deserialize;
 use std::env;
 
@@ -22,18 +22,44 @@ pub struct OtelConfig {
     pub logger_level: Option<String>,
 }
 
+// 辅助函数：从环境变量中提取可观测性配置
+fn extract_otel_config_from_env() -> OtelConfig {
+    OtelConfig {
+        endpoint: env::var("RUSTFS_OBSERVABILITY_ENDPOINT").unwrap_or_else(|_| "".to_string()),
+        use_stdout: env::var("RUSTFS_OBSERVABILITY_USE_STDOUT")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .or(Some(USE_STDOUT)),
+        sample_ratio: env::var("RUSTFS_OBSERVABILITY_SAMPLE_RATIO")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .or(Some(SAMPLE_RATIO)),
+        meter_interval: env::var("RUSTFS_OBSERVABILITY_METER_INTERVAL")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .or(Some(METER_INTERVAL)),
+        service_name: env::var("RUSTFS_OBSERVABILITY_SERVICE_NAME")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .or(Some(SERVICE_NAME.to_string())),
+        service_version: env::var("RUSTFS_OBSERVABILITY_SERVICE_VERSION")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .or(Some(SERVICE_VERSION.to_string())),
+        environment: env::var("RUSTFS_OBSERVABILITY_ENVIRONMENT")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .or(Some(ENVIRONMENT.to_string())),
+        logger_level: env::var("RUSTFS_OBSERVABILITY_LOGGER_LEVEL")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .or(Some(LOGGER_LEVEL.to_string())),
+    }
+}
+
 impl Default for OtelConfig {
     fn default() -> Self {
-        OtelConfig {
+        extract_otel_config_from_env()
-            endpoint: "".to_string(),
-            use_stdout: Some(true),
-            sample_ratio: Some(SAMPLE_RATIO),
-            meter_interval: Some(METER_INTERVAL),
-            service_name: Some(SERVICE_NAME.to_string()),
-            service_version: Some(SERVICE_VERSION.to_string()),
-            environment: Some(ENVIRONMENT.to_string()),
-            logger_level: Some(LOGGER_LEVEL.to_string()),
-        }
     }
 }
 
@@ -69,14 +95,18 @@ pub struct FileSinkConfig {
 
 impl FileSinkConfig {
     pub fn get_default_log_path() -> String {
-        let temp_dir = env::temp_dir().join("rustfs").join("logs");
+        let temp_dir = env::temp_dir().join("rustfs");
 
         if let Err(e) = std::fs::create_dir_all(&temp_dir) {
             eprintln!("Failed to create log directory: {}", e);
-            return "logs/app.log".to_string();
+            return "rustfs/rustfs.log".to_string();
         }
 
-        temp_dir.join("app.log").to_str().unwrap_or("logs/app.log").to_string()
+        temp_dir
+            .join("rustfs.log")
+            .to_str()
+            .unwrap_or("rustfs/rustfs.log")
+            .to_string()
     }
 }
 
@@ -84,7 +114,10 @@ impl Default for FileSinkConfig {
     fn default() -> Self {
         FileSinkConfig {
             enabled: true,
-            path: Self::get_default_log_path(),
+            path: env::var("RUSTFS_SINKS_FILE_PATH")
+                .ok()
+                .filter(|s| !s.trim().is_empty())
+                .unwrap_or_else(|| Self::get_default_log_path()),
             buffer_size: Some(8192),
             flush_interval_ms: Some(1000),
             flush_threshold: Some(100),
@@ -93,11 +126,21 @@ impl Default for FileSinkConfig {
 }
 
 /// Sink configuration collection
-#[derive(Debug, Deserialize, Clone, Default)]
+#[derive(Debug, Deserialize, Clone)]
 pub struct SinkConfig {
-    pub kafka: KafkaSinkConfig,
+    pub kafka: Option<KafkaSinkConfig>,
-    pub webhook: WebhookSinkConfig,
+    pub webhook: Option<WebhookSinkConfig>,
-    pub file: FileSinkConfig,
+    pub file: Option<FileSinkConfig>,
+}
+
+impl Default for SinkConfig {
+    fn default() -> Self {
+        SinkConfig {
+            kafka: None,
+            webhook: None,
+            file: Some(FileSinkConfig::default()),
+        }
+    }
 }
 
 ///Logger Configuration
@@ -109,7 +152,7 @@ pub struct LoggerConfig {
 impl Default for LoggerConfig {
     fn default() -> Self {
         LoggerConfig {
-            queue_capacity: Some(1000),
+            queue_capacity: Some(10000),
         }
     }
 }
@@ -128,11 +171,28 @@ impl Default for LoggerConfig {
 ///
 /// let config = load_config(None);
 /// ```
-#[derive(Debug, Deserialize, Clone, Default)]
+#[derive(Debug, Deserialize, Clone)]
 pub struct AppConfig {
     pub observability: OtelConfig,
     pub sinks: SinkConfig,
-    pub logger: LoggerConfig,
+    pub logger: Option<LoggerConfig>,
+}
+
+// 为 AppConfig 实现 Default
+impl AppConfig {
+    pub fn new() -> Self {
+        Self {
+            observability: OtelConfig::default(),
+            sinks: SinkConfig::default(),
+            logger: Some(LoggerConfig::default()),
+        }
+    }
+}
+
+impl Default for AppConfig {
+    fn default() -> Self {
+        Self::new()
+    }
 }
 
 const DEFAULT_CONFIG_FILE: &str = "obs";
@@ -187,9 +247,25 @@ pub fn load_config(config_dir: Option<String>) -> AppConfig {
     let config = Config::builder()
         .add_source(File::with_name(config_dir.as_str()).format(FileFormat::Toml).required(false))
         .add_source(File::with_name(config_dir.as_str()).format(FileFormat::Yaml).required(false))
-        .add_source(config::Environment::with_prefix(""))
+        .add_source(
+            Environment::default()
+                .prefix("RUSTFS")
+                .prefix_separator("__")
+                .separator("__")
+                .with_list_parse_key("volumes")
+                .try_parsing(true),
+        )
         .build()
         .unwrap_or_default();
 
-    config.try_deserialize().unwrap_or_default()
+    match config.try_deserialize::<AppConfig>() {
+        Ok(app_config) => {
+            println!("Parsed AppConfig: {:?}", app_config);
+            app_config
+        }
+        Err(e) => {
+            println!("Failed to deserialize config: {}", e);
+            AppConfig::default()
+        }
+    }
 }

crates/obs/src/logger.rs

@@ -21,7 +21,7 @@ impl Logger {
     /// Returns Logger and corresponding Receiver
     pub fn new(config: &AppConfig) -> (Self, Receiver<UnifiedLogEntry>) {
         // Get queue capacity from configuration, or use default values 10000
-        let queue_capacity = config.logger.queue_capacity.unwrap_or(10000);
+        let queue_capacity = config.logger.as_ref().and_then(|l| l.queue_capacity).unwrap_or(10000);
         let (sender, receiver) = mpsc::channel(queue_capacity);
         (Logger { sender, queue_capacity }, receiver)
     }

crates/obs/src/sink.rs

@@ -4,7 +4,6 @@ use std::sync::Arc;
 use tokio::fs::OpenOptions;
 use tokio::io;
 use tokio::io::AsyncWriteExt;
-use tracing::debug;
 
 /// Sink Trait definition, asynchronously write logs
 #[async_trait]
@@ -274,15 +273,15 @@ impl FileSink {
         // if the file not exists, create it
         if !file_exists {
             tokio::fs::create_dir_all(std::path::Path::new(&path).parent().unwrap()).await?;
-            debug!("the file not exists,create if. path: {:?}", path)
+            tracing::debug!("the file not exists,create if. path: {:?}", path)
         }
         let file = if file_exists {
             // If the file exists, open it in append mode
-            debug!("FileSink: File exists, opening in append mode.");
+            tracing::debug!("FileSink: File exists, opening in append mode.");
             OpenOptions::new().append(true).create(true).open(&path).await?
         } else {
             // If the file does not exist, create it
-            debug!("FileSink: File does not exist, creating a new file.");
+            tracing::debug!("FileSink: File does not exist, creating a new file.");
             // Create the file and write a header or initial content if needed
             OpenOptions::new().create(true).truncate(true).write(true).open(&path).await?
         };
@@ -414,53 +413,85 @@ pub async fn create_sinks(config: &AppConfig) -> Vec<Arc<dyn Sink>> {
     let mut sinks: Vec<Arc<dyn Sink>> = Vec::new();
 
     #[cfg(feature = "kafka")]
-    if config.sinks.kafka.enabled {
+    {
+        match &config.sinks.kafka {
+            Some(sink_kafka) => {
+                if sink_kafka.enabled {
                     match rdkafka::config::ClientConfig::new()
-            .set("bootstrap.servers", &config.sinks.kafka.bootstrap_servers)
+                        .set("bootstrap.servers", &sink_kafka.bootstrap_servers)
                         .set("message.timeout.ms", "5000")
                         .create()
                     {
                         Ok(producer) => {
                             sinks.push(Arc::new(KafkaSink::new(
                                 producer,
-                    config.sinks.kafka.topic.clone(),
+                                sink_kafka.topic.clone(),
-                    config.sinks.kafka.batch_size.unwrap_or(100),
+                                sink_kafka.batch_size.unwrap_or(100),
-                    config.sinks.kafka.batch_timeout_ms.unwrap_or(1000),
+                                sink_kafka.batch_timeout_ms.unwrap_or(1000),
                             )));
                         }
-            Err(e) => eprintln!("Failed to create Kafka producer: {}", e),
+                        Err(e) => {
+                            tracing::error!("Failed to create Kafka producer: {}", e);
+                        }
+                    }
+                } else {
+                    tracing::info!("Kafka sink is disabled in the configuration");
+                }
+            }
+            _ => {
+                tracing::info!("Kafka sink is not configured or disabled");
+            }
         }
     }
-
     #[cfg(feature = "webhook")]
-    if config.sinks.webhook.enabled {
+    {
+        match &config.sinks.webhook {
+            Some(sink_webhook) => {
+                if sink_webhook.enabled {
                     sinks.push(Arc::new(WebhookSink::new(
-            config.sinks.webhook.endpoint.clone(),
+                        sink_webhook.endpoint.clone(),
-            config.sinks.webhook.auth_token.clone(),
+                        sink_webhook.auth_token.clone(),
-            config.sinks.webhook.max_retries.unwrap_or(3),
+                        sink_webhook.max_retries.unwrap_or(3),
-            config.sinks.webhook.retry_delay_ms.unwrap_or(100),
+                        sink_webhook.retry_delay_ms.unwrap_or(100),
                     )));
+                } else {
+                    tracing::info!("Webhook sink is disabled in the configuration");
+                }
+            }
+            _ => {
+                tracing::info!("Webhook sink is not configured or disabled");
+            }
+        }
     }
 
     #[cfg(feature = "file")]
     {
-        let path = if config.sinks.file.enabled {
+        // let config = config.clone();
-            config.sinks.file.path.clone()
+        match &config.sinks.file {
+            Some(sink_file) => {
+                tracing::info!("File sink is enabled in the configuration");
+                let path = if sink_file.enabled {
+                    sink_file.path.clone()
                 } else {
-            "default.log".to_string()
+                    "rustfs.log".to_string()
                 };
-        debug!("FileSink: Using path: {}", path);
+                tracing::debug!("FileSink: Using path: {}", path);
                 sinks.push(Arc::new(
                     FileSink::new(
                         path.clone(),
-                config.sinks.file.buffer_size.unwrap_or(8192),
+                        sink_file.buffer_size.unwrap_or(8192),
-                config.sinks.file.flush_interval_ms.unwrap_or(1000),
+                        sink_file.flush_interval_ms.unwrap_or(1000),
-                config.sinks.file.flush_threshold.unwrap_or(100),
+                        sink_file.flush_threshold.unwrap_or(100),
                     )
                     .await
                     .unwrap(),
                 ));
             }
+            _ => {
+                tracing::info!("File sink is not configured or disabled");
+            }
+        }
+    }
 
     sinks
 }

crates/obs/src/telemetry.rs

@@ -293,7 +293,10 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
 
     registry.with(ErrorLayer::default()).with(fmt_layer).init();
     if !config.endpoint.is_empty() {
-        info!("OpenTelemetry telemetry initialized with OTLP endpoint: {}", config.endpoint);
+        info!(
+            "OpenTelemetry telemetry initialized with OTLP endpoint: {}, logger_level: {}",
+            config.endpoint, logger_level
+        );
     }
 
     OtelGuard {

deploy/config/.example.env

@@ -0,0 +1,27 @@
+OBSERVABILITY__ENDPOINT=http://localhost:4317
+OBSERVABILITY__USE_STDOUT=true
+OBSERVABILITY__SAMPLE_RATIO=2.0
+OBSERVABILITY__METER_INTERVAL=30
+OBSERVABILITY__SERVICE_NAME=rustfs
+OBSERVABILITY__SERVICE_VERSION=0.1.0
+OBSERVABILITY__ENVIRONMENT=develop
+OBSERVABILITY__LOGGER_LEVEL=debug
+
+SINKS__KAFKA__ENABLED=false
+SINKS__KAFKA__BOOTSTRAP_SERVERS=localhost:9092
+SINKS__KAFKA__TOPIC=logs
+SINKS__KAFKA__BATCH_SIZE=100
+SINKS__KAFKA__BATCH_TIMEOUT_MS=1000
+
+SINKS__WEBHOOK__ENABLED=false
+SINKS__WEBHOOK__ENDPOINT=http://localhost:8080/webhook
+SINKS__WEBHOOK__AUTH_TOKEN=
+SINKS__WEBHOOK__BATCH_SIZE=100
+SINKS__WEBHOOK__BATCH_TIMEOUT_MS=1000
+
+SINKS__FILE__ENABLED=true
+SINKS__FILE__PATH=./deploy/logs/app.log
+SINKS__FILE__BATCH_SIZE=10
+SINKS__FILE__BATCH_TIMEOUT_MS=1000
+
+LOGGER__QUEUE_CAPACITY=10

deploy/config/obs.example.toml

@@ -6,7 +6,7 @@ meter_interval = 30
 service_name = "rustfs"
 service_version = "0.1.0"
 environment = "develop"
-looger_level = "info"
+logger_level = "info"
 
 [sinks]
 [sinks.kafka] # Kafka sink is disabled by default
@@ -25,7 +25,7 @@ batch_timeout_ms = 1000 # Default is 100ms if not specified
 
 [sinks.file]
 enabled = true
-path = "logs/app.log"
+path = "./deploy/logs/app.log"
 batch_size = 100
 batch_timeout_ms = 1000 # Default is 8192 bytes if not specified
 

rustfs/Cargo.toml

@@ -19,6 +19,7 @@ madmin.workspace = true
 api = { path = "../s3select/api" }
 appauth = { version = "0.0.1", path = "../appauth" }
 atoi = { workspace = true }
+atomic_enum = { workspace = true }
 axum.workspace = true
 axum-extra = { workspace = true }
 axum-server = { workspace = true }
@@ -87,7 +88,7 @@ url.workspace = true
 uuid = "1.15.1"
 
 [target.'cfg(target_os = "linux")'.dependencies]
-libsystemd = "0.7"
+libsystemd.workspace = true
 
 [build-dependencies]
 prost-build.workspace = true

rustfs/src/main.rs

@@ -5,14 +5,15 @@ mod console;
 mod grpc;
 pub mod license;
 mod logging;
+mod server;
 mod service;
 mod storage;
 mod utils;
-
 use crate::auth::IAMAuth;
 use crate::console::{init_console_cfg, CONSOLE_CONFIG};
-use crate::utils::error;
 // Ensure the correct path for parse_license is imported
+use crate::server::{wait_for_shutdown, ServiceState, ServiceStateManager, ShutdownSignal, SHUTDOWN_TIMEOUT};
+use crate::utils::error;
 use chrono::Datelike;
 use clap::Parser;
 use common::{
@@ -31,6 +32,7 @@ use ecstore::{
 };
 use ecstore::{global::set_global_rustfs_port, notification_sys::new_global_notification_sys};
 use grpc::make_server;
+use hyper_util::server::graceful::GracefulShutdown;
 use hyper_util::{
     rt::{TokioExecutor, TokioIo},
     server::conn::auto::Builder as ConnBuilder,
@@ -44,41 +46,18 @@ use rustls::ServerConfig;
 use s3s::{host::MultiDomain, service::S3ServiceBuilder};
 use service::hybrid;
 use std::sync::Arc;
+use std::time::Duration;
 use std::{io::IsTerminal, net::SocketAddr};
 use tokio::net::TcpListener;
 use tokio::signal::unix::{signal, SignalKind};
 use tokio_rustls::TlsAcceptor;
 use tonic::{metadata::MetadataValue, Request, Status};
 use tower_http::cors::CorsLayer;
-use tracing::{debug, error, info, info_span, warn};
+use tracing::log::warn;
+use tracing::{debug, error, info, info_span};
 use tracing_error::ErrorLayer;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
-#[cfg(target_os = "linux")]
-fn notify_systemd(state: &str) {
-    use libsystemd::daemon::{notify, NotifyState};
-    let notify_state = match state {
-        "ready" => NotifyState::Ready,
-        "stopping" => NotifyState::Stopping,
-        _ => {
-            warn!("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);
-}
-
 #[allow(dead_code)]
 fn setup_tracing() {
     use tracing_subscriber::EnvFilter;
@@ -286,8 +265,14 @@ async fn run(opt: config::Opt) -> Result<()> {
         None
     };
 
-    // Create an oneshot channel to wait for the service to start
+    let state_manager = ServiceStateManager::new();
-    let (tx, rx) = tokio::sync::oneshot::channel();
+    let worker_state_manager = state_manager.clone();
+    // 更新服务状态为启动中
+    state_manager.update(ServiceState::Starting);
+
+    // Create shutdown channel
+    let (shutdown_tx, mut shutdown_rx) = tokio::sync::broadcast::channel(1);
+    let shutdown_tx_clone = shutdown_tx.clone();
 
     tokio::spawn(async move {
         // 错误处理改进
@@ -317,11 +302,11 @@ async fn run(opt: config::Opt) -> Result<()> {
 
         let http_server = ConnBuilder::new(TokioExecutor::new());
         let mut ctrl_c = std::pin::pin!(tokio::signal::ctrl_c());
-        let graceful = hyper_util::server::graceful::GracefulShutdown::new();
+        let graceful = GracefulShutdown::new();
         debug!("graceful initiated");
 
-        // Send a message to the main thread to indicate that the server has started
+        // 服务准备就绪
-        let _ = tx.send(());
+        worker_state_manager.update(ServiceState::Ready);
 
         loop {
             debug!("waiting for SIGINT or SIGTERM has_tls_certs: {}", has_tls_certs);
@@ -337,17 +322,23 @@ async fn run(opt: config::Opt) -> Result<()> {
                     }
                 }
                 _ = ctrl_c.as_mut() => {
-                    drop(listener);
+                    info!("Ctrl-C received in worker thread");
-                    eprintln!("Ctrl-C received, starting shutdown");
+                    let _ = shutdown_tx_clone.send(());
                     break;
                 }
 
                 _ = sigint_inner.recv() => {
                     info!("SIGINT received in worker thread");
+                    let _ = shutdown_tx_clone.send(());
                     break;
                 }
                 _ = sigterm_inner.recv() => {
                     info!("SIGTERM received in worker thread");
+                    let _ = shutdown_tx_clone.send(());
+                    break;
+                }
+                _ = shutdown_rx.recv() => {
+                    info!("Shutdown signal received in worker thread");
                     break;
                 }
             };
@@ -391,7 +382,7 @@ async fn run(opt: config::Opt) -> Result<()> {
                 debug!("Http handshake success");
             }
         }
-
+        worker_state_manager.update(ServiceState::Stopping);
         tokio::select! {
             () = graceful.shutdown() => {
                  debug!("Gracefully shutdown!");
@@ -400,6 +391,7 @@ async fn run(opt: config::Opt) -> Result<()> {
                  debug!("Waited 10 seconds for graceful shutdown, aborting...");
             }
         }
+        worker_state_manager.update(ServiceState::Stopped);
     });
 
     // init store
@@ -449,97 +441,24 @@ async fn run(opt: config::Opt) -> Result<()> {
         });
     }
 
-    // 执行休眠 1 秒钟
+    // Perform hibernation for 1 second
-    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+    tokio::time::sleep(SHUTDOWN_TIMEOUT).await;
-    // Wait for the HTTP service to finish starting
+    // listen to the shutdown signal
-    if rx.await.is_ok() {
+    match wait_for_shutdown().await {
-        notify_systemd("ready");
+        ShutdownSignal::CtrlC | ShutdownSignal::Sigint | ShutdownSignal::Sigterm => {
-    } else {
+            info!("Shutdown signal received in main thread");
-        info!("Failed to start the server");
+            // update the status to stopping first
-    }
+            state_manager.update(ServiceState::Stopping);
-
+            info!("Server is stopping...");
-    // 主线程中监听信号
+            let _ = shutdown_tx.send(());
-    let mut sigterm = signal(SignalKind::terminate())?;
+            // Wait for the worker thread to complete the cleaning work
-    let mut sigint = signal(SignalKind::interrupt())?;
+            tokio::time::sleep(SHUTDOWN_TIMEOUT).await;
-    tokio::select! {
+            // the last updated status is stopped
-        _ = tokio::signal::ctrl_c() => {
+            state_manager.update(ServiceState::Stopped);
-            eprintln!("Ctrl-C received, starting shutdown");
+            info!("Server stopped current ");
-            notify_systemd("stopping");
-        }
-
-        _ = sigint.recv() => {
-            info!("SIGINT received, starting shutdown");
-            notify_systemd("stopping");
-        }
-        _ = sigterm.recv() => {
-            info!("SIGTERM received, starting shutdown");
-            notify_systemd("stopping");
         }
     }
 
-    info!("server is stopped");
+    info!("server is stopped state: {:?}", state_manager.current_state());
     Ok(())
 }
-
-// #[allow(dead_code)]
-// #[derive(Debug)]
-// enum ShutdownSignal {
-//     CtrlC,
-//     Sigterm,
-//     Sigint,
-// }
-// #[allow(dead_code)]
-// async fn wait_for_shutdown() -> ShutdownSignal {
-//     let mut sigterm = signal(SignalKind::terminate()).unwrap();
-//     let mut sigint = signal(SignalKind::interrupt()).unwrap();
-//
-//     tokio::select! {
-//         _ = tokio::signal::ctrl_c() => {
-//             info!("Received Ctrl-C signal");
-//             ShutdownSignal::CtrlC
-//         }
-//         _ = sigint.recv() => {
-//             info!("Received SIGINT signal");
-//             ShutdownSignal::Sigint
-//         }
-//         _ = sigterm.recv() => {
-//             info!("Received SIGTERM signal");
-//             ShutdownSignal::Sigterm
-//         }
-//     }
-// }
-// #[allow(dead_code)]
-// #[derive(Debug)]
-// enum ServiceState {
-//     Starting,
-//     Ready,
-//     Stopping,
-//     Stopped,
-// }
-// #[allow(dead_code)]
-// fn notify_service_state(state: ServiceState) {
-//     match state {
-//         ServiceState::Starting => {
-//             info!("Service is starting...");
-//             #[cfg(target_os = "linux")]
-//             if let Err(e) = libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Starting...")]) {
-//                 error!("Failed to notify systemd of starting state: {}", e);
-//             }
-//         }
-//         ServiceState::Ready => {
-//             info!("Service is ready");
-//             notify_systemd("ready");
-//         }
-//         ServiceState::Stopping => {
-//             info!("Service is stopping...");
-//             notify_systemd("stopping");
-//         }
-//         ServiceState::Stopped => {
-//             info!("Service has stopped");
-//             #[cfg(target_os = "linux")]
-//             if let Err(e) = libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Stopped")]) {
-//                 error!("Failed to notify systemd of stopped state: {}", e);
-//             }
-//         }
-//     }
-// }

rustfs/src/server/mod.rs

@@ -0,0 +1,6 @@
+mod service_state;
+pub(crate) use service_state::wait_for_shutdown;
+pub(crate) use service_state::ServiceState;
+pub(crate) use service_state::ServiceStateManager;
+pub(crate) use service_state::ShutdownSignal;
+pub(crate) use service_state::SHUTDOWN_TIMEOUT;

rustfs/src/server/service_state.rs

@@ -0,0 +1,152 @@
+use atomic_enum::atomic_enum;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::signal::unix::{signal, SignalKind};
+use tracing::info;
+
+// a configurable shutdown timeout
+pub(crate) const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(1);
+
+#[cfg(target_os = "linux")]
+fn notify_systemd(state: &str) {
+    use libsystemd::daemon::{notify, NotifyState};
+    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);
+}
+
+#[derive(Debug)]
+pub enum ShutdownSignal {
+    CtrlC,
+    Sigterm,
+    Sigint,
+}
+
+#[atomic_enum]
+#[derive(PartialEq)]
+pub(crate) enum ServiceState {
+    Starting,
+    Ready,
+    Stopping,
+    Stopped,
+}
+
+pub(crate) async fn wait_for_shutdown() -> ShutdownSignal {
+    let mut sigterm = signal(SignalKind::terminate()).expect("failed to create SIGTERM signal handler");
+    let mut sigint = signal(SignalKind::interrupt()).expect("failed to create SIGINT signal handler");
+
+    tokio::select! {
+        _ = tokio::signal::ctrl_c() => {
+            info!("Received Ctrl-C signal");
+            ShutdownSignal::CtrlC
+        }
+        _ = sigint.recv() => {
+            info!("Received SIGINT signal");
+            ShutdownSignal::Sigint
+        }
+        _ = sigterm.recv() => {
+            info!("Received SIGTERM signal");
+            ShutdownSignal::Sigterm
+        }
+    }
+}
+
+#[derive(Clone)]
+pub(crate) struct ServiceStateManager {
+    state: Arc<AtomicServiceState>,
+}
+
+impl ServiceStateManager {
+    pub fn new() -> Self {
+        Self {
+            state: Arc::new(AtomicServiceState::new(ServiceState::Starting)),
+        }
+    }
+
+    pub fn update(&self, new_state: ServiceState) {
+        self.state.store(new_state, Ordering::SeqCst);
+        self.notify_systemd(&new_state);
+    }
+
+    pub fn current_state(&self) -> ServiceState {
+        self.state.load(Ordering::SeqCst)
+    }
+
+    fn notify_systemd(&self, state: &ServiceState) {
+        match state {
+            ServiceState::Starting => {
+                info!("Service is starting...");
+                #[cfg(target_os = "linux")]
+                if let Err(e) = libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Starting...")]) {
+                    tracing::error!("Failed to notify systemd of starting state: {}", e);
+                }
+            }
+            ServiceState::Ready => {
+                info!("Service is ready");
+                notify_systemd("ready");
+            }
+            ServiceState::Stopping => {
+                info!("Service is stopping...");
+                notify_systemd("stopping");
+            }
+            ServiceState::Stopped => {
+                info!("Service has stopped");
+                #[cfg(target_os = "linux")]
+                if let Err(e) = libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Stopped")]) {
+                    tracing::error!("Failed to notify systemd of stopped state: {}", e);
+                }
+            }
+        }
+    }
+}
+
+impl Default for ServiceStateManager {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+// 使用示例
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_service_state_manager() {
+        let manager = ServiceStateManager::new();
+
+        // 初始状态应该是 Starting
+        assert_eq!(manager.current_state(), ServiceState::Starting);
+
+        // 更新状态到 Ready
+        manager.update(ServiceState::Ready);
+        assert_eq!(manager.current_state(), ServiceState::Ready);
+
+        // 更新状态到 Stopping
+        manager.update(ServiceState::Stopping);
+        assert_eq!(manager.current_state(), ServiceState::Stopping);
+
+        // 更新状态到 Stopped
+        manager.update(ServiceState::Stopped);
+        assert_eq!(manager.current_state(), ServiceState::Stopped);
+    }
+}

scripts/run.sh

@@ -33,8 +33,27 @@ export RUSTFS_CONSOLE_ENABLE=true
 export RUSTFS_CONSOLE_ADDRESS="0.0.0.0:9002"
 # export RUSTFS_SERVER_DOMAINS="localhost:9000"
 
-# 具体路径修改为配置文件真实路径，obs.example.toml 仅供参考
+# 具体路径修改为配置文件真实路径，obs.example.toml 仅供参考 其中`RUSTFS_OBS_CONFIG` 和下面变量二选一
-export RUSTFS_OBS_CONFIG="./config/obs.example.toml"
+export RUSTFS_OBS_CONFIG="./deploy/config/obs.example.toml"
+
+# 如下变量需要必须参数都有值才可以，以及会覆盖配置文件中的值
+export RUSTFS__OBSERVABILITY__ENDPOINT=http://localhost:43178
+export RUSTFS__OBSERVABILITY__USE_STDOUT=true
+export RUSTFS__OBSERVABILITY__SAMPLE_RATIO=2.0
+export RUSTFS__OBSERVABILITY__METER_INTERVAL=30
+export RUSTFS__OBSERVABILITY__SERVICE_NAME=rustfs
+export RUSTFS__OBSERVABILITY__SERVICE_VERSION=0.1.0
+export RUSTFS__OBSERVABILITY__ENVIRONMENT=develop
+export RUSTFS__OBSERVABILITY__LOGGER_LEVEL=info
+export RUSTFS__SINKS__FILE__ENABLED=true
+export RUSTFS__SINKS__FILE__PATH="./deploy/logs/app.log"
+export RUSTFS__SINKS__WEBHOOK__ENABLED=false
+export RUSTFS__SINKS__WEBHOOK__ENDPOINT=""
+export RUSTFS__SINKS__WEBHOOK__AUTH_TOKEN=""
+export RUSTFS__SINKS__KAFKA__ENABLED=false
+export RUSTFS__SINKS__KAFKA__BOOTSTRAP_SERVERS=""
+export RUSTFS__SINKS__KAFKA__TOPIC=""
+export RUSTFS__LOGGER__QUEUE_CAPACITY=10
 
 if [ -n "$1" ]; then
 	export RUSTFS_VOLUMES="$1"