feat(traces): migrate trace propagation to dd-trace-rs

bottlecap/Cargo.toml

@@ -76,6 +76,7 @@ libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog", rev = "c812
 libdd-trace-normalization = { git = "https://github.com/DataDog/libdatadog",  rev = "c8121f422d2c8d219f8d421ff3cdb1fcbe9e8b09" }
 libdd-trace-obfuscation = { git = "https://github.com/DataDog/libdatadog", rev = "c8121f422d2c8d219f8d421ff3cdb1fcbe9e8b09"  }
 libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog", rev = "c8121f422d2c8d219f8d421ff3cdb1fcbe9e8b09"  }
+datadog-opentelemetry = { git = "https://github.com/DataDog/dd-trace-rs", rev = "f51cefc4ad24bec81b38fb2f36b1ed93f21ae913", default-features = false }
 dogstatsd = { git = "https://github.com/DataDog/serverless-components", rev = "28f796bf767fff56caf08153ade5cd80c8e8f705", default-features = false }
 datadog-fips = { git = "https://github.com/DataDog/serverless-components", rev = "28f796bf767fff56caf08153ade5cd80c8e8f705", default-features = false }
 libddwaf = { version = "1.28.1", git = "https://github.com/DataDog/libddwaf-rust", rev = "d1534a158d976bd4f747bf9fcc58e0712d2d17fc", default-features = false, features = ["serde"] }

bottlecap/src/config/env.rs

@@ -24,10 +24,11 @@ use crate::{
         },
         processing_rule::{ProcessingRule, deserialize_processing_rules},
         service_mapping::deserialize_service_mapping,
-        trace_propagation_style::{TracePropagationStyle, deserialize_trace_propagation_style},
+        trace_propagation_style::deserialize_trace_propagation_style,
     },
     merge_hashmap, merge_option, merge_option_to_value, merge_string, merge_vec,
 };
+use datadog_opentelemetry::propagation::TracePropagationStyle;
 
 #[derive(Debug, PartialEq, Deserialize, Clone, Default)]
 #[serde(default)]
@@ -719,8 +720,8 @@ mod tests {
         flush_strategy::{FlushStrategy, PeriodicStrategy},
         log_level::LogLevel,
         processing_rule::{Kind, ProcessingRule},
-        trace_propagation_style::TracePropagationStyle,
     };
+    use datadog_opentelemetry::propagation::TracePropagationStyle;
 
     #[test]
     #[allow(clippy::too_many_lines)]
@@ -802,7 +803,7 @@ mod tests {
             jail.set_env("DD_METRICS_CONFIG_COMPRESSION_LEVEL", "3");
             // Trace Propagation
             jail.set_env("DD_TRACE_PROPAGATION_STYLE", "datadog");
-            jail.set_env("DD_TRACE_PROPAGATION_STYLE_EXTRACT", "b3");
+            jail.set_env("DD_TRACE_PROPAGATION_STYLE_EXTRACT", "tracecontext");
             jail.set_env("DD_TRACE_PROPAGATION_EXTRACT_FIRST", "true");
             jail.set_env("DD_TRACE_PROPAGATION_HTTP_BAGGAGE_ENABLED", "true");
             jail.set_env("DD_TRACE_AWS_SERVICE_REPRESENTATION_ENABLED", "true");
@@ -984,7 +985,7 @@ mod tests {
                     "debug:^true$".to_string(),
                 ]),
                 trace_propagation_style: vec![TracePropagationStyle::Datadog],
-                trace_propagation_style_extract: vec![TracePropagationStyle::B3],
+                trace_propagation_style_extract: vec![TracePropagationStyle::TraceContext],
                 trace_propagation_extract_first: true,
                 trace_propagation_http_baggage_enabled: true,
                 trace_aws_service_representation_enabled: true,

bottlecap/src/config/mod.rs

@@ -29,9 +29,9 @@ use crate::config::{
     log_level::LogLevel,
     logs_additional_endpoints::LogsAdditionalEndpoint,
     processing_rule::{ProcessingRule, deserialize_processing_rules},
-    trace_propagation_style::TracePropagationStyle,
     yaml::YamlConfigSource,
 };
+use datadog_opentelemetry::propagation::TracePropagationStyle;
 
 /// Helper macro to merge Option<String> fields to String fields
 ///
@@ -488,6 +488,38 @@ impl Default for Config {
     }
 }
 
