Introduce TraceData to unify text and binary data

Also move Span structures to v04, to make space for v1 spans (which will eventually become the new default).
TraceData is also going to be used in the V1 implementation, to carry around byte arrays and strings alike, separate from the indexed offsets into the big vector.

Signed-off-by: Bob Weinand <[email protected]>

Author: bwoebi

data-pipeline-ffi/src/trace_exporter.rs

@@ -544,7 +544,7 @@ pub unsafe extern "C" fn ddog_trace_exporter_send(
 mod tests {
     use super::*;
     use crate::error::ddog_trace_exporter_error_free;
-    use datadog_trace_utils::span::SpanSlice;
+    use datadog_trace_utils::span::v04::SpanSlice;
     use httpmock::prelude::*;
     use httpmock::MockServer;
     use std::{borrow::Borrow, mem::MaybeUninit};

data-pipeline/benches/span_concentrator_bench.rs

@@ -7,7 +7,7 @@ use std::{
 
 use criterion::{criterion_group, Criterion};
 use data_pipeline::span_concentrator::SpanConcentrator;
-use datadog_trace_utils::span::SpanBytes;
+use datadog_trace_utils::span::v04::SpanBytes;
 
 fn get_bucket_start(now: SystemTime, n: u64) -> i64 {
     let start = now.duration_since(time::UNIX_EPOCH).unwrap() + Duration::from_secs(10 * n);

data-pipeline/src/span_concentrator/aggregation.rs

@@ -4,10 +4,10 @@
 //! This includes the aggregation key to group spans together and the computation of stats from a
 //! span.
 use datadog_trace_protobuf::pb;
-use datadog_trace_utils::span::trace_utils;
-use datadog_trace_utils::span::Span;
-use datadog_trace_utils::span::SpanText;
+use datadog_trace_utils::span::v04::Span;
+use datadog_trace_utils::span::{trace_utils, SpanText, TraceData};
 use hashbrown::HashMap;
+use std::borrow::Borrow;
 
 const TAG_STATUS_CODE: &str = "http.status_code";
 const TAG_SYNTHETICS: &str = "synthetics";
@@ -101,10 +101,7 @@ impl<'a> BorrowedAggregationKey<'a> {
     ///
     /// If `peer_tags_keys` is not empty then the peer tags of the span will be included in the
     /// key.
-    pub(super) fn from_span<T>(span: &'a Span<T>, peer_tag_keys: &'a [String]) -> Self
-    where
-        T: SpanText,
-    {
+    pub(super) fn from_span<T: TraceData>(span: &'a Span<T>, peer_tag_keys: &'a [String]) -> Self {
         let span_kind = span
             .meta
             .get(TAG_SPANKIND)
@@ -176,7 +173,7 @@ impl From<pb::ClientGroupedStats> for OwnedAggregationKey {
 /// Return the status code of a span based on the metrics and meta tags.
 fn get_status_code<T>(span: &Span<T>) -> u32
 where
-    T: SpanText,
+    T: TraceData,
 {
     if let Some(status_code) = span.metrics.get(TAG_STATUS_CODE) {
         *status_code as u32
@@ -205,7 +202,7 @@ fn get_peer_tags<'k, 'v, T>(
     peer_tag_keys: &'k [String],
 ) -> Vec<(&'k str, &'v str)>
 where
-    T: SpanText,
+    T: TraceData,
 {
     peer_tag_keys
         .iter()
@@ -228,7 +225,7 @@ impl GroupedStats {
     /// Update the stats of a GroupedStats by inserting a span.
     fn insert<T>(&mut self, value: &Span<T>)
     where
-        T: SpanText,
+        T: TraceData,
     {
         self.hits += 1;
         self.duration += value.duration as u64;
@@ -266,7 +263,7 @@ impl StatsBucket {
     /// not exist it creates it.
     pub(super) fn insert<T>(&mut self, key: BorrowedAggregationKey<'_>, value: &Span<T>)
     where
-        T: SpanText,
+        T: TraceData,
     {
         self.data.entry_ref(&key).or_default().insert(value);
     }
@@ -326,7 +323,7 @@ fn encode_grouped_stats(key: OwnedAggregationKey, group: GroupedStats) -> pb::Cl
 
 #[cfg(test)]
 mod tests {
-    use datadog_trace_utils::span::{SpanBytes, SpanSlice};
+    use datadog_trace_utils::span::v04::{SpanBytes, SpanSlice};
 
     use super::*;
     use std::{collections::HashMap, hash::Hash};

data-pipeline/src/span_concentrator/mod.rs

@@ -1,11 +1,12 @@
 // Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 //! This module implements the SpanConcentrator used to aggregate spans into stats
+use std::borrow::Borrow;
 use std::collections::HashMap;
 use std::time::{self, Duration, SystemTime};
 
 use datadog_trace_protobuf::pb;
-use datadog_trace_utils::span::{trace_utils, Span, SpanText};
+use datadog_trace_utils::span::{trace_utils, v04::Span, TraceData};
 
 use aggregation::{BorrowedAggregationKey, StatsBucket};
 
@@ -27,7 +28,7 @@ fn align_timestamp(t: u64, bucket_size: u64) -> u64 {
 /// Return true if the span has a span.kind that is eligible for stats computation
 fn compute_stats_for_span_kind<T>(span: &Span<T>, span_kinds_stats_computed: &[String]) -> bool
 where
-    T: SpanText,
+    T: TraceData,
 {
     !span_kinds_stats_computed.is_empty()
         && span.meta.get("span.kind").is_some_and(|span_kind| {
@@ -38,7 +39,7 @@ where
 /// Return true if the span should be ignored for stats computation
 fn should_ignore_span<T>(span: &Span<T>, span_kinds_stats_computed: &[String]) -> bool
 where
-    T: SpanText,
+    T: TraceData,
 {
     !(trace_utils::has_top_level(span)
         || trace_utils::is_measured(span)
@@ -121,7 +122,7 @@ impl SpanConcentrator {
     /// computation.
     pub fn add_span<T>(&mut self, span: &Span<T>)
     where
-        T: SpanText,
+        T: TraceData,
     {
         // If the span is eligible for stats computation
         if !should_ignore_span(span, self.span_kinds_stats_computed.as_slice()) {

data-pipeline/src/span_concentrator/tests.rs

@@ -4,7 +4,7 @@
 use crate::span_concentrator::aggregation::OwnedAggregationKey;
 
 use super::*;
-use datadog_trace_utils::span::{trace_utils::compute_top_level_span, SpanSlice};
+use datadog_trace_utils::span::{trace_utils::compute_top_level_span, v04::SpanSlice};
 use rand::{thread_rng, Rng};
 
 const BUCKET_SIZE: u64 = Duration::from_secs(2).as_nanos() as u64;

data-pipeline/src/stats_exporter.rs

@@ -188,7 +188,7 @@ pub fn stats_url_from_agent_url(agent_url: &str) -> anyhow::Result<hyper::Uri> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use datadog_trace_utils::span::{trace_utils, SpanSlice};
+    use datadog_trace_utils::span::{trace_utils, v04::SpanSlice};
     use datadog_trace_utils::test_utils::poll_for_mock_hit;
     use httpmock::prelude::*;
     use httpmock::MockServer;

data-pipeline/src/trace_exporter/mod.rs

@@ -34,7 +34,7 @@ use datadog_trace_utils::msgpack_decoder;
 use datadog_trace_utils::send_with_retry::{
     send_with_retry, RetryStrategy, SendWithRetryError, SendWithRetryResult,
 };
-use datadog_trace_utils::span::{Span, SpanText};
+use datadog_trace_utils::span::{v04::Span, TraceData};
 use datadog_trace_utils::trace_utils::TracerHeaderTags;
 use ddcommon::MutexExt;
 use ddcommon::{hyper_migration, Endpoint};
@@ -577,7 +577,7 @@ impl TraceExporter {
     /// # Returns
     /// * Ok(String): The response from the agent
     /// * Err(TraceExporterError): An error detailing what went wrong in the process
-    pub fn send_trace_chunks<T: SpanText>(
+    pub fn send_trace_chunks<T: TraceData>(
         &self,
         trace_chunks: Vec<Vec<Span<T>>>,
     ) -> Result<AgentResponse, TraceExporterError> {
@@ -658,7 +658,7 @@ impl TraceExporter {
         self.handle_send_result(result, chunks, payload_len).await
     }
 
-    fn send_trace_chunks_inner<T: SpanText>(
+    fn send_trace_chunks_inner<T: TraceData>(
         &self,
         mut traces: Vec<Vec<Span<T>>>,
     ) -> Result<AgentResponse, TraceExporterError> {
@@ -982,8 +982,8 @@ mod tests {
     use self::error::AgentErrorKind;
     use super::*;
     use datadog_trace_utils::msgpack_encoder;
+    use datadog_trace_utils::span::v04::SpanBytes;
     use datadog_trace_utils::span::v05;
-    use datadog_trace_utils::span::SpanBytes;
     use httpmock::prelude::*;
     use httpmock::MockServer;
     use std::collections::HashMap;
@@ -1993,7 +1993,7 @@ mod single_threaded_tests {
     use super::*;
     use crate::agent_info;
     use datadog_trace_utils::msgpack_encoder;
-    use datadog_trace_utils::span::SpanBytes;
+    use datadog_trace_utils::span::v04::SpanBytes;
     use httpmock::prelude::*;
     use std::time::Duration;
     use tokio::time::sleep;

data-pipeline/src/trace_exporter/stats.rs

@@ -200,9 +200,9 @@ pub(crate) fn handle_stats_enabled(
 /// Add all spans from the given iterator into the stats concentrator
 /// # Panic
 /// Will panic if another thread panicked will holding the lock on `stats_concentrator`
-fn add_spans_to_stats<T: datadog_trace_utils::span::SpanText>(
+fn add_spans_to_stats<T: datadog_trace_utils::span::TraceData>(
     stats_concentrator: &Mutex<SpanConcentrator>,
-    traces: &[Vec<datadog_trace_utils::span::Span<T>>],
+    traces: &[Vec<datadog_trace_utils::span::v04::Span<T>>],
 ) {
     let mut stats_concentrator = stats_concentrator.lock_or_panic();
 
@@ -213,8 +213,8 @@ fn add_spans_to_stats<T: datadog_trace_utils::span::SpanText>(
 }
 
 /// Process traces for stats computation and update header tags accordingly
-pub(crate) fn process_traces_for_stats<T: datadog_trace_utils::span::SpanText>(
-    traces: &mut Vec<Vec<datadog_trace_utils::span::Span<T>>>,
+pub(crate) fn process_traces_for_stats<T: datadog_trace_utils::span::TraceData>(
+    traces: &mut Vec<Vec<datadog_trace_utils::span::v04::Span<T>>>,
     header_tags: &mut datadog_trace_utils::trace_utils::TracerHeaderTags,
     client_side_stats: &ArcSwap<StatsComputationStatus>,
     client_computed_top_level: bool,

data-pipeline/src/trace_exporter/trace_serializer.rs

@@ -8,7 +8,7 @@ use crate::trace_exporter::error::TraceExporterError;
 use crate::trace_exporter::TraceExporterOutputFormat;
 use datadog_trace_utils::msgpack_decoder::decode::error::DecodeError;
 use datadog_trace_utils::msgpack_encoder;
-use datadog_trace_utils::span::{Span, SpanText};
+use datadog_trace_utils::span::{v04::Span, TraceData};
 use datadog_trace_utils::trace_utils::{self, TracerHeaderTags};
 use datadog_trace_utils::tracer_payload;
 use ddcommon::header::{
@@ -46,7 +46,7 @@ impl<'a> TraceSerializer<'a> {
     }
 
     /// Prepare traces payload and HTTP headers for sending to agent
-    pub(super) fn prepare_traces_payload<T: SpanText>(
+    pub(super) fn prepare_traces_payload<T: TraceData>(
         &self,
         traces: Vec<Vec<Span<T>>>,
         header_tags: TracerHeaderTags,
@@ -64,7 +64,7 @@ impl<'a> TraceSerializer<'a> {
     }
 
     /// Collect trace chunks based on output format
-    fn collect_and_process_traces<T: SpanText>(
+    fn collect_and_process_traces<T: TraceData>(
         &self,
         traces: Vec<Vec<Span<T>>>,
     ) -> Result<tracer_payload::TraceChunks<T>, TraceExporterError> {
@@ -97,7 +97,7 @@ impl<'a> TraceSerializer<'a> {
     }
 
     /// Serialize payload to msgpack format
-    fn serialize_payload<T: SpanText>(
+    fn serialize_payload<T: TraceData>(
         &self,
         payload: &tracer_payload::TraceChunks<T>,
     ) -> Result<Vec<u8>, TraceExporterError> {
@@ -114,7 +114,7 @@ impl<'a> TraceSerializer<'a> {
 mod tests {
     use super::*;
     use crate::trace_exporter::agent_response::AgentResponsePayloadVersion;
-    use datadog_trace_utils::span::SpanBytes;
+    use datadog_trace_utils::span::v04::SpanBytes;
     use datadog_trace_utils::trace_utils::TracerHeaderTags;
     use ddcommon::header::{
         APPLICATION_MSGPACK_STR, DATADOG_SEND_REAL_HTTP_STATUS_STR, DATADOG_TRACE_COUNT_STR,

datadog-sidecar-ffi/src/span.rs

@@ -1,7 +1,7 @@
 // Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 
-use datadog_trace_utils::span::{
+use datadog_trace_utils::span::v04::{
     AttributeAnyValueBytes, AttributeArrayValueBytes, SpanBytes, SpanEventBytes, SpanLinkBytes,
 };
 use ddcommon_ffi::slice::{AsBytes, CharSlice};