Global Log handler cleanup - Logs SDK

opentelemetry-sdk/src/logs/log_emitter.rs

@@ -1,9 +1,8 @@
 use super::{BatchLogProcessor, LogProcessor, LogRecord, SimpleLogProcessor, TraceContext};
 use crate::{export::logs::LogExporter, runtime::RuntimeChannel, Resource};
-use opentelemetry::otel_warn;
 use opentelemetry::{
-    global,
     logs::{LogError, LogResult},
+    otel_debug,
     trace::TraceContextExt,
     Context, InstrumentationLibrary,
 };
@@ -24,15 +23,14 @@
     inner: Arc::new(LoggerProviderInner {
         processors: Vec::new(),
         resource: Resource::empty(),
+        is_shutdown: AtomicBool::new(true),
     }),
-    is_shutdown: Arc::new(AtomicBool::new(true)),
 });
 
 #[derive(Debug, Clone)]
 /// Creator for `Logger` instances.
 pub struct LoggerProvider {
     inner: Arc<LoggerProviderInner>,
-    is_shutdown: Arc<AtomicBool>,
 }
 
 /// Default logger name if empty string is provided.
@@ -73,7 +71,7 @@
 
     fn library_logger(&self, library: Arc<InstrumentationLibrary>) -> Self::Logger {
         // If the provider is shutdown, new logger will refer a no-op logger provider.
-        if self.is_shutdown.load(Ordering::Relaxed) {
+        if self.inner.is_shutdown.load(Ordering::Relaxed) {
             return Logger::new(library, NOOP_LOGGER_PROVIDER.clone());
         }
         Logger::new(library, self.clone())
@@ -105,6 +103,7 @@
     /// Shuts down this `LoggerProvider`
     pub fn shutdown(&self) -> LogResult<()> {
         if self
+            .inner
             .is_shutdown
             .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
             .is_ok()
@@ -114,24 +113,36 @@
             let mut errs = vec![];
             for processor in &self.inner.processors {
                 if let Err(err) = processor.shutdown() {
-                    otel_warn!(
-                        name: "logger_provider_shutdown_error",
-                        error = format!("{:?}", err)
-                    );
+                    match err {
+                        // Specific handling for mutex poisoning
+                        LogError::MutexPoisoned(_) => {
+                            otel_debug!(
+                                name: "LoggerProvider.Shutdown.MutexPoisoned",
+                            );
+                        }
+                        _ => {
+                            otel_debug!(
+                                name: "LoggerProvider.Shutdown.Error",
+                                error = format!("{err}")
+                            );
+                        }
+                    }
                     errs.push(err);
                 }
             }
 
             if errs.is_empty() {
                 Ok(())
             } else {
+                // consolidate errors from all the processors - not all may be user errors
                 Err(LogError::Other(format!("{errs:?}").into()))
             }
         } else {
-            otel_warn!(
-                name: "logger_provider_already_shutdown"
+            let error = LogError::AlreadyShutdown("LoggerProvider".to_string());
+            otel_debug!(
+                name: "LoggerProvider.Shutdown.AlreadyShutdown",
             );
-            Err(LogError::Other("logger provider already shut down".into()))
+            Err(error)
         }
     }
 }
@@ -140,14 +151,38 @@
 struct LoggerProviderInner {
     processors: Vec<Box<dyn LogProcessor>>,
     resource: Resource,
+    is_shutdown: AtomicBool,
 }
 
 impl Drop for LoggerProviderInner {
     fn drop(&mut self) {
-        for processor in &mut self.processors {
-            if let Err(err) = processor.shutdown() {
-                global::handle_error(err);
+        if self
+            .is_shutdown
+            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
+            .is_ok()
+        {
+            for processor in &mut self.processors {
+                if let Err(err) = processor.shutdown() {
+                    match err {
+                        // Specific handling for mutex poisoning
+                        LogError::MutexPoisoned(_) => {
+                            otel_debug!(
+                                name: "LoggerProvider.Drop.ShutdownMutexPoisoned",
+                            );
+                        }
+                        _ => {
+                            otel_debug!(
+                                name: "LoggerProvider.Drop.ShutdownError",
+                                error = format!("{err}")
+                            );
+                        }
+                    }
+                }
             }
+        } else {
+            otel_debug!(
+                name: "LoggerProvider.Drop.AlreadyShutdown",
+            );
         }
     }
 }
@@ -202,8 +237,8 @@
             inner: Arc::new(LoggerProviderInner {
                 processors: self.processors,
                 resource,
+                is_shutdown: AtomicBool::new(false),
             }),
-            is_shutdown: Arc::new(AtomicBool::new(false)),
         };
 
         // invoke set_resource on all the processors

opentelemetry-sdk/src/logs/log_processor.rs

@@ -12,9 +12,8 @@
 #[cfg(feature = "logs_level_enabled")]
 use opentelemetry::logs::Severity;
 use opentelemetry::{
-    global,
     logs::{LogError, LogResult},
-    otel_error, otel_warn, InstrumentationLibrary,
+    otel_debug, otel_error, otel_warn, InstrumentationLibrary,
 };
 
 use std::sync::atomic::AtomicBool;
@@ -99,26 +98,36 @@
     fn emit(&self, record: &mut LogRecord, instrumentation: &InstrumentationLibrary) {
         // noop after shutdown
         if self.is_shutdown.load(std::sync::atomic::Ordering::Relaxed) {
+            // this is a warning, as the user is trying to log after the processor has been shutdown
             otel_warn!(
-                name: "simple_log_processor_emit_after_shutdown"
+                name: "SimpleLogProcessor.Emit.ProcessorShutdown",
             );
             return;
         }
 
         let result = self
             .exporter
             .lock()
-            .map_err(|_| LogError::Other("simple logprocessor mutex poison".into()))
+            .map_err(|_| LogError::MutexPoisoned("SimpleLogProcessor".into()))
             .and_then(|mut exporter| {
                 let log_tuple = &[(record as &LogRecord, instrumentation)];
                 futures_executor::block_on(exporter.export(LogBatch::new(log_tuple)))
             });
-        if let Err(err) = result {
-            otel_error!(
-                name: "simple_log_processor_emit_error",
-                error = format!("{:?}", err)
-            );
-            global::handle_error(err);
+        // Handle errors with specific static names
+        match result {
+            Err(LogError::MutexPoisoned(_)) => {
+                // logging as debug as this is not a user error
+                otel_debug!(
+                    name: "SimpleLogProcessor.Emit.MutexPoisoning",
+                );
+            }
+            Err(err) => {
+                otel_error!(
+                    name: "SimpleLogProcessor.Emit.ExportError",
+                    error = format!("{}",err)
+                );
+            }
+            _ => {}
         }
     }
 
@@ -133,12 +142,7 @@
             exporter.shutdown();
             Ok(())
         } else {
-            otel_error!(
-                name: "simple_log_processor_shutdown_error"
-            );
-            Err(LogError::Other(
-                "simple logprocessor mutex poison during shutdown".into(),
-            ))
+            Err(LogError::MutexPoisoned("SimpleLogProcessor".into()))
         }
     }
 
@@ -170,12 +174,12 @@
             instrumentation.clone(),
         )));
 
