feat(data-pipeline)!: CSS Trace Filters

libdd-common/src/regex_engine.rs

@@ -13,7 +13,7 @@
 //! regexes requiring Unicode character class support.
 
 #[cfg(all(feature = "regex-lite", not(feature = "require-regex-full")))]
-pub use regex_lite::{escape, Captures, Regex, RegexBuilder, Replacer};
+pub use regex_lite::{escape, Captures, Error, Regex, RegexBuilder, Replacer};
 
 #[cfg(not(all(feature = "regex-lite", not(feature = "require-regex-full"))))]
-pub use regex::{escape, Captures, Regex, RegexBuilder, Replacer};
+pub use regex::{escape, Captures, Error, Regex, RegexBuilder, Replacer};

libdd-data-pipeline/Cargo.toml

@@ -36,6 +36,7 @@ libdd-common = { version = "4.2.0", path = "../libdd-common", default-features =
 libdd-shared-runtime = { version = "1.0.0", path = "../libdd-shared-runtime", default-features = false }
 libdd-telemetry = { version = "5.0.1", path = "../libdd-telemetry", default-features = false, optional = true}
 libdd-trace-protobuf = { version = "3.0.2", path = "../libdd-trace-protobuf" }
+libdd-trace-normalization = { version = "2.0.0", path = "../libdd-trace-normalization" }
 libdd-trace-stats = { version = "5.0.0", path = "../libdd-trace-stats", default-features = false }
 libdd-trace-utils = { version = "8.0.0", path = "../libdd-trace-utils", default-features = false }
 libdd-trace-obfuscation = { version = "4.0.0", path = "../libdd-trace-obfuscation", default-features = false, optional = true }

libdd-data-pipeline/src/agent_info/schema.rs

@@ -40,20 +40,25 @@ pub struct AgentInfoStruct {
     /// Container tags hash from HTTP response header
     pub container_tags_hash: Option<String>,
     /// Exact-match tag filters applied before stats computation (root span only).
-    pub filter_tags: Option<FilterTagsConfig>,
+    #[serde(default)]
+    pub filter_tags: FilterTagsConfig,
     /// Regex-match tag filters applied before stats computation (root span only).
-    pub filter_tags_regex: Option<FilterTagsConfig>,
+    #[serde(default)]
+    pub filter_tags_regex: FilterTagsConfig,
     /// Regex patterns for root-span resource names; matching traces are excluded from stats.
-    pub ignore_resources: Option<Vec<String>>,
+    #[serde(default)]
+    pub ignore_resources: Vec<String>,
 }
 
 /// Require/reject lists for tag-based trace filters exposed by the agent /info endpoint.
 #[derive(Clone, Serialize, Deserialize, Default, Debug, PartialEq)]
 pub struct FilterTagsConfig {
     /// All listed filters must match at least one root-span tag for the trace to be accepted.
-    pub require: Option<Vec<String>>,
+    #[serde(default)]
+    pub require: Vec<String>,
     /// If any listed filter matches a root-span tag the trace is rejected.
-    pub reject: Option<Vec<String>>,
+    #[serde(default)]
+    pub reject: Vec<String>,
 }
 
 #[allow(missing_docs)]

libdd-data-pipeline/src/telemetry/metrics.rs

@@ -27,6 +27,8 @@ pub enum MetricKind {
     ChunksSent,
     /// trace_chunks_dropped metric (reason: p0_drop)
     ChunksDroppedP0,
+    /// trace_chunks_dropped metric (reason: trace_filters)
+    ChunksDroppedByTraceFilter,
     /// trace_chunks_dropped metric (reason: serialization_error)
     ChunksDroppedSerializationError,
     /// trace_chunks_dropped metric (reason: send_failure)
@@ -102,6 +104,15 @@ const METRICS: &[Metric] = &[
         namespace: MetricNamespace::Tracers,
         tags: &[tag!["src_library", "libdatadog"], tag!["reason", "p0_drop"]],
     },
+    Metric {
+        name: CHUNKS_DROPPED_STR,
+        metric_type: MetricType::Count,
+        namespace: MetricNamespace::Tracers,
+        tags: &[
+            tag!["src_library", "libdatadog"],
+            tag!["reason", "trace_filters"],
+        ],
+    },
     Metric {
         name: CHUNKS_DROPPED_STR,
         metric_type: MetricType::Count,

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

@@ -13,6 +13,7 @@ use libdd_telemetry::worker::{
 };
 use libdd_trace_utils::{
     send_with_retry::{SendWithRetryError, SendWithRetryResult},
+    span::trace_utils::DroppedStats,
     trace_utils::SendDataResult,
 };
 use std::{collections::HashMap, time::Duration};
@@ -165,6 +166,7 @@ pub struct SendPayloadTelemetry {
     bytes_sent: u64,
     chunks_sent: u64,
     chunks_dropped_p0: u64,
+    chunks_dropped_by_trace_filter: u64,
     chunks_dropped_serialization_error: u64,
     chunks_dropped_send_failure: u64,
     responses_count_per_code: HashMap<u16, u64>,
@@ -193,15 +195,16 @@ impl SendPayloadTelemetry {
     /// * `value` - The result of sending traces with retry
     /// * `bytes_sent` - The number of bytes in the payload
     /// * `chunks` - The number of trace chunks in the payload
-    /// * `chunks_dropped_p0` - The number of P0 trace chunks dropped due to sampling
+    /// * `dropped_stats` - Trace dropped stats from `stats::process_traces_for_stats`
     pub fn from_retry_result(
         value: &SendWithRetryResult,
         bytes_sent: u64,
         chunks: u64,
-        chunks_dropped_p0: u64,
+        dropped_stats: DroppedStats,
     ) -> Self {
         let mut telemetry = Self {
-            chunks_dropped_p0,
+            chunks_dropped_p0: dropped_stats.dropped_p0_traces as u64,
+            chunks_dropped_by_trace_filter: dropped_stats.dropped_by_trace_filter as u64,
             ..Default::default()
         };
         match value {
@@ -288,6 +291,13 @@ impl TelemetryClient {
             self.worker
                 .add_point(data.chunks_dropped_p0 as f64, key, vec![])?;
         }
+        if data.chunks_dropped_by_trace_filter > 0 {
+            let key = self
+                .metrics
+                .get(metrics::MetricKind::ChunksDroppedByTraceFilter);
+            self.worker
+                .add_point(data.chunks_dropped_by_trace_filter as f64, key, vec![])?;
+        }
         if data.chunks_dropped_serialization_error > 0 {
             let key = self
                 .metrics
@@ -704,7 +714,16 @@ mod tests {
                 .unwrap(),
             3,
         ));
-        let telemetry = SendPayloadTelemetry::from_retry_result(&result, 4, 5, 0);
+        let telemetry = SendPayloadTelemetry::from_retry_result(
+            &result,
+            4,
+            5,
+            DroppedStats {
+                dropped_p0_traces: 0,
+                dropped_p0_spans: 0,
+                dropped_by_trace_filter: 0,
+            },
+        );
         assert_eq!(
             telemetry,
             SendPayloadTelemetry {
@@ -726,7 +745,16 @@ mod tests {
                 .unwrap(),
             3,
         ));
-        let telemetry = SendPayloadTelemetry::from_retry_result(&result, 4, 5, 10);
+        let telemetry = SendPayloadTelemetry::from_retry_result(
+            &result,
+            4,
+            5,
+            DroppedStats {
+                dropped_p0_traces: 10,
+                dropped_p0_spans: 0,
+                dropped_by_trace_filter: 0,
+            },
+        );
         assert_eq!(
             telemetry,
             SendPayloadTelemetry {
@@ -747,7 +775,16 @@ mod tests {
             .body(Bytes::new())
             .unwrap();
         let result = Err(SendWithRetryError::Http(error_response, 5));
-        let telemetry = SendPayloadTelemetry::from_retry_result(&result, 1, 2, 0);
+        let telemetry = SendPayloadTelemetry::from_retry_result(
+            &result,
+            1,
+            2,
+            DroppedStats {
+                dropped_p0_traces: 0,
+                dropped_p0_spans: 0,
+                dropped_by_trace_filter: 0,
+            },
+        );
         assert_eq!(
             telemetry,
             SendPayloadTelemetry {
@@ -766,7 +803,16 @@ mod tests {
             HttpError::Network(anyhow::anyhow!("connection refused")),
             5,
         ));
-        let telemetry = SendPayloadTelemetry::from_retry_result(&result, 1, 2, 0);
+        let telemetry = SendPayloadTelemetry::from_retry_result(
+            &result,
+            1,
+            2,
+            DroppedStats {
+                dropped_p0_traces: 0,
+                dropped_p0_spans: 0,
+                dropped_by_trace_filter: 0,
+            },
+        );
         assert_eq!(
             telemetry,
             SendPayloadTelemetry {
@@ -781,7 +827,16 @@ mod tests {
     #[test]
     fn telemetry_from_timeout_error_test() {
         let result = Err(SendWithRetryError::Timeout(5));
-        let telemetry = SendPayloadTelemetry::from_retry_result(&result, 1, 2, 0);
+        let telemetry = SendPayloadTelemetry::from_retry_result(
+            &result,
+            1,
+            2,
+            DroppedStats {
+                dropped_p0_traces: 0,
+                dropped_p0_spans: 0,
+                dropped_by_trace_filter: 0,
+            },
+        );
         assert_eq!(
             telemetry,
             SendPayloadTelemetry {
@@ -797,7 +852,16 @@ mod tests {
     #[test]
     fn telemetry_from_build_error_test() {
         let result = Err(SendWithRetryError::Build(5));
-        let telemetry = SendPayloadTelemetry::from_retry_result(&result, 1, 2, 0);
+        let telemetry = SendPayloadTelemetry::from_retry_result(
+            &result,
+            1,
+            2,
+            DroppedStats {
+                dropped_p0_traces: 0,
+                dropped_p0_spans: 0,
+                dropped_by_trace_filter: 0,
+            },
+        );
         assert_eq!(
             telemetry,
             SendPayloadTelemetry {

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

@@ -22,6 +22,7 @@ use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability};
 use libdd_common::{parse_uri, tag, Endpoint};
 use libdd_dogstatsd_client::new;
 use libdd_shared_runtime::SharedRuntime;
+use libdd_trace_utils::trace_filter::TraceFilterer;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -514,6 +515,7 @@ impl TraceExporterBuilder {
                 .agent_rates_payload_version_enabled
                 .then(AgentResponsePayloadVersion::new),
             otlp_config,
+            trace_filterer: ArcSwap::from_pointee(TraceFilterer::with_empty_conf()),
         })
     }
 

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

@@ -9,6 +9,7 @@ mod trace_serializer;
 
 // Re-export the builder
 pub use builder::TraceExporterBuilder;
+use libdd_trace_utils::{span::trace_utils::DroppedStats, trace_filter::TraceFilterer};
 
 use self::agent_response::AgentResponse;
 use self::metrics::MetricsEmitter;
@@ -30,7 +31,7 @@ use crate::{
     health_metrics,
     health_metrics::{HealthMetric, SendResult, TransportErrorType},
 };
-use arc_swap::ArcSwapOption;
+use arc_swap::{ArcSwap, ArcSwapOption};
 use bytes::Bytes;
 use http::header::HeaderMap;
 use http::uri::PathAndQuery;
@@ -213,6 +214,7 @@ pub struct TraceExporter<C: HttpClientCapability + SleepCapability + MaybeSend +
     agent_payload_response_version: Option<AgentResponsePayloadVersion>,
     /// When set, traces are exported via OTLP HTTP/JSON instead of the Datadog agent.
     otlp_config: Option<OtlpTraceConfig>,
+    trace_filterer: ArcSwap<TraceFilterer>,
 }
 
 impl<C: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static> TraceExporter<C> {
@@ -348,9 +350,9 @@ impl<C: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static> Tra
                 .map(|s| s.as_str())
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     /// Reconcile in-process stats state with the latest agent info.
     /// Async so the `Enabled` arm can await a stats-worker shutdown without `block_on`.
+    #[cfg(not(target_arch = "wasm32"))]
     async fn check_agent_info(&self) {
         let Some(agent_info) = agent_info::get_agent_info() else {
             return;
@@ -363,6 +365,14 @@ impl<C: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static> Tra
             self.refresh_v1_active(&agent_info);
         }
 
+        self.trace_filterer.store(Arc::new(TraceFilterer::new(
+            &agent_info.info.filter_tags.require,
+            &agent_info.info.filter_tags.reject,
+            &agent_info.info.filter_tags_regex.require,
+            &agent_info.info.filter_tags_regex.reject,
+            &agent_info.info.ignore_resources,
+        )));
+
         // load_full() avoids holding an ArcSwap Guard (!Send) across .await.
         let status = self.client_side_stats.status.load_full();
         match &*status {
@@ -567,7 +577,8 @@ impl<C: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static> Tra
         mp_payload: Vec<u8>,
         headers: HeaderMap,
         chunks: usize,
-        #[cfg_attr(not(feature = "telemetry"), allow(unused_variables))] chunks_dropped_p0: usize,
+        #[cfg_attr(not(feature = "telemetry"), allow(unused_variables))]
+        dropped_stats: DroppedStats,
     ) -> Result<AgentResponse, TraceExporterError> {
         let strategy = RetryStrategy::default();
         let payload_len = mp_payload.len();
@@ -588,7 +599,7 @@ impl<C: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static> Tra
                 &result,
                 payload_len as u64,
                 chunks as u64,
-                chunks_dropped_p0 as u64,
+                dropped_stats,
             )) {
                 error!(?e, "Error sending telemetry");
             }
@@ -605,11 +616,12 @@ impl<C: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static> Tra
 
         // Process stats computation and drop non-sampled (p0) chunks.
         // This must run before the OTLP path so that unsampled spans are not exported.
-        let dropped_p0_stats = stats::process_traces_for_stats(
+        let dropped_stats = stats::process_traces_for_stats(
             &mut traces,
             &mut header_tags,
             &self.client_side_stats.status,
             self.client_computed_top_level,
+            &self.trace_filterer.load(),
         );
 
         for chunk in &mut traces {
@@ -662,7 +674,7 @@ impl<C: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static> Tra
                 prepared.data,
                 prepared.headers,
                 prepared.chunk_count,
-                dropped_p0_stats.dropped_p0_traces,
+                dropped_stats,
             )
             .await;