feat(data-pipeline): port dd-trace-rs trace buffer implementation (#1826)

# Motivation

As we implement native spans in ruby in python, we have opportuinities
to replace more components of the span pipeline with native code.

One of them i the span buffer which we is before the trace exporter.

# Changes

Add a TraceBuffer implementation.
The TraceBuffer is split in two part, a Sender which allows putting
trace chunks in the queue and a background task, spawned on the shared
runtime which periodically pulls data from the chunk and submit them
using the TraceExporter.

Author: paullegranddc

libdd-data-pipeline/Cargo.toml

@@ -26,6 +26,8 @@ sha2 = "0.10"
 either = "1.13.0"
 tokio = { version = "1.23", features = [
     "rt",
+    "sync",
+    "time",
 ], default-features = false }
 uuid = { version = "1.10.0", features = ["v4"] }
 tokio-util = "0.7.11"
@@ -53,9 +55,14 @@ uuid = { version = "1", features = ["js"] }
 [lib]
 bench = false
 
+[[bench]]
+name = "trace_buffer"
+harness = false
+
 [dev-dependencies]
 libdd-capabilities-impl = { version = "0.1.0", path = "../libdd-capabilities-impl" }
 libdd-log = { path = "../libdd-log" }
+libdd-shared-runtime = { version = "0.1.0", path = "../libdd-shared-runtime" }
 clap = { version = "4.0", features = ["derive"] }
 criterion = "0.5.1"
 libdd-trace-utils = { path = "../libdd-trace-utils", features = [

libdd-data-pipeline/benches/trace_buffer.rs

@@ -0,0 +1,99 @@
+// Copyright 2025-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+
+use std::pin::Pin;
+use std::sync::Arc;
+use std::time::Duration;
+
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
+use libdd_data_pipeline::trace_buffer::{Export, TraceBuffer, TraceBufferConfig, TraceChunk};
+use libdd_data_pipeline::trace_exporter::{
+    agent_response::AgentResponse, error::TraceExporterError,
+};
+use libdd_shared_runtime::SharedRuntime;
+
+type Span = [u8; 100];
+
+// Number of chunks each sender thread sends per benchmark iteration.
+const CHUNKS_PER_SENDER: usize = 900;
+
+// Simulates async IO by sleeping 2ms per export batch.
+#[derive(Debug)]
+struct SleepExport;
+
+impl Export<Span> for SleepExport {
+    fn export_trace_chunks(
+        &mut self,
+        _trace_chunks: Vec<TraceChunk<Span>>,
+    ) -> Pin<
+        Box<
+            dyn std::future::Future<Output = Result<AgentResponse, TraceExporterError>> + Send + '_,
+        >,
+    > {
+        Box::pin(async {
+            tokio::time::sleep(Duration::from_millis(2)).await;
+            Ok(AgentResponse::Unchanged)
+        })
+    }
+}
+
+fn setup_buffer() -> (Arc<SharedRuntime>, Arc<TraceBuffer<Span>>) {
+    let rt = Arc::new(SharedRuntime::new().expect("SharedRuntime::new"));
+    let cfg = TraceBufferConfig::new()
+        .max_buffered_spans(1_000)
+        .span_flush_threshold(500)
+        .max_flush_interval(Duration::from_secs(2));
+    let (buf, worker) = TraceBuffer::new(cfg, Box::new(|_| {}), Box::new(SleepExport));
+    rt.spawn_worker(worker, true).expect("spawn_worker");
+    (rt, Arc::new(buf))
+}
+
+fn bench_trace_buffer(c: &mut Criterion) {
+    let mut group = c.benchmark_group("trace_buffer");
+
+    // (label, inter-send delay)
+    let workloads: &[(&str, Option<Duration>)] = &[
+        ("no_delay", None),
+        ("1us_delay", Some(Duration::from_micros(1))),
+        ("10us_delay", Some(Duration::from_micros(10))),
+    ];
+
+    for &(delay_label, delay) in workloads {
+        for num_senders in [1_usize, 2, 4, 8] {
+            let (rt, sender) = setup_buffer();
+
+            group.throughput(Throughput::Elements(
+                (num_senders * CHUNKS_PER_SENDER) as u64,
+            ));
+
+            group.bench_function(
+                BenchmarkId::new(format!("{}_senders", num_senders), delay_label),
+                |b| {
+                    b.iter(|| {
+                        std::thread::scope(|s| {
+                            for _ in 0..num_senders {
+                                let sender = sender.clone();
+                                s.spawn(move || {
+                                    for _ in 0..CHUNKS_PER_SENDER {
+                                        // BatchFull errors are expected under high load.
+                                        let _ = sender.send_chunk(vec![[0u8; 100]]);
+                                        if let Some(d) = delay {
+                                            std::thread::sleep(d);
+                                        }
+                                    }
+                                });
+                            }
+                        });
+                    });
+                },
+            );
+
+            rt.shutdown(None).expect("runtime shutdown");
+        }
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, bench_trace_buffer);
+criterion_main!(benches);

libdd-data-pipeline/src/lib.rs

@@ -15,5 +15,7 @@ mod health_metrics;
 pub(crate) mod otlp;
 #[cfg(feature = "telemetry")]
 pub(crate) mod telemetry;
+#[cfg(not(target_arch = "wasm32"))]
+pub mod trace_buffer;
 #[allow(missing_docs)]
 pub mod trace_exporter;