+        // TODO - Implement throttling to prevent error flooding when the queue is full or closed.
         if let Err(err) = result {
             otel_error!(
-                name: "batch_log_processor_emit_error",
-                error = format!("{:?}", err)
+                name: "BatchLogProcessor.Export.Error",
+                error = format!("{}", err)
             );
-            global::handle_error(LogError::Other(err.into()));
         }
     }
 
@@ -243,10 +247,9 @@
 
                             if let Err(err) = result {
                                 otel_error!(
-                                    name: "batch_log_processor_export_error",
-                                    error = format!("{:?}", err)
+                                    name: "BatchLogProcessor.Export.Error",
+                                    error = format!("{}", err)
                                 );
-                                global::handle_error(err);
                             }
                         }
                     }
@@ -261,24 +264,12 @@
                         .await;
 
                         if let Some(channel) = res_channel {
-                            if let Err(result) = channel.send(result) {
-                                global::handle_error(LogError::from(format!(
-                                    "failed to send flush result: {:?}",
-                                    result
-                                )));
-                                otel_error!(
-                                    name: "batch_log_processor_flush_error",
-                                    error = format!("{:?}", result),
-                                    message = "Failed to send flush result"
+                            if let Err(send_error) = channel.send(result) {
+                                otel_debug!(
+                                    name: "BatchLogProcessor.Flush.SendResultError",
+                                    error = format!("{:?}", send_error),
                                 );
                             }
-                        } else if let Err(err) = result {
-                            otel_error!(
-                                name: "batch_log_processor_flush_error",
-                                error = format!("{:?}", err),
-                                message = "Flush failed"
-                            );
-                            global::handle_error(err);
                         }
                     }
                     // Stream has terminated or processor is shutdown, return to finish execution.
@@ -293,21 +284,14 @@
 
                         exporter.shutdown();
 
-                        if let Err(result) = ch.send(result) {
-                            otel_error!(
-                                name: "batch_log_processor_shutdown_error",
-                                error = format!("{:?}", result),
-                                message = "Failed to send shutdown result"
+                        if let Err(send_error) = ch.send(result) {
+                            otel_debug!(
+                                name: "BatchLogProcessor.Shutdown.SendResultError",
+                                error = format!("{:?}", send_error),
                             );
-                            global::handle_error(LogError::from(format!(
-                                "failed to send batch processor shutdown result: {:?}",
-                                result
-                            )));
                         }
-
                         break;
                     }
-
                     // propagate the resource
                     BatchMessage::SetResource(resource) => {
                         exporter.set_resource(&resource);
@@ -357,13 +341,7 @@
     pin_mut!(timeout);
     match future::select(export, timeout).await {
         Either::Left((export_res, _)) => export_res,
-        Either::Right((_, _)) => {
-            otel_error!(
-                name: "export_with_timeout_timeout",
-                timeout_duration = time_out.as_millis()
-            );
-            ExportResult::Err(LogError::ExportTimedOut(time_out))
-        }
+        Either::Right((_, _)) => ExportResult::Err(LogError::ExportTimedOut(time_out)),
     }
 }
 

opentelemetry/src/global/internal_logging.rs

@@ -50,11 +50,30 @@ macro_rules! otel_warn {
         {
             tracing::warn!(name: $name, target: env!("CARGO_PKG_NAME"), "");
         }
+        #[cfg(not(feature = "internal-logs"))]
+        {
+            #[allow(unused_variables)]
+            {
+
+            }
+        }
     };
     (name: $name:expr, $($key:ident = $value:expr),+ $(,)?) => {
         #[cfg(feature = "internal-logs")]
         {
-            tracing::warn!(name: $name, target: env!("CARGO_PKG_NAME"), $($key = $value),+, "");
+            tracing::warn!(name: $name,
+                            target: env!("CARGO_PKG_NAME"),
+                            $($key = {
+                                    $value
+                            }),+,
+                            ""
+                    )
+        }
+        #[cfg(not(feature = "internal-logs"))]
+        {
+            {
+                let _ = ($name, $($value),+);
+            }
         }
     };
 }
@@ -104,11 +123,31 @@ macro_rules! otel_error {
         {
             tracing::error!(name: $name, target: env!("CARGO_PKG_NAME"), "");
         }
+        #[cfg(not(feature = "internal-logs"))]
+        {
+            #[allow(unused_variables)]
+            {
+
+            }
+        }
     };
     (name: $name:expr, $($key:ident = $value:expr),+ $(,)?) => {
         #[cfg(feature = "internal-logs")]
         {
-            tracing::error!(name: $name, target: env!("CARGO_PKG_NAME"), $($key = $value),+, "");
+            tracing::error!(name: $name,
+                            target: env!("CARGO_PKG_NAME"),
+                            $($key = {
+                                    $value
+                            }),+,
+                            ""
+                    )
+        }
+        #[cfg(not(feature = "internal-logs"))]
+        {
+            {
+                let _ = ($name, $($value),+);
+
+            }
         }
     };
 }

opentelemetry/src/logs/mod.rs

@@ -30,6 +30,14 @@ pub enum LogError {
     #[error("Exporter timed out after {} seconds", .0.as_secs())]
     ExportTimedOut(Duration),
 
+    /// Processor is already shutdown
+    #[error("{0} already shutdown")]
+    AlreadyShutdown(String),
+
+    /// Mutex lock poisoning
+    #[error("mutex lock poisioning for {0}")]
+    MutexPoisoned(String),
+
     /// Other errors propagated from log SDK that weren't covered above.
     #[error(transparent)]
     Other(#[from] Box<dyn std::error::Error + Send + Sync + 'static>),