Span Exporter async native (#2685)

Author: lalitb

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

@@ -223,7 +223,7 @@ impl HttpExporterBuilder {
             OTEL_EXPORTER_OTLP_TRACES_HEADERS,
         )?;
 
-        Ok(crate::SpanExporter::new(client))
+        Ok(crate::SpanExporter::from_http(client))
     }
 
     /// Create a log exporter with the current configuration

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

@@ -1,7 +1,6 @@
 use std::sync::Arc;
 
 use super::OtlpHttpClient;
-use futures_core::future::BoxFuture;
 use http::{header::CONTENT_TYPE, Method};
 use opentelemetry::otel_debug;
 use opentelemetry_sdk::{
@@ -10,7 +9,7 @@ use opentelemetry_sdk::{
 };
 
 impl SpanExporter for OtlpHttpClient {
-    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
+    async fn export(&mut self, batch: Vec<SpanData>) -> OTelSdkResult {
         let client = match self
             .client
             .lock()
@@ -20,16 +19,12 @@ impl SpanExporter for OtlpHttpClient {
                 _ => Err(OTelSdkError::AlreadyShutdown),
             }) {
             Ok(client) => client,
-            Err(err) => return Box::pin(std::future::ready(Err(err))),
+            Err(err) => return 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(OTelSdkError::InternalFailure(
-                    e.to_string(),
-                ))))
-            }
+            Err(e) => return Err(OTelSdkError::InternalFailure(e.to_string())),
         };
 
         let mut request = match http::Request::builder()
@@ -39,37 +34,31 @@ impl SpanExporter for OtlpHttpClient {
             .body(body.into())
         {
             Ok(req) => req,
-            Err(e) => {
-                return Box::pin(std::future::ready(Err(OTelSdkError::InternalFailure(
-                    e.to_string(),
-                ))))
-            }
+            Err(e) => return Err(OTelSdkError::InternalFailure(e.to_string())),
         };
 
         for (k, v) in &self.headers {
             request.headers_mut().insert(k.clone(), v.clone());
         }
 
-        Box::pin(async move {
-            let request_uri = request.uri().to_string();
-            otel_debug!(name: "HttpTracesClient.CallingExport");
-            let response = client
-                .send_bytes(request)
-                .await
-                .map_err(|e| OTelSdkError::InternalFailure(format!("{e:?}")))?;
+        let request_uri = request.uri().to_string();
+        otel_debug!(name: "HttpTracesClient.CallingExport");
+        let response = client
+            .send_bytes(request)
+            .await
+            .map_err(|e| OTelSdkError::InternalFailure(format!("{e:?}")))?;
 
-            if !response.status().is_success() {
-                let error = format!(
-                    "OpenTelemetry trace export failed. Url: {}, Status Code: {}, Response: {:?}",
-                    response.status().as_u16(),
-                    request_uri,
-                    response.body()
-                );
-                return Err(OTelSdkError::InternalFailure(error));
-            }
+        if !response.status().is_success() {
+            let error = format!(
+                "OpenTelemetry trace export failed. Url: {}, Status Code: {}, Response: {:?}",
+                response.status().as_u16(),
+                request_uri,
+                response.body()
+            );
+            return Err(OTelSdkError::InternalFailure(error));
+        }
 
-            Ok(())
-        })
+        Ok(())
     }
 
     fn shutdown(&mut self) -> OTelSdkResult {

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

@@ -317,7 +317,7 @@ impl TonicExporterBuilder {
 
         let client = TonicTracesClient::new(channel, interceptor, compression);
 
-        Ok(crate::SpanExporter::new(client))
+        Ok(crate::SpanExporter::from_tonic(client))
     }
 }
 

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

@@ -1,6 +1,5 @@
 use core::fmt;
 
-use futures_core::future::BoxFuture;
 use opentelemetry::otel_debug;
 use opentelemetry_proto::tonic::collector::trace::v1::{
     trace_service_client::TraceServiceClient, ExportTraceServiceRequest,
@@ -59,39 +58,33 @@ impl TonicTracesClient {
 }
 
 impl SpanExporter for TonicTracesClient {
-    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
+    async fn export(&mut self, batch: Vec<SpanData>) -> 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(OTelSdkError::InternalFailure(
-                            e.to_string(),
-                        ))))
-                    }
+                    Err(e) => return Err(OTelSdkError::InternalFailure(e.to_string())),
                 };
                 (inner.client.clone(), m, e)
             }
             None => {
-                return Box::pin(std::future::ready(Err(OTelSdkError::AlreadyShutdown)));
+                return Err(OTelSdkError::AlreadyShutdown);
             }
         };
 
         let resource_spans = group_spans_by_resource_and_scope(batch, &self.resource);
 
         otel_debug!(name: "TonicsTracesClient.CallingExport");
 
