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-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, LogResult, ShutdownError};
 use crate::Resource;
 use async_trait::async_trait;
 #[cfg(feature = "spec_unstable_logs_enabled")]
@@ -83,7 +83,10 @@ pub trait LogExporter: Send + Sync + Debug {
     ///
     async fn export(&self, batch: LogBatch<'_>) -> LogResult<()>;
     /// Shuts down the exporter.
-    fn shutdown(&mut self) {}
+    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 +99,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,31 @@
     /// 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),
+
+    /// An unexpected error occurred during shutdown.
+    #[error(transparent)]
+    Other(#[from] Box<dyn std::error::Error + Send + Sync + 'static>),
+}
+
+/// Custom error wrapper for string messages.
+#[derive(Error, Debug)]
+#[error("{0}")]
+struct CustomError(String);
+
+impl<T> From<PoisonError<T>> for ShutdownError {
+    fn from(err: PoisonError<T>) -> Self {
+        ShutdownError::Other(Box::new(CustomError(err.to_string())))
+    }
+}

opentelemetry-sdk/src/logs/error.rs

@@ -61,3 +61,22 @@
 #[derive(Error, Debug)]
 #[error("{0}")]
 struct Custom(String);
+
+/// Errors returned during shutdown
+#[derive(Error, Debug)]
+#[non_exhaustive]
+pub enum ShutdownError {
+    /// 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>),
+}
+
+impl<T> From<PoisonError<T>> for ShutdownError {
+    fn from(err: PoisonError<T>) -> Self {
+        ShutdownError::Other(err.to_string().into())
+    }
+}

opentelemetry-sdk/src/logs/log_processor.rs

@@ -144,7 +144,9 @@
         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()))
@@ -650,7 +652,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 +942,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,

opentelemetry-sdk/src/testing/logs/in_memory_exporter.rs

@@ -1,6 +1,6 @@
 use crate::export::logs::{LogBatch, LogExporter};
-use crate::logs::LogRecord;
 use crate::logs::{LogError, LogResult};
+use crate::logs::{LogRecord, ShutdownError};
 use crate::Resource;
 use async_trait::async_trait;
 use opentelemetry::InstrumentationScope;
@@ -195,10 +195,11 @@ impl LogExporter for InMemoryLogExporter {
         Ok(())
     }
 
-    fn shutdown(&mut self) {
+    fn shutdown(&mut self) -> Result<(), ShutdownError> {
         if self.should_reset_on_shutdown {
             self.reset();
         }
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &Resource) {

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

@@ -1,4 +1,4 @@
-use crate::export::trace::{ExportResult, SpanData, SpanExporter};
+use crate::export::trace::{ExportResult, ShutdownResult, SpanData, SpanExporter};
 use crate::resource::Resource;
 use futures_util::future::BoxFuture;
 use opentelemetry::trace::{TraceError, TraceResult};
@@ -142,8 +142,9 @@ impl SpanExporter for InMemorySpanExporter {
         Box::pin(std::future::ready(Ok(())))
     }
 
-    fn shutdown(&mut self) {
-        self.reset()
+    fn shutdown(&mut self) -> ShutdownResult {
+        self.reset();
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &Resource) {

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

@@ -1,3 +1,4 @@
+use crate::export::trace::{ShutdownError, ShutdownResult};
 use crate::{
     export::trace::{ExportResult, SpanData, SpanExporter},
     trace::{SpanEvents, SpanLinks},
@@ -53,8 +54,11 @@ impl SpanExporter for TokioSpanExporter {
         Box::pin(std::future::ready(Ok(())))
     }
 
-    fn shutdown(&mut self) {
-        self.tx_shutdown.send(()).unwrap();
+    fn shutdown(&mut self) -> ShutdownResult {
+        self.tx_shutdown
+            .send(())
+            .map_err::<ShutdownError, _>(|err| ShutdownError::Other(Box::new(err)))?;
+        Ok(())
     }
 }
 

opentelemetry-sdk/src/trace/span_processor.rs

@@ -149,14 +149,10 @@
     }
 
     fn shutdown(&self) -> TraceResult<()> {
-        if let Ok(mut exporter) = self.exporter.lock() {
-            exporter.shutdown();
-            Ok(())
-        } else {
-            Err(TraceError::Other(
-                "SimpleSpanProcessor mutex poison at shutdown".into(),
-            ))
-        }
+        self.exporter
+            .lock()?
+            .shutdown()
+            .map_err(|e| TraceError::Other(Box::new(e)))
     }
 
     fn set_resource(&mut self, resource: &Resource) {
@@ -432,7 +428,12 @@
             // Stream has terminated or processor is shutdown, return to finish execution.
             BatchMessage::Shutdown(ch) => {
                 self.flush(Some(ch)).await;
-                self.exporter.shutdown();
+                if let Err(e) = self.exporter.shutdown() {
+                    otel_warn!(
+                       name: "SpanProcessor.Shutdown.Failed",
+                       message = "failed shutting down exporter cleanly",
+                       error = format!("{:?}", e));
+                }
                 return false;
             }
             // propagate the resource

opentelemetry-stdout/CHANGELOG.md

@@ -5,6 +5,9 @@
 - Bump msrv to 1.75.0.
 - *Breaking* time fields, `StartTime` and `EndTime` is printed on aggregation (Sum, Gauge, Histogram, ExpoHistogram) with 2 tabs, previously it was on aggregation data point, with 3 tabs, see [#2377](https://github.com/open-telemetry/opentelemetry-rust/pull/2377) and [#2411](https://github.com/open-telemetry/opentelemetry-rust/pull/2411).
 
+- `LogExporter.shutdown` and `SpanExporter.shutdown`  now explicitly return a result
+
+
 ## 0.27.0
 
 Released 2024-Nov-11