diff --git a/crates/obs/src/error.rs b/crates/obs/src/error.rs index c3dcff80b..f18728199 100644 --- a/crates/obs/src/error.rs +++ b/crates/obs/src/error.rs @@ -27,9 +27,11 @@ pub enum GlobalError { #[error("Failed to set a global recorder: {0}")] SetRecorder(#[from] metrics::SetRecorderError), #[error("Failed to set global guard: {0}")] - SetError(#[from] SetError>>), + SetError(#[from] SetError>>>), #[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}")] diff --git a/crates/obs/src/global.rs b/crates/obs/src/global.rs index aea785725..5d6ea7708 100644 --- a/crates/obs/src/global.rs +++ b/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>> = OnceCell::const_new(); +static GLOBAL_GUARD: OnceCell>>> = OnceCell::const_new(); /// Flag indicating if observability metric is enabled pub(crate) static OBSERVABILITY_METRIC_ENABLED: OnceCell = 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>)` if guard exists +/// * `Ok(Arc>>>)` 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>, GlobalError> { +pub fn get_global_guard() -> Result>>, 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::*; diff --git a/crates/obs/src/telemetry/local.rs b/crates/obs/src/telemetry/local.rs index 65f01c4a0..28a7a4da3 100644 --- a/crates/obs/src/telemetry/local.rs +++ b/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 { 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); diff --git a/crates/obs/src/telemetry/otel.rs b/crates/obs/src/telemetry/otel.rs index d955e180e..467ba8691 100644 --- a/crates/obs/src/telemetry/otel.rs +++ b/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!( diff --git a/rustfs/src/startup_entrypoint.rs b/rustfs/src/startup_entrypoint.rs index 203f6ac89..4d226ff97 100644 --- a/rustfs/src/startup_entrypoint.rs +++ b/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); } }; diff --git a/rustfs/src/startup_lifecycle.rs b/rustfs/src/startup_lifecycle.rs index ecc45340d..21ac47fa2 100644 --- a/rustfs/src/startup_lifecycle.rs +++ b/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(()) } diff --git a/rustfs/src/startup_runtime_sources.rs b/rustfs/src/startup_runtime_sources.rs index e27f260d2..121ea785a 100644 --- a/rustfs/src/startup_runtime_sources.rs +++ b/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() } diff --git a/rustfs/src/startup_shutdown.rs b/rustfs/src/startup_shutdown.rs index 28197cfb4..8b89f0f3b 100644 --- a/rustfs/src/startup_shutdown.rs +++ b/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)]