diff --git a/Cargo.lock b/Cargo.lock index 59767fd02b..ab707641eb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1182,9 +1182,9 @@ checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" [[package]] name = "crossbeam-channel" -version = "0.5.15" +version = "0.5.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2" +checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" dependencies = [ "crossbeam-utils", ] @@ -3645,6 +3645,7 @@ dependencies = [ "cargo-platform", "cargo_metadata", "criterion", + "crossbeam-channel", "flate2", "futures", "getrandom 0.2.15", @@ -3674,6 +3675,7 @@ dependencies = [ "serde_json", "tempfile", "thin-vec", + "thread_local", "tokio", "tracing", "urlencoding", @@ -6073,12 +6075,11 @@ dependencies = [ [[package]] name = "thread_local" -version = "1.1.8" +version = "1.1.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b9ef9bad013ada3808854ceac7b46812a6465ba368859a37e2100283d2d719c" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" dependencies = [ "cfg-if", - "once_cell", ] [[package]] diff --git a/libdd-data-pipeline-core/src/agentless/exporter.rs b/libdd-data-pipeline-core/src/agentless/exporter.rs index bf4c3709d3..9abc32c1fc 100644 --- a/libdd-data-pipeline-core/src/agentless/exporter.rs +++ b/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( capabilities: &C, - traces: Vec>>, + 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( capabilities: &C, - mut traces: Vec>>, + 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![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) -> Vec { diff --git a/libdd-data-pipeline-ffi/src/tracer.rs b/libdd-data-pipeline-ffi/src/tracer.rs index eabfd11e74..14a394aaf1 100644 --- a/libdd-data-pipeline-ffi/src/tracer.rs +++ b/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))); diff --git a/libdd-data-pipeline/examples/send-traces-agentless.rs b/libdd-data-pipeline/examples/send-traces-agentless.rs index 8c097de461..48c7487963 100644 --- a/libdd-data-pipeline/examples/send-traces-agentless.rs +++ b/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}"); diff --git a/libdd-data-pipeline/examples/send-traces-with-stats.rs b/libdd-data-pipeline/examples/send-traces-with-stats.rs index 76684fe589..25f913fcd0 100644 --- a/libdd-data-pipeline/examples/send-traces-with-stats.rs +++ b/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) diff --git a/libdd-data-pipeline/src/agentless/exporter.rs b/libdd-data-pipeline/src/agentless/exporter.rs index fda230ddb7..5c3a810551 100644 --- a/libdd-data-pipeline/src/agentless/exporter.rs +++ b/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( capabilities: &C, - traces: Vec>>, + traces: PooledChunks<'_, T>, metadata: &TracerMetadata, config: &AgentlessTraceConfig, client_side_stats: bool, diff --git a/libdd-data-pipeline/src/trace_buffer/mod.rs b/libdd-data-pipeline/src/trace_buffer/mod.rs index 715c087857..f33c91fe17 100644 --- a/libdd-data-pipeline/src/trace_buffer/mod.rs +++ b/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,6 +811,7 @@ where R: SharedRuntime + std::fmt::Debug + Send + Sync + 'static, { trace_exporter: TraceExporter, + span_pool: Option>, } impl DefaultExport @@ -814,8 +819,14 @@ where C: HttpClientCapability + SleepCapability + LogWriterCapability + MaybeSend + Sync + 'static, R: SharedRuntime + std::fmt::Debug + Send + Sync + 'static, { - pub fn new(trace_exporter: TraceExporter) -> Self { - Self { trace_exporter } + pub fn new( + trace_exporter: TraceExporter, + span_pool: Option>, + ) -> 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 }) } diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index 7e522a4d3a..3cb14655a9 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/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( &self, - trace_chunks: Vec>>, + trace_chunks: PooledChunks<'_, T>, cancellation_token: Option<&CancellationToken>, ) -> Result where @@ -639,7 +642,7 @@ impl< /// * Err(TraceExporterError): An error detailing what went wrong in the process pub async fn send_trace_chunks_async( &self, - trace_chunks: Vec>>, + trace_chunks: PooledChunks<'_, T>, ) -> Result { // 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( &self, - traces: Vec>>, + traces: PooledChunks<'_, T>, config: &AgentlessTraceConfig, client_side_stats: bool, ) -> Result { @@ -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( &self, - traces: Vec>>, + traces: &[Vec>], config: &OtlpTraceConfig, ) -> Result { #[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( &self, - mut traces: Vec>>, + mut traces: PooledChunks<'_, T>, ) -> Result { + // `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(), diff --git a/libdd-data-pipeline/src/trace_exporter/stats.rs b/libdd-data-pipeline/src/trace_exporter/stats.rs index d38165bc4b..9f83520058 100644 --- a/libdd-data-pipeline/src/trace_exporter/stats.rs +++ b/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>>, + traces: &mut PooledChunks<'_, T>, header_tags: &mut libdd_trace_utils::trace_utils::TracerHeaderTags, client_side_stats: &ArcSwap, client_computed_top_level: bool, diff --git a/libdd-data-pipeline/src/trace_exporter/trace_serializer.rs b/libdd-data-pipeline/src/trace_exporter/trace_serializer.rs index 1a5d42d15d..ab9fce122c 100644 --- a/libdd-data-pipeline/src/trace_exporter/trace_serializer.rs +++ b/libdd-data-pipeline/src/trace_exporter/trace_serializer.rs @@ -49,52 +49,24 @@ impl TraceSerializer { /// Prepare traces payload and HTTP headers for sending to agent pub(super) fn prepare_traces_payload( &self, - traces: Vec>>, + traces: &[Vec>], header_tags: TracerHeaderTags, metadata: &TracerMetadata, agent_payload_response_version: Option<&AgentResponsePayloadVersion>, output_format: TraceExporterOutputFormat, ) -> Result { - let payload = self.collect_and_process_traces(traces, output_format)?; - let chunks = payload.size(); + let chunk_count = traces.len(); let headers = - self.build_traces_headers(header_tags, chunks, agent_payload_response_version); - let mp_payload = self.serialize_payload(&payload, metadata, output_format)?; + self.build_traces_headers(header_tags, chunk_count, agent_payload_response_version); + let mp_payload = self.serialize_payload(traces, metadata, output_format)?; Ok(PreparedTracesPayload { data: mp_payload, headers, - chunk_count: chunks, + chunk_count, }) } - /// Collect trace chunks based on output format - fn collect_and_process_traces( - &self, - traces: Vec>>, - output_format: TraceExporterOutputFormat, - ) -> Result, TraceExporterError> { - let map_err = |e: anyhow::Error| { - TraceExporterError::Deserialization(DecodeError::InvalidFormat(e.to_string())) - }; - match output_format { - // v0.4 input spans are kept as-is in `TraceChunks::V04`. Whether they go out as v0.4 - // or are cross-encoded into V1 on the wire is decided in `serialize_payload`. - // - // APMSP-2812 - TODO: when the data-pipeline gains a V1-native input model (its own - // `v1::Span`-shaped builder), route `OutputFormat::V1` to - // `TraceChunks::V1(v1::TracerPayload)` instead and serialize via - // `to_vec_from_payload_v1`. A `StatSpan` impl on `v1::Span` will also be needed - // if client-side stats are enabled on the V1-native path. - TraceExporterOutputFormat::V04 | TraceExporterOutputFormat::V1 => { - Ok(tracer_payload::TraceChunks::V04(traces)) - } - TraceExporterOutputFormat::V05 => { - trace_utils::convert_trace_chunks_v04_to_v05(traces).map_err(map_err) - } - } - } - /// Build HTTP headers for traces request fn build_traces_headers( &self, @@ -115,10 +87,14 @@ impl TraceSerializer { headers } - /// Serialize payload to msgpack format + /// Serialize the borrowed traces to the msgpack payload for `output_format`. + // + // APMSP-2812 - TODO: when the data-pipeline gains a V1-native input model (its own + // `v1::Span`-shaped builder), serialize `OutputFormat::V1` from a native + // `v1::TracerPayload` via `to_vec_from_payload_v1` instead of cross-encoding v0.4 spans. fn serialize_payload( &self, - payload: &tracer_payload::TraceChunks, + traces: &[Vec>], metadata: &TracerMetadata, output_format: TraceExporterOutputFormat, ) -> Result, TraceExporterError> { @@ -126,35 +102,39 @@ impl TraceSerializer { .previous_serialised_len .load(Ordering::Relaxed) .max(MIN_BUFFER_CAPACITY); - let buff = match (payload, output_format) { - (tracer_payload::TraceChunks::V04(p), TraceExporterOutputFormat::V04) => { - msgpack_encoder::v04::to_vec_with_capacity_from_v04(p, capacity as u32) + let buff = match output_format { + TraceExporterOutputFormat::V04 => { + msgpack_encoder::v04::to_vec_with_capacity_from_v04(traces, capacity as u32) } // v0.4 spans cross-encoded as V1 on the wire (used when the agent advertises // /v1.0/traces). - (tracer_payload::TraceChunks::V04(p), TraceExporterOutputFormat::V1) => { - msgpack_encoder::v1::to_vec_with_capacity_from_v04(p, capacity as u32, metadata) - } - (tracer_payload::TraceChunks::V05(p), TraceExporterOutputFormat::V05) => { - let mut buff = Vec::with_capacity(capacity); - rmp_serde::encode::write(&mut buff, p) - .map_err(TraceExporterError::Serialization)?; - buff - } - // Native V1 input model, serialized directly — the payload carries its own - // tracer-level metadata, so no `TracerMetadata` is needed here. - (tracer_payload::TraceChunks::V1(p), TraceExporterOutputFormat::V1) => { - msgpack_encoder::v1::to_vec_with_capacity_from_v1(p, capacity as u32) - } - // `collect_and_process_traces` only produces (V04, V04|V1), (V05, V05), - // or (V1, V1) — any other combination here is a programming error. - _ => { - return Err(TraceExporterError::Deserialization( - DecodeError::InvalidFormat( - "Unsupported (TraceChunks, OutputFormat) combination for serialization" - .to_owned(), - ), - )); + TraceExporterOutputFormat::V1 => msgpack_encoder::v1::to_vec_with_capacity_from_v04( + traces, + capacity as u32, + metadata, + ), + TraceExporterOutputFormat::V05 => { + let map_err = |e: anyhow::Error| { + TraceExporterError::Deserialization(DecodeError::InvalidFormat(e.to_string())) + }; + let payload = + trace_utils::convert_trace_chunks_v04_to_v05(traces).map_err(map_err)?; + match payload { + tracer_payload::TraceChunks::V05(p) => { + let mut buff = Vec::with_capacity(capacity); + rmp_serde::encode::write(&mut buff, &p) + .map_err(TraceExporterError::Serialization)?; + buff + } + // `convert_trace_chunks_v04_to_v05` always returns `V05`. + _ => { + return Err(TraceExporterError::Deserialization( + DecodeError::InvalidFormat( + "v0.5 conversion produced an unexpected payload variant".to_owned(), + ), + )); + } + } } }; self.previous_serialised_len @@ -255,58 +235,13 @@ mod tests { assert_eq!(headers.get(DATADOG_TRACE_COUNT).unwrap(), "2"); } - #[test] - fn test_collect_and_process_traces_v04() { - let serializer = TraceSerializer::new(); - let traces = vec![vec![create_test_span()]]; - - let result = serializer.collect_and_process_traces(traces, TraceExporterOutputFormat::V04); - assert!(result.is_ok()); - - let payload = result.unwrap(); - assert!(matches!(payload, tracer_payload::TraceChunks::V04(_))); - assert_eq!(payload.size(), 1); - } - - #[test] - fn test_collect_and_process_traces_v05() { - let serializer = TraceSerializer::new(); - let traces = vec![vec![create_test_span()]]; - - let result = serializer.collect_and_process_traces(traces, TraceExporterOutputFormat::V05); - assert!(result.is_ok()); - - let payload = result.unwrap(); - assert!(matches!(payload, tracer_payload::TraceChunks::V05(_))); - assert_eq!(payload.size(), 1); - } - - #[test] - fn test_collect_and_process_traces_multiple_chunks() { - let serializer = TraceSerializer::new(); - let traces = vec![ - vec![create_test_span()], - vec![create_test_span(), create_test_span()], - vec![create_test_span()], - ]; - - let result = serializer.collect_and_process_traces(traces, TraceExporterOutputFormat::V04); - assert!(result.is_ok()); - - let payload = result.unwrap(); - assert_eq!(payload.size(), 3); - } - #[test] fn test_serialize_payload_v04() { let serializer = TraceSerializer::new(); let original_traces = vec![vec![create_test_span()]]; - let payload = serializer - .collect_and_process_traces(original_traces.clone(), TraceExporterOutputFormat::V04) - .unwrap(); let result = serializer.serialize_payload( - &payload, + &original_traces, &TracerMetadata::default(), TraceExporterOutputFormat::V04, ); @@ -340,12 +275,9 @@ mod tests { fn test_serialize_payload_v05() { let serializer = TraceSerializer::new(); let original_traces = vec![vec![create_test_span()]]; - let payload = serializer - .collect_and_process_traces(original_traces.clone(), TraceExporterOutputFormat::V05) - .unwrap(); let result = serializer.serialize_payload( - &payload, + &original_traces, &TracerMetadata::default(), TraceExporterOutputFormat::V05, ); @@ -385,7 +317,7 @@ mod tests { let header_tags = create_test_header_tags(); let result = serializer.prepare_traces_payload( - traces, + &traces, header_tags, &TracerMetadata::default(), None, @@ -410,7 +342,7 @@ mod tests { let header_tags = create_test_header_tags(); let result = serializer.prepare_traces_payload( - traces, + &traces, header_tags, &TracerMetadata::default(), None, @@ -432,7 +364,7 @@ mod tests { let header_tags = create_test_header_tags(); let result = serializer.prepare_traces_payload( - traces, + &traces, header_tags, &TracerMetadata::default(), Some(&agent_version), @@ -452,7 +384,7 @@ mod tests { let header_tags = create_test_header_tags(); let result = serializer.prepare_traces_payload( - traces, + &traces, header_tags, &TracerMetadata::default(), None, diff --git a/libdd-data-pipeline/tests/test_agentless_stats.rs b/libdd-data-pipeline/tests/test_agentless_stats.rs index 480a5eaeb1..cb4432330d 100644 --- a/libdd-data-pipeline/tests/test_agentless_stats.rs +++ b/libdd-data-pipeline/tests/test_agentless_stats.rs @@ -17,7 +17,10 @@ use libdd_data_pipeline::trace_exporter::TraceExporterBuilder; use libdd_shared_runtime::ForkSafeRuntime; use libdd_tinybytes::BytesString; use libdd_trace_protobuf::pb; -use libdd_trace_utils::span::v04::{SpanBytes, VecMap}; +use libdd_trace_utils::span::{ + span_pool::PooledChunks, + v04::{SpanBytes, VecMap}, +}; use std::time::Duration; use tokio::task; @@ -81,7 +84,7 @@ async fn run_agentless_with_stats( for chunks in chunks_per_call { exporter - .send_trace_chunks(chunks, None) + .send_trace_chunks(PooledChunks::unpooled(chunks), None) .expect("send_trace_chunks failed"); } exporter.shutdown(None).expect("shutdown failed"); @@ -160,7 +163,10 @@ async fn test_agentless_stats_sent_to_correct_endpoint() { .expect("TraceExporter::build failed"); exporter - .send_trace_chunks(vec![vec![make_root_span(1, Some(1.0), 0)]], None) + .send_trace_chunks( + PooledChunks::unpooled(vec![vec![make_root_span(1, Some(1.0), 0)]]), + None, + ) .expect("send_trace_chunks failed"); exporter.shutdown(None).expect("shutdown failed"); }) @@ -233,7 +239,10 @@ async fn test_agentless_stats_payload_structure() { .expect("build failed"); exporter - .send_trace_chunks(vec![vec![make_root_span(1, Some(1.0), 0)]], None) + .send_trace_chunks( + PooledChunks::unpooled(vec![vec![make_root_span(1, Some(1.0), 0)]]), + None, + ) .expect("send_trace_chunks failed"); exporter.shutdown(None).expect("shutdown failed"); }) @@ -581,7 +590,10 @@ async fn test_agentless_stats_preserves_container_id() { .expect("build failed"); exporter - .send_trace_chunks(vec![vec![make_root_span(1, Some(1.0), 0)]], None) + .send_trace_chunks( + PooledChunks::unpooled(vec![vec![make_root_span(1, Some(1.0), 0)]]), + None, + ) .expect("send_trace_chunks failed"); exporter.shutdown(None).expect("shutdown failed"); }) @@ -639,7 +651,7 @@ async fn test_agentless_stats_honors_additional_metric_tag_keys() { .expect("build failed"); exporter - .send_trace_chunks(vec![vec![span]], None) + .send_trace_chunks(PooledChunks::unpooled(vec![vec![span]]), None) .expect("send_trace_chunks failed"); exporter.shutdown(None).expect("shutdown failed"); }) diff --git a/libdd-trace-utils/Cargo.toml b/libdd-trace-utils/Cargo.toml index f1aff74533..edde1554f9 100644 --- a/libdd-trace-utils/Cargo.toml +++ b/libdd-trace-utils/Cargo.toml @@ -39,11 +39,13 @@ tracing.workspace = true serde_json = { workspace = true, features = ["std"] } serde-transcode = "1.1" futures.workspace = true -rand = "0.8.5" -bytes = { workspace = true, features = ["std"] } +rand = { version = "0.8.5", features = ["small_rng"] } +bytes = "1.11.1" rmpv = { version = "1.3.0", default-features = false } rmp = { version = "0.8.14", default-features = false } rustc-hash = "2.1.1" +crossbeam-channel = "0.5.16" +thread_local = "1.1" libdd-capabilities = { path = "../libdd-capabilities", version = "3.0.0" } libdd-common = { version = "5.2.0", path = "../libdd-common", default-features = false } diff --git a/libdd-trace-utils/benches/deserialization_v05.rs b/libdd-trace-utils/benches/deserialization_v05.rs index 226bd934ad..0553dc8870 100644 --- a/libdd-trace-utils/benches/deserialization_v05.rs +++ b/libdd-trace-utils/benches/deserialization_v05.rs @@ -150,7 +150,7 @@ fn build_v05_payload(num_traces: usize, spans_per_trace: usize, unique_per_span: .into_iter() .map(|trace| { trace - .into_iter() + .iter() .map(|span| from_v04_span(span, &mut dict)) .collect::>>() }) diff --git a/libdd-trace-utils/benches/main.rs b/libdd-trace-utils/benches/main.rs index 348115b7de..7f98e9242b 100644 --- a/libdd-trace-utils/benches/main.rs +++ b/libdd-trace-utils/benches/main.rs @@ -13,6 +13,7 @@ mod deserialization; mod deserialization_v05; mod otlp_encoding; mod serialization; +mod span_pool; criterion_main!( serialization::serialize_benches, @@ -20,5 +21,7 @@ criterion_main!( deserialization::deserialize_alloc_benches, deserialization_v05::deserialize_v05_benches, deserialization_v05::deserialize_v05_alloc_benches, - otlp_encoding::otlp_benches + otlp_encoding::otlp_benches, + span_pool::span_pool_benches, + span_pool::span_pool_alloc_benches ); diff --git a/libdd-trace-utils/benches/otlp_encoding.rs b/libdd-trace-utils/benches/otlp_encoding.rs index 9c99144752..a4af1ed5cd 100644 --- a/libdd-trace-utils/benches/otlp_encoding.rs +++ b/libdd-trace-utils/benches/otlp_encoding.rs @@ -84,13 +84,13 @@ pub fn otlp_encoding_benches(c: &mut Criterion) { c.bench_function(&format!("otlp/map_to_prost/{id}"), |b| { b.iter_batched( || spans.clone(), - |s| black_box(map_traces_to_otlp(black_box(s), &info, false)), + |s| black_box(map_traces_to_otlp(black_box(&s), &info, false)), BatchSize::SmallInput, ) }); // Pre-built IR for the encode-only benches (owned prost; no borrow of `bytes`). - let req = map_traces_to_otlp(spans.clone(), &info, false); + let req = map_traces_to_otlp(&spans, &info, false); // 2) prost IR -> HTTP/protobuf bytes. c.bench_function(&format!("otlp/encode_protobuf/{id}"), |b| { @@ -107,7 +107,7 @@ pub fn otlp_encoding_benches(c: &mut Criterion) { b.iter_batched( || spans.clone(), |s| { - let req = map_traces_to_otlp(s, &info, false); + let req = map_traces_to_otlp(&s, &info, false); black_box(encode_otlp_protobuf(&req)) }, BatchSize::SmallInput, @@ -119,7 +119,7 @@ pub fn otlp_encoding_benches(c: &mut Criterion) { b.iter_batched( || spans.clone(), |s| { - let req = map_traces_to_otlp(s, &info, false); + let req = map_traces_to_otlp(&s, &info, false); black_box(encode_otlp_json(&req).expect("json")) }, BatchSize::SmallInput, diff --git a/libdd-trace-utils/benches/span_pool.rs b/libdd-trace-utils/benches/span_pool.rs new file mode 100644 index 0000000000..9844624444 --- /dev/null +++ b/libdd-trace-utils/benches/span_pool.rs @@ -0,0 +1,169 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use criterion::measurement::Measurement; +use criterion::{black_box, criterion_group, Criterion, Throughput}; +use libdd_common::bench_utils::{memory_allocated_measurement, MeasurementName}; +use libdd_tinybytes::BytesString; +use libdd_trace_utils::span::span_pool::SpanPool; +use libdd_trace_utils::span::v04::Span; +use libdd_trace_utils::span::BytesData; + +/// Configurations exercised by the benchmarks: `(number of chunks, spans per chunk)`. +/// The small one resembles a single trace flush; the large one resembles a full payload. +const CONFIGS: &[(usize, usize)] = &[(1, 10), (20, 100)]; + +/// Pool capacity: large enough to hold a full iteration's worth of recycled spans, so the +/// bounded channel never drops spans because it is full (only the drop policy does). +const POOL_CAPACITY: usize = 4_096; + +/// Populate a span with realistic fields and a few meta/metrics entries. +/// +/// Inserting into `meta`/`metrics` is where the pool pays off: a recycled span keeps its +/// `VecMap`'s backing `Vec` capacity, so the inserts reuse it instead of growing a fresh +/// allocation from zero. +fn populate_span(span: &mut Span, span_idx: u64, trace_id: u128) { + span.service = BytesString::from_static("test-service"); + span.name = BytesString::from_static("http.request"); + span.resource = BytesString::from_static("GET /api/resource"); + span.r#type = BytesString::from_static("http"); + span.trace_id = trace_id; + span.span_id = span_idx; + span.parent_id = if span_idx == 0 { 0 } else { span_idx - 1 }; + span.start = 1_000_000 + span_idx as i64; + span.duration = 5_000; + span.error = 0; + span.meta.insert( + BytesString::from_static("http.method"), + BytesString::from_static("GET"), + ); + span.meta.insert( + BytesString::from_static("http.route"), + BytesString::from_static("/api/echo"), + ); + span.meta.insert( + BytesString::from_static("http.status_code"), + BytesString::from_static("200"), + ); + span.meta.insert( + BytesString::from_static("_dd.p.dm"), + BytesString::from_static("-0"), + ); + span.meta.insert( + BytesString::from_static("language"), + BytesString::from_static("python"), + ); + span.meta.insert( + BytesString::from_static("runtime-id"), + BytesString::from_static("bcc8589f1d534d2abf2bd7eb4a8eba2d"), + ); + span.metrics + .insert(BytesString::from_static("_sampling_priority_v1"), 2.0); + span.metrics + .insert(BytesString::from_static("_dd.top_level"), 1.0); + span.metrics + .insert(BytesString::from_static("_dd.tracer_kr"), 1.0); + span.metrics + .insert(BytesString::from_static("process_id"), 80474.0); +} + +/// Build `num_chunks` chunks of `spans_per_chunk` spans, populating each span via +/// [`populate_span`]. `get_span` decides whether spans come from the pool or are freshly +/// allocated; `pull_empty_chunk` supplies the backing `Vec` (recycled by the pooled path). +fn build_chunks Span, G: Fn() -> Vec>>( + num_chunks: usize, + spans_per_chunk: usize, + get_span: F, + pull_empty_chunk: G, +) -> Vec>> { + let mut chunks = Vec::with_capacity(num_chunks); + for chunk_idx in 0..num_chunks { + let trace_id = (chunk_idx as u128) << 64 | chunk_idx as u128; + let mut chunk = pull_empty_chunk(); + chunk.reserve(spans_per_chunk); + let base = chunk_idx * spans_per_chunk; + for i in 0..spans_per_chunk { + let mut span = get_span(); + populate_span(&mut span, base as u64 + i as u64, trace_id); + chunk.push(span); + } + chunks.push(chunk); + } + chunks +} + +/// Warm the pool with `count` freshly-allocated, populated spans so that the first +/// measured iterations dequeue recycled spans (steady state) rather than allocating. +fn warm_pool(pool: &SpanPool, count: usize) { + let chunks = build_chunks(1, count, Span::::default, Vec::new); + // Returning the chunks to the pool recycles the spans (minus the ~10% dropped by the + // drop policy). + drop(pool.wrap_chunks(chunks)); +} + +fn bench_iter< + M: Measurement, + F: Fn() -> Span, + G: Fn() -> Vec>, + H: Fn(Vec>>), +>( + group: &mut criterion::BenchmarkGroup<'_, M>, + variant_name: &'static str, + num_chunks: usize, + spans_per_chunk: usize, + get_span: F, + get_chunk: G, + return_chunks: H, +) { + group.throughput(Throughput::Elements((num_chunks * spans_per_chunk) as u64)); + group.bench_with_input( + format!("{num_chunks}x{spans_per_chunk}/{variant_name}"), + &(num_chunks, spans_per_chunk), + |b, &(num_chunks, spans_per_chunk)| { + b.iter(|| { + let chunks = build_chunks(num_chunks, spans_per_chunk, &get_span, &get_chunk); + // Enqueue: returning the chunks to the pool recycles the spans. + return_chunks(black_box(chunks)); + }); + }, + ); +} + +/// Dequeue spans from the pool, populate them, build chunks, then enqueue the chunks back +/// into the pool (recycling the spans on drop). Throughput is reported per span. +fn enqueue_dequeue(c: &mut Criterion) { + let mut group = c.benchmark_group(format!("span_pool/{}", M::name())); + for &(num_chunks, spans_per_chunk) in CONFIGS { + let total = num_chunks * spans_per_chunk; + let pool = SpanPool::::new(POOL_CAPACITY); + warm_pool(&pool, total); + + bench_iter( + &mut group, + "pooled", + num_chunks, + spans_per_chunk, + || pool.get_span(), + || pool.pull_empty_chunk(), + |chunks| drop(pool.wrap_chunks(chunks)), + ); + + bench_iter( + &mut group, + "allocating", + num_chunks, + spans_per_chunk, + Span::::default, + Vec::new, + drop, + ); + } + group.finish(); +} + +criterion_group!(span_pool_benches, enqueue_dequeue,); +criterion_group!( + name = span_pool_alloc_benches; + config = memory_allocated_measurement(&super::GLOBAL); + targets = enqueue_dequeue, +); diff --git a/libdd-trace-utils/src/otlp_encoder/mapper.rs b/libdd-trace-utils/src/otlp_encoder/mapper.rs index 8275ecdbff..cea2a1dd55 100644 --- a/libdd-trace-utils/src/otlp_encoder/mapper.rs +++ b/libdd-trace-utils/src/otlp_encoder/mapper.rs @@ -318,7 +318,7 @@ fn collect_event_attributes(ev: &SpanEvent) -> Vec( - trace_chunks: Vec>>, + trace_chunks: &[Vec>], resource_info: &OtlpResourceInfo, otel_trace_semantics_enabled: bool, ) -> ProtoReq { @@ -326,7 +326,7 @@ pub fn map_traces_to_otlp( // Pre-size to the total span count so the per-span push loop never reallocates. let total_spans: usize = trace_chunks.iter().map(|chunk| chunk.len()).sum(); let mut all_spans: Vec = Vec::with_capacity(total_spans); - for chunk in &trace_chunks { + for chunk in trace_chunks { // Resolve the high 64 bits of the 128-bit trace ID once per chunk. For each span, // prefer the native u128 `trace_id` field (e.g. Python's native spans hold the full // 128-bit ID there) and fall back to its RFC #85 `_dd.p.tid` meta tag. @@ -538,7 +538,7 @@ mod tests { ); span.metrics .insert(libdd_tinybytes::BytesString::from_static("count"), 42.0); - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; assert_eq!(s.trace_id, 0xD269B633813FC60C_u128.to_be_bytes().to_vec()); assert_eq!(s.span_id, 0xEEE19B7EC3C1B174u64.to_be_bytes().to_vec()); @@ -576,7 +576,7 @@ mod tests { ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let scope = req.resource_spans[0].scope_spans[0].scope.as_ref().unwrap(); assert_eq!(scope.name, "dd-trace-js"); assert_eq!(scope.version, "7.0.0-pre"); @@ -599,7 +599,7 @@ mod tests { duration: 1000000000, ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; assert_eq!( s.trace_id, @@ -634,7 +634,7 @@ mod tests { duration: 0, ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; assert_eq!( s.start_time_unix_nano, 0, @@ -662,7 +662,7 @@ mod tests { libdd_tinybytes::BytesString::from_static("error.msg"), libdd_tinybytes::BytesString::from_static("something broke"), ); - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; let status = s.status.as_ref().unwrap(); assert_eq!(status.code, status_code::ERROR); @@ -687,7 +687,7 @@ mod tests { libdd_tinybytes::BytesString::from_static("rate"), std::f64::consts::PI, ); - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; let count = s.attributes.iter().find(|a| a.key == "count").unwrap(); assert!(matches!( @@ -718,7 +718,7 @@ mod tests { "_dd.p.tid".into(), libdd_tinybytes::BytesString::from_static("5b8efff798038103"), ); - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; assert_eq!( s.trace_id, @@ -753,7 +753,7 @@ mod tests { duration: 1, ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![root, child]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![root, child]], &resource_info, false); let spans = &req.resource_spans[0].scope_spans[0].spans; let expected = full.to_be_bytes().to_vec(); assert_eq!(spans[0].trace_id, expected); @@ -773,7 +773,7 @@ mod tests { duration: 1, ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; assert_eq!(s.trace_id, 0xD269B633813FC60C_u128.to_be_bytes().to_vec()); } @@ -815,7 +815,7 @@ mod tests { duration: 1, ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![root, child_a, child_b]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![root, child_a, child_b]], &resource_info, false); let spans = &req.resource_spans[0].scope_spans[0].spans; assert_eq!(spans.len(), 3); let expected = 0x5b8efff798038103_d269b633813fc60c_u128 @@ -876,7 +876,7 @@ mod tests { ..Default::default() }; let req = map_traces_to_otlp( - vec![vec![root_a, child_a], vec![root_b, child_b]], + &[vec![root_a, child_a], vec![root_b, child_b]], &resource_info, false, ); @@ -936,7 +936,7 @@ mod tests { libdd_tinybytes::BytesString::from_static("dddddddddddddddd"), ); let req = map_traces_to_otlp( - vec![vec![root, child_no_tag, child_valid]], + &[vec![root, child_no_tag, child_valid]], &resource_info, false, ); @@ -965,7 +965,7 @@ mod tests { duration: 1, ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let resource_attrs = &req.resource_spans[0].resource.as_ref().unwrap().attributes; let kv = resource_attrs .iter() @@ -992,7 +992,7 @@ mod tests { duration: 1, ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let resource_attrs = &req.resource_spans[0].resource.as_ref().unwrap().attributes; assert!( !resource_attrs.iter().any(|a| a.key == "_dd.stats_computed"), @@ -1018,7 +1018,7 @@ mod tests { flags: 1, ..Default::default() }); - let req = map_traces_to_otlp(vec![vec![span]], &OtlpResourceInfo::default(), false); + let req = map_traces_to_otlp(&[vec![span]], &OtlpResourceInfo::default(), false); let link = &req.resource_spans[0].scope_spans[0].spans[0].links[0]; assert_eq!( link.flags, 1, @@ -1046,7 +1046,7 @@ mod tests { flags, ..Default::default() }); - let req = map_traces_to_otlp(vec![vec![span]], &OtlpResourceInfo::default(), false); + let req = map_traces_to_otlp(&[vec![span]], &OtlpResourceInfo::default(), false); req.resource_spans[0].scope_spans[0].spans[0].links[0].flags } @@ -1090,7 +1090,7 @@ mod tests { libdd_tinybytes::BytesString::from_static("http.method"), libdd_tinybytes::BytesString::from_static("GET"), ); - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, true); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, true); let attrs = &req.resource_spans[0].scope_spans[0].spans[0].attributes; let keys: Vec<&str> = attrs.iter().map(|kv| kv.key.as_str()).collect(); for omitted in [ @@ -1130,7 +1130,7 @@ mod tests { libdd_tinybytes::BytesString::from_static("error.message"), libdd_tinybytes::BytesString::from_static("boom"), ); - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, true); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, true); let otlp_span = &req.resource_spans[0].scope_spans[0].spans[0]; let status = otlp_span .status @@ -1155,7 +1155,7 @@ mod tests { // Defensive: an empty chunk should produce no spans and not panic. let resource_info = OtlpResourceInfo::default(); let empty: Vec>> = vec![vec![]]; - let req = map_traces_to_otlp(empty, &resource_info, false); + let req = map_traces_to_otlp(&empty, &resource_info, false); let spans = &req.resource_spans[0].scope_spans[0].spans; assert!(spans.is_empty()); } @@ -1175,7 +1175,7 @@ mod tests { "tracestate".into(), libdd_tinybytes::BytesString::from_static("vendor1=abc,rojo=00f067"), ); - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; assert_eq!(s.trace_state, "vendor1=abc,rojo=00f067"); } @@ -1195,7 +1195,7 @@ mod tests { }; span.meta_struct .insert("my_key".into(), Bytes::from(vec![1u8, 2, 3])); - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; let kv = s .attributes @@ -1203,7 +1203,7 @@ mod tests { .find(|a| a.key == "my_key") .expect("my_key attribute not found"); match kv.value.as_ref().unwrap().value { - Some(PV::BytesValue(ref b)) => assert_eq!(b, &vec![1u8, 2, 3]), + Some(PV::BytesValue(ref b)) => assert_eq!(b, &[1u8, 2, 3]), ref other => panic!("expected bytes, got {other:?}"), } } @@ -1220,7 +1220,7 @@ mod tests { duration: 1, ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; let kv = s .attributes @@ -1246,7 +1246,7 @@ mod tests { duration: 1, ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; let kv = s .attributes @@ -1272,7 +1272,7 @@ mod tests { duration: 1, ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; // resource maps to the OTLP span name assert_eq!(s.name, "GET /api/users"); @@ -1301,7 +1301,7 @@ mod tests { duration: 1, ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; assert!( !s.attributes.iter().any(|a| a.key == "resource.name"), @@ -1327,7 +1327,7 @@ mod tests { duration: 1, ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; let kv = s .attributes @@ -1353,7 +1353,7 @@ mod tests { ..Default::default() }; span.metrics.insert("_sampling_priority_v1".into(), 0.0); - let req = map_traces_to_otlp(vec![vec![span]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![span]], &resource_info, false); let s = &req.resource_spans[0].scope_spans[0].spans[0]; assert_eq!(s.flags, 0); } @@ -1379,7 +1379,7 @@ mod tests { ..Default::default() }; - let req = map_traces_to_otlp(vec![vec![root, child]], &resource_info, true); + let req = map_traces_to_otlp(&[vec![root, child]], &resource_info, true); let spans = &req.resource_spans[0].scope_spans[0].spans; assert!(spans.iter().all(|span| span.flags == 0)); } @@ -1405,7 +1405,7 @@ mod tests { ..Default::default() }; child.metrics.insert("_sampling_priority_v1".into(), 1.0); - let req = map_traces_to_otlp(vec![vec![root, child]], &resource_info, true); + let req = map_traces_to_otlp(&[vec![root, child]], &resource_info, true); let spans = &req.resource_spans[0].scope_spans[0].spans; assert!(spans.iter().all(|span| span.flags == 1)); } @@ -1433,7 +1433,7 @@ mod tests { }; child.metrics.insert("_sampling_priority_v1".into(), 1.0); - let req = map_traces_to_otlp(vec![vec![child, root]], &resource_info, true); + let req = map_traces_to_otlp(&[vec![child, root]], &resource_info, true); let spans = &req.resource_spans[0].scope_spans[0].spans; assert!(spans.iter().all(|span| span.flags == 1)); } @@ -1461,7 +1461,7 @@ mod tests { }; second.metrics.insert("_sampling_priority_v1".into(), 1.0); - let req = map_traces_to_otlp(vec![vec![first, second]], &resource_info, true); + let req = map_traces_to_otlp(&[vec![first, second]], &resource_info, true); let spans = &req.resource_spans[0].scope_spans[0].spans; assert!(spans.iter().all(|span| span.flags == 0)); } @@ -1500,7 +1500,7 @@ mod tests { .metrics .insert("_sampling_priority_v1".into(), 0.5); - let req = map_traces_to_otlp(vec![vec![root, child, fractional]], &resource_info, false); + let req = map_traces_to_otlp(&[vec![root, child, fractional]], &resource_info, false); let spans = &req.resource_spans[0].scope_spans[0].spans; assert_eq!(spans[0].flags, 0); assert_eq!(spans[1].flags, 1); diff --git a/libdd-trace-utils/src/otlp_encoder/mod.rs b/libdd-trace-utils/src/otlp_encoder/mod.rs index 70ef928a1f..af5e0164fe 100644 --- a/libdd-trace-utils/src/otlp_encoder/mod.rs +++ b/libdd-trace-utils/src/otlp_encoder/mod.rs @@ -86,7 +86,7 @@ mod encode_tests { // Decisive guard: JSON and protobuf are encoded from the *same* prost IR, so the two // wire formats cannot drift. let (chunks, info) = sample_native(); - let req = map_traces_to_otlp(chunks, &info, false); + let req = map_traces_to_otlp(&chunks, &info, false); let json = encode_otlp_json(&req).unwrap(); let pb = encode_otlp_protobuf(&req); @@ -133,7 +133,7 @@ mod encode_tests { // would need a deserializer mirroring `json_serializer`, which this crate doesn't ship; // `json_and_protobuf_carry_same_span` guards that the JSON matches this same IR.) let (chunks, info) = sample_native(); - let req = map_traces_to_otlp(chunks, &info, false); + let req = map_traces_to_otlp(&chunks, &info, false); let decoded = ProtoReq::decode(encode_otlp_protobuf(&req).as_slice()).unwrap(); assert_eq!(decoded, req); } diff --git a/libdd-trace-utils/src/span/mod.rs b/libdd-trace-utils/src/span/mod.rs index be2a3d3caf..f9e592a8f5 100644 --- a/libdd-trace-utils/src/span/mod.rs +++ b/libdd-trace-utils/src/span/mod.rs @@ -1,6 +1,7 @@ // Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 +pub mod span_pool; pub mod trace_utils; pub mod trace_utils_v1; pub mod v04; @@ -37,7 +38,7 @@ pub(crate) const SPAN_LINK_FLAGS_SET_SENTINEL: u32 = 1 << 31; /// Trait representing the requirements for a type to be used as a Span "string" type. /// Note: Borrow is not required by the derived traits, but allows to access HashMap elements /// from a static str and check if the string is empty. -pub trait SpanText: Debug + Eq + Hash + Borrow + Serialize + Default { +pub trait SpanText: Debug + Eq + Hash + Borrow + Serialize + Default + Send { fn from_static_str(value: &'static str) -> Self; /// Copies this text into an owned [`BytesString`]. @@ -77,7 +78,7 @@ impl SpanText for BytesString { } } -pub trait SpanBytes: Debug + Eq + Hash + Borrow<[u8]> + Serialize + Default + Clone { +pub trait SpanBytes: Debug + Eq + Hash + Borrow<[u8]> + Serialize + Default + Clone + Send { fn from_static_bytes(value: &'static [u8]) -> Self; } diff --git a/libdd-trace-utils/src/span/span_pool.rs b/libdd-trace-utils/src/span/span_pool.rs new file mode 100644 index 0000000000..d4b59305bf --- /dev/null +++ b/libdd-trace-utils/src/span/span_pool.rs @@ -0,0 +1,421 @@ +// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use super::v04::Span; +use super::TraceData; +use rand::{Rng as _, SeedableRng as _}; +use std::cell::RefCell; +use std::ops::{Deref, DerefMut}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use thread_local::ThreadLocal; + +/// When this function returns true, do not add the returned chunk to the queue. +/// +/// Why are we doing this? +/// +/// If we keep recylcing spans forever, two things are going to happen +/// * We will keep the **maximum** number of spans ever used by the program alive, even if memory +/// usage scales down +/// * As spans get reused and atributes data structure are pushed and popped, they will tend to grow +/// to have the maximum size of attributes +/// +/// The policy operates at chunk granularity: `add_chunks` draws once per chunk and either +/// recycles the whole chunk or drops it. Dropping a fixed pct of chunks returned ensures that +/// we eventually free memory if span usage spikes, and then goes down. +fn drop_policy() -> bool { + const PCT_OF_SPANS_RETURNED_DROPPED: f64 = 0.1; + thread_local! { + static RNG: RefCell = RefCell::new(rand::rngs::SmallRng::from_entropy()); + } + RNG.with_borrow_mut(|r| r.gen_bool(PCT_OF_SPANS_RETURNED_DROPPED)) +} + +/// Reset fields to default, keeping collection capacity for reuse. +fn reset_span( + Span { + service, + name, + resource, + r#type, + trace_id, + span_id, + parent_id, + start, + duration, + error, + meta, + metrics, + meta_struct, + span_links, + span_events, + }: &mut Span, +) { + *service = Default::default(); + *name = Default::default(); + *resource = Default::default(); + *r#type = Default::default(); + *trace_id = Default::default(); + *span_id = Default::default(); + *parent_id = Default::default(); + *start = Default::default(); + *duration = Default::default(); + *error = Default::default(); + meta.clear(); + metrics.clear(); + meta_struct.clear(); + span_links.clear(); + span_events.clear(); +} + +fn reset_chunk(chunk: &mut Vec>) { + for span in chunk { + reset_span(span); + } +} + +/// Max spans per recycled chunk. Larger chunks are split so no thread hoards a big chunk in its +/// local cache. +const MAX_CHUNK_SIZE: usize = 20; + +/// Split a chunk into pieces of at most [`MAX_CHUNK_SIZE`] spans, keeping each piece's spans' +/// buffer capacity for reuse +fn split_chunk(chunk: Vec>) -> impl Iterator>> { + let mut remaining = chunk; + std::iter::from_fn(move || { + if remaining.is_empty() { + return None; + } + if remaining.len() <= MAX_CHUNK_SIZE { + return Some(std::mem::take(&mut remaining)); + } + let at = remaining.len() - MAX_CHUNK_SIZE; + Some(remaining.split_off(at)) + }) +} + +/// Thread-safe pool of recyclable [`Span`] allocations. +/// +/// Spans come back as whole chunks (`Vec>`) via [`SpanPool::add_chunks`] (usually by +/// dropping a [`PooledChunks`]) and are handed out by [`SpanPool::get_span`]. Reuse keeps the +/// pre-allocated `meta`/`metrics`/... buffers alive across flushes, skipping alloc churn. +/// +/// Backed by an unbounded crossbeam channel of chunks (one send per chunk, not per span). The +/// capacity (in spans) bounds the channel only. Thread-local caches are not counted, so idle +/// threads holding cached spans don't reduce the pool's headroom. +/// +/// The capacity bound is best-effort: `add_chunks` checks-and-increments `len` with relaxed +/// atomics, so concurrent producers can briefly exceed `capacity` (the channel itself is +/// unbounded). +/// Since there is a single task returning chunks to the queue (the exporter task) the bound is +/// exact. `get_span` first hits a per-thread +/// chunk cache ([`ThreadLocal`]) and only when empty does it dequeue a fresh chunk. This keeps +/// the single-producer path lock-free and gives each thread a local chunk under contention. +#[derive(Debug, Clone)] +pub struct SpanPool { + inner: Arc>, +} + +#[derive(Debug)] +struct SpanPoolInner { + queue: crossbeam_channel::Sender>>, + receiver: crossbeam_channel::Receiver>>, + /// Per-thread cache: the last chunk pulled from the channel plus one recycled empty `Vec`. + thread_cache: ThreadLocal>>, + /// Total spans currently held in the global queue (channel); the capacity bound is in spans. + len: AtomicUsize, + /// Maximum number of recycled spans the pool will hold. + capacity: usize, +} + +#[derive(Debug, Default)] +struct ThreadCache { + /// Last chunk pulled from the channel; spans are popped from it by `get_span`. + last_chunk: Option>>, + /// One recycled empty `Vec` kept for `pull_empty_chunk`; never returned to the pool. + empty_chunk: Option>>, +} + +impl SpanPool { + /// New pool holding at most `pool_capacity` recycled spans. + pub fn new(pool_capacity: usize) -> Self { + Self::with_capacity(pool_capacity) + } + + /// New pool holding at most `capacity` recycled spans. + pub fn with_capacity(capacity: usize) -> Self { + let (queue, receiver) = crossbeam_channel::unbounded(); + Self { + inner: Arc::new(SpanPoolInner { + queue, + receiver, + thread_cache: ThreadLocal::new(), + len: AtomicUsize::new(0), + capacity, + }), + } + } + + /// Reset and return the given chunks to the pool for reuse. + /// + /// Spans are dropped (not pooled) when the drop policy fires or the pool is full. + pub fn add_chunks>>>(&self, chunks: I) { + for mut chunk in chunks { + if chunk.is_empty() || drop_policy() { + continue; + } + reset_chunk(&mut chunk); + for piece in split_chunk(chunk) { + let piece_len = piece.len(); + // Reserve span-count atomically against the cap; drop the piece if it won't fit. + let current = self.inner.len.load(Ordering::Relaxed); + if current + piece_len > self.inner.capacity { + return; + } + self.inner.len.fetch_add(piece_len, Ordering::Relaxed); + if self.inner.queue.send(piece).is_err() { + return; + } + } + } + } + + /// Get a span from the pool, or a fresh default if empty. + /// Tries the per-thread cache first (lock-free), dequeues a new chunk only when it's empty. + pub fn get_span(&self) -> Span { + loop { + let cell = self.inner.thread_cache.get_or_default(); + { + let mut slot = cell.borrow_mut(); + if let Some(chunk) = slot.last_chunk.as_mut() { + if let Some(span) = chunk.pop() { + if chunk.is_empty() { + // Recycle the now-empty `Vec` for `pull_empty_chunk`. + let empty = std::mem::take(chunk); + slot.last_chunk = None; + if slot.empty_chunk.is_none() { + slot.empty_chunk = Some(empty); + } + } + return span; + } + } + } + match self.inner.receiver.try_recv() { + Ok(chunk) => { + self.inner.len.fetch_sub(chunk.len(), Ordering::Relaxed); + self.inner + .thread_cache + .get_or_default() + .borrow_mut() + .last_chunk = Some(chunk); + } + Err(_) => return Span::default(), + } + } + } + + /// Get an empty `Vec>` with retained capacity for building a chunk, or a fresh one if + /// the thread has none cached. Not counted in the pool's length; never returned to the pool. + pub fn pull_empty_chunk(&self) -> Vec> { + self.inner + .thread_cache + .get_or_default() + .borrow_mut() + .empty_chunk + .take() + .unwrap_or_default() + } + + /// Spans currently held in the global queue (channel only, not thread-local caches). + /// Decremented by chunk when a chunk is dequeued, so idle threads holding cached spans don't + /// count against the capacity. + pub fn len(&self) -> usize { + self.inner.len.load(Ordering::Relaxed) + } + + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// Wrap chunks so they're returned to this pool on drop of the [`PooledChunks`]. + pub fn wrap_chunks(&self, chunks: Vec>>) -> PooledChunks<'_, T> { + PooledChunks::new(chunks, Some(self)) + } +} + +/// A reference to a `SpanPool` that might be enabled or disabled +pub struct MaybePool<'a, T: TraceData> { + pool: Option<&'a SpanPool>, +} + +impl MaybePool<'_, T> { + pub fn add_spans>>(&self, spans: I) { + let Some(pool) = self.pool else { + // No pool: still drive the iterator so lazy `extract_if` side-effects run. + spans.into_iter().for_each(drop); + return; + }; + let chunk: Vec> = spans.into_iter().collect(); + if !chunk.is_empty() { + pool.add_chunks(std::iter::once(chunk)); + } + } + + pub fn add_chunks>>>(&self, chunks: I) { + let Some(pool) = self.pool else { + // No pool: still drive the iterator so lazy `extract_if` side-effects run. + chunks.into_iter().for_each(drop); + return; + }; + pool.add_chunks(chunks); + } + + /// Get an empty chunk: recycled from the pool's thread-local cache when a pool is attached, + /// or a fresh `Vec` otherwise. + pub fn pull_empty_chunk(&self) -> Vec> { + match self.pool { + Some(pool) => pool.pull_empty_chunk(), + None => Vec::new(), + } + } +} + +/// Owned trace chunks that return to a [`SpanPool`] on drop. Deref's to the inner +/// `Vec>>` for in-place processing. With no pool ([`PooledChunks::unpooled`]) spans +/// just drop normally. +#[derive(Debug)] +pub struct PooledChunks<'a, T: TraceData> { + chunks: Vec>>, + pool: Option<&'a SpanPool>, +} + +impl<'a, T: TraceData> PooledChunks<'a, T> { + pub fn new(chunks: Vec>>, pool: Option<&'a SpanPool>) -> Self { + Self { chunks, pool } + } + + /// Wrap `chunks` with no pool, spans drop normally. For span types with no reusable + /// allocations (e.g. borrowed slice-backed spans). + pub fn unpooled(chunks: Vec>>) -> Self { + Self::new(chunks, None) + } + + /// Take the inner chunks, disabling pooling. For call sites that consume the chunks (e.g. + /// formats transforming spans into another representation). + pub fn into_chunks(mut self) -> Vec>> { + std::mem::take(&mut self.chunks) + } + + pub fn chunks_mut(&mut self) -> (MaybePool<'a, T>, &mut Vec>>) { + (MaybePool { pool: self.pool }, &mut self.chunks) + } +} + +impl Deref for PooledChunks<'_, T> { + type Target = Vec>>; + + fn deref(&self) -> &Self::Target { + &self.chunks + } +} + +impl DerefMut for PooledChunks<'_, T> { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.chunks + } +} + +impl Drop for PooledChunks<'_, T> { + fn drop(&mut self) { + if let Some(pool) = self.pool { + let chunks = std::mem::take(&mut self.chunks); + pool.add_chunks(chunks); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::span::v04::SpanBytes; + use libdd_tinybytes::BytesString; + + fn span(name: &str) -> SpanBytes { + SpanBytes { + name: BytesString::from_slice(name.as_bytes()).unwrap(), + ..Default::default() + } + } + + #[test] + fn returned_spans_are_recycled_and_reset() { + let pool = SpanPool::::with_capacity(100); + { + // No drop-policy control here, but a single span is very likely retained. + // If we drop 10% of spans, the likelyhood all spans are dropped is 1/10**100 + // which is basically never happening if we ran this test until the heat death of + // this universe + let chunks = pool.wrap_chunks(vec![vec![span("a"); 100]]); + drop(chunks); + } + for _ in 0..100 { + let s = pool.get_span(); + assert_eq!(s.name, BytesString::default()); + } + } + + #[test] + fn unpooled_chunks_do_not_feed_the_pool() { + let pool = SpanPool::::with_capacity(100); + drop(PooledChunks::unpooled(vec![vec![span("a")]])); + assert!(pool.is_empty()); + } + + #[test] + fn into_chunks_disables_pooling() { + let pool = SpanPool::::with_capacity(100); + let chunks = pool.wrap_chunks(vec![vec![span("a")]]); + let inner = chunks.into_chunks(); + assert_eq!(inner.len(), 1); + assert!(pool.is_empty()); + } + + #[test] + fn pool_is_bounded() { + let pool = SpanPool::::with_capacity(2); + // drop_policy keeps ~90%; push far more than capacity and check the bound holds. + for _ in 0..1000 { + pool.add_chunks(std::iter::once(vec![span("x")])); + } + assert!(pool.len() <= 2); + } + + #[test] + fn large_chunks_are_split_into_max_size_pieces() { + // MAX_CHUNK_SIZE=20, so 50 spans => 20 + 20 + 10. Capacity holds all pieces; we want the + // split, not the bound. Drop policy may drop the whole chunk, so retry until one makes it. + let pool = SpanPool::::with_capacity(100); + loop { + let big_chunk: Vec = (0..50).map(|_| span("x")).collect(); + pool.add_chunks(std::iter::once(big_chunk)); + if !pool.is_empty() { + break; + } + } + + let mut count = 0; + while let Ok(chunk) = pool.inner.receiver.try_recv() { + assert!( + chunk.len() <= MAX_CHUNK_SIZE, + "chunk of {} spans exceeds max", + chunk.len() + ); + count += 1; + } + assert!( + count >= 2, + "expected the 50-span chunk to be split into >= 2 pieces, got {count}" + ); + } +} diff --git a/libdd-trace-utils/src/span/trace_utils.rs b/libdd-trace-utils/src/span/trace_utils.rs index 60790aa3cb..f3bd691d3b 100644 --- a/libdd-trace-utils/src/span/trace_utils.rs +++ b/libdd-trace-utils/src/span/trace_utils.rs @@ -5,6 +5,8 @@ use tracing::debug; +use crate::span::span_pool::PooledChunks; + use super::{v04::Span, SpanText, TraceData}; use std::collections::{HashMap, HashSet}; @@ -145,18 +147,19 @@ const SAMPLING_ANALYTICS_RATE_KEY: &str = "_dd1.sr.eausr"; /// /// # Trace-level attributes /// Some attributes related to the whole trace are stored in the root span of the chunk. -pub fn drop_chunks(traces: &mut Vec>>) -> DroppedP0Stats +pub fn drop_chunks(traces: &mut PooledChunks) -> DroppedP0Stats where T: TraceData, { let mut dropped_p0_traces = 0; let mut dropped_p0_spans = 0; - traces.retain_mut(|chunk| { + let (pool, traces) = traces.chunks_mut(); + + let dropped = traces.extract_if(.., |chunk| { // ErrorSampler if chunk.iter().any(|s| s.error == 1) { - // We send chunks containing an error - return true; + return false; } // PrioritySampler and NoPrioritySampler @@ -164,8 +167,7 @@ where .iter() .find_map(|s| s.metrics.get(SAMPLING_PRIORITY_KEY)); if chunk_priority.is_none_or(|p| *p > 0.0) { - // We send chunks with positive priority or no priority - return true; + return false; } // SingleSpanSampler and AnalyzedSpansSampler @@ -186,15 +188,25 @@ where if sampled_indexes.is_empty() { // If no spans were sampled we can drop the whole chunk dropped_p0_traces += 1; - return false; + return true; } - let sampled_spans = sampled_indexes - .iter() - .map(|i| std::mem::take(&mut chunk[*i])) - .collect(); - *chunk = sampled_spans; - true + + let mut sampled_indices = sampled_indexes.iter().copied().peekable(); + let mut i: usize = 0; + let dropped_spans = chunk.extract_if(.., |_span| { + let drop = if sampled_indices.peek().copied() == Some(i) { + sampled_indices.next(); + false + } else { + true + }; + i += 1; + drop + }); + pool.add_spans(dropped_spans); + false }); + pool.add_chunks(dropped); DroppedP0Stats { dropped_p0_traces, @@ -426,14 +438,16 @@ mod tests { (chunk_with_analyzed_span, 1), ]; - for (chunk, expected_count) in chunks_and_expected_sampled_spans.into_iter() { - let mut traces = vec![chunk]; + for (i, (chunk, expected_count)) in + chunks_and_expected_sampled_spans.into_iter().enumerate() + { + let mut traces = PooledChunks::unpooled(vec![chunk]); drop_chunks(&mut traces); if expected_count == 0 { - assert!(traces.is_empty()); + assert!(traces.is_empty(), "failed at item {i}"); } else { - assert_eq!(traces[0].len(), expected_count); + assert_eq!(traces[0].len(), expected_count, "failed at item {i}"); } } } diff --git a/libdd-trace-utils/src/span/v05/mod.rs b/libdd-trace-utils/src/span/v05/mod.rs index 0dbdfdc2e4..135ac4948a 100644 --- a/libdd-trace-utils/src/span/v05/mod.rs +++ b/libdd-trace-utils/src/span/v05/mod.rs @@ -217,14 +217,14 @@ fn get_or_insert( /// dictionary always owns its strings ([`SharedDictBytes`]). Borrowed input text is copied into /// the dictionary; owned text is reference-counted. pub fn from_v04_span( - span: crate::span::v04::Span, + span: &crate::span::v04::Span, dict: &mut SharedDictBytes, ) -> Result { let meta_len = span.meta.len(); let metrics_len = span.metrics.len(); - // Serialize span links / span events before `span` is consumed below. v0.5 has no - // dedicated slots for them, so they are flattened into `meta` as JSON strings. + // Serialize span links / span events + // v0.5 has no dedicated slots for them, so they are flattened into `meta` as JSON strings. let serialized_span_links = if span.span_links.is_empty() { None } else { @@ -248,10 +248,10 @@ pub fn from_v04_span( let service = get_or_insert(dict, &span.service)?; let name = get_or_insert(dict, &span.name)?; let resource = get_or_insert(dict, &span.resource)?; - let mut meta = span.meta.into_iter().try_fold( + let mut meta = span.meta.iter().try_fold( HashMap::with_capacity(meta_len + extra_meta), |mut meta, (k, v)| -> anyhow::Result> { - meta.insert(get_or_insert(dict, &k)?, get_or_insert(dict, &v)?); + meta.insert(get_or_insert(dict, k)?, get_or_insert(dict, v)?); Ok(meta) }, )?; @@ -267,10 +267,10 @@ pub fn from_v04_span( meta.insert(key, value); } - let metrics = span.metrics.into_iter().try_fold( + let metrics = span.metrics.iter().try_fold( HashMap::with_capacity(metrics_len), |mut metrics, (k, v)| -> anyhow::Result> { - metrics.insert(get_or_insert(dict, &k)?, v); + metrics.insert(get_or_insert(dict, k)?, *v); Ok(metrics) }, )?; @@ -332,7 +332,7 @@ mod tests { }; let mut dict = SharedDictBytes::default(); - let v05_span = from_v04_span(span, &mut dict).unwrap(); + let v05_span = from_v04_span(&span, &mut dict).unwrap(); let get_index_from_str = |str: &str| -> u32 { dict.iter() @@ -420,7 +420,7 @@ mod tests { }]; let mut dict = SharedDictBytes::default(); - let v05_span = from_v04_span(span, &mut dict).unwrap(); + let v05_span = from_v04_span(&span, &mut dict).unwrap(); let links_json = meta_json(&dict, &v05_span, "_dd.span_links").unwrap(); assert_eq!( @@ -441,7 +441,7 @@ mod tests { #[test] fn from_v04_span_empty_links_events_no_meta_keys_test() { let mut dict = SharedDictBytes::default(); - let v05_span = from_v04_span(base_span(), &mut dict).unwrap(); + let v05_span = from_v04_span(&base_span(), &mut dict).unwrap(); assert_eq!(v05_span.meta.len(), 1); assert!(meta_json(&dict, &v05_span, "_dd.span_links").is_none()); assert!(meta_json(&dict, &v05_span, "events").is_none()); @@ -459,7 +459,7 @@ mod tests { .into(); let mut dict = SharedDictBytes::default(); - let v05_span = from_v04_span(span, &mut dict).unwrap(); + let v05_span = from_v04_span(&span, &mut dict).unwrap(); assert_eq!(v05_span.meta.len(), 1); assert!(meta_json(&dict, &v05_span, "appsec").is_none()); assert!(meta_json(&dict, &v05_span, "meta_struct").is_none()); diff --git a/libdd-trace-utils/src/trace_filter.rs b/libdd-trace-utils/src/trace_filter.rs index 0f59959d7a..29d72aac29 100644 --- a/libdd-trace-utils/src/trace_filter.rs +++ b/libdd-trace-utils/src/trace_filter.rs @@ -8,6 +8,7 @@ use libdd_common::regex_engine::Regex; use libdd_trace_normalization::{normalize_utils, normalizer}; use tracing::{debug, error}; +use crate::span::span_pool::PooledChunks; use crate::span::v1::{AttributeValue, SpanKind, TraceChunk}; use crate::span::vec_map::VecMap; use crate::span::{self, trace_utils::get_root_span_index, trace_utils_v1, TraceData}; @@ -274,18 +275,20 @@ impl TraceFilterer { } /// Removes traces that fail filter checks in-place. Returns the number of traces dropped. - pub fn filter_traces(&self, traces: &mut Vec>>) -> usize { + pub fn filter_traces(&self, traces: &mut PooledChunks<'_, T>) -> usize { let traces_count_before = traces.len(); - traces.retain(|trace| { + let (pool, traces) = traces.chunks_mut(); + let dropped_spans = traces.extract_if(.., |trace| { let Ok(root_span_index) = get_root_span_index(trace) else { - return true; + return false; }; let should_drop = self.should_drop(&trace[root_span_index]); if should_drop { debug!("Trace rejected as it fails to meet tag requirements. root: %v"); } - !should_drop + should_drop }); + pool.add_chunks(dropped_spans); let traces_count_after = traces.len(); traces_count_before - traces_count_after @@ -401,6 +404,7 @@ impl TraceFilterer { #[cfg(test)] mod tests { use super::TraceFilterer; + use crate::span::span_pool::PooledChunks; use crate::span::v04::{SpanBytes, VecMap}; use crate::span::v1::{ AttributeValue as AttributeValueV1, SpanBytes as SpanBytesV1, TraceChunk, @@ -423,8 +427,8 @@ mod tests { } } - fn one_trace(s: SpanBytes) -> Vec> { - vec![vec![s]] + fn one_trace(s: SpanBytes) -> PooledChunks<'static, crate::span::BytesData> { + PooledChunks::unpooled(vec![vec![s]]) } fn v1_chunk_with( @@ -701,10 +705,10 @@ mod tests { #[test] fn multiple_traces_partial_rejection() { let f = reject_str(&["env:prod"]); - let mut traces = vec![ + let mut traces = PooledChunks::unpooled(vec![ vec![span_with("r", &[("env", "prod")])], // dropped vec![span_with("r", &[("env", "staging")])], // kept - ]; + ]); f.filter_traces(&mut traces); assert_eq!(traces.len(), 1); } @@ -712,10 +716,10 @@ mod tests { #[test] fn no_filters_keeps_all_traces() { let f = TraceFilterer::new(&[], &[], &[], &[], &[]); - let mut traces = vec![ + let mut traces = PooledChunks::unpooled(vec![ vec![span_with("r1", &[])], vec![span_with("r2", &[("env", "prod")])], - ]; + ]); f.filter_traces(&mut traces); assert_eq!(traces.len(), 2); } diff --git a/libdd-trace-utils/src/trace_utils.rs b/libdd-trace-utils/src/trace_utils.rs index e7d03dbd67..1d71123055 100644 --- a/libdd-trace-utils/src/trace_utils.rs +++ b/libdd-trace-utils/src/trace_utils.rs @@ -598,13 +598,13 @@ pub fn enrich_span_with_azure_function_metadata(span: &mut pb::Span) { /// Returns `Err` if any span fails to convert (e.g. unsupported field value); the partial /// dictionary built so far is discarded. pub fn convert_trace_chunks_v04_to_v05( - traces: Vec>>, + traces: &[Vec>], ) -> anyhow::Result> { let mut shared_dict = SharedDict::default(); let mut v05_traces: Vec> = Vec::with_capacity(traces.len()); for trace in traces { let v05_trace = trace - .into_iter() + .iter() .map(|span| v05::from_v04_span(span, &mut shared_dict)) .collect::>>()?; v05_traces.push(v05_trace); @@ -1130,7 +1130,7 @@ mod tests { fn test_convert_trace_chunks_v04_to_v05() { let chunk = vec![create_test_no_alloc_span(123, 456, 789, 1, true)]; - let collection = convert_trace_chunks_v04_to_v05(vec![chunk]).unwrap(); + let collection = convert_trace_chunks_v04_to_v05(&[chunk]).unwrap(); let (dict, traces) = match collection { TraceChunks::V05(payload) => payload, diff --git a/libdd-trace-utils/src/tracer_payload.rs b/libdd-trace-utils/src/tracer_payload.rs index b68c343d6c..f14e75e4b7 100644 --- a/libdd-trace-utils/src/tracer_payload.rs +++ b/libdd-trace-utils/src/tracer_payload.rs @@ -267,7 +267,7 @@ pub fn decode_to_trace_chunks( let (data, size) = msgpack_decoder::v05::from_bytes(data).map_err(|e| { anyhow::format_err!("Error deserializing trace from request body: {e}") })?; - Ok((convert_trace_chunks_v04_to_v05(data)?, size)) + Ok((convert_trace_chunks_v04_to_v05(&data)?, size)) } TraceEncoding::V1 => { let (data, size) = msgpack_decoder::v1::from_bytes(data).map_err(|e| {