chore: modify LogExporter and TraceExporter interfaces to support returning failure

opentelemetry-otlp/CHANGELOG.md

@@ -1,9 +1,8 @@
 # Changelog
 
 ## vNext
-
 - Bump msrv to 1.75.0.
-
+- `OtlpHttpClient.shutdown` `TonicLogsClient.shutdown`, and `TonicTracesClient.shutdown` now explicitly return a result. 
 
 ## 0.27.0
 

opentelemetry-otlp/src/exporter/http/logs.rs

@@ -3,7 +3,7 @@
 use async_trait::async_trait;
 use http::{header::CONTENT_TYPE, Method};
 use opentelemetry::otel_debug;
-use opentelemetry_sdk::export::logs::{LogBatch, LogExporter};
+use opentelemetry_sdk::export::logs::{LogBatch, LogExporter, ShutdownResult};
 use opentelemetry_sdk::logs::{LogError, LogResult};
 
 use super::OtlpHttpClient;
@@ -49,8 +49,9 @@
         Ok(())
     }
 
-    fn shutdown(&mut self) {
-        let _ = self.client.lock().map(|mut c| c.take());
+    fn shutdown(&mut self) -> ShutdownResult {
+        let _ = self.client.lock()?.take();
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {

opentelemetry-otlp/src/exporter/http/trace.rs

@@ -2,8 +2,9 @@
 
 use futures_core::future::BoxFuture;
 use http::{header::CONTENT_TYPE, Method};
-use opentelemetry::{otel_debug, trace::TraceError};
-use opentelemetry_sdk::export::trace::{ExportResult, SpanData, SpanExporter};
+use opentelemetry::otel_debug;
+use opentelemetry::trace::TraceError;
+use opentelemetry_sdk::export::trace::{ExportResult, ShutdownResult, SpanData, SpanExporter};
 
 use super::OtlpHttpClient;
 
@@ -64,8 +65,9 @@
         })
     }
 