+impl datadog_opentelemetry::propagation::PropagationConfig for Config {
+    fn trace_propagation_style(&self) -> Option<&[TracePropagationStyle]> {
+        if self.trace_propagation_style.is_empty() {
+            None
+        } else {
+            Some(&self.trace_propagation_style)
+        }
+    }
+
+    fn trace_propagation_style_extract(&self) -> Option<&[TracePropagationStyle]> {
+        if self.trace_propagation_style_extract.is_empty() {
+            None
+        } else {
+            Some(&self.trace_propagation_style_extract)
+        }
+    }
+
+    fn trace_propagation_style_inject(&self) -> Option<&[TracePropagationStyle]> {
+        // Bottlecap does not configure injection styles separately
+        None
+    }
+
+    fn trace_propagation_extract_first(&self) -> bool {
+        self.trace_propagation_extract_first
+    }
+
+    fn datadog_tags_max_length(&self) -> usize {
+        // Default max length matching dd-trace-rs
+        512
+    }
+}
+
 #[allow(clippy::module_name_repetitions)]
 #[inline]
 #[must_use]
@@ -819,8 +851,8 @@ pub mod tests {
         flush_strategy::{FlushStrategy, PeriodicStrategy},
         log_level::LogLevel,
         processing_rule::ProcessingRule,
-        trace_propagation_style::TracePropagationStyle,
     };
