Add support for concurrent exports

Applications generating significant span volume can end up dropping data
due to the synchronous export step. According to the opentelemetry spec,

    This function will never be called concurrently for the same exporter
    instance. It can be called again only after the current call returns.

However, it does not place a restriction on concurrent I/O or anything
of that nature. There is an [ongoing discussion] about tweaking the
language to make this more clear.

With that in mind, this commit makes the exporters return a future that
can be spawned concurrently. Unfortunately, this means that the
`export()` method can no longer be async while taking &mut self. The
latter is desirable to enforce the no concurrent calls line of the spec,
so the choice is made here to return a future instead with the lifetime
decoupled from self. This resulted in a bit of additional verbosity, but
for the most part the async code can still be shoved into an async fn
for the ergonomics.

The main exception to this is the `jaeger` exporter which internally
requires a bunch of mutable references. I plan to discuss with the
opentelemetry team the overall goal of this PR and get buy-in before
making more invasive changes to support this in the jaeger exporter.

[ongoing discussion]: open-telemetry/opentelemetry-specification#2434

Cargo.toml

@@ -4,31 +4,31 @@ members = [
     "opentelemetry-contrib",
     "opentelemetry-datadog",
     "opentelemetry-http",
-    "opentelemetry-jaeger",
+# "opentelemetry-jaeger",
     "opentelemetry-otlp",
     "opentelemetry-prometheus",
     "opentelemetry-aws",
     "opentelemetry-semantic-conventions",
     "opentelemetry-stackdriver",
     "opentelemetry-zipkin",
     "opentelemetry-zpages",
-    "examples/actix-http",
-    "examples/actix-http-tracing",
-    "examples/actix-udp",
-    "examples/async",
+# "examples/actix-http",
+# "examples/actix-http-tracing",
+# "examples/actix-udp",
+# "examples/async",
     "examples/aws-xray",
-    "examples/basic",
+# "examples/basic",
     "examples/basic-otlp",
     "examples/basic-otlp-with-selector",
     "examples/basic-otlp-http",
     "examples/datadog",
     "examples/external-otlp-tonic-tokio",
-    "examples/grpc",
+# "examples/grpc",
     "examples/http",
     "examples/hyper-prometheus",
-    "examples/tracing-grpc",
+# "examples/tracing-grpc",
     "examples/zipkin",
-    "examples/multiple-span-processors",
+# "examples/multiple-span-processors",
     "examples/zpages"
 ]
 exclude = ["examples/external-otlp-grpcio-async-std"]

opentelemetry-datadog/Cargo.toml

@@ -36,6 +36,7 @@ thiserror = "1.0"
 itertools = "0.10"
 http = "0.2"
 lazy_static = "1.4"
+futures = "0.3"
 
 [dev-dependencies]
 base64 = "0.13"

opentelemetry-datadog/src/exporter/mod.rs

@@ -5,6 +5,7 @@ pub use model::ApiVersion;
 pub use model::Error;
 
 use async_trait::async_trait;
+use futures::future::BoxFuture;
 use http::{Method, Request, Uri};
 use itertools::Itertools;
 use opentelemetry::sdk::export::trace;
