Add datadog exporter (#216) (#222)

Author: tustvold

Cargo.toml

@@ -61,6 +61,7 @@ members = [
     "examples/aws-xray",
     "examples/basic",
     "examples/basic-otlp",
+    "examples/datadog",
     "examples/grpc",
     "examples/http",
     "examples/zipkin",

benches/ddsketch.rs

@@ -1,12 +1,8 @@
-use criterion::{
-    criterion_group, criterion_main, Criterion
-};
+use criterion::{criterion_group, criterion_main, Criterion};
 use opentelemetry::api::metrics::{InstrumentKind, Number, NumberKind};
 use opentelemetry::sdk::export::metrics::Aggregator;
 use opentelemetry::{
-    api::{
-        metrics::Descriptor,
-    },
+    api::metrics::Descriptor,
     sdk::{
         export::metrics::Quantile,
         metrics::aggregators::{ArrayAggregator, DDSKetchAggregator},
@@ -40,7 +36,9 @@ fn ddsketch(data: Vec<f64>) {
     }
     let new_aggregator: Arc<(dyn Aggregator + Send + Sync)> =
         Arc::new(DDSKetchAggregator::new(0.001, 2048, 1e-9, NumberKind::F64));
-    aggregator.synchronized_move(&new_aggregator, &descriptor).unwrap();
+    aggregator
+        .synchronized_move(&new_aggregator, &descriptor)
+        .unwrap();
     for quantile in get_test_quantile() {
         if let Some(new_aggregator) = new_aggregator.as_any().downcast_ref::<DDSKetchAggregator>() {
             let _ = new_aggregator.quantile(*quantile);
@@ -60,7 +58,9 @@ fn array(data: Vec<f64>) {
         aggregator.update(&Number::from(f), &descriptor).unwrap();
     }
     let new_aggregator: Arc<(dyn Aggregator + Send + Sync)> = Arc::new(ArrayAggregator::default());
-    aggregator.synchronized_move(&new_aggregator, &descriptor).unwrap();
+    aggregator
+        .synchronized_move(&new_aggregator, &descriptor)
+        .unwrap();
     for quantile in get_test_quantile() {
         if let Some(new_aggregator) = new_aggregator.as_any().downcast_ref::<ArrayAggregator>() {
             let _ = new_aggregator.quantile(*quantile);

examples/datadog/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "datadog"
+version = "0.1.0"
+edition = "2018"
+
+[dependencies]
+opentelemetry = { path = "../../" }
+opentelemetry-contrib = { path = "../../opentelemetry-contrib", features = ["datadog"] }

examples/datadog/README.md

@@ -0,0 +1,15 @@
+# Datadog Exporter Example
+
+Sends spans to a datadog-agent collector.
+
+## Usage 
+
+First run version 7.22.0 or above of the datadog-agent locally as described [here](https://docs.datadoghq.com/agent/)
+
+Then run the example to report spans:
+
+```shell
+$ cargo run
+```
+
+Traces should appear in the datadog APM dashboard

examples/datadog/src/main.rs

@@ -0,0 +1,35 @@
+use opentelemetry::api::{Key, Span, TraceContextExt, Tracer};
+use opentelemetry::global;
+use std::thread;
+use std::time::Duration;
+use opentelemetry_contrib::datadog::ApiVersion;
+
+fn bar() {
+    let tracer = global::tracer("component-bar");
+    let span = tracer.start("bar");
+    span.set_attribute(Key::new("span.type").string("sql"));
+    span.set_attribute(Key::new("sql.query").string("SELECT * FROM table"));
+    thread::sleep(Duration::from_millis(6));
+    span.end()
+}
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let tracer = opentelemetry_contrib::datadog::new_pipeline()
+        .with_service_name("trace-demo")
+        .with_version(ApiVersion::Version05)
+        .install()?;
+
+    tracer.in_span("foo", |cx| {
+        let span = cx.span();
+        span.set_attribute(Key::new("span.type").string("web"));
+        span.set_attribute(Key::new("http.url").string("http://localhost:8080/foo"));
+        span.set_attribute(Key::new("http.method").string("GET"));
+        span.set_attribute(Key::new("http.status_code").i64(200));
+
+        thread::sleep(Duration::from_millis(6));
+        bar();
+        thread::sleep(Duration::from_millis(6));
+    });
+
+    Ok(())
+}

opentelemetry-contrib/Cargo.toml

@@ -16,6 +16,16 @@ edition = "2018"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
+[features]
+default = []
+datadog = ["indexmap", "reqwest", "rmp"]
+
 [dependencies]
+indexmap = { version = "1.6.0", optional = true }
 opentelemetry = { version = "0.8.0", path = ".." }
+reqwest = { version = "0.10", features = ["blocking"], optional = true }
+rmp = { version = "0.8", optional = true }
 lazy_static = "1.4"
+
+[dev-dependencies]
+base64 = "0.12.3"

opentelemetry-contrib/src/datadog/intern.rs

@@ -0,0 +1,53 @@
+use indexmap::set::IndexSet;
+
+pub(crate) struct StringInterner {
+    data: IndexSet<String>,
+}
+
+impl StringInterner {
+    pub(crate) fn new() -> StringInterner {
+        StringInterner {
+            data: Default::default(),
+        }
+    }
+
+    pub(crate) fn intern(&mut self, data: &str) -> u32 {
+        if let Some(idx) = self.data.get_index_of(data) {
+            return idx as u32;
+        }
+        self.data.insert_full(data.to_string()).0 as u32
+    }
+
+    pub(crate) fn iter(&self) -> impl Iterator<Item = &String> {
+        self.data.iter()
+    }
+
+    pub(crate) fn len(&self) -> u32 {
+        self.data.len() as u32
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_intern() {
+        let a = "a".to_string();
+        let b = "b";
+        let c = "c";
+
+        let mut intern = StringInterner::new();
+        let a_idx = intern.intern(a.as_str());
+        let b_idx = intern.intern(b);
+        let c_idx = intern.intern(c);
+        let d_idx = intern.intern(a.as_str());
+        let e_idx = intern.intern(c);
+
+        assert_eq!(a_idx, 0);
+        assert_eq!(b_idx, 1);
+        assert_eq!(c_idx, 2);
+        assert_eq!(d_idx, a_idx);
+        assert_eq!(e_idx, c_idx);
+    }
+}

opentelemetry-contrib/src/datadog/mod.rs

@@ -0,0 +1,213 @@
+//! # OpenTelemetry Datadog Exporter
+//!
+//! An OpenTelemetry exporter implementation
+//!
+//! See the [Datadog Docs](https://docs.datadoghq.com/agent/) for information on how to run the datadog-agent
+//!
+//! ## Quirks
+//!
+//! There are currently some incompatibilities between Datadog and OpenTelemetry, and this manifests
+//! as minor quirks to this exporter.
+//!
+//! Firstly Datadog uses operation_name to describe what OpenTracing would call a component.
+//! Or to put it another way, in OpenTracing the operation / span name's are relatively
+//! granular and might be used to identify a specific endpoint. In datadog, however, they
+//! are less granular - it is expected in Datadog that a service will have single
+//! primary span name that is the root of all traces within that service, with an additional piece of
+//! metadata called resource_name providing granularity - https://docs.datadoghq.com/tracing/guide/configuring-primary-operation/
+//!
+//! The Datadog Golang API takes the approach of using a `resource.name` OpenTelemetry attribute to set the
+//! resource_name - https://github.com/DataDog/dd-trace-go/blob/ecb0b805ef25b00888a2fb62d465a5aa95e7301e/ddtrace/opentracer/tracer.go#L10
+//!
+//! Unfortunately, this breaks compatibility with other OpenTelemetry exporters which expect
+//! a more granular operation name - as per the OpenTracing specification.
+//!
+//! This exporter therefore takes a different approach of naming the span with the name of the
+//! tracing provider, and using the span name to set the resource_name. This should in most cases
+//! lead to the behaviour that users expect.
+//!
+//! Datadog additionally has a span_type string that alters the rendering of the spans in the web UI.
+//! This can be set as the `span.type` OpenTelemetry span attribute.
+//!
+//! For standard values see here - https://github.com/DataDog/dd-trace-go/blob/ecb0b805ef25b00888a2fb62d465a5aa95e7301e/ddtrace/ext/app_types.go#L31
+//!
+//! ## Performance
+//!
+//! For optimal performance, a batch exporter is recommended as the simple
+//! exporter will export each span synchronously on drop. You can enable the
+//! [`tokio`] or [`async-std`] features to have a batch exporter configured for
+//! you automatically for either executor when you install the pipeline.
+//!
+//! ```toml
+//! [dependencies]
+//! opentelemetry = { version = "*", features = ["tokio"] }
+//! opentelemetry-datadog = "*"
+//! ```
+//!
+//! [`tokio`]: https://tokio.rs
+//! [`async-std`]: https://async.rs
+//!
+//! ## Kitchen Sink Full Configuration
+//!
+//! Example showing how to override all configuration options. See the
+//! [`DatadogPipelineBuilder`] docs for details of each option.
+//!
+//! [`DatadogPipelineBuilder`]: struct.DatadogPipelineBuilder.html
+//!
+//! ```no_run
+//! use opentelemetry::api::{KeyValue, Tracer};
+//! use opentelemetry::sdk::{trace, IdGenerator, Resource, Sampler};
+//!
+//! fn main() -> Result<(), Box<dyn std::error::Error>> {
+//!     let tracer = opentelemetry_contrib::datadog::new_pipeline()
+//!         .with_service_name("my_app")
+//!         .with_version(opentelemetry_contrib::datadog::ApiVersion::Version05)
+//!         .with_agent_endpoint("http://localhost:8126")
+//!         .with_trace_config(
+//!             trace::config()
+//!                 .with_default_sampler(Sampler::AlwaysOn)
+//!                 .with_id_generator(IdGenerator::default())
+//!         )
+//!         .install()?;
+//!
+//!     tracer.in_span("doing_work", |cx| {
+//!         // Traced app logic here...
+//!     });
+//!
+//!     Ok(())
+//! }
+//! ```
+#![deny(missing_docs, unreachable_pub, missing_debug_implementations)]
+#![cfg_attr(test, deny(warnings))]
+
+mod intern;
+mod model;
+
+pub use model::ApiVersion;
+
+use opentelemetry::{api::TracerProvider, exporter::trace, global, sdk};
+use reqwest::header::CONTENT_TYPE;
+use reqwest::Url;
+use std::error::Error;
+use std::sync::Arc;
+
+/// Default Datadog collector endpoint
+const DEFAULT_AGENT_ENDPOINT: &str = "http://127.0.0.1:8126";
+
+/// Default service name if no service is configured.
+const DEFAULT_SERVICE_NAME: &str = "OpenTelemetry";
+
+/// Datadog span exporter
+#[derive(Debug)]
+pub struct DatadogExporter {
+    client: reqwest::blocking::Client,
+    request_url: Url,
+    service_name: String,
+    version: ApiVersion,
+}
+
+impl DatadogExporter {
+    fn new(service_name: String, agent_endpoint: Url, version: ApiVersion) -> Self {
+        let mut request_url = agent_endpoint;
+        request_url.set_path(version.path());
+
+        DatadogExporter {
+            client: reqwest::blocking::Client::new(),
+            request_url,
+            service_name,
+            version,
+        }
+    }
+}
+
+/// Create a new Datadog exporter pipeline builder.
+pub fn new_pipeline() -> DatadogPipelineBuilder {
+    DatadogPipelineBuilder::default()
+}
+
+/// Builder for `ExporterConfig` struct.
+#[derive(Debug)]
+pub struct DatadogPipelineBuilder {
+    service_name: String,
+    agent_endpoint: String,
+    trace_config: Option<sdk::Config>,
+    version: ApiVersion,
+}
+
+impl Default for DatadogPipelineBuilder {
+    fn default() -> Self {
+        DatadogPipelineBuilder {
+            service_name: DEFAULT_SERVICE_NAME.to_string(),
+            agent_endpoint: DEFAULT_AGENT_ENDPOINT.to_string(),
+            trace_config: None,
+            version: ApiVersion::Version05,
+        }
+    }
+}
+
+impl DatadogPipelineBuilder {
+    /// Create `ExporterConfig` struct from current `ExporterConfigBuilder`
+    pub fn install(mut self) -> Result<sdk::Tracer, Box<dyn Error>> {
+        let exporter = DatadogExporter::new(
+            self.service_name.clone(),
+            self.agent_endpoint.parse()?,
+            self.version,
+        );
+
+        let mut provider_builder = sdk::TracerProvider::builder().with_exporter(exporter);
+        if let Some(config) = self.trace_config.take() {
+            provider_builder = provider_builder.with_config(config);
+        }
+        let provider = provider_builder.build();
+        let tracer = provider.get_tracer("opentelemetry-datadog", Some(env!("CARGO_PKG_VERSION")));
+        global::set_provider(provider);
+
+        Ok(tracer)
+    }
+
+    /// Assign the service name under which to group traces
+    pub fn with_service_name<T: Into<String>>(mut self, name: T) -> Self {
+        self.service_name = name.into();
+        self
+    }
+
+    /// Assign the Datadog collector endpoint
+    pub fn with_agent_endpoint<T: Into<String>>(mut self, endpoint: T) -> Self {
+        self.agent_endpoint = endpoint.into();
+        self
+    }
+
+    /// Assign the SDK trace configuration
+    pub fn with_trace_config(mut self, config: sdk::Config) -> Self {
+        self.trace_config = Some(config);
+        self
+    }
+
+    /// Set version of Datadog trace ingestion API
+    pub fn with_version(mut self, version: ApiVersion) -> Self {
+        self.version = version;
+        self
+    }
+}
+
+impl trace::SpanExporter for DatadogExporter {
+    /// Export spans to datadog-agent
+    fn export(&self, batch: Vec<Arc<trace::SpanData>>) -> trace::ExportResult {
+        let data = match self.version.encode(&self.service_name, batch) {
+            Ok(data) => data,
+            Err(_) => return trace::ExportResult::FailedNotRetryable,
+        };
+
+        let resp = self
+            .client
+            .post(self.request_url.clone())
+            .header(CONTENT_TYPE, self.version.content_type())
+            .body(data)
+            .send();
+
+        match resp {
+            Ok(response) if response.status().is_success() => trace::ExportResult::Success,
+            _ => trace::ExportResult::FailedRetryable,
+        }
+    }
+}