chore: modify LogExporter and SpanExporter interfaces to support retu…

…rning failure
open-telemetry · Dec 5, 2024 · 2df2353 · 2df2353
1 parent 957659f
commit 2df2353
Show file tree

Hide file tree

Showing 16 changed files with 67 additions and 32 deletions.
diff --git a/opentelemetry-otlp/CHANGELOG.md b/opentelemetry-otlp/CHANGELOG.md
@@ -2,6 +2,9 @@
 
 ## vNext
 
+- `OtlpHttpClient.shutdown` `TonicLogsClient.shutdown`, and `TonicTracesClient.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.
+
 ## 0.27.0
 
 Released 2024-Nov-11

diff --git a/opentelemetry-otlp/src/exporter/http/logs.rs b/opentelemetry-otlp/src/exporter/http/logs.rs
@@ -47,8 +47,9 @@ impl LogExporter for OtlpHttpClient {
         Ok(())
     }
 
-    fn shutdown(&mut self) {
-        let _ = self.client.lock().map(|mut c| c.take());
+    fn shutdown(&mut self) -> LogResult<()> {
+        let _ = self.client.lock()?.take();
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {

diff --git a/opentelemetry-otlp/src/exporter/http/trace.rs b/opentelemetry-otlp/src/exporter/http/trace.rs
@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use futures_core::future::BoxFuture;
 use http::{header::CONTENT_TYPE, Method};
-use opentelemetry::trace::TraceError;
+use opentelemetry::trace::{TraceError, TraceResult};
 use opentelemetry_sdk::export::trace::{ExportResult, SpanData, SpanExporter};
 
 use super::OtlpHttpClient;
@@ -63,8 +63,9 @@ impl SpanExporter for OtlpHttpClient {
         })
     }
 
-    fn shutdown(&mut self) {
-        let _ = self.client.lock().map(|mut c| c.take());
+    fn shutdown(&mut self) -> TraceResult<()> {
+        let _ = self.client.lock()?.take();
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {

diff --git a/opentelemetry-otlp/src/exporter/tonic/logs.rs b/opentelemetry-otlp/src/exporter/tonic/logs.rs
@@ -81,8 +81,9 @@ impl LogExporter for TonicLogsClient {
         Ok(())
     }
 
-    fn shutdown(&mut self) {
+    fn shutdown(&mut self) -> LogResult<()> {
         let _ = self.inner.take();
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {

diff --git a/opentelemetry-otlp/src/exporter/tonic/trace.rs b/opentelemetry-otlp/src/exporter/tonic/trace.rs
@@ -1,7 +1,7 @@
 use core::fmt;
 
 use futures_core::future::BoxFuture;
-use opentelemetry::trace::TraceError;
+use opentelemetry::trace::{TraceError, TraceResult};
 use opentelemetry_proto::tonic::collector::trace::v1::{
     trace_service_client::TraceServiceClient, ExportTraceServiceRequest,
 };
@@ -88,8 +88,9 @@ impl SpanExporter for TonicTracesClient {
         })
     }
 
-    fn shutdown(&mut self) {
+    fn shutdown(&mut self) -> TraceResult<()> {
         let _ = self.inner.take();
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {

diff --git a/opentelemetry-sdk/CHANGELOG.md b/opentelemetry-sdk/CHANGELOG.md
@@ -10,6 +10,8 @@
     [#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()`)
+  - 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.
 
 ## 0.27.1
 

