Switch to async exporters (#232)

Author: jtescher

Cargo.toml

@@ -18,6 +18,7 @@ edition = "2018"
 
 [dependencies]
 async-std = { version = "1.6", features = ["unstable"], optional = true }
+async-trait = { version = "0.1", optional = true }
 base64 = { version = "0.12", optional = true }
 bincode = { version = "1.2", optional = true }
 dashmap = { version = "4.0.0-rc6", optional = true }
@@ -42,7 +43,7 @@ tokio = { version = "0.2", features = ["full"] }
 [features]
 default = ["metrics", "trace"]
 base64_format = ["base64", "binary_propagator"]
-trace = ["rand", "pin-project"]
+trace = ["rand", "pin-project", "async-trait"]
 metrics = ["thiserror", "dashmap", "fnv"]
 serialize = ["serde", "bincode"]
 binary_propagator = []
@@ -78,4 +79,4 @@ harness = false
 
 [[bench]]
 name = "ddsketch"
-harness = false
+harness = false

examples/basic/Cargo.toml

@@ -6,8 +6,8 @@ edition = "2018"
 [dependencies]
 futures = "0.3"
 lazy_static = "1.4"
-opentelemetry = { path = "../../", features = ["serialize"] }
-opentelemetry-jaeger = { path = "../../opentelemetry-jaeger" }
+opentelemetry = { path = "../../", features = ["serialize", "tokio"] }
+opentelemetry-jaeger = { path = "../../opentelemetry-jaeger", features = ["tokio"] }
 serde_json = "1.0"
 thrift = "0.13"
 tokio = { version = "0.2", features = ["full"] }

examples/stdout.rs

@@ -1,15 +1,15 @@
-use opentelemetry::exporter::trace::stdout;
-use opentelemetry::{api::Tracer, sdk};
+use opentelemetry::{
+    api::Tracer,
+    exporter::trace::stdout,
+    sdk::{trace, Sampler},
+};
 
 fn main() {
     // Install stdout exporter pipeline to be able to retrieve collected spans.
     // For the demonstration, use `Sampler::AlwaysOn` sampler to sample all traces. In a production
     // application, use `Sampler::ParentBased` or `Sampler::TraceIdRatioBased` with a desired ratio.
     let (tracer, _uninstall) = stdout::new_pipeline()
-        .with_trace_config(sdk::Config {
-            default_sampler: Box::new(sdk::Sampler::AlwaysOn),
-            ..Default::default()
-        })
+        .with_trace_config(trace::config().with_default_sampler(Sampler::AlwaysOn))
         .install();
 
     tracer.in_span("operation", |_cx| {});

opentelemetry-contrib/Cargo.toml

@@ -18,12 +18,13 @@ edition = "2018"
 
 [features]
 default = []
-datadog = ["indexmap", "reqwest", "rmp"]
+datadog = ["indexmap", "reqwest", "rmp", "async-trait"]
 
 [dependencies]
+async-trait = { version = "0.1", optional = true }
 indexmap = { version = "1.6.0", optional = true }
 opentelemetry = { version = "0.8.0", path = ".." }
-reqwest = { version = "0.10", features = ["blocking"], optional = true }
+reqwest = { version = "0.10", optional = true }
 rmp = { version = "0.8", optional = true }
 lazy_static = "1.4"
 

opentelemetry-contrib/src/datadog/mod.rs

@@ -85,6 +85,7 @@ mod model;
 
 pub use model::ApiVersion;
 
+use async_trait::async_trait;
 use opentelemetry::{api::TracerProvider, exporter::trace, global, sdk};
 use reqwest::header::CONTENT_TYPE;
 use reqwest::Url;
@@ -100,7 +101,7 @@ const DEFAULT_SERVICE_NAME: &str = "OpenTelemetry";
 /// Datadog span exporter
 #[derive(Debug)]
 pub struct DatadogExporter {
-    client: reqwest::blocking::Client,
+    client: reqwest::Client,
     request_url: Url,
     service_name: String,
     version: ApiVersion,
@@ -112,7 +113,7 @@ impl DatadogExporter {
         request_url.set_path(version.path());
 
         DatadogExporter {
-            client: reqwest::blocking::Client::new(),
+            client: reqwest::Client::new(),
             request_url,
             service_name,
             version,
@@ -190,9 +191,10 @@ impl DatadogPipelineBuilder {
     }
 }
 
+#[async_trait]
 impl trace::SpanExporter for DatadogExporter {
     /// Export spans to datadog-agent
-    fn export(&self, batch: Vec<Arc<trace::SpanData>>) -> trace::ExportResult {
+    async fn export(&self, batch: &[Arc<trace::SpanData>]) -> trace::ExportResult {
         let data = match self.version.encode(&self.service_name, batch) {
             Ok(data) => data,
             Err(_) => return trace::ExportResult::FailedNotRetryable,
@@ -203,7 +205,8 @@ impl trace::SpanExporter for DatadogExporter {
             .post(self.request_url.clone())
             .header(CONTENT_TYPE, self.version.content_type())
             .body(data)
-            .send();
+            .send()
+            .await;
 
         match resp {
             Ok(response) if response.status().is_success() => trace::ExportResult::Success,

opentelemetry-contrib/src/datadog/model/mod.rs

@@ -53,7 +53,7 @@ impl ApiVersion {
     pub(crate) fn encode(
         self,
         service_name: &str,
-        spans: Vec<Arc<trace::SpanData>>,
+        spans: &[Arc<trace::SpanData>],
     ) -> Result<Vec<u8>, Error> {
         match self {
             Self::Version03 => v03::encode(service_name, spans),
@@ -115,7 +115,7 @@ mod tests {
     #[test]
     fn test_encode_v03() -> Result<(), Box<dyn std::error::Error>> {
         let spans = get_spans();
-        let encoded = base64::encode(ApiVersion::Version03.encode("service_name", spans)?);
+        let encoded = base64::encode(ApiVersion::Version03.encode("service_name", &spans)?);
 
         assert_eq!(encoded.as_str(), "kZGLpHR5cGWjd2Vip3NlcnZpY2Wsc2VydmljZV9uYW1lpG5hbWWpY29tcG9uZW50qHJlc291cmNlqHJlc291cmNlqHRyYWNlX2lkzwAAAAAAAAAHp3NwYW5faWTPAAAAAAAAAGOpcGFyZW50X2lkzwAAAAAAAAABpXN0YXJ00wAAAAAAAAAAqGR1cmF0aW9u0wAAAAA7msoApWVycm9y0gAAAACkbWV0YYGpc3Bhbi50eXBlo3dlYg==");
 
@@ -125,7 +125,7 @@ mod tests {
     #[test]
     fn test_encode_v05() -> Result<(), Box<dyn std::error::Error>> {
         let spans = get_spans();
-        let encoded = base64::encode(ApiVersion::Version05.encode("service_name", spans)?);
+        let encoded = base64::encode(ApiVersion::Version05.encode("service_name", &spans)?);
 
         assert_eq!(encoded.as_str(), "kpWsc2VydmljZV9uYW1lo3dlYqljb21wb25lbnSocmVzb3VyY2Wpc3Bhbi50eXBlkZGczgAAAADOAAAAAs4AAAADzwAAAAAAAAAHzwAAAAAAAABjzwAAAAAAAAAB0wAAAAAAAAAA0wAAAAA7msoA0gAAAACBzgAAAATOAAAAAYDOAAAAAQ==");
 

opentelemetry-contrib/src/datadog/model/v03.rs

@@ -4,10 +4,7 @@ use opentelemetry::exporter::trace;
 use std::sync::Arc;
 use std::time::SystemTime;
 
-pub(crate) fn encode(
-    service_name: &str,
-    spans: Vec<Arc<trace::SpanData>>,
-) -> Result<Vec<u8>, Error> {
+pub(crate) fn encode(service_name: &str, spans: &[Arc<trace::SpanData>]) -> Result<Vec<u8>, Error> {
     let mut encoded = Vec::new();
     rmp::encode::write_array_len(&mut encoded, spans.len() as u32)?;
 

opentelemetry-contrib/src/datadog/model/v05.rs

@@ -50,10 +50,7 @@ use std::time::SystemTime;
 //
 // 		The dictionary in this case would be []string{""}, having only the empty string at index 0.
 //
-pub(crate) fn encode(
-    service_name: &str,
-    spans: Vec<Arc<trace::SpanData>>,
-) -> Result<Vec<u8>, Error> {
+pub(crate) fn encode(service_name: &str, spans: &[Arc<trace::SpanData>]) -> Result<Vec<u8>, Error> {
     let mut interner = StringInterner::new();
     let mut encoded_spans = encode_spans(&mut interner, service_name, spans)?;
 
@@ -73,7 +70,7 @@ pub(crate) fn encode(
 fn encode_spans(
     interner: &mut StringInterner,
     service_name: &str,
-    spans: Vec<Arc<trace::SpanData>>,
+    spans: &[Arc<trace::SpanData>],
 ) -> Result<Vec<u8>, Error> {
     let mut encoded = Vec::new();
     rmp::encode::write_array_len(&mut encoded, spans.len() as u32)?;

opentelemetry-jaeger/Cargo.toml

@@ -16,10 +16,14 @@ license = "Apache-2.0"
 edition = "2018"
 
 [dependencies]
+async-std = { version = "1.6", optional = true }
+async-trait = "0.1"
+http = { version = "0.2", optional = true }
+isahc = { version = "0.9", default-features = false, optional = true }
 opentelemetry = { version = "0.8", default-features = false, features = ["trace"], path = ".." }
-ureq = { version = "1.4", optional = true }
 thrift = "0.13"
+tokio = { version = "0.2", features = ["udp", "sync"], optional = true }
 
 [features]
 default = []
-collector_client = ["ureq"]
+collector_client = ["isahc", "http"]

opentelemetry-jaeger/README.md

@@ -44,7 +44,7 @@ for either executor when you install the pipeline.
 ```toml
 [dependencies]
 opentelemetry = { version = "*", features = ["tokio"] }
-opentelemetry-jaeger = "*"
+opentelemetry-jaeger = { version = "*", features = ["tokio"] }
 ```
 
 [`tokio`]: https://tokio.rs

opentelemetry-jaeger/src/agent.rs

@@ -1,57 +1,120 @@
 //! # UDP Jaeger Agent Client
-use crate::thrift::{agent, jaeger, zipkincore};
-use crate::transport::TUdpChannel;
+use crate::thrift::{
+    agent::{self, TAgentSyncClient},
+    jaeger,
+};
+use crate::transport::TNoopChannel;
 use std::fmt;
-use std::net::ToSocketAddrs;
-use thrift::protocol;
-use thrift::protocol::{TCompactInputProtocol, TCompactOutputProtocol};
-use thrift::transport::{ReadHalf, TIoChannel, WriteHalf};
+use std::net::{ToSocketAddrs, UdpSocket};
+use std::sync::Mutex;
+use thrift::{
+    protocol::{TCompactInputProtocol, TCompactOutputProtocol},
+    transport::{ReadHalf, TBufferChannel, TIoChannel, WriteHalf},
+};
 
-/// The max size of UDP packet we want to send, synced with jaeger-agent.
-const UDP_MAX_PACKET_SIZE: usize = 65000;
-
-/// `AgentSyncClientUDP` implements the `TAgentSyncClient` interface over UDP
-pub(crate) struct AgentSyncClientUDP {
+struct BufferClient {
+    buffer: ReadHalf<TBufferChannel>,
     client: agent::AgentSyncClient<
-        TCompactInputProtocol<ReadHalf<TUdpChannel>>,
-        TCompactOutputProtocol<WriteHalf<TUdpChannel>>,
+        TCompactInputProtocol<TNoopChannel>,
+        TCompactOutputProtocol<WriteHalf<TBufferChannel>>,
     >,
 }
 
-impl fmt::Debug for AgentSyncClientUDP {
+impl fmt::Debug for BufferClient {
     /// Debug info
     fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
-        fmt.debug_struct("AgentClientUDP")
+        fmt.debug_struct("BufferClient")
+            .field("buffer", &self.buffer)
             .field("client", &"AgentSyncClient")
             .finish()
     }
 }
 
-impl AgentSyncClientUDP {
+/// `AgentAsyncClientUDP` implements an async version of the `TAgentSyncClient`
+/// interface over UDP.
+#[derive(Debug)]
+pub(crate) struct AgentAsyncClientUDP {
+    #[cfg(all(not(feature = "async-std"), not(feature = "tokio")))]
+    conn: UdpSocket,
+    #[cfg(feature = "tokio")]
+    conn: tokio::sync::Mutex<tokio::net::UdpSocket>,
+    #[cfg(all(feature = "async-std", not(feature = "tokio")))]
+    conn: async_std::sync::Mutex<async_std::net::UdpSocket>,
+    buffer_client: Mutex<BufferClient>,
+}
+
+impl AgentAsyncClientUDP {
     /// Create a new UDP agent client
-    pub(crate) fn new<T: ToSocketAddrs>(
-        host_port: T,
-        max_packet_size: Option<usize>,
-    ) -> thrift::Result<Self> {
-        let transport = TUdpChannel::new(host_port, max_packet_size.or(Some(UDP_MAX_PACKET_SIZE)))?;
-        let (read, write) = transport.split()?;
+    pub(crate) fn new<T: ToSocketAddrs>(host_port: T) -> thrift::Result<Self> {
+        let (buffer, write) = TBufferChannel::with_capacity(0, 512).split()?;
         let client = agent::AgentSyncClient::new(
-            protocol::TCompactInputProtocol::new(read),
-            protocol::TCompactOutputProtocol::new(write),
+            TCompactInputProtocol::new(TNoopChannel),
+            TCompactOutputProtocol::new(write),
         );
 
-        Ok(AgentSyncClientUDP { client })
-    }
-}
+        let conn = UdpSocket::bind("0.0.0.0:0")?;
+        conn.connect(host_port)?;
 
-impl agent::TAgentSyncClient for AgentSyncClientUDP {
-    /// Emit zipkin batch (Deprecated)
-    fn emit_zipkin_batch(&mut self, spans: Vec<zipkincore::Span>) -> thrift::Result<()> {
-        self.client.emit_zipkin_batch(spans)
+        Ok(AgentAsyncClientUDP {
+            #[cfg(all(not(feature = "async-std"), not(feature = "tokio")))]
+            conn,
+            #[cfg(feature = "tokio")]
+            conn: tokio::sync::Mutex::new(tokio::net::UdpSocket::from_std(conn)?),
+            #[cfg(all(feature = "async-std", not(feature = "tokio")))]
+            conn: async_std::sync::Mutex::new(async_std::net::UdpSocket::from(conn)),
+            buffer_client: Mutex::new(BufferClient { buffer, client }),
+        })
     }
 
     /// Emit standard Jaeger batch
-    fn emit_batch(&mut self, batch: jaeger::Batch) -> thrift::Result<()> {
-        self.client.emit_batch(batch)
+    pub(crate) async fn emit_batch(&self, batch: jaeger::Batch) -> thrift::Result<()> {
+        // Write payload to buffer
+        let payload = self
+            .buffer_client
+            .lock()
+            .map_err(|err| {
+                thrift::Error::from(std::io::Error::new(
+                    std::io::ErrorKind::Other,
+                    err.to_string(),
+                ))
+            })
+            .and_then(|mut buffer_client| {
+                // Write to tmp buffer
+                buffer_client.client.emit_batch(batch)?;
+                // extract written payload
+                let payload = buffer_client.buffer.write_bytes();
+                // reset
+                buffer_client.buffer.empty_write_buffer();
+
+                Ok(payload)
+            })?;
+
+        // Write async to socket, reading from buffer
+        write_to_socket(self, payload).await?;
+
+        Ok(())
     }
 }
+
+#[cfg(all(not(feature = "async-std"), not(feature = "tokio")))]
+async fn write_to_socket(client: &AgentAsyncClientUDP, payload: Vec<u8>) -> thrift::Result<()> {
+    client.conn.send(&payload)?;
+
+    Ok(())
+}
+
+#[cfg(feature = "tokio")]
+async fn write_to_socket(client: &AgentAsyncClientUDP, payload: Vec<u8>) -> thrift::Result<()> {
+    let mut conn = client.conn.lock().await;
+    conn.send(&payload).await?;
+
+    Ok(())
+}
+
+#[cfg(all(feature = "async-std", not(feature = "tokio")))]
+async fn write_to_socket(client: &AgentAsyncClientUDP, payload: Vec<u8>) -> thrift::Result<()> {
+    let conn = client.conn.lock().await;
+    conn.send(&payload).await?;
+
+    Ok(())
+}