feat: [WIP] Implement datadog v0.7 traces api

opentelemetry-datadog/benches/datadog_exporter.rs

@@ -195,10 +195,10 @@ fn generate_traces(number_of_traces: usize, spans_per_trace: usize) -> Vec<SpanD
     result
 }
 
-fn criterion_benchmark(c: &mut Criterion) {
+fn bench_export(c: &mut Criterion, api_version: ApiVersion) {
     let exporter = new_pipeline()
         .with_service_name("trace-demo")
-        .with_api_version(ApiVersion::Version05)
+        .with_api_version(api_version)
         .with_http_client(DummyClient)
         .build_exporter()
         .unwrap();
@@ -210,11 +210,22 @@ fn criterion_benchmark(c: &mut Criterion) {
         let data_ref = &data;
 
         c.bench_function(
-            format!("export {number_of_traces} traces with {spans_per_trace} spans").as_str(),
+            format!(
+                "export {number_of_traces} traces with {spans_per_trace} spans for {api_version:?}"
+            )
+            .as_str(),
             |b| b.iter(|| exporter.export(black_box(data_ref.clone()))),
         );
     }
 }
 
-criterion_group!(benches, criterion_benchmark);
+fn bench_export_v05(c: &mut Criterion) {
+    bench_export(c, ApiVersion::Version05);
+}
+
+fn bench_export_v07(c: &mut Criterion) {
+    bench_export(c, ApiVersion::Version07);
+}
+
+criterion_group!(benches, bench_export_v05, bench_export_v07);
 criterion_main!(benches);

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

@@ -1,5 +1,6 @@
-use crate::exporter::ModelConfig;
+use crate::{exporter::ModelConfig, DatadogTraceState};
 use http::uri;
+use opentelemetry::Value;
 use opentelemetry_sdk::{
     trace::{self, SpanData},
     ExportError, Resource,
@@ -14,15 +15,38 @@ use super::Mapping;
 pub mod unified_tags;
 mod v03;
 mod v05;
+mod v07;
 
 // todo: we should follow the same mapping defined in https://github.com/DataDog/datadog-agent/blob/main/pkg/trace/api/otlp.go
 
 // https://github.com/DataDog/dd-trace-js/blob/c89a35f7d27beb4a60165409376e170eacb194c5/packages/dd-trace/src/constants.js#L4
 static SAMPLING_PRIORITY_KEY: &str = "_sampling_priority_v1";
 
+#[cfg(not(feature = "agent-sampling"))]
+fn get_sampling_priority(_span: &SpanData) -> f64 {
+    1.0
+}
+
+#[cfg(feature = "agent-sampling")]
+fn get_sampling_priority(span: &SpanData) -> f64 {
+    if span.span_context.trace_state().priority_sampling_enabled() {
+        1.0
+    } else {
+        0.0
+    }
+}
+
 // https://github.com/DataDog/datadog-agent/blob/ec96f3c24173ec66ba235bda7710504400d9a000/pkg/trace/traceutil/span.go#L20
 static DD_MEASURED_KEY: &str = "_dd.measured";
 
+fn get_measuring(span: &SpanData) -> f64 {
+    if span.span_context.trace_state().measuring_enabled() {
+        1.0
+    } else {
+        0.0
+    }
+}
+
 /// Custom mapping between opentelemetry spans and datadog spans.
 ///
 /// User can provide custom function to change the mapping. It currently supports customizing the following
@@ -77,6 +101,16 @@ fn default_resource_mapping<'a>(span: &'a SpanData, _config: &'a ModelConfig) ->
     span.name.as_ref()
 }
 
+fn get_span_type(span: &SpanData) -> Option<&Value> {
+    for kv in &span.attributes {
+        if kv.key.as_str() == "span.type" {
+            return Some(&kv.value);
+        }
+    }
+
+    None
+}
+
 /// Wrap type for errors from opentelemetry datadog exporter
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
@@ -129,20 +163,24 @@ pub enum ApiVersion {
     Version03,
     /// Version 0.5 - requires datadog-agent v7.22.0 or above
     Version05,
+    /// Version 0.7
+    Version07,
 }
 
 impl ApiVersion {
     pub(crate) fn path(self) -> &'static str {
         match self {
             ApiVersion::Version03 => "/v0.3/traces",
             ApiVersion::Version05 => "/v0.5/traces",
+            ApiVersion::Version07 => "/v0.7/traces",
         }
     }
 
     pub(crate) fn content_type(self) -> &'static str {
         match self {
             ApiVersion::Version03 => "application/msgpack",
             ApiVersion::Version05 => "application/msgpack",
+            ApiVersion::Version07 => "application/msgpack",
         }
     }
 
