.

Author: scottgerring

opentelemetry-sdk/src/logs/log_processor.rs

@@ -32,7 +32,9 @@
 //! ```
 
 use crate::{
-    error::ShutdownResult, logs::{ExportResult, LogBatch, LogError, LogExporter, LogRecord, LogResult}, Resource
+    error::{ShutdownError, ShutdownResult},
+    logs::{ExportResult, LogBatch, LogError, LogExporter, LogRecord, LogResult},
+    Resource,
 };
 use std::sync::mpsc::{self, RecvTimeoutError, SyncSender};
 
@@ -90,7 +92,7 @@ pub trait LogProcessor: Send + Sync + Debug {
     /// After shutdown returns the log processor should stop processing any logs.
     /// It's up to the implementation on when to drop the LogProcessor.
     fn shutdown(&self) -> ShutdownResult;
-    
+
     #[cfg(feature = "spec_unstable_logs_enabled")]
     /// Check if logging is enabled
     fn event_enabled(&self, _level: Severity, _target: &str, _name: &str) -> bool {
@@ -189,14 +191,11 @@ impl<T: LogExporter> LogProcessor for SimpleLogProcessor<T> {
     fn shutdown(&self) -> ShutdownResult {
         self.is_shutdown
             .store(true, std::sync::atomic::Ordering::Relaxed);
-        if let Ok(mut exporter) = self.exporter.lock() {
-            exporter
-                .shutdown()
-                .map_err(|e| LogError:(Box::new(e)))?;
-            Ok(())
-        } else {
-            Err(LogError::MutexPoisoned("SimpleLogProcessor".into()))
-        }
+        self.exporter
+            .lock()
+            .map_err(|e| ShutdownError::InternalFailure(format!("Failed to acquire lock: {}", e)))?
+            .shutdown()?;
+        Ok(())
     }
 
     fn set_resource(&self, resource: &Resource) {
@@ -398,7 +397,7 @@ impl LogProcessor for BatchLogProcessor {
         }
     }
 
-    fn shutdown(&self) -> LogResult<()> {
+    fn shutdown(&self) -> ShutdownResult {
         let dropped_logs = self.dropped_logs_count.load(Ordering::Relaxed);
         let max_queue_size = self.max_queue_size;
         if dropped_logs > 0 {
@@ -421,22 +420,22 @@ impl LogProcessor for BatchLogProcessor {
                         if let Some(handle) = self.handle.lock().unwrap().take() {
                             handle.join().unwrap();
                         }
-                        LogResult::Ok(())
+                        Ok(())
                     })
                     .map_err(|err| match err {
                         RecvTimeoutError::Timeout => {
                             otel_error!(
                                 name: "BatchLogProcessor.Shutdown.Timeout",
                                 message = "BatchLogProcessor shutdown timing out."
                             );
-                            LogError::ExportTimedOut(self.shutdown_timeout)
+                            ShutdownError::Timeout(self.shutdown_timeout)
                         }
                         _ => {
                             otel_error!(
                                 name: "BatchLogProcessor.Shutdown.Error",
                                 error = format!("{}", err)
                             );
-                            LogError::Other(err.into())
+                            ShutdownError::InternalFailure(err.to_string())
                         }
                     })?
             }
@@ -446,7 +445,7 @@ impl LogProcessor for BatchLogProcessor {
                     name: "BatchLogProcessor.Shutdown.ControlChannelFull",
                     message = "Control message to shutdown the worker thread could not be sent as the control channel is full. This can occur if user repeatedily calls force_flush/shutdown without finishing the previous call."
                 );
-                LogResult::Err(LogError::Other("Shutdown cannot be performed as Control channel is full. This can occur if user repeatedily calls force_flush/shutdown without finishing the previous call.".into()))
+                Err(ShutdownError::InternalFailure("Shutdown cannot be performed as Control channel is full. This can occur if user repeatedily calls force_flush/shutdown without finishing the previous call.".into()))
             }
             Err(mpsc::TrySendError::Disconnected(_)) => {
                 // Given background thread is the only receiver, and it's
@@ -456,7 +455,7 @@ impl LogProcessor for BatchLogProcessor {
                     message = "Shutdown is being invoked more than once. This is noop, but indicates a potential issue in the application's lifecycle management."
                 );
 
-                LogResult::Err(LogError::Other(
+                Err(ShutdownError::InternalFailure(
                     "BatchLogProcessor is already shutdown".into(),
                 ))
             }
@@ -1217,7 +1216,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> LogResult<()> {
+        fn shutdown(&self) -> ShutdownResult {
             Ok(())
         }
     }
@@ -1247,7 +1246,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> LogResult<()> {
+        fn shutdown(&self) -> ShutdownResult {
             Ok(())
         }
     }

opentelemetry-sdk/src/logs/log_processor_with_async_runtime.rs

@@ -1,4 +1,5 @@
 use crate::{
+    error::{ShutdownError, ShutdownResult},
     logs::{ExportResult, LogBatch, LogError, LogExporter, LogRecord, LogResult},
     Resource,
 };
@@ -86,7 +87,7 @@ impl<R: RuntimeChannel> LogProcessor for BatchLogProcessor<R> {
             .and_then(std::convert::identity)
     }
 
-    fn shutdown(&self) -> LogResult<()> {
+    fn shutdown(&self) -> ShutdownResult {
         let dropped_logs = self.dropped_logs_count.load(Ordering::Relaxed);
         let max_queue_size = self.max_queue_size;
         if dropped_logs > 0 {
@@ -100,10 +101,10 @@ impl<R: RuntimeChannel> LogProcessor for BatchLogProcessor<R> {
         let (res_sender, res_receiver) = oneshot::channel();
         self.message_sender
             .try_send(BatchMessage::Shutdown(res_sender))
-            .map_err(|err| LogError::Other(err.into()))?;
+            .map_err(|err| ShutdownError::InternalFailure(err.to_string()))?;
 
         futures_executor::block_on(res_receiver)
-            .map_err(|err| LogError::Other(err.into()))
+            .map_err(|err| ShutdownError::InternalFailure(err.to_string()))
             .and_then(std::convert::identity)
     }
 
@@ -655,7 +656,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> LogResult<()> {
+        fn shutdown(&self) -> ShutdownResult {
             Ok(())
         }
     }
@@ -685,7 +686,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> LogResult<()> {
+        fn shutdown(&self) -> ShutdownResult {
             Ok(())
         }
     }

opentelemetry-sdk/src/logs/logger_provider.rs

@@ -1,4 +1,5 @@
 use super::{BatchLogProcessor, LogProcessor, LogRecord, SimpleLogProcessor, TraceContext};
+use crate::error::ShutdownError;
 use crate::logs::{LogError, LogExporter, LogResult};
 use crate::Resource;
 use opentelemetry::{otel_debug, otel_info, trace::TraceContextExt, Context, InstrumentationScope};
@@ -131,7 +132,7 @@ struct LoggerProviderInner {
 
 impl LoggerProviderInner {
     /// Shuts down the `LoggerProviderInner` and returns any errors.
-    pub(crate) fn shutdown(&self) -> Vec<LogError> {
+    pub(crate) fn shutdown(&self) -> Vec<ShutdownError> {
         let mut errs = vec![];
         for processor in &self.processors {
             if let Err(err) = processor.shutdown() {
@@ -141,9 +142,15 @@ impl LoggerProviderInner {
                 //    which is non-actionable by the user
                 match err {
                     // specific handling for mutex poisoning
-                    LogError::MutexPoisoned(_) => {
+                    ShutdownError::AlreadyShutdown => {
                         otel_debug!(
-                            name: "LoggerProvider.Drop.ShutdownMutexPoisoned",
+                            name: "LoggerProvider.Drop.AlreadyShutdown",
+                        );
+                    }
+                    ShutdownError::Timeout(t) => {
+                        otel_debug!(
+                            name: "LoggerProvider.Drop.ExportTimedOut",
+                            error = format!("timed out shutting down; t={:?}", t)
                         );
                     }
                     _ => {
@@ -333,6 +340,7 @@ impl opentelemetry::logs::Logger for Logger {
 #[cfg(test)]
 mod tests {
     use crate::{
+        error::ShutdownResult,
         logs::InMemoryLogExporter,
         resource::{
             SERVICE_NAME, TELEMETRY_SDK_LANGUAGE, TELEMETRY_SDK_NAME, TELEMETRY_SDK_VERSION,
@@ -387,7 +395,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> LogResult<()> {
+        fn shutdown(&self) -> ShutdownResult {
             self.is_shutdown
                 .lock()
                 .map(|mut is_shutdown| *is_shutdown = true)
@@ -800,7 +808,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> LogResult<()> {
+        fn shutdown(&self) -> ShutdownResult {
             *self.shutdown_called.lock().unwrap() = true;
             Ok(())
         }
@@ -831,7 +839,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> LogResult<()> {
+        fn shutdown(&self) -> ShutdownResult {
             let mut count = self.shutdown_count.lock().unwrap();
             *count += 1;
             Ok(())

opentelemetry-sdk/src/testing/trace/span_exporters.rs

@@ -1,7 +1,4 @@
-use crate::{
-    trace::{ExportResult, SpanData, SpanExporter},
-    trace::{SpanEvents, SpanLinks},
-};
+use crate::trace::{ExportResult, SpanData, SpanEvents, SpanExporter, SpanLinks};
 use futures_util::future::BoxFuture;
 pub use opentelemetry::testing::trace::TestSpan;
 use opentelemetry::{

opentelemetry-sdk/src/trace/export.rs

@@ -1,5 +1,4 @@
 //! Trace exporters
-use crate::error::ShutdownResult;
 use crate::Resource;
 use futures_util::future::BoxFuture;
 use opentelemetry::trace::{SpanContext, SpanId, SpanKind, Status, TraceError};
@@ -44,9 +43,7 @@ pub trait SpanExporter: Send + Sync + Debug {
     /// flush the data and the destination is unavailable). SDK authors
     /// can decide if they want to make the shutdown timeout
     /// configurable.
-    fn shutdown(&mut self) -> ShutdownResult {
-        Ok(())
-    }
+    fn shutdown(&mut self) {}
 
     /// This is a hint to ensure that the export of any Spans the exporter
     /// has received prior to the call to this function SHOULD be completed

stress/src/logs.rs

@@ -9,8 +9,11 @@
     ~40 M /sec
 */
 
+use std::ascii::AsciiExt;
+
 use opentelemetry::InstrumentationScope;
 use opentelemetry_appender_tracing::layer;
+use opentelemetry_sdk::error::ShutdownResult;
 use opentelemetry_sdk::logs::{LogBatch, LogExporter};
 use opentelemetry_sdk::logs::{LogProcessor, LogRecord, LogResult, LoggerProvider};
 
@@ -46,7 +49,7 @@ impl LogProcessor for MockLogProcessor {
         Ok(())
     }
 
-    fn shutdown(&self) -> LogResult<()> {
+    fn shutdown(&self) -> ShutdownResult {
         Ok(())
     }
 }