diff --git a/opentelemetry-sdk/src/export/logs/mod.rs b/opentelemetry-sdk/src/export/logs/mod.rs
@@ -82,8 +82,12 @@ pub trait LogExporter: Send + Sync + Debug {
     /// `Ok(())`) or an error (`Err(LogError)`) if the export operation failed.
     ///
     async fn export(&mut self, batch: LogBatch<'_>) -> LogResult<()>;
+
     /// Shuts down the exporter.
-    fn shutdown(&mut self) {}
+    fn shutdown(&mut self) -> LogResult<()> {
+        Ok(())
+    }
+
     #[cfg(feature = "spec_unstable_logs_enabled")]
     /// Chek if logs are enabled.
     fn event_enabled(&self, _level: Severity, _target: &str, _name: &str) -> bool {

diff --git a/opentelemetry-sdk/src/export/trace.rs b/opentelemetry-sdk/src/export/trace.rs
@@ -10,6 +10,9 @@ use std::time::SystemTime;
 /// Describes the result of an export.
 pub type ExportResult = Result<(), TraceError>;
 
+/// Describes the results of other operations on the trace API.
+pub type TraceResult<T> = Result<T, TraceError>;
+
 /// `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.
@@ -43,7 +46,9 @@ 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) {}
+    fn shutdown(&mut self) -> TraceResult<()> {
+        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

diff --git a/opentelemetry-sdk/src/logs/log_processor.rs b/opentelemetry-sdk/src/logs/log_processor.rs
@@ -136,7 +136,7 @@ impl<T: LogExporter> LogProcessor for SimpleLogProcessor<T> {
         self.is_shutdown
             .store(true, std::sync::atomic::Ordering::Relaxed);
         if let Ok(mut exporter) = self.exporter.lock() {
-            exporter.shutdown();
+            exporter.shutdown()?;
             Ok(())
         } else {
             Err(LogError::MutexPoisoned("SimpleLogProcessor".into()))
@@ -299,7 +299,13 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
                         )
                         .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!(
@@ -590,7 +596,9 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&mut self) {}
+        fn shutdown(&mut self) -> LogResult<()> {
+            Ok(())
+        }
 
         fn set_resource(&mut self, resource: &Resource) {
             self.resource

diff --git a/opentelemetry-sdk/src/testing/logs/in_memory_exporter.rs b/opentelemetry-sdk/src/testing/logs/in_memory_exporter.rs
@@ -195,10 +195,11 @@ impl LogExporter for InMemoryLogExporter {
         Ok(())
     }
 
-    fn shutdown(&mut self) {
+    fn shutdown(&mut self) -> LogResult<()> {
         if self.should_reset_on_shutdown {
             self.reset();
         }
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &Resource) {

diff --git a/opentelemetry-sdk/src/testing/trace/in_memory_exporter.rs b/opentelemetry-sdk/src/testing/trace/in_memory_exporter.rs
@@ -142,8 +142,9 @@ impl SpanExporter for InMemorySpanExporter {
         Box::pin(std::future::ready(Ok(())))
     }
 
-    fn shutdown(&mut self) {
-        self.reset()
+    fn shutdown(&mut self) -> TraceResult<()> {
+        self.reset();
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &Resource) {

diff --git a/opentelemetry-sdk/src/testing/trace/span_exporters.rs b/opentelemetry-sdk/src/testing/trace/span_exporters.rs
@@ -1,11 +1,11 @@
 use crate::{
-    export::trace::{ExportResult, SpanData, SpanExporter},
+    export::trace::{ExportResult, SpanData, SpanExporter, TraceResult},
     trace::{SpanEvents, SpanLinks},
 };
 use futures_util::future::BoxFuture;
 pub use opentelemetry::testing::trace::TestSpan;
 use opentelemetry::{
-    trace::{SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState},
+    trace::{SpanContext, SpanId, SpanKind, Status, TraceError, TraceFlags, TraceId, TraceState},
     InstrumentationScope,
 };
 use std::fmt::{Display, Formatter};
@@ -53,8 +53,11 @@ impl SpanExporter for TokioSpanExporter {
         Box::pin(std::future::ready(Ok(())))
     }
 
-    fn shutdown(&mut self) {
-        self.tx_shutdown.send(()).unwrap();
+    fn shutdown(&mut self) -> TraceResult<()> {
+        self.tx_shutdown
+            .send(())
+            .map_err::<TraceError, _>(|err| TraceError::Other(Box::new(err)))?;
+        Ok(())
     }
 }
 

diff --git a/opentelemetry-sdk/src/trace/span_processor.rs b/opentelemetry-sdk/src/trace/span_processor.rs
@@ -149,14 +149,7 @@ impl SpanProcessor for SimpleSpanProcessor {
     }
 
     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()
     }
 
     fn set_resource(&mut self, resource: &Resource) {
@@ -432,7 +425,12 @@ impl<R: RuntimeChannel> BatchSpanProcessorInternal<R> {
             // 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

diff --git a/opentelemetry-stdout/CHANGELOG.md b/opentelemetry-stdout/CHANGELOG.md
@@ -2,6 +2,10 @@
 
 ## vNext
 
+- `LogExporter.shutdown` and `SpanExporter.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.
+
+
 ## 0.27.0
 
 Released 2024-Nov-11

diff --git a/opentelemetry-stdout/src/logs/exporter.rs b/opentelemetry-stdout/src/logs/exporter.rs
@@ -56,8 +56,9 @@ impl opentelemetry_sdk::export::logs::LogExporter for LogExporter {
         }
     }
 
-    fn shutdown(&mut self) {
+    fn shutdown(&mut self) -> LogResult<()> {
         self.is_shutdown.store(true, atomic::Ordering::SeqCst);
+        Ok(())
     }
 
     fn set_resource(&mut self, res: &opentelemetry_sdk::Resource) {

diff --git a/opentelemetry-stdout/src/trace/exporter.rs b/opentelemetry-stdout/src/trace/exporter.rs
@@ -1,7 +1,7 @@
 use chrono::{DateTime, Utc};
 use core::fmt;
 use futures_util::future::BoxFuture;
-use opentelemetry::trace::TraceError;
+use opentelemetry::trace::{TraceError, TraceResult};
 use opentelemetry_sdk::export::{self, trace::ExportResult};
 use std::sync::atomic;
 
@@ -59,8 +59,9 @@ impl opentelemetry_sdk::export::trace::SpanExporter for SpanExporter {
         }
     }
 
-    fn shutdown(&mut self) {
+    fn shutdown(&mut self) -> TraceResult<()> {
         self.is_shutdown.store(true, atomic::Ordering::SeqCst);
+        Ok(())
     }
 
     fn set_resource(&mut self, res: &opentelemetry_sdk::Resource) {