+    use datadog_opentelemetry::propagation::TracePropagationStyle;
 
     #[test]
     fn test_default_logs_intake_url() {
@@ -1299,17 +1331,12 @@ pub mod tests {
     fn test_parse_trace_propagation_style() {
         figment::Jail::expect_with(|jail| {
             jail.clear_env();
-            jail.set_env(
-                "DD_TRACE_PROPAGATION_STYLE",
-                "datadog,tracecontext,b3,b3multi",
-            );
+            jail.set_env("DD_TRACE_PROPAGATION_STYLE", "datadog,tracecontext");
             let config = get_config(Path::new(""));
 
             let expected_styles = vec![
                 TracePropagationStyle::Datadog,
                 TracePropagationStyle::TraceContext,
-                TracePropagationStyle::B3,
-                TracePropagationStyle::B3Multi,
             ];
             assert_eq!(config.trace_propagation_style, expected_styles);
             assert_eq!(config.trace_propagation_style_extract, expected_styles);

bottlecap/src/config/trace_propagation_style.rs

@@ -1,48 +1,9 @@
-use std::{fmt::Display, str::FromStr};
+use std::str::FromStr;
 
+use datadog_opentelemetry::propagation::TracePropagationStyle;
 use serde::{Deserialize, Deserializer};
 use tracing::error;
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum TracePropagationStyle {
-    Datadog,
-    B3Multi,
-    B3,
-    TraceContext,
-    None,
-}
-
-impl FromStr for TracePropagationStyle {
-    type Err = String;
-
-    fn from_str(s: &str) -> Result<Self, Self::Err> {
-        match s.to_lowercase().as_str() {
-            "datadog" => Ok(TracePropagationStyle::Datadog),
-            "b3multi" => Ok(TracePropagationStyle::B3Multi),
-            "b3" => Ok(TracePropagationStyle::B3),
-            "tracecontext" => Ok(TracePropagationStyle::TraceContext),
-            "none" => Ok(TracePropagationStyle::None),
-            _ => {
-                error!("Trace propagation style is invalid: {:?}, using None", s);
-                Ok(TracePropagationStyle::None)
-            }
-        }
-    }
-}
-
-impl Display for TracePropagationStyle {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let style = match self {
-            TracePropagationStyle::Datadog => "datadog",
-            TracePropagationStyle::B3Multi => "b3multi",
-            TracePropagationStyle::B3 => "b3",
-            TracePropagationStyle::TraceContext => "tracecontext",
-            TracePropagationStyle::None => "none",
-        };
-        write!(f, "{style}")
-    }
-}
-
 #[allow(clippy::module_name_repetitions)]
 pub fn deserialize_trace_propagation_style<'de, D>(
     deserializer: D,
@@ -57,7 +18,7 @@ where
             |style| match TracePropagationStyle::from_str(style.trim()) {
                 Ok(parsed_style) => Some(parsed_style),
                 Err(e) => {
-                    tracing::error!("Failed to parse trace propagation style: {}, ignoring", e);
+                    error!("Failed to parse trace propagation style: {}, ignoring", e);
                     None
                 }
             },

bottlecap/src/config/yaml.rs

@@ -1,21 +1,20 @@
 use std::time::Duration;
 use std::{collections::HashMap, path::PathBuf};
 
+use datadog_opentelemetry::propagation::TracePropagationStyle;
+
 use crate::{
     config::{
         Config, ConfigError, ConfigSource, ProcessingRule,
-        additional_endpoints::deserialize_additional_endpoints,
-        deserialize_apm_replace_rules, deserialize_key_value_pair_array_to_hashmap,
-        deserialize_option_lossless, deserialize_optional_bool_from_anything,
-        deserialize_optional_duration_from_microseconds,
+        additional_endpoints::deserialize_additional_endpoints, deserialize_apm_replace_rules,
+        deserialize_key_value_pair_array_to_hashmap, deserialize_option_lossless,
+        deserialize_optional_bool_from_anything, deserialize_optional_duration_from_microseconds,
         deserialize_optional_duration_from_seconds,
         deserialize_optional_duration_from_seconds_ignore_zero, deserialize_optional_string,
-        deserialize_processing_rules, deserialize_string_or_int,
-        flush_strategy::FlushStrategy,
-        log_level::LogLevel,
-        logs_additional_endpoints::LogsAdditionalEndpoint,
+        deserialize_processing_rules, deserialize_string_or_int, flush_strategy::FlushStrategy,
+        log_level::LogLevel, logs_additional_endpoints::LogsAdditionalEndpoint,
         service_mapping::deserialize_service_mapping,
-        trace_propagation_style::{TracePropagationStyle, deserialize_trace_propagation_style},
+        trace_propagation_style::deserialize_trace_propagation_style,
     },
     merge_hashmap, merge_option, merge_option_to_value, merge_string, merge_vec,
 };
@@ -826,7 +825,7 @@ service_mapping: old-service:new-service
 
 # Trace Propagation
 trace_propagation_style: "datadog"
-trace_propagation_style_extract: "b3"
+trace_propagation_style_extract: "tracecontext"
 trace_propagation_extract_first: true
 trace_propagation_http_baggage_enabled: true
 trace_aws_service_representation_enabled: true
@@ -972,7 +971,7 @@ api_security_sample_delay: 60 # Seconds
                     ),
                 ]),
                 trace_propagation_style: vec![TracePropagationStyle::Datadog],
-                trace_propagation_style_extract: vec![TracePropagationStyle::B3],
+                trace_propagation_style_extract: vec![TracePropagationStyle::TraceContext],
                 trace_propagation_extract_first: true,
                 trace_propagation_http_baggage_enabled: true,
                 trace_aws_service_representation_enabled: true,

bottlecap/src/lifecycle/invocation/context.rs

@@ -1,7 +1,7 @@
 use crate::{
     lifecycle::invocation::processor::MS_TO_NS, metrics::enhanced::lambda::EnhancedMetricData,
-    traces::context::SpanContext,
 };
+use datadog_opentelemetry::propagation::context::SpanContext;
 use std::{
     collections::{HashMap, VecDeque},
     time::{SystemTime, UNIX_EPOCH},

bottlecap/src/lifecycle/invocation/processor.rs

@@ -5,6 +5,13 @@ use std::{
 };
 
 use chrono::{DateTime, Utc};
+use datadog_opentelemetry::propagation::{
+    context::SpanContext,
+    datadog::{
+        DATADOG_PARENT_ID_KEY, DATADOG_SAMPLING_PRIORITY_KEY, DATADOG_TAGS_KEY,
+        DATADOG_TRACE_ID_KEY,
+    },
+};
 use libdd_trace_protobuf::pb::Span;
 use libdd_trace_utils::tracer_header_tags;
 use serde_json::Value;
@@ -31,14 +38,7 @@ use crate::{
     },
     tags::{lambda::tags::resolve_runtime_from_proc, provider},
     traces::{
-        context::SpanContext,
-        propagation::{
-            DatadogCompositePropagator, Propagator,
-            text_map_propagator::{
-                DATADOG_PARENT_ID_KEY, DATADOG_SAMPLING_PRIORITY_KEY, DATADOG_SPAN_ID_KEY,
-                DATADOG_TRACE_ID_KEY, DatadogHeaderPropagator,
-            },
-        },
+        propagation::{DatadogCompositePropagator, carrier::JsonCarrier},
         trace_processor::SendingTraceProcessor,
     },
 };
@@ -52,6 +52,7 @@ pub const DATADOG_INVOCATION_ERROR_MESSAGE_KEY: &str = "x-datadog-invocation-err
 pub const DATADOG_INVOCATION_ERROR_TYPE_KEY: &str = "x-datadog-invocation-error-type";
 pub const DATADOG_INVOCATION_ERROR_STACK_KEY: &str = "x-datadog-invocation-error-stack";
 pub const DATADOG_INVOCATION_ERROR_KEY: &str = "x-datadog-invocation-error";
+const DATADOG_SPAN_ID_KEY: &str = "x-datadog-span-id";
 const TAG_SAMPLING_PRIORITY: &str = "_sampling_priority_v1";
 
 pub struct Processor {
@@ -979,7 +980,10 @@ impl Processor {
 
         // Set the extracted trace context to the spans
         if let Some(sc) = &context.extracted_span_context {
-            context.invocation_span.trace_id = sc.trace_id;
+            #[allow(clippy::cast_possible_truncation)] // Datadog protocol uses lower 64 bits
+            {
+                context.invocation_span.trace_id = sc.trace_id as u64;
+            }
             context.invocation_span.parent_id = sc.span_id;
 
             // Set the right data to the correct root level span,
@@ -1082,7 +1086,7 @@ impl Processor {
     pub fn extract_span_context(
         headers: &HashMap<String, String>,
         payload_value: &Value,
-        propagator: Arc<impl Propagator>,
+        propagator: Arc<DatadogCompositePropagator>,
     ) -> Option<SpanContext> {
         if let Some(sc) =
             span_inferrer::extract_span_context(payload_value, Arc::clone(&propagator))
@@ -1093,14 +1097,14 @@ impl Processor {
         if let Some(sc) = payload_value
             .get("request")
             .and_then(|req| req.get("headers"))
-            .and_then(|headers| propagator.extract(headers))
+            .and_then(|headers| propagator.extract(&JsonCarrier(headers)))
         {
             debug!("Extracted trace context from event.request.headers");
             return Some(sc);
         }
 
         if let Some(payload_headers) = payload_value.get("headers")
-            && let Some(sc) = propagator.extract(payload_headers)
+            && let Some(sc) = propagator.extract(&JsonCarrier(payload_headers))
         {
             debug!("Extracted trace context from event headers");
             return Some(sc);
@@ -1202,14 +1206,17 @@ impl Processor {
             self.inferrer.set_status_code(status_code_as_string);
         }
 
-        let mut trace_id = 0;
-        let mut parent_id = 0;
+        let mut trace_id: u64 = 0;
+        let mut parent_id: u64 = 0;
         let mut tags: HashMap<String, String> = HashMap::new();
 
         // If we have a trace context, this means we got it from
         // distributed tracing
         if let Some(sc) = &context.extracted_span_context {
-            trace_id = sc.trace_id;
+            #[allow(clippy::cast_possible_truncation)] // Datadog protocol uses lower 64 bits
+            {
+                trace_id = sc.trace_id as u64;
+            }
             parent_id = sc.span_id;
             tags.extend(sc.tags.clone());
         }
@@ -1235,9 +1242,9 @@ impl Processor {
                     .insert(TAG_SAMPLING_PRIORITY.to_string(), priority);
             }
 
-            // Extract tags from headers
-            // Used for 128 bit trace ids
-            tags = DatadogHeaderPropagator::extract_tags(&headers);
+            // Extract _dd.p.* propagation tags from x-datadog-tags header
+            let carrier_tags = headers.get(DATADOG_TAGS_KEY).map_or("", String::as_str);
+            tags = crate::traces::propagation::extract_propagation_tags(carrier_tags);
         }
 
         // We should always use the generated span id from the tracer
@@ -2048,8 +2055,8 @@ mod tests {
     fn test_extract_span_context_priority_order() {
         let config = Arc::new(config::Config {
             trace_propagation_style_extract: vec![
-                config::trace_propagation_style::TracePropagationStyle::Datadog,
-                config::trace_propagation_style::TracePropagationStyle::TraceContext,
+                datadog_opentelemetry::propagation::TracePropagationStyle::Datadog,
+                datadog_opentelemetry::propagation::TracePropagationStyle::TraceContext,
             ],
             ..config::Config::default()
         });
@@ -2086,8 +2093,8 @@ mod tests {
     fn test_extract_span_context_no_request_headers() {
         let config = Arc::new(config::Config {
             trace_propagation_style_extract: vec![
-                config::trace_propagation_style::TracePropagationStyle::Datadog,
-                config::trace_propagation_style::TracePropagationStyle::TraceContext,
+                datadog_opentelemetry::propagation::TracePropagationStyle::Datadog,
+                datadog_opentelemetry::propagation::TracePropagationStyle::TraceContext,
             ],
             ..config::Config::default()
         });

bottlecap/src/lifecycle/invocation/processor_service.rs

@@ -1,6 +1,7 @@
 use std::{collections::HashMap, sync::Arc};
 
 use chrono::{DateTime, Utc};
+use datadog_opentelemetry::propagation::context::SpanContext;
 use dogstatsd::aggregator::AggregatorHandle;
 use libdd_trace_protobuf::pb::Span;
 use serde_json::Value;
@@ -18,10 +19,7 @@ use crate::{
         processor::Processor,
     },
     tags::provider,
-    traces::{
-        context::SpanContext, propagation::DatadogCompositePropagator,
-        trace_processor::SendingTraceProcessor,
-    },
+    traces::{propagation::DatadogCompositePropagator, trace_processor::SendingTraceProcessor},
 };
 
 #[derive(Error, Debug)]