@@ -190,6 +228,24 @@ impl ApiVersion {
                 unified_tags,
                 resource,
             ),
+            Self::Version07 => v07::encode(
+                model_config,
+                traces,
+                |span, config| match &mapping.service_name {
+                    Some(f) => f(span, config),
+                    None => default_service_name_mapping(span, config),
+                },
+                |span, config| match &mapping.name {
+                    Some(f) => f(span, config),
+                    None => default_name_mapping(span, config),
+                },
+                |span, config| match &mapping.resource {
+                    Some(f) => f(span, config),
+                    None => default_resource_mapping(span, config),
+                },
+                unified_tags,
+                resource,
+            ),
         }
     }
 }
@@ -198,6 +254,7 @@ impl ApiVersion {
 pub(crate) mod tests {
     use super::*;
     use base64::{engine::general_purpose::STANDARD, Engine};
+    use opentelemetry::trace::Event;
     use opentelemetry::InstrumentationScope;
     use opentelemetry::{
         trace::{SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState},
@@ -213,7 +270,43 @@ pub(crate) mod tests {
         vec![vec![get_span(7, 1, 99)]]
     }
 
+    fn get_traces_with_events() -> Vec<Vec<trace::SpanData>> {
+        let event = Event::new(
+            "myevent",
+            SystemTime::UNIX_EPOCH
+                .checked_add(Duration::from_secs(5))
+                .unwrap(),
+            vec![
+                KeyValue::new("mykey", 1),
+                KeyValue::new(
+                    "myarray",
+                    Value::Array(opentelemetry::Array::String(vec![
+                        "myvalue1".into(),
+                        "myvalue2".into(),
+                    ])),
+                ),
+                KeyValue::new("mybool", true),
+                KeyValue::new("myint", 2.5),
+                KeyValue::new("myboolfalse", false),
+            ],
+            0,
+        );
+        let mut events = SpanEvents::default();
+        events.events.push(event);
+
+        vec![vec![get_span_with_events(7, 1, 99, events)]]
+    }
+
     pub(crate) fn get_span(trace_id: u128, parent_span_id: u64, span_id: u64) -> trace::SpanData {
+        get_span_with_events(trace_id, parent_span_id, span_id, SpanEvents::default())
+    }
+
+    pub(crate) fn get_span_with_events(
+        trace_id: u128,
+        parent_span_id: u64,
+        span_id: u64,
+        events: SpanEvents,
+    ) -> trace::SpanData {
         let span_context = SpanContext::new(
             TraceId::from_u128(trace_id),
             SpanId::from_u64(span_id),
@@ -226,7 +319,6 @@ pub(crate) mod tests {
         let end_time = start_time.checked_add(Duration::from_secs(1)).unwrap();
 
         let attributes = vec![KeyValue::new("span.type", "web")];
-        let events = SpanEvents::default();
         let links = SpanLinks::default();
         let instrumentation_scope = InstrumentationScope::builder("component").build();
 
@@ -305,4 +397,71 @@ pub(crate) mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_encode_v07() {
+        let traces = get_traces_with_events();
+        let model_config = ModelConfig {
+            service_name: "service_name".to_string(),
+            ..Default::default()
+        };
+
+        // we use an empty builder with a single attribute because the attributes are in a hashmap
+        // which causes the order to change every test
+        let resource = Resource::builder_empty()
+            .with_attribute(KeyValue::new("host.name", "test"))
+            .build();
+
+        let mut unified_tags = UnifiedTags::new();
+        unified_tags.set_env(Some(String::from("test-env")));
+        unified_tags.set_version(Some(String::from("test-version")));
+        unified_tags.set_service(Some(String::from("test-service")));
+
+        let encoded = STANDARD.encode(
+            ApiVersion::Version07
+                .encode(
+                    &model_config,
+                    traces.iter().map(|x| &x[..]).collect(),
+                    &Mapping::empty(),
+                    &unified_tags,
+                    Some(&resource),
+                )
+                .unwrap(),
+        );
+
+        // A very nice way to check the encoded values is to use
+        // https://github.com/DataDog/dd-apm-test-agent
+        // Which is a test http server that receives and validates sent traces
+        let expected = "ha1sYW5ndWFnZV9uYW1lpHJ1c3SmY2h1bmtzkYOocHJpb3JpdHnSAAAAAaZvcmlnaW6gpXNwY\
+        W5zkY6kbmFtZaljb21wb25lbnSnc3Bhbl9pZM8AAAAAAAAAY6h0cmFjZV9pZM8AAAAAAAAAB6VzdGFydNMAAAAAAAAAAKhk\
+        dXJhdGlvbtMAAAAAO5rKAKlwYXJlbnRfaWTPAAAAAAAAAAGnc2VydmljZaxzZXJ2aWNlX25hbWWocmVzb3VyY2WocmVzb3V\
+        yY2WkdHlwZaN3ZWKlZXJyb3LSAAAAAKRtZXRhgqlob3N0Lm5hbWWkdGVzdKlzcGFuLnR5cGWjd2Vip21ldHJpY3OCtV9zYW\
+        1wbGluZ19wcmlvcml0eV92Mcs/8AAAAAAAAKxfZGQubWVhc3VyZWTLAAAAAAAAAACqc3Bhbl9saW5rc5Crc3Bhbl9ldmVud\
+        HORg6RuYW1lp215ZXZlbnSudGltZV91bml4X25hbm/TAAAAASoF8gCqYXR0cmlidXRlc4WlbXlrZXmCpHR5cGXSAAAAAqlp\
+        bnRfdmFsdWXTAAAAAAAAAAGnbXlhcnJheYKkdHlwZdIAAAAEq2FycmF5X3ZhbHVlkoKkdHlwZQCsc3RyaW5nX3ZhbHVlqG1\
+        5dmFsdWUxgqR0eXBlAKxzdHJpbmdfdmFsdWWobXl2YWx1ZTKmbXlib29sgqR0eXBl0gAAAAGqYm9vbF92YWx1ZcOlbXlpbn\
+        SCpHR5cGXSAAAAA6xkb3VibGVfdmFsdWXLQAQAAAAAAACrbXlib29sZmFsc2WCpHR5cGXSAAAAAapib29sX3ZhbHVlwqR0Y\
+        Wdzg6dzZXJ2aWNlrHRlc3Qtc2VydmljZad2ZXJzaW9urHRlc3QtdmVyc2lvbqNlbnaodGVzdC1lbnajZW52qHRlc3QtZW52\
+        q2FwcF92ZXJzaW9urHRlc3QtdmVyc2lvbg==";
+        assert_eq!(encoded.as_str(), expected);
+
+        // change to a different resource and make sure the encoded value changes and that we actually encode stuff
+        let other_resource = Resource::builder_empty()
+            .with_attribute(KeyValue::new("host.name", "thisissometingelse"))
+            .build();
+
+        let encoded = STANDARD.encode(
+            ApiVersion::Version07
+                .encode(
+                    &model_config,
+                    traces.iter().map(|x| &x[..]).collect(),
+                    &Mapping::empty(),
+                    &unified_tags,
+                    Some(&other_resource),
+                )
+                .unwrap(),
+        );
+
+        assert_ne!(encoded.as_str(), expected);
+    }
 }

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

@@ -1,13 +1,13 @@
 use crate::exporter::intern::StringInterner;
 use crate::exporter::model::{DD_MEASURED_KEY, SAMPLING_PRIORITY_KEY};
 use crate::exporter::{Error, ModelConfig};
-use crate::propagator::DatadogTraceState;
 use opentelemetry::trace::Status;
 use opentelemetry_sdk::trace::SpanData;
 use opentelemetry_sdk::Resource;
 use std::time::SystemTime;
 
 use super::unified_tags::{UnifiedTagField, UnifiedTags};
+use super::{get_measuring, get_sampling_priority, get_span_type};
 
 const SPAN_NUM_ELEMENTS: u32 = 12;
 const METRICS_LEN: u32 = 2;
@@ -127,28 +127,6 @@ fn write_unified_tag<'a>(
     Ok(())
 }
 
-#[cfg(not(feature = "agent-sampling"))]
-fn get_sampling_priority(_span: &SpanData) -> f64 {
-    1.0
-}
-
-#[cfg(feature = "agent-sampling")]
-fn get_sampling_priority(span: &SpanData) -> f64 {
-    if span.span_context.trace_state().priority_sampling_enabled() {
-        1.0
-    } else {
-        0.0
-    }
-}
-
-fn get_measuring(span: &SpanData) -> f64 {
-    if span.span_context.trace_state().measuring_enabled() {
-        1.0
-    } else {
-        0.0
-    }
-}
-
 #[allow(clippy::too_many_arguments)]
 fn encode_traces<'interner, S, N, R>(
     interner: &mut StringInterner<'interner>,
@@ -186,11 +164,8 @@ where
                 .unwrap_or(0);
 
             let mut span_type = interner.intern("");
-            for kv in &span.attributes {
-                if kv.key.as_str() == "span.type" {
-                    span_type = interner.intern_value(&kv.value);
-                    break;
-                }
+            if let Some(value) = get_span_type(span) {
+                span_type = interner.intern_value(value);
             }
 
             // Datadog span name is OpenTelemetry component name - see module docs for more information