-        Box::pin(async move {
-            client
-                .export(Request::from_parts(
-                    metadata,
-                    extensions,
-                    ExportTraceServiceRequest { resource_spans },
-                ))
-                .await
-                .map_err(|e| OTelSdkError::InternalFailure(e.to_string()))?;
-            Ok(())
-        })
+        client
+            .export(Request::from_parts(
+                metadata,
+                extensions,
+                ExportTraceServiceRequest { resource_spans },
+            ))
+            .await
+            .map_err(|e| OTelSdkError::InternalFailure(e.to_string()))?;
+        Ok(())
     }
 
     fn shutdown(&mut self) -> OTelSdkResult {

opentelemetry-otlp/src/span.rs

@@ -4,7 +4,6 @@
 
 use std::fmt::Debug;
 
-use futures_core::future::BoxFuture;
 use opentelemetry_sdk::error::OTelSdkResult;
 use opentelemetry_sdk::trace::SpanData;
 
@@ -66,15 +65,15 @@ impl SpanExporterBuilder<TonicExporterBuilderSet> {
     pub fn build(self) -> Result<SpanExporter, opentelemetry::trace::TraceError> {
         let span_exporter = self.client.0.build_span_exporter()?;
         opentelemetry::otel_debug!(name: "SpanExporterBuilt");
-        Ok(SpanExporter::new(span_exporter))
+        Ok(span_exporter)
     }
 }
 
 #[cfg(any(feature = "http-proto", feature = "http-json"))]
 impl SpanExporterBuilder<HttpExporterBuilderSet> {
     pub fn build(self) -> Result<SpanExporter, opentelemetry::trace::TraceError> {
         let span_exporter = self.client.0.build_span_exporter()?;
-        Ok(SpanExporter::new(span_exporter))
+        Ok(span_exporter)
     }
 }
 
@@ -106,28 +105,57 @@ impl HasHttpConfig for SpanExporterBuilder<HttpExporterBuilderSet> {
     }
 }
 
-/// OTLP exporter that sends tracing information
+/// OTLP exporter that sends tracing data
 #[derive(Debug)]
