Trace subsystem - use OTelSdkResult/OTelSdkError (#2613)

Author: lalitb

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

@@ -1,29 +1,35 @@
 use std::sync::Arc;
 
+use super::OtlpHttpClient;
 use futures_core::future::BoxFuture;
 use http::{header::CONTENT_TYPE, Method};
-use opentelemetry::{otel_debug, trace::TraceError};
-use opentelemetry_sdk::trace::{ExportResult, SpanData, SpanExporter};
-
-use super::OtlpHttpClient;
+use opentelemetry::otel_debug;
+use opentelemetry_sdk::{
+    error::{OTelSdkError, OTelSdkResult},
+    trace::{SpanData, SpanExporter},
+};
 
 impl SpanExporter for OtlpHttpClient {
-    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
+    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
         let client = match self
             .client
             .lock()
-            .map_err(|e| TraceError::Other(e.to_string().into()))
+            .map_err(|e| OTelSdkError::InternalFailure(format!("Mutex lock failed: {}", e)))
             .and_then(|g| match &*g {
                 Some(client) => Ok(Arc::clone(client)),
-                _ => Err(TraceError::Other("exporter is already shut down".into())),
+                _ => Err(OTelSdkError::AlreadyShutdown),
             }) {
             Ok(client) => client,
             Err(err) => return Box::pin(std::future::ready(Err(err))),
         };
 
         let (body, content_type) = match self.build_trace_export_body(batch) {
             Ok(body) => body,
-            Err(e) => return Box::pin(std::future::ready(Err(e))),
+            Err(e) => {
+                return Box::pin(std::future::ready(Err(OTelSdkError::InternalFailure(
+                    e.to_string(),
+                ))))
+            }
         };
 
         let mut request = match http::Request::builder()
@@ -34,10 +40,9 @@ impl SpanExporter for OtlpHttpClient {
         {
             Ok(req) => req,
             Err(e) => {
-                return Box::pin(std::future::ready(Err(crate::Error::RequestFailed(
-                    Box::new(e),
-                )
-                .into())))
+                return Box::pin(std::future::ready(Err(OTelSdkError::InternalFailure(
+                    e.to_string(),
+                ))))
             }
         };
 
@@ -48,7 +53,10 @@ impl SpanExporter for OtlpHttpClient {
         Box::pin(async move {
             let request_uri = request.uri().to_string();
             otel_debug!(name: "HttpTracesClient.CallingExport");
-            let response = client.send_bytes(request).await?;
+            let response = client
+                .send_bytes(request)
+                .await
+                .map_err(|e| OTelSdkError::InternalFailure(format!("{e:?}")))?;
 
             if !response.status().is_success() {
                 let error = format!(
@@ -57,15 +65,23 @@ impl SpanExporter for OtlpHttpClient {
                     request_uri,
                     response.body()
                 );
-                return Err(TraceError::Other(error.into()));
+                return Err(OTelSdkError::InternalFailure(error));
             }
 
             Ok(())
         })
     }
 
-    fn shutdown(&mut self) {
-        let _ = self.client.lock().map(|mut c| c.take());
+    fn shutdown(&mut self) -> OTelSdkResult {
+        let mut client_guard = self.client.lock().map_err(|e| {
+            OTelSdkError::InternalFailure(format!("Failed to acquire client lock: {}", e))
+        })?;
+
+        if client_guard.take().is_none() {
+            return Err(OTelSdkError::AlreadyShutdown);
+        }
+
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {

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

@@ -1,14 +1,17 @@
 use core::fmt;
 
 use futures_core::future::BoxFuture;
-use opentelemetry::{otel_debug, trace::TraceError};
+use opentelemetry::otel_debug;
 use opentelemetry_proto::tonic::collector::trace::v1::{
     trace_service_client::TraceServiceClient, ExportTraceServiceRequest,
 };
-use opentelemetry_sdk::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::error::OTelSdkError;
+use opentelemetry_sdk::{
+    error::OTelSdkResult,
+    trace::{SpanData, SpanExporter},
+};
+use tonic::{codegen::CompressionEncoding, service::Interceptor, transport::Channel, Request};
 
 use super::BoxInterceptor;
 
@@ -56,21 +59,21 @@ impl TonicTracesClient {
 }
 
 impl SpanExporter for TonicTracesClient {
-    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
+    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
         let (mut client, metadata, extensions) = match &mut self.inner {
             Some(inner) => {
                 let (m, e, _) = match inner.interceptor.call(Request::new(())) {
                     Ok(res) => res.into_parts(),
                     Err(e) => {
-                        return Box::pin(std::future::ready(Err(TraceError::Other(Box::new(e)))))
+                        return Box::pin(std::future::ready(Err(OTelSdkError::InternalFailure(
+                            e.to_string(),
+                        ))))
                     }
                 };
                 (inner.client.clone(), m, e)
             }
             None => {
-                return Box::pin(std::future::ready(Err(TraceError::Other(
-                    "exporter is already shut down".into(),
-                ))))
+                return Box::pin(std::future::ready(Err(OTelSdkError::AlreadyShutdown)));
             }
         };
 
@@ -86,14 +89,16 @@ impl SpanExporter for TonicTracesClient {
                     ExportTraceServiceRequest { resource_spans },
                 ))
                 .await
-                .map_err(crate::Error::from)?;
-
+                .map_err(|e| OTelSdkError::InternalFailure(e.to_string()))?;
             Ok(())
         })
     }
 
-    fn shutdown(&mut self) {
-        let _ = self.inner.take();
+    fn shutdown(&mut self) -> OTelSdkResult {
+        match self.inner.take() {
+            Some(_) => Ok(()), // Successfully took `inner`, indicating a successful shutdown.
+            None => Err(OTelSdkError::AlreadyShutdown), // `inner` was already `None`, meaning it's already shut down.
+        }
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {

opentelemetry-otlp/src/span.rs

@@ -5,7 +5,8 @@
 use std::fmt::Debug;
 
 use futures_core::future::BoxFuture;
-use opentelemetry_sdk::trace::{ExportResult, SpanData};
+use opentelemetry_sdk::error::OTelSdkResult;
+use opentelemetry_sdk::trace::SpanData;
 
 #[cfg(feature = "grpc-tonic")]
 use crate::{
@@ -122,7 +123,7 @@ impl SpanExporter {
 }
 
 impl opentelemetry_sdk::trace::SpanExporter for SpanExporter {
-    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
+    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
         self.0.export(batch)
     }
 

opentelemetry-sdk/CHANGELOG.md

@@ -373,6 +373,24 @@ let processor = BatchSpanProcessor::builder(exporter)
     .build();
 ```
 
+- **Breaking**
+ - The public API changes in the Tracing:
+   - Before:
+      ```rust
+        fn SpanExporter::export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, ExportResult>;
+        fn SpanExporter::shutdown(&mut self);
+        fn SpanExporter::force_flush(&mut self) -> BoxFuture<'static, ExportResult>
+        fn TraerProvider::shutdown(&self) -> TraceResult<()>
+        fn TracerProvider::force_flush(&self) -> Vec<TraceResult<()>>
+      ```
+    - After:
+      ```rust
+        fn SpanExporter::export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult>;
+        fn SpanExporter::shutdown(&mut self) -> OTelSdkResult;
+        fn SpanExporter::force_flush(&mut self) -> BoxFuture<'static, OTelSdkResult>
+        fn TraerProvider::shutdown(&self) -> OTelSdkResult;
+        fn TracerProvider::force_flush(&self) -> OTelSdkResult;
+      ```
 - **Breaking** Renamed `LoggerProvider` and `Logger` to `SdkLoggerProvider` and
   `SdkLogger` respectively to avoid name collision with public API types.
   [#2612](https://github.com/open-telemetry/opentelemetry-rust/pull/2612)

opentelemetry-sdk/benches/context.rs

@@ -9,8 +9,8 @@ use opentelemetry::{
     Context, ContextGuard,
 };
 use opentelemetry_sdk::{
-    trace::{ExportResult, SpanData, SpanExporter},
-    trace::{Sampler, TracerProvider},
+    error::OTelSdkResult,
+    trace::{Sampler, SpanData, SpanExporter, TracerProvider},
 };
 #[cfg(not(target_os = "windows"))]
 use pprof::criterion::{Output, PProfProfiler};
@@ -137,7 +137,7 @@ fn parent_sampled_tracer(inner_sampler: Sampler) -> (TracerProvider, BoxedTracer
 struct NoopExporter;
 
 impl SpanExporter for NoopExporter {
-    fn export(&mut self, _spans: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
+    fn export(&mut self, _spans: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
         Box::pin(futures_util::future::ready(Ok(())))
     }
 }

opentelemetry-sdk/benches/span_builder.rs

@@ -4,9 +4,10 @@ use opentelemetry::{
     trace::{Span, Tracer, TracerProvider},
     KeyValue,
 };
+use opentelemetry_sdk::error::OTelSdkResult;
 use opentelemetry_sdk::{
     trace as sdktrace,
-    trace::{ExportResult, SpanData, SpanExporter},
+    trace::{SpanData, SpanExporter},
 };
 #[cfg(not(target_os = "windows"))]
 use pprof::criterion::{Output, PProfProfiler};
@@ -65,7 +66,7 @@ fn not_sampled_provider() -> (sdktrace::TracerProvider, sdktrace::Tracer) {
 struct NoopExporter;
 
 impl SpanExporter for NoopExporter {
-    fn export(&mut self, _spans: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
+    fn export(&mut self, _spans: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
         Box::pin(futures_util::future::ready(Ok(())))
     }
 }

opentelemetry-sdk/benches/trace.rs

@@ -5,8 +5,8 @@ use opentelemetry::{
     KeyValue,
 };
 use opentelemetry_sdk::{
-    trace as sdktrace,
-    trace::{ExportResult, SpanData, SpanExporter},
+    error::OTelSdkResult,
+    trace::{self as sdktrace, SpanData, SpanExporter},
 };
 #[cfg(not(target_os = "windows"))]
 use pprof::criterion::{Output, PProfProfiler};
@@ -60,7 +60,7 @@ fn criterion_benchmark(c: &mut Criterion) {
 struct VoidExporter;
 
 impl SpanExporter for VoidExporter {
-    fn export(&mut self, _spans: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
+    fn export(&mut self, _spans: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
         Box::pin(futures_util::future::ready(Ok(())))
     }
 }

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

@@ -1,5 +1,6 @@
+use crate::error::{OTelSdkError, OTelSdkResult};
 use crate::{
-    trace::{ExportResult, SpanData, SpanExporter},
+    trace::{SpanData, SpanExporter},
     trace::{SpanEvents, SpanLinks},
 };
 use futures_util::future::BoxFuture;
@@ -40,21 +41,20 @@ pub struct TokioSpanExporter {
 }
 
 impl SpanExporter for TokioSpanExporter {
-    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
-        for span_data in batch {
-            if let Err(err) = self
-                .tx_export
+    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
+        let result = batch.into_iter().try_for_each(|span_data| {
+            self.tx_export
                 .send(span_data)
-                .map_err::<TestExportError, _>(Into::into)
-            {
-                return Box::pin(std::future::ready(Err(Into::into(err))));
-            }
-        }
-        Box::pin(std::future::ready(Ok(())))
+                .map_err(|err| OTelSdkError::InternalFailure(format!("Export failed: {:?}", err)))
+        });
+
+        Box::pin(std::future::ready(result))
     }
 
-    fn shutdown(&mut self) {
-        self.tx_shutdown.send(()).unwrap();
+    fn shutdown(&mut self) -> OTelSdkResult {
+        self.tx_shutdown.send(()).map_err(|_| {
+            OTelSdkError::InternalFailure("Failed to send shutdown signal".to_string())
+        })
     }
 }
 
@@ -113,7 +113,7 @@ impl NoopSpanExporter {
 
 #[async_trait::async_trait]
 impl SpanExporter for NoopSpanExporter {
-    fn export(&mut self, _: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
+    fn export(&mut self, _: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
         Box::pin(std::future::ready(Ok(())))
     }
 }

opentelemetry-sdk/src/trace/export.rs

@@ -1,4 +1,5 @@
 //! Trace exporters
+use crate::error::OTelSdkResult;
 use crate::Resource;
 use futures_util::future::BoxFuture;
 use opentelemetry::trace::{SpanContext, SpanId, SpanKind, Status, TraceError};
@@ -30,7 +31,7 @@ pub trait SpanExporter: Send + Sync + Debug {
     ///
     /// 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, OTelSdkResult>;
 
     /// Shuts down the exporter. Called when SDK is shut down. This is an
     /// opportunity for exporter to do any cleanup required.
@@ -43,7 +44,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) -> OTelSdkResult {
+        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
@@ -60,8 +63,8 @@ pub trait SpanExporter: Send + Sync + Debug {
     /// implemented as a blocking API or an asynchronous API which notifies the caller via
     /// a callback or an event. OpenTelemetry client authors can decide if they want to
     /// make the flush timeout configurable.
-    fn force_flush(&mut self) -> BoxFuture<'static, ExportResult> {
-        Box::pin(async { Ok(()) })
+    fn force_flush(&mut self) -> OTelSdkResult {
+        Ok(())
     }
 
     /// Set the resource for the exporter.

opentelemetry-sdk/src/trace/in_memory_exporter.rs

@@ -1,5 +1,6 @@
+use crate::error::{OTelSdkError, OTelSdkResult};
 use crate::resource::Resource;
-use crate::trace::{ExportResult, SpanData, SpanExporter};
+use crate::trace::{SpanData, SpanExporter};
 use futures_util::future::BoxFuture;
 use opentelemetry::trace::{TraceError, TraceResult};
 use std::sync::{Arc, Mutex};
@@ -130,20 +131,20 @@ impl InMemorySpanExporter {
 }
 
 impl SpanExporter for InMemorySpanExporter {
-    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
-        if let Err(err) = self
+    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
+        let result = self
             .spans
             .lock()
             .map(|mut spans_guard| spans_guard.append(&mut batch.clone()))
-            .map_err(TraceError::from)
-        {
-            return Box::pin(std::future::ready(Err(Into::into(err))));
-        }
-        Box::pin(std::future::ready(Ok(())))
+            .map_err(|err| {
+                OTelSdkError::InternalFailure(format!("Failed to lock spans: {:?}", err))
+            });
+        Box::pin(std::future::ready(result))
     }
 
-    fn shutdown(&mut self) {
-        self.reset()
+    fn shutdown(&mut self) -> OTelSdkResult {
+        self.reset();
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &Resource) {