-    fn shutdown(&mut self) {
-        let _ = self.client.lock().map(|mut c| c.take());
+    fn shutdown(&mut self) -> ShutdownResult {
+        let _ = self.client.lock()?.take();
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {

opentelemetry-otlp/src/exporter/tonic/logs.rs

@@ -4,7 +4,7 @@
 use opentelemetry_proto::tonic::collector::logs::v1::{
     logs_service_client::LogsServiceClient, ExportLogsServiceRequest,
 };
-use opentelemetry_sdk::export::logs::{LogBatch, LogExporter};
+use opentelemetry_sdk::export::logs::{LogBatch, LogExporter, ShutdownResult};
 use opentelemetry_sdk::logs::{LogError, LogResult};
 use tonic::{codegen::CompressionEncoding, service::Interceptor, transport::Channel, Request};
 
@@ -89,8 +89,9 @@
         Ok(())
     }
 
-    fn shutdown(&mut self) {
+    fn shutdown(&mut self) -> ShutdownResult {
         let _ = self.inner.take();
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {

opentelemetry-otlp/src/exporter/tonic/trace.rs

@@ -5,10 +5,9 @@
 use opentelemetry_proto::tonic::collector::trace::v1::{
     trace_service_client::TraceServiceClient, ExportTraceServiceRequest,
 };
-use opentelemetry_sdk::export::trace::{ExportResult, SpanData, SpanExporter};
-use tonic::{codegen::CompressionEncoding, service::Interceptor, transport::Channel, Request};
-
 use opentelemetry_proto::transform::trace::tonic::group_spans_by_resource_and_scope;
+use opentelemetry_sdk::export::trace::{ExportResult, ShutdownResult, SpanData, SpanExporter};
+use tonic::{codegen::CompressionEncoding, service::Interceptor, transport::Channel, Request};
 
 use super::BoxInterceptor;
 
@@ -92,8 +91,9 @@
         })
     }
 
-    fn shutdown(&mut self) {
+    fn shutdown(&mut self) -> ShutdownResult {
         let _ = self.inner.take();
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {

opentelemetry-otlp/tests/integration_test/tests/logs.rs

@@ -94,7 +94,7 @@ mod logtests {
 
         tokio::time::sleep(Duration::from_secs(10)).await;
 
-        assert_logs_results(test_utils::LOGS_FILE, "expected/logs.json");
+        let _ = assert_logs_results(test_utils::LOGS_FILE, "expected/logs.json");
 
         Ok(())
     }
@@ -122,7 +122,7 @@ mod logtests {
         }
         let _ = logger_provider.shutdown();
         // tokio::time::sleep(Duration::from_secs(10)).await;
-        assert_logs_results(test_utils::LOGS_FILE, "expected/logs.json");
+        let _ = assert_logs_results(test_utils::LOGS_FILE, "expected/logs.json");
 
         Ok(())
     }

opentelemetry-sdk/CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## vNext
 
+- If you are an exporter author, the trait functions `LogExporter.shutdown` and `TraceExporter.shutdown` must now return a result. Note that implementing shutdown is optional as the trait provides a default implementation that returns Ok(()).
+
+- The trait functions `LogExporter.shutdown` and `TraceExporter.shutdown` now explicitly return a result. The
+  semantics of the method have not changed, but you will have a new lint encouraging you to consume these results.
+
 - *Breaking(Affects custom metric exporter authors only)* `start_time` and `time` is moved from DataPoints to aggregations (Sum, Gauge, Histogram, ExpoHistogram) see [#2377](https://github.com/open-telemetry/opentelemetry-rust/pull/2377) and [#2411](https://github.com/open-telemetry/opentelemetry-rust/pull/2411), to reduce memory.
 
 - *Breaking* `start_time` is no longer optional for `Sum` aggregation, see [#2367](https://github.com/open-telemetry/opentelemetry-rust/pull/2367), but is still optional for `Gauge` aggregation see [#2389](https://github.com/open-telemetry/opentelemetry-rust/pull/2389).
@@ -14,6 +19,7 @@
     [#2338](https://github.com/open-telemetry/opentelemetry-rust/pull/2338)
   - `ResourceDetector.detect()` no longer supports timeout option.
   - `opentelemetry::global::shutdown_tracer_provider()` Removed from the API, should now use `tracer_provider.shutdown()` see [#2369](https://github.com/open-telemetry/opentelemetry-rust/pull/2369) for a migration example. "Tracer provider" is cheaply cloneable, so users are encouraged to set a clone of it as the global (ex: `global::set_tracer_provider(provider.clone()))`, so that instrumentations and other components can obtain tracers from `global::tracer()`. The tracer_provider must be kept around to call shutdown on it at the end of application (ex: `tracer_provider.shutdown()`)
+
 - *Feature*: Add `ResourceBuilder` for an easy way to create new `Resource`s
 - *Breaking*: Remove `Resource::{new,empty,from_detectors,new_with_defaults,from_schema_url,merge,default}` from public api. To create Resources you should only use `Resource::builder()` or `Resource::builder_empty()`. See [#2322](https://github.com/open-telemetry/opentelemetry-rust/pull/2322) for a migration guide.
   Example Usage:
@@ -156,6 +162,7 @@ metadata, a feature introduced in version 0.1.40. [#2418](https://github.com/ope
     - Continue enabling one of the async runtime feature flags: `rt-tokio`,
       `rt-tokio-current-thread`, or `rt-async-std`.
 
+
 ## 0.27.1
 
 Released 2024-Nov-27

opentelemetry-sdk/src/export/logs/mod.rs

@@ -1,6 +1,6 @@
 //! Log exporters
 use crate::logs::LogRecord;
-use crate::logs::{LogError, LogResult};
+use crate::logs::{LogError, ShutdownError};
 use crate::Resource;
 use async_trait::async_trait;
 #[cfg(feature = "spec_unstable_logs_enabled")]
@@ -81,9 +81,14 @@ pub trait LogExporter: Send + Sync + Debug {
     /// A `LogResult<()>`, which is a result type indicating either a successful export (with
     /// `Ok(())`) or an error (`Err(LogError)`) if the export operation failed.
     ///
-    async fn export(&self, batch: LogBatch<'_>) -> LogResult<()>;
-    /// Shuts down the exporter.
-    fn shutdown(&mut self) {}
+    async fn export(&self, batch: LogBatch<'_>) -> ExportResult;
+
+    /// Shuts down the exporter. This function is idempotent; calling it
+    /// more than once has no additional effect.
+    fn shutdown(&mut self) -> ShutdownResult {
+        Ok(())
+    }
+
     #[cfg(feature = "spec_unstable_logs_enabled")]
     /// Chek if logs are enabled.
     fn event_enabled(&self, _level: Severity, _target: &str, _name: &str) -> bool {
@@ -96,3 +101,6 @@ pub trait LogExporter: Send + Sync + Debug {
 
 /// Describes the result of an export.
 pub type ExportResult = Result<(), LogError>;
+
+/// Describes the result of a shutdown in the log SDK.
+pub type ShutdownResult = Result<(), ShutdownError>;

opentelemetry-sdk/src/export/trace.rs

@@ -5,11 +5,16 @@
 use opentelemetry::{InstrumentationScope, KeyValue};
 use std::borrow::Cow;
 use std::fmt::Debug;
-use std::time::SystemTime;
+use std::sync::PoisonError;
+use std::time::{Duration, SystemTime};
+use thiserror::Error;
 
-/// Describes the result of an export.
+/// Results of an export operation
 pub type ExportResult = Result<(), TraceError>;
 
+/// Result of a  shutdown operation
+pub type ShutdownResult = Result<(), ShutdownError>;
+
 /// `SpanExporter` defines the interface that protocol-specific exporters must
 /// implement so that they can be plugged into OpenTelemetry SDK and support
 /// sending of telemetry data.
@@ -30,7 +35,7 @@
     ///
     /// Any retry logic that is required by the exporter is the responsibility
     /// of the exporter.
-    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, ExportResult>;
+    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, Result<(), TraceError>>;
 
     /// Shuts down the exporter. Called when SDK is shut down. This is an
     /// opportunity for exporter to do any cleanup required.
@@ -43,7 +48,9 @@
     /// 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) {}
+    fn shutdown(&mut self) -> ShutdownResult {
+        Ok(())
+    }
 
     /// 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
@@ -98,3 +105,32 @@
     /// Instrumentation scope that produced this span
     pub instrumentation_scope: InstrumentationScope,
 }
+
+/// Errors returned by shutdown operations in the Export API.
+#[derive(Error, Debug)]
+#[non_exhaustive]
+pub enum ShutdownError {
+    /// The exporter has already been shut down.
+    #[error("Shutdown already performed")]
+    AlreadyShutdown,
+
+    /// Shutdown timed out before completing.
+    #[error("Shutdown timed out after {0:?}")]
+    Timeout(Duration),
+
+    /// The export client failed while holding the client lock. It is not
+    /// possible to complete the shutdown and a retry will not help.
+    /// This is something that should not happen and should likely emit some diagnostic.
+    #[error("export client failed while holding lock; cannot retry.")]
+    ClientFailed(String),
+
+    /// An unexpected error occurred during shutdown.
+    #[error(transparent)]
+    Other(#[from] Box<dyn std::error::Error + Send + Sync + 'static>),
+}
+
+impl<T> From<PoisonError<T>> for ShutdownError {
+    fn from(err: PoisonError<T>) -> Self {
+        ShutdownError::ClientFailed(format!("Mutex poisoned during shutdown: {}", err))
+    }
+}

opentelemetry-sdk/src/logs/error.rs

@@ -18,14 +18,16 @@
     #[error("Exporter timed out after {} seconds", .0.as_secs())]
     ExportTimedOut(Duration),
 
+    /// The export client failed while holding the client lock. It is not
+    /// possible to complete the shutdown and a retry will not help.
+    /// This is something that should not happen and should likely emit some diagnostic.
+    #[error("export client failed while holding lock; cannot retry.")]
+    ClientFailed(String),
+
     /// 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>),
@@ -54,10 +56,32 @@
 
 impl<T> From<PoisonError<T>> for LogError {
     fn from(err: PoisonError<T>) -> Self {
-        LogError::Other(err.to_string().into())
+        LogError::ClientFailed(err.to_string().into())
     }
 }
+
 /// Wrap type for string
 #[derive(Error, Debug)]
 #[error("{0}")]
 struct Custom(String);
+
+/// Errors returned during shutdown
+#[derive(Error, Debug)]
+#[non_exhaustive]
+pub enum ShutdownError {
+    /// The export client failed while holding the client lock. It is not
+    /// possible to complete the shutdown and a retry will not help.
+    /// This is something that should not happen and should likely emit some diagnostic.
+    #[error("export client failed while holding lock; cannot retry.")]
+    ClientFailed(String),
+
+    /// Other errors propagated from log SDK that weren't covered above.
+    #[error(transparent)]
+    Other(#[from] Box<dyn std::error::Error + Send + Sync + 'static>),
+}
+
+impl<T> From<PoisonError<T>> for ShutdownError {
+    fn from(err: PoisonError<T>) -> Self {
+        ShutdownError::ClientFailed(format!("Mutex poisoned during shutdown: {}", err))
+    }
+}

opentelemetry-sdk/src/logs/log_emitter.rs

@@ -141,9 +141,9 @@
                 //    which is non-actionable by the user
                 match err {
                     // specific handling for mutex poisioning
-                    LogError::MutexPoisoned(_) => {
+                    LogError::ClientFailed(_) => {
                         otel_debug!(
-                            name: "LoggerProvider.Drop.ShutdownMutexPoisoned",
+                            name: "LoggerProvider.Drop.ShutdownClientFailed",
                         );
                     }
                     _ => {

opentelemetry-sdk/src/logs/log_processor.rs

@@ -113,14 +113,14 @@
         let result = self
             .exporter
             .lock()
-            .map_err(|_| LogError::MutexPoisoned("SimpleLogProcessor".into()))
+            .map_err(|_| LogError::ClientFailed("SimpleLogProcessor".into()))
             .and_then(|exporter| {
                 let log_tuple = &[(record as &LogRecord, instrumentation)];
                 futures_executor::block_on(exporter.export(LogBatch::new(log_tuple)))
             });
         // Handle errors with specific static names
         match result {
-            Err(LogError::MutexPoisoned(_)) => {
+            Err(LogError::ClientFailed(_)) => {
                 // logging as debug as this is not a user error
                 otel_debug!(
                     name: "SimpleLogProcessor.Emit.MutexPoisoning",
@@ -144,10 +144,13 @@
         self.is_shutdown
             .store(true, std::sync::atomic::Ordering::Relaxed);
         if let Ok(mut exporter) = self.exporter.lock() {
-            exporter.shutdown();
+            exporter
+                .shutdown()
+                .map_err(|e| LogError::Other(Box::new(e)))?;
             Ok(())
         } else {
-            Err(LogError::MutexPoisoned("SimpleLogProcessor".into()))
+            // Failing to get the mutex means the export client failed whilst holding it
+            Err(LogError::ClientFailed("SimpleLogProcessor".into()))
         }
     }
 
@@ -650,7 +653,13 @@
                         )
                         .await;
 
-                        exporter.shutdown();
+                        if let Err(e) = exporter.shutdown() {
+                            otel_warn!(
+                               name: "BatchLogProcessor.Shutdown.Failed",
+                               message = "failed shutting down exporter cleanly",
+                               error = format!("{:?}", e)
+                            );
+                        };
 
                         if let Err(send_error) = ch.send(result) {
                             otel_debug!(
@@ -934,8 +943,6 @@
             Ok(())
         }
 
-        fn shutdown(&mut self) {}
-
         fn set_resource(&mut self, resource: &Resource) {
             self.resource
                 .lock()

opentelemetry-sdk/src/logs/mod.rs

@@ -4,7 +4,7 @@ mod log_emitter;
 mod log_processor;
 pub(crate) mod record;
 
-pub use error::{LogError, LogResult};
+pub use error::{LogError, LogResult, ShutdownError};
 pub use log_emitter::{Builder, Logger, LoggerProvider};
 pub use log_processor::{
     BatchConfig, BatchConfigBuilder, BatchLogProcessor, BatchLogProcessorBuilder, LogProcessor,