@@ -29,7 +30,7 @@ const DATADOG_TRACE_COUNT_HEADER: &str = "X-Datadog-Trace-Count";
 /// Datadog span exporter
 #[derive(Debug)]
 pub struct DatadogExporter {
-    client: Box<dyn HttpClient>,
+    client: Arc<dyn HttpClient>,
     request_url: Uri,
     service_name: String,
     version: ApiVersion,
@@ -40,7 +41,7 @@ impl DatadogExporter {
         service_name: String,
         request_url: Uri,
         version: ApiVersion,
-        client: Box<dyn HttpClient>,
+        client: Arc<dyn HttpClient>,
     ) -> Self {
         DatadogExporter {
             client,
@@ -49,6 +50,21 @@ impl DatadogExporter {
             version,
         }
     }
+
+    fn build_request(&self, batch: Vec<SpanData>) -> Result<http::Request<Vec<u8>>, TraceError> {
+        let traces: Vec<Vec<SpanData>> = group_into_traces(batch);
+        let trace_count = traces.len();
+        let data = self.version.encode(&self.service_name, traces)?;
+        let req = Request::builder()
+            .method(Method::POST)
+            .uri(self.request_url.clone())
+            .header(http::header::CONTENT_TYPE, self.version.content_type())
+            .header(DATADOG_TRACE_COUNT_HEADER, trace_count)
+            .body(data)
+            .map_err::<Error, _>(Into::into)?;
+
+        Ok(req)
+    }
 }
 
 /// Create a new Datadog exporter pipeline builder.
@@ -63,7 +79,7 @@ pub struct DatadogPipelineBuilder {
     agent_endpoint: String,
     trace_config: Option<sdk::trace::Config>,
     version: ApiVersion,
-    client: Option<Box<dyn HttpClient>>,
+    client: Option<Arc<dyn HttpClient>>,
 }
 
 impl Default for DatadogPipelineBuilder {
@@ -84,15 +100,15 @@ impl Default for DatadogPipelineBuilder {
                 not(feature = "reqwest-blocking-client"),
                 feature = "surf-client"
             ))]
-            client: Some(Box::new(surf::Client::new())),
+            client: Some(Arc::new(surf::Client::new())),
             #[cfg(all(
                 not(feature = "surf-client"),
                 not(feature = "reqwest-blocking-client"),
                 feature = "reqwest-client"
             ))]
-            client: Some(Box::new(reqwest::Client::new())),
+            client: Some(Arc::new(reqwest::Client::new())),
             #[cfg(feature = "reqwest-blocking-client")]
-            client: Some(Box::new(reqwest::blocking::Client::new())),
+            client: Some(Arc::new(reqwest::blocking::Client::new())),
         }
     }
 }
@@ -206,7 +222,7 @@ impl DatadogPipelineBuilder {
     /// Choose the http client used by uploader
     pub fn with_http_client<T: HttpClient + 'static>(
         mut self,
-        client: Box<dyn HttpClient>,
+        client: Arc<dyn HttpClient>,
     ) -> Self {
         self.client = Some(client);
         self
@@ -234,22 +250,25 @@ fn group_into_traces(spans: Vec<SpanData>) -> Vec<Vec<SpanData>> {
         .collect()
 }
 
+async fn send_request(
+    client: Arc<dyn HttpClient>,
+    request: http::Request<Vec<u8>>,
+) -> trace::ExportResult {
+    let _ = client.send(request).await?.error_for_status()?;
+    Ok(())
+}
+
 #[async_trait]
 impl trace::SpanExporter for DatadogExporter {
     /// Export spans to datadog-agent
-    async fn export(&mut self, batch: Vec<SpanData>) -> trace::ExportResult {
-        let traces: Vec<Vec<SpanData>> = group_into_traces(batch);
-        let trace_count = traces.len();
-        let data = self.version.encode(&self.service_name, traces)?;
-        let req = Request::builder()
-            .method(Method::POST)
-            .uri(self.request_url.clone())
-            .header(http::header::CONTENT_TYPE, self.version.content_type())
-            .header(DATADOG_TRACE_COUNT_HEADER, trace_count)
-            .body(data)
-            .map_err::<Error, _>(Into::into)?;
-        let _ = self.client.send(req).await?.error_for_status()?;
-        Ok(())
+    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, trace::ExportResult> {
+        let request = match self.build_request(batch) {
+            Ok(req) => req,
+            Err(err) => return Box::pin(std::future::ready(Err(err))),
+        };
+
+        let client = self.client.clone();
+        Box::pin(send_request(client, request))
     }
 }
 

opentelemetry-otlp/src/exporter/http.rs

@@ -1,6 +1,7 @@
 use crate::{ExportConfig, Protocol};
 use opentelemetry_http::HttpClient;
 use std::collections::HashMap;
+use std::sync::Arc;
 
 /// Configuration of the http transport
 #[cfg(feature = "http-proto")]
@@ -15,7 +16,7 @@ use std::collections::HashMap;
 )]
 pub struct HttpConfig {
     /// Select the HTTP client
-    pub client: Option<Box<dyn HttpClient>>,
+    pub client: Option<Arc<dyn HttpClient>>,
 
     /// Additional headers to send to the collector.
     pub headers: Option<HashMap<String, String>>,
@@ -30,19 +31,19 @@ impl Default for HttpConfig {
     fn default() -> Self {
         HttpConfig {
             #[cfg(feature = "reqwest-blocking-client")]
-            client: Some(Box::new(reqwest::blocking::Client::new())),
+            client: Some(Arc::new(reqwest::blocking::Client::new())),
             #[cfg(all(
                 not(feature = "reqwest-blocking-client"),
                 not(feature = "surf-client"),
                 feature = "reqwest-client"
             ))]
-            client: Some(Box::new(reqwest::Client::new())),
+            client: Some(Arc::new(reqwest::Client::new())),
             #[cfg(all(
                 not(feature = "reqwest-client"),
                 not(feature = "reqwest-blocking-client"),
                 feature = "surf-client"
             ))]
