perf(trace-utils)!: span pool to reduce alloc churn on the send path

libdd-data-pipeline-core/src/agentless/exporter.rs

@@ -11,6 +11,7 @@ use libdd_trace_utils::send_with_retry::{
     send_with_retry_and_size, CompressionStrategy, RetryBackoffType, RetryStrategy,
     SendWithRetryError, SendWithRetryResult,
 };
+use libdd_trace_utils::span::span_pool::PooledChunks;
 use libdd_trace_utils::span::{trace_utils::compute_top_level_span, TraceData};
 use libdd_trace_utils::tracer_metadata::TracerMetadata;
 use thiserror::Error;
@@ -43,7 +44,7 @@ pub enum AgentlessError {
 /// does not double-count the same traces.
 pub async fn send_agentless_traces<C, T>(
     capabilities: &C,
-    traces: Vec<Vec<libdd_trace_utils::span::v04::Span<T>>>,
+    traces: PooledChunks<'_, T>,
     metadata: &TracerMetadata,
     config: &AgentlessTraceConfig,
     client_side_stats: bool,
@@ -68,7 +69,7 @@ where
 /// The observer is invoked once after the retry loop with the post-compression payload size.
 pub async fn send_agentless_traces_with_observer<C, T, F>(
     capabilities: &C,
-    mut traces: Vec<Vec<libdd_trace_utils::span::v04::Span<T>>>,
+    mut traces: PooledChunks<'_, T>,
     metadata: &TracerMetadata,
     config: &AgentlessTraceConfig,
     client_side_stats: bool,
@@ -82,7 +83,7 @@ where
     // Top-level tagging is already done before sending when client-side stats
     // are active
     if !metadata.client_computed_top_level && !client_side_stats {
-        for chunk in &mut traces {
+        for chunk in traces.as_mut_slice() {
             compute_top_level_span(chunk);
         }
     }
@@ -186,7 +187,7 @@ mod tests {
     use bytes::Bytes;
     use libdd_tinybytes::BytesString;
     use libdd_trace_obfuscation::obfuscation_config::ObfuscationConfig;
-    use libdd_trace_utils::span::v04::SpanBytes;
+    use libdd_trace_utils::span::{v04::SpanBytes, BytesData};
     use std::{
         sync::{Arc, Mutex},
         time::Duration,
@@ -247,8 +248,8 @@ mod tests {
         }
     }
 
-    fn v04_traces() -> Vec<Vec<SpanBytes>> {
-        vec![vec![SpanBytes {
+    fn v04_traces() -> PooledChunks<'static, BytesData> {
+        PooledChunks::unpooled(vec![vec![SpanBytes {
             name: BytesString::from_static("operation"),
             service: BytesString::from_static("service-1"),
             resource: BytesString::from_static("resource-1"),
@@ -257,7 +258,7 @@ mod tests {
             start: 1,
             duration: 2,
             ..Default::default()
-        }]]
+        }]])
     }
 
     fn request_body(request: &http::Request<Bytes>) -> Vec<u8> {

libdd-data-pipeline-ffi/src/tracer.rs

@@ -17,6 +17,7 @@ use crate::{catch_panic, gen_error};
 use libdd_common_ffi::slice::{AsBytes, ByteSlice, Slice};
 use libdd_common_ffi::CharSlice;
 use libdd_tinybytes::{Bytes, BytesString};
+use libdd_trace_utils::span::span_pool::PooledChunks;
 use libdd_trace_utils::span::v04::{
     AttributeAnyValueBytes, AttributeArrayValueBytes, SpanBytes, SpanEventBytes, SpanLinkBytes,
 };
@@ -776,7 +777,7 @@ pub unsafe extern "C" fn ddog_trace_exporter_send_trace_chunks(
     };
 
     catch_panic!(
-        match exporter.send_trace_chunks(chunks.0, cancel) {
+        match exporter.send_trace_chunks(PooledChunks::unpooled(chunks.0), cancel) {
             Ok(resp) => {
                 if let Some(out) = response_out {
                     out.as_ptr().write(Box::new(ExporterResponse::from(resp)));

libdd-data-pipeline/examples/send-traces-agentless.rs

@@ -20,7 +20,10 @@ use libdd_log::logger::{
     logger_configure_std, logger_set_log_level, LogEventLevel, StdConfig, StdTarget,
 };
 use libdd_shared_runtime::{ForkSafeRuntime, SharedRuntime};
-use libdd_trace_utils::span::v04::{SpanBytes, SpanEvent, SpanLink, VecMap};
+use libdd_trace_utils::span::{
+    span_pool::PooledChunks,
+    v04::{SpanBytes, SpanEvent, SpanLink, VecMap},
+};
 use rand::random;
 use std::{collections::HashMap, sync::Arc, time::UNIX_EPOCH};
 
@@ -108,7 +111,7 @@ fn main() {
     let traces = vec![trace];
 
     exporter
-        .send_trace_chunks(traces, None)
+        .send_trace_chunks(PooledChunks::unpooled(traces), None)
         .expect("Failed to send traces");
     println!("Trace sent to agentless intake at {intake_url}");
 

libdd-data-pipeline/examples/send-traces-with-stats.rs

@@ -16,7 +16,10 @@ mod example {
     };
     use libdd_shared_runtime::{ForkSafeRuntime, SharedRuntime};
     use libdd_tinybytes::BytesString;
-    use libdd_trace_utils::span::v04::{Span, SpanBytes, VecMap};
+    use libdd_trace_utils::span::{
+        span_pool::PooledChunks,
+        v04::{Span, SpanBytes, VecMap},
+    };
     use std::{
         sync::Arc,
         time::{Duration, UNIX_EPOCH},
@@ -153,7 +156,7 @@ mod example {
         dbg!(&traces);
 
         exporter
-            .send_trace_chunks(traces, None)
+            .send_trace_chunks(PooledChunks::unpooled(traces), None)
             .expect("Failed to send traces");
         shared_runtime
             .shutdown(None)

libdd-data-pipeline/src/agentless/exporter.rs

@@ -9,13 +9,14 @@ use libdd_data_pipeline_core::{
     send_agentless_traces_with_observer as send_traces, AgentlessError, AgentlessTraceConfig,
 };
 use libdd_trace_utils::send_with_retry::{SendWithRetryError, SendWithRetryResult};
+use libdd_trace_utils::span::span_pool::PooledChunks;
 use libdd_trace_utils::span::TraceData;
 use libdd_trace_utils::tracer_metadata::TracerMetadata;
 use tracing::error;
 
 pub(crate) async fn send_agentless_traces_with_observer<T, C, F, S>(
     capabilities: &C,
-    traces: Vec<Vec<libdd_trace_utils::span::v04::Span<T>>>,
+    traces: PooledChunks<'_, T>,
     metadata: &TracerMetadata,
     config: &AgentlessTraceConfig,
     client_side_stats: bool,

libdd-data-pipeline/src/trace_buffer/mod.rs

@@ -15,6 +15,10 @@ use std::{
 
 use libdd_capabilities::{HttpClientCapability, LogWriterCapability, MaybeSend, SleepCapability};
 use libdd_shared_runtime::{SharedRuntime, Worker};
+use libdd_trace_utils::span::{
+    span_pool::{PooledChunks, SpanPool},
+    BytesData,
+};
 
 use crate::trace_exporter::{
     agent_response::AgentResponse, error::TraceExporterError, TraceExporter,
@@ -807,15 +811,22 @@ where
     R: SharedRuntime + std::fmt::Debug + Send + Sync + 'static,
 {
     trace_exporter: TraceExporter<C, R>,
+    span_pool: Option<SpanPool<BytesData>>,
 }
 
 impl<C, R> DefaultExport<C, R>
 where
     C: HttpClientCapability + SleepCapability + LogWriterCapability + MaybeSend + Sync + 'static,
     R: SharedRuntime + std::fmt::Debug + Send + Sync + 'static,
 {
-    pub fn new(trace_exporter: TraceExporter<C, R>) -> Self {
-        Self { trace_exporter }
+    pub fn new(
+        trace_exporter: TraceExporter<C, R>,
+        span_pool: Option<SpanPool<BytesData>>,
+    ) -> Self {
+        Self {
+            trace_exporter,
+            span_pool,
+        }
     }
 }
 
@@ -834,7 +845,10 @@ where
     > {
         Box::pin(async {
             self.trace_exporter
-                .send_trace_chunks_async(trace_chunks)
+                .send_trace_chunks_async(match &self.span_pool {
+                    Some(p) => p.wrap_chunks(trace_chunks),
+                    None => PooledChunks::unpooled(trace_chunks),
+                })
                 .await
         })
     }

libdd-data-pipeline/src/trace_exporter/mod.rs

@@ -55,6 +55,7 @@ use libdd_trace_utils::msgpack_decoder;
 use libdd_trace_utils::send_with_retry::{
     send_with_retry, CompressionStrategy, RetryStrategy, SendWithRetryError, SendWithRetryResult,
 };
+use libdd_trace_utils::span::span_pool::PooledChunks;
 use libdd_trace_utils::span::{v04::Span, TraceData};
 use libdd_trace_utils::trace_utils::TracerHeaderTags;
 use std::io;
@@ -421,7 +422,9 @@ impl<
             None,
         );
 
-        let res = self.send_trace_chunks_inner(traces).await?;
+        let res = self
+            .send_trace_chunks_inner(PooledChunks::unpooled(traces))
+            .await?;
         if matches!(&res, AgentResponse::Changed { body } if body.is_empty()) {
             return Err(TraceExporterError::Agent(
                 error::AgentErrorKind::EmptyResponse,
@@ -607,7 +610,7 @@ impl<
     #[cfg(not(target_arch = "wasm32"))]
     pub fn send_trace_chunks<T: TraceData>(
         &self,
-        trace_chunks: Vec<Vec<Span<T>>>,
+        trace_chunks: PooledChunks<'_, T>,
         cancellation_token: Option<&CancellationToken>,
     ) -> Result<AgentResponse, TraceExporterError>
     where
@@ -639,7 +642,7 @@ impl<
     /// * Err(TraceExporterError): An error detailing what went wrong in the process
     pub async fn send_trace_chunks_async<T: TraceData>(
         &self,
-        trace_chunks: Vec<Vec<Span<T>>>,
+        trace_chunks: PooledChunks<'_, T>,
     ) -> Result<AgentResponse, TraceExporterError> {
         // There is no agent to negotiate with, skip the poll.
         if self.log_output.is_none() && self.agentless_config.is_none() {
@@ -651,7 +654,7 @@ impl<
     /// Sends trace chunks to the Datadog agentless intake (`/v1/input`) as JSON.
     async fn send_agentless_traces_inner<T: TraceData>(
         &self,
-        traces: Vec<Vec<Span<T>>>,
+        traces: PooledChunks<'_, T>,
         config: &AgentlessTraceConfig,
         client_side_stats: bool,
     ) -> Result<AgentResponse, TraceExporterError> {
@@ -679,11 +682,11 @@ impl<
     /// Sends trace chunks via OTLP HTTP (JSON or protobuf) when OTLP config is enabled.
     async fn send_otlp_traces_inner<T: TraceData>(
         &self,
-        traces: Vec<Vec<Span<T>>>,
+        traces: &[Vec<Span<T>>],
         config: &OtlpTraceConfig,
     ) -> Result<AgentResponse, TraceExporterError> {
         #[cfg(feature = "telemetry")]
-        let counts = PayloadCounts::from_traces(&traces);
+        let counts = PayloadCounts::from_traces(traces);
         let resource_info = {
             let mut r = OtlpResourceInfo::default();
             r.service = self.metadata.service.clone();
@@ -802,8 +805,13 @@ impl<
 
     async fn send_trace_chunks_inner<T: TraceData>(
         &self,
-        mut traces: Vec<Vec<Span<T>>>,
+        mut traces: PooledChunks<'_, T>,
     ) -> Result<AgentResponse, TraceExporterError> {
+        // `traces` is a `PooledChunks`: keeping it owned (rather than moving its inner `Vec`
+        // into the consuming code paths) is what lets its spans be recycled into the pool when
+        // it is dropped at the end of this function. Paths that must consume the spans use
+        // `into_chunks()` and forgo pooling.
+        //
         // TODO(APMSP-3608): log-output silently takes precedence over OTLP/agent here.
         // The builder should reject conflicting destinations at build time instead.
         if let Some(max_line_size) = self.log_output {
@@ -824,7 +832,7 @@ impl<
             self.telemetry.load_full().as_deref(),
         );
 
-        for chunk in &mut traces {
+        for chunk in traces.iter_mut() {
             for span in chunk.iter_mut() {
                 span.dedup();
             }
@@ -847,7 +855,7 @@ impl<
             if traces.is_empty() {
                 return Ok(AgentResponse::Unchanged);
             }
-            return self.send_otlp_traces_inner(traces, config).await;
+            return self.send_otlp_traces_inner(&traces, config).await;
         }
 
         // Snapshot the effective format once so the serializer and the URL agree even if
@@ -856,7 +864,7 @@ impl<
         let counts = PayloadCounts::from_traces(&traces);
 
         let prepared = match self.serializer.prepare_traces_payload(
-            traces,
+            &traces,
             header_tags,
             &self.metadata,
             self.agent_payload_response_version.as_ref(),

libdd-data-pipeline/src/trace_exporter/stats.rs

@@ -8,6 +8,7 @@
 //! and processing traces for stats collection.
 
 pub use libdd_trace_stats::span_concentrator::CardinalityLimitConfig;
+use libdd_trace_utils::span::span_pool::PooledChunks;
 
 use super::add_path;
 use super::TracerMetadata;
@@ -331,7 +332,7 @@ pub(crate) fn process_traces_for_stats<
         + Sync
         + 'static,
 >(
-    traces: &mut Vec<Vec<libdd_trace_utils::span::v04::Span<T>>>,
+    traces: &mut PooledChunks<'_, T>,
     header_tags: &mut libdd_trace_utils::trace_utils::TracerHeaderTags,
     client_side_stats: &ArcSwap<StatsComputationStatus>,
     client_computed_top_level: bool,