fix(obs): shut down global telemetry guard (#4430)

Co-authored-by: heihutu <heihutu@gmail.com>

crates/obs/src/error.rs

@@ -27,9 +27,11 @@ pub enum GlobalError {
     #[error("Failed to set a global recorder: {0}")]
     SetRecorder(#[from] metrics::SetRecorderError<crate::Recorder>),
     #[error("Failed to set global guard: {0}")]
-    SetError(#[from] SetError<Arc<Mutex<OtelGuard>>>),
+    SetError(#[from] SetError<Arc<Mutex<Option<OtelGuard>>>>),
     #[error("Global guard not initialized")]
     NotInitialized,
+    #[error("Global guard lock poisoned: {0}")]
+    GuardPoisoned(&'static str),
     #[error("Global system metrics err: {0}")]
     MetricsError(String),
     #[error("Failed to get current PID: {0}")]

crates/obs/src/global.rs

@@ -22,7 +22,7 @@ const LOG_SUBSYSTEM_GLOBAL: &str = "global";
 const EVENT_OBS_GLOBAL_STATE: &str = "obs_global_state";
 
 /// Global guard for OpenTelemetry tracing
-static GLOBAL_GUARD: OnceCell<Arc<Mutex<OtelGuard>>> = OnceCell::const_new();
+static GLOBAL_GUARD: OnceCell<Arc<Mutex<Option<OtelGuard>>>> = OnceCell::const_new();
 
 /// Flag indicating if observability metric is enabled
 pub(crate) static OBSERVABILITY_METRIC_ENABLED: OnceCell<bool> = OnceCell::const_new();
@@ -158,13 +158,15 @@ pub fn set_global_guard(guard: OtelGuard) -> Result<(), GlobalError> {
         state = "guard_initializing",
         "obs global state changed"
     );
-    GLOBAL_GUARD.set(Arc::new(Mutex::new(guard))).map_err(GlobalError::SetError)
+    GLOBAL_GUARD
+        .set(Arc::new(Mutex::new(Some(guard))))
+        .map_err(GlobalError::SetError)
 }
 
 /// Get the global guard for OtelGuard
 ///
 /// # Returns
-/// * `Ok(Arc<Mutex<OtelGuard>>)` if guard exists
+/// * `Ok(Arc<Mutex<Option<OtelGuard>>>>)` if guard exists
 /// * `Err(GuardError)` if guard not initialized
 ///
 /// # Example
@@ -178,10 +180,23 @@ pub fn set_global_guard(guard: OtelGuard) -> Result<(), GlobalError> {
 /// #    Ok(())
 /// # }
 /// ```
-pub fn get_global_guard() -> Result<Arc<Mutex<OtelGuard>>, GlobalError> {
+pub fn get_global_guard() -> Result<Arc<Mutex<Option<OtelGuard>>>, GlobalError> {
     GLOBAL_GUARD.get().cloned().ok_or(GlobalError::NotInitialized)
 }
 
+/// Explicitly drop the global guard so telemetry providers can flush before exit.
+pub fn shutdown_global_guard() -> Result<(), GlobalError> {
+    let guard_handle = get_global_guard()?;
+    let guard = {
+        let mut slot = guard_handle
+            .lock()
+            .map_err(|_| GlobalError::GuardPoisoned("global observability guard"))?;
+        slot.take()
+    };
+    drop(guard);
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/obs/src/telemetry/local.rs

@@ -213,14 +213,14 @@ pub(super) fn init_local_logging(
         match init_file_logging_internal(config, log_directory, logger_level, is_production) {
             Ok(guard) => Ok(guard),
             Err(error) if should_fallback_to_stdout(&error) => {
-                let guard = init_stdout_only(config, logger_level, is_production);
+                let guard = init_stdout_only(config, logger_level, is_production)?;
                 emit_file_logging_fallback_warning(log_directory, &error);
                 Ok(guard)
             }
             Err(error) => Err(error),
         }
     } else {
-        Ok(init_stdout_only(config, logger_level, is_production))
+        init_stdout_only(config, logger_level, is_production)
     }
 }
 
@@ -236,7 +236,7 @@ pub(super) fn init_local_logging(
 /// * `_config` - Unused at the moment; reserved for future configuration.
 /// * `logger_level` - Effective log level string.
 /// * `is_production` - Controls span event verbosity.
-fn init_stdout_only(_config: &OtelConfig, logger_level: &str, is_production: bool) -> OtelGuard {
+fn init_stdout_only(_config: &OtelConfig, logger_level: &str, is_production: bool) -> Result<OtelGuard, TelemetryError> {
     let env_filter = build_env_filter(logger_level, None);
     let (nb, guard) = tracing_appender::non_blocking(std::io::stdout());
     let span_events = if is_production { FmtSpan::CLOSE } else { FmtSpan::FULL };
@@ -249,7 +249,8 @@ fn init_stdout_only(_config: &OtelConfig, logger_level: &str, is_production: boo
         .with(env_filter)
         .with(ErrorLayer::default())
         .with(fmt_layer)
-        .init();
+        .try_init()
+        .map_err(|err| TelemetryError::SubscriberInit(err.to_string()))?;
 
     set_observability_metric_enabled(false);
     counter!("rustfs_start_total").increment(1);
@@ -266,7 +267,7 @@ fn init_stdout_only(_config: &OtelConfig, logger_level: &str, is_production: boo
         "local logging state"
     );
 
-    OtelGuard {
+    Ok(OtelGuard {
         tracer_provider: None,
         meter_provider: None,
         logger_provider: None,
@@ -274,7 +275,7 @@ fn init_stdout_only(_config: &OtelConfig, logger_level: &str, is_production: boo
         tracing_guard: Some(guard),
         stdout_guard: None,
         cleanup_handle: None,
-    }
+    })
 }
 
 // ─── File-rolling ─────────────────────────────────────────────────────────────
@@ -359,7 +360,8 @@ fn init_file_logging_internal(
         .with(ErrorLayer::default())
         .with(file_layer)
         .with(stdout_layer)
-        .init();
+        .try_init()
+        .map_err(|err| TelemetryError::SubscriberInit(err.to_string()))?;
 
     set_observability_metric_enabled(false);
 

crates/obs/src/telemetry/otel.rs

@@ -295,7 +295,8 @@ pub(super) fn init_observability_http(
         .with(tracer_layer)
         .with(otel_bridge)
         .with(metrics_layer)
-        .init();
+        .try_init()
+        .map_err(|err| TelemetryError::SubscriberInit(err.to_string()))?;
 
     counter!("rustfs_start_total").increment(1);
     info!(

rustfs/src/startup_entrypoint.rs

@@ -37,6 +37,7 @@ pub fn run_process() {
             // Tracing may not be initialized when startup fails this early.
             emit_fatal_stderr("Server runtime failed", e);
         }
+        let _ = crate::startup_runtime_sources::shutdown_observability_guard();
         std::process::exit(1);
     }
 }
@@ -59,6 +60,7 @@ async fn async_main() -> Result<()> {
         Ok(result) => result,
         Err(e) => {
             emit_fatal_stderr("Command parse failed", e);
+            let _ = crate::startup_runtime_sources::shutdown_observability_guard();
             std::process::exit(1);
         }
     };

rustfs/src/startup_lifecycle.rs

@@ -23,7 +23,7 @@ use crate::{
 use rustfs_common::GlobalReadiness;
 use rustfs_scanner::init_data_scanner;
 use std::{
-    io::Result,
+    io::{Error, Result},
     net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
     sync::{
         Arc,
@@ -149,6 +149,8 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec
         result = "stopped",
         "RustFS server stopped"
     );
+    startup_runtime_sources::shutdown_observability_guard()
+        .map_err(|err| Error::other(format!("shutdown_global_guard: {err}")))?;
     Ok(())
 }
 

rustfs/src/startup_runtime_sources.rs

@@ -66,6 +66,10 @@ pub(crate) fn set_observability_guard(guard: OtelGuard) -> Result<(), Observabil
     rustfs_obs::set_global_guard(guard)
 }
 
+pub(crate) fn shutdown_observability_guard() -> Result<(), ObservabilityError> {
+    rustfs_obs::shutdown_global_guard()
+}
+
 pub(crate) fn observability_metric_enabled() -> bool {
     rustfs_obs::observability_metric_enabled()
 }

rustfs/src/startup_shutdown.rs

@@ -18,6 +18,7 @@ use crate::{
     startup_optional_runtime_sidecars::{
         OptionalRuntimeServices, prepare_optional_runtime_shutdowns, shutdown_optional_runtime_services,
     },
+    startup_runtime_sources,
 };
 use rustfs_heal::shutdown_ahm_services;
 use rustfs_utils::get_env_bool_with_aliases;
@@ -299,6 +300,17 @@ pub(crate) async fn run_embedded_server_shutdown(
         state = "stopped",
         "Embedded server state changed"
     );
+
+    if let Err(err) = startup_runtime_sources::shutdown_observability_guard() {
+        warn!(
+            component = LOG_COMPONENT_EMBEDDED,
+            subsystem = LOG_SUBSYSTEM_EMBEDDED,
+            event = EVENT_EMBEDDED_SHUTDOWN_CLEANUP_FAILED,
+            service = "observability",
+            error = %err,
+            "Embedded shutdown cleanup failed"
+        );
+    }
 }
 
 #[cfg(test)]