-            client: Some(Box::new(surf::Client::new())),
+            client: Some(Arc::new(surf::Client::new())),
             #[cfg(all(
                 not(feature = "reqwest-client"),
                 not(feature = "surf-client"),
@@ -78,7 +79,7 @@ impl Default for HttpExporterBuilder {
 impl HttpExporterBuilder {
     /// Assign client implementation
     pub fn with_http_client<T: HttpClient + 'static>(mut self, client: T) -> Self {
-        self.http_config.client = Some(Box::new(client));
+        self.http_config.client = Some(Arc::new(client));
         self
     }
 

opentelemetry-otlp/src/span.rs

@@ -67,7 +67,7 @@ use opentelemetry::trace::{TraceError, TracerProvider};
 
 use async_trait::async_trait;
 
-#[cfg(feature = "grpc-sys")]
+#[cfg(any(feature = "grpc-sys", feature = "http-proto"))]
 use std::sync::Arc;
 
 use std::time::Duration;
@@ -280,7 +280,7 @@ pub enum SpanExporter {
         /// The Collector URL
         collector_endpoint: Uri,
         /// The HTTP trace exporter
-        trace_exporter: Option<Box<dyn HttpClient>>,
+        trace_exporter: Option<Arc<dyn HttpClient>>,
     },
 }
 
@@ -408,9 +408,74 @@ impl SpanExporter {
     }
 }
 