-pub struct SpanExporter(Box<dyn opentelemetry_sdk::trace::SpanExporter>);
+pub struct SpanExporter {
+    client: SupportedTransportClient,
+}
+
+#[derive(Debug)]
+enum SupportedTransportClient {
+    #[cfg(feature = "grpc-tonic")]
+    Tonic(crate::exporter::tonic::trace::TonicTracesClient),
+    #[cfg(any(feature = "http-proto", feature = "http-json"))]
+    Http(crate::exporter::http::OtlpHttpClient),
+}
 
 impl SpanExporter {
     /// Obtain a builder to configure a [SpanExporter].
     pub fn builder() -> SpanExporterBuilder<NoExporterBuilderSet> {
         SpanExporterBuilder::default()
     }
 
-    /// Build a new span exporter from a client
-    pub fn new(client: impl opentelemetry_sdk::trace::SpanExporter + 'static) -> Self {
-        SpanExporter(Box::new(client))
+    #[cfg(any(feature = "http-proto", feature = "http-json"))]
+    pub(crate) fn from_http(client: crate::exporter::http::OtlpHttpClient) -> Self {
+        SpanExporter {
+            client: SupportedTransportClient::Http(client),
+        }
+    }
+
+    #[cfg(feature = "grpc-tonic")]
+    pub(crate) fn from_tonic(client: crate::exporter::tonic::trace::TonicTracesClient) -> Self {
+        SpanExporter {
+            client: SupportedTransportClient::Tonic(client),
+        }
     }
 }
 
 impl opentelemetry_sdk::trace::SpanExporter for SpanExporter {
-    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
-        self.0.export(batch)
+    async fn export(&mut self, batch: Vec<SpanData>) -> OTelSdkResult {
+        match &mut self.client {
+            #[cfg(feature = "grpc-tonic")]
+            SupportedTransportClient::Tonic(client) => client.export(batch).await,
+            #[cfg(any(feature = "http-proto", feature = "http-json"))]
+            SupportedTransportClient::Http(client) => client.export(batch).await,
+        }
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
-        self.0.set_resource(resource);
+        match &mut self.client {
+            #[cfg(feature = "grpc-tonic")]
+            SupportedTransportClient::Tonic(client) => client.set_resource(resource),
+            #[cfg(any(feature = "http-proto", feature = "http-json"))]
+            SupportedTransportClient::Http(client) => client.set_resource(resource),
+        }
     }
 }

opentelemetry-sdk/benches/context.rs

@@ -1,5 +1,4 @@
 use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
-use futures_util::future::BoxFuture;
 use opentelemetry::{
     global::BoxedTracer,
     trace::{
@@ -137,8 +136,8 @@ fn parent_sampled_tracer(inner_sampler: Sampler) -> (SdkTracerProvider, BoxedTra
 struct NoopExporter;
 
 impl SpanExporter for NoopExporter {
-    fn export(&mut self, _spans: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
-        Box::pin(futures_util::future::ready(Ok(())))
+    async fn export(&mut self, _spans: Vec<SpanData>) -> OTelSdkResult {
+        Ok(())
     }
 }
 

opentelemetry-sdk/benches/span_builder.rs

@@ -1,5 +1,4 @@
 use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
-use futures_util::future::BoxFuture;
 use opentelemetry::{
     trace::{Span, Tracer, TracerProvider},
     KeyValue,
@@ -66,8 +65,8 @@ fn not_sampled_provider() -> (sdktrace::SdkTracerProvider, sdktrace::SdkTracer)
 struct NoopExporter;
 
 impl SpanExporter for NoopExporter {
-    fn export(&mut self, _spans: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
-        Box::pin(futures_util::future::ready(Ok(())))
+    async fn export(&mut self, _spans: Vec<SpanData>) -> OTelSdkResult {
+        Ok(())
     }
 }
 

opentelemetry-sdk/benches/trace.rs

@@ -1,5 +1,4 @@
 use criterion::{criterion_group, criterion_main, Criterion};
-use futures_util::future::BoxFuture;
 use opentelemetry::{
     trace::{Span, Tracer, TracerProvider},
     KeyValue,
@@ -60,8 +59,8 @@ fn criterion_benchmark(c: &mut Criterion) {
 struct VoidExporter;
 
 impl SpanExporter for VoidExporter {
-    fn export(&mut self, _spans: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
-        Box::pin(futures_util::future::ready(Ok(())))
+    async fn export(&mut self, _spans: Vec<SpanData>) -> OTelSdkResult {
+        Ok(())
     }
 }
 

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

@@ -3,7 +3,6 @@ use crate::{
     trace::{SpanData, SpanExporter},
     trace::{SpanEvents, SpanLinks},
 };
-use futures_util::future::BoxFuture;
 pub use opentelemetry::testing::trace::TestSpan;
 use opentelemetry::{
     trace::{SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState},
@@ -41,14 +40,12 @@ pub struct TokioSpanExporter {
 }
 
 impl SpanExporter for TokioSpanExporter {
-    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
-        let result = batch.into_iter().try_for_each(|span_data| {
+    async fn export(&mut self, batch: Vec<SpanData>) -> OTelSdkResult {
+        batch.into_iter().try_for_each(|span_data| {
             self.tx_export
                 .send(span_data)
                 .map_err(|err| OTelSdkError::InternalFailure(format!("Export failed: {:?}", err)))
-        });
-
-        Box::pin(std::future::ready(result))
+        })
     }
 
     fn shutdown(&mut self) -> OTelSdkResult {
@@ -112,7 +109,7 @@ impl NoopSpanExporter {
 }
 
 impl SpanExporter for NoopSpanExporter {
-    fn export(&mut self, _: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult> {
-        Box::pin(std::future::ready(Ok(())))
+    async fn export(&mut self, _: Vec<SpanData>) -> OTelSdkResult {
+        Ok(())
     }
 }

opentelemetry-sdk/src/trace/export.rs

@@ -1,7 +1,6 @@
 //! Trace exporters
 use crate::error::OTelSdkResult;
 use crate::Resource;
-use futures_util::future::BoxFuture;
 use opentelemetry::trace::{SpanContext, SpanId, SpanKind, Status};
 use opentelemetry::{InstrumentationScope, KeyValue};
 use std::borrow::Cow;
@@ -28,7 +27,10 @@ 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, OTelSdkResult>;
+    fn export(
+        &mut self,
+        batch: Vec<SpanData>,
+    ) -> impl std::future::Future<Output = OTelSdkResult> + Send;
 
     /// Shuts down the exporter. Called when SDK is shut down. This is an
     /// opportunity for exporter to do any cleanup required.