+#[cfg(feature = "grpc-sys")]
+async fn grpcio_send_request(
+    trace_exporter: GrpcioTraceServiceClient,
+    request: GrpcRequest,
+    call_options: CallOption,
+) -> ExportResult {
+    let receiver = trace_exporter
+        .export_async_opt(&request, call_options)
+        .map_err::<crate::Error, _>(Into::into)?;
+    receiver.await.map_err::<crate::Error, _>(Into::into)?;
+    Ok(())
+}
+
+#[cfg(feature = "tonic")]
+async fn tonic_send_request(
+    trace_exporter: TonicTraceServiceClient<TonicChannel>,
+    request: Request<TonicRequest>,
+) -> ExportResult {
+    trace_exporter
+        .to_owned()
+        .export(request)
+        .await
+        .map_err::<crate::Error, _>(Into::into)?;
+
+    Ok(())
+}
+
+#[cfg(feature = "http-proto")]
+async fn http_send_request(
+    batch: Vec<SpanData>,
+    client: std::sync::Arc<dyn HttpClient>,
+    headers: Option<HashMap<String, String>>,
+    collector_endpoint: Uri,
+) -> ExportResult {
+    let req = ProstRequest {
+        resource_spans: batch.into_iter().map(Into::into).collect(),
+    };
+
+    let mut buf = vec![];
+    req.encode(&mut buf)
+        .map_err::<crate::Error, _>(Into::into)?;
+
+    let mut request = http::Request::builder()
+        .method(Method::POST)
+        .uri(collector_endpoint)
+        .header(CONTENT_TYPE, "application/x-protobuf")
+        .body(buf)
+        .map_err::<crate::Error, _>(Into::into)?;
+
+    if let Some(headers) = headers {
+        for (k, val) in headers {
+            let value =
+                HeaderValue::from_str(val.as_ref()).map_err::<crate::Error, _>(Into::into)?;
+            let key = HeaderName::try_from(&k).map_err::<crate::Error, _>(Into::into)?;
+            request.headers_mut().insert(key, value);
+        }
+    }
+
+    client.send(request).await?;
+    Ok(())
+}
+
 #[async_trait]
 impl opentelemetry::sdk::export::trace::SpanExporter for SpanExporter {
-    async fn export(&mut self, batch: Vec<SpanData>) -> ExportResult {
+    fn export(
+        &mut self,
+        batch: Vec<SpanData>,
+    ) -> futures::future::BoxFuture<'static, ExportResult> {
         match self {
             #[cfg(feature = "grpc-sys")]
             SpanExporter::Grpcio {
@@ -438,11 +503,11 @@ impl opentelemetry::sdk::export::trace::SpanExporter for SpanExporter {
                     call_options = call_options.headers(metadata_builder.build());
                 }
 
-                let receiver = trace_exporter
-                    .export_async_opt(&request, call_options)
-                    .map_err::<crate::Error, _>(Into::into)?;
-                receiver.await.map_err::<crate::Error, _>(Into::into)?;
-                Ok(())
+                Box::pin(grpcio_send_request(
+                    trace_exporter.clone(),
+                    request,
+                    call_options,
+                ))
             }
 
             #[cfg(feature = "tonic")]
@@ -468,13 +533,7 @@ impl opentelemetry::sdk::export::trace::SpanExporter for SpanExporter {
                     }
                 }
 
-                trace_exporter
-                    .to_owned()
-                    .export(request)
-                    .await
-                    .map_err::<crate::Error, _>(Into::into)?;
-
-                Ok(())
+                Box::pin(tonic_send_request(trace_exporter.to_owned(), request))
             }
 
             #[cfg(feature = "http-proto")]
@@ -484,36 +543,16 @@ impl opentelemetry::sdk::export::trace::SpanExporter for SpanExporter {
                 headers,
                 ..
             } => {
-                let req = ProstRequest {
-                    resource_spans: batch.into_iter().map(Into::into).collect(),
-                };
-
-                let mut buf = vec![];
-                req.encode(&mut buf)
-                    .map_err::<crate::Error, _>(Into::into)?;
-
-                let mut request = http::Request::builder()
-                    .method(Method::POST)
-                    .uri(collector_endpoint.clone())
-                    .header(CONTENT_TYPE, "application/x-protobuf")
-                    .body(buf)
-                    .map_err::<crate::Error, _>(Into::into)?;
-
-                if let Some(headers) = headers.clone() {
-                    for (k, val) in headers {
-                        let value = HeaderValue::from_str(val.as_ref())
-                            .map_err::<crate::Error, _>(Into::into)?;
-                        let key =
-                            HeaderName::try_from(&k).map_err::<crate::Error, _>(Into::into)?;
-                        request.headers_mut().insert(key, value);
-                    }
-                }
-
-                if let Some(client) = trace_exporter {
-                    client.send(request).await?;
-                    Ok(())
+                if let Some(ref client) = trace_exporter {
+                    let client = Arc::clone(client);
+                    Box::pin(http_send_request(
+                        batch,
+                        client,
+                        headers.clone(),
+                        collector_endpoint.clone(),
+                    ))
                 } else {
-                    Err(crate::Error::NoHttpClient.into())
+                    Box::pin(std::future::ready(Err(crate::Error::NoHttpClient.into())))
                 }
             }
         }