feat(stats)!: per-key cardinality limits

datadog-ipc/src/shm_stats.rs

@@ -56,7 +56,9 @@ use zwohash::ZwoHasher;
 
 use libdd_ddsketch::DDSketch;
 use libdd_trace_protobuf::pb;
-use libdd_trace_stats::span_concentrator::{FixedAggregationKey, FlushableConcentrator};
+use libdd_trace_stats::span_concentrator::{
+    FixedAggregationKey, FlushResult, FlushableConcentrator,
+};
 
 use crate::platform::{FileBackedHandle, MappedMem, NamedShmHandle};
 
@@ -820,8 +822,13 @@ impl ShmSpanConcentrator {
 }
 
 impl FlushableConcentrator for ShmSpanConcentrator {
-    fn flush_buckets(&mut self, force: bool) -> (Vec<pb::ClientStatsBucket>, u64) {
-        (self.drain_buckets(force), 0)
+    fn flush_buckets(&mut self, force: bool) -> FlushResult {
+        // The SHM concentrator does not perform client-side obfuscation.
+        FlushResult {
+            obfuscated_buckets: vec![],
+            unobfuscated_buckets: self.drain_buckets(force),
+            collapsed_spans: 0,
+        }
     }
 }
 

datadog-sidecar/src/service/stats_flusher.rs

@@ -111,12 +111,6 @@ fn make_exporter(
         s.meta.clone(),
         endpoint,
         NativeCapabilities::new_client(),
-        // Sidecar does not perform client-side stats obfuscation. Pass a disabled
-        // default so the `datadog-obfuscation-version` header is never sent.
-        #[cfg(feature = "stats-obfuscation")]
-        Arc::new(arc_swap::ArcSwap::from_pointee(
-            libdd_trace_stats::span_concentrator::StatsComputationObfuscationConfig::default(),
-        )),
         #[cfg(feature = "stats-obfuscation")]
         "0",
         None,

libdd-data-pipeline-ffi/cbindgen.toml

@@ -47,4 +47,4 @@ must_use = "DDOG_CHECK_RETURN"
 
 [parse]
 parse_deps = true
-include = ["libdd-common", "libdd-common-ffi", "libdd-shared-runtime", "libdd-data-pipeline", "libdd-trace-utils"]
+include = ["libdd-common", "libdd-common-ffi", "libdd-shared-runtime", "libdd-data-pipeline", "libdd-trace-utils", "libdd-trace-stats"]

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

@@ -9,6 +9,7 @@ use libdd_common_ffi::{
     CharSlice,
     {slice::AsBytes, slice::ByteSlice},
 };
+use libdd_data_pipeline::trace_exporter::stats::CardinalityLimitConfig;
 use libdd_data_pipeline::trace_exporter::{
     TelemetryConfig, TelemetryInstrumentationSessions, TraceExporter as GenericTraceExporter,
     TraceExporterInputFormat, TraceExporterOutputFormat,
@@ -90,7 +91,7 @@ pub struct TraceExporterConfig {
     otlp_protocol: Option<OtlpProtocol>,
     output_to_log: bool,
     log_max_line_size: Option<usize>,
-    stats_cardinality_limit: Option<usize>,
+    stats_cardinality_limits: Option<CardinalityLimitConfig>,
 }
 
 #[no_mangle]
@@ -555,11 +556,11 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_set_otlp_protocol(
 #[no_mangle]
 pub unsafe extern "C" fn ddog_trace_exporter_config_set_stats_cardinality_limit(
     config: Option<&mut TraceExporterConfig>,
-    limit: usize,
+    limits: CardinalityLimitConfig,
 ) -> Option<Box<ExporterError>> {
     catch_panic!(
         if let Some(handle) = config {
-            handle.stats_cardinality_limit = Some(limit);
+            handle.stats_cardinality_limits = Some(limits);
             None
         } else {
             gen_error!(ErrorCode::InvalidArgument)
@@ -645,7 +646,7 @@ pub unsafe extern "C" fn ddog_trace_exporter_new(
                 .set_output_format(config.output_format)
                 .set_connection_timeout(config.connection_timeout);
 
-            if let Some(limit) = config.stats_cardinality_limit {
+            if let Some(limit) = config.stats_cardinality_limits {
                 builder.set_stats_cardinality_limit(limit);
             }
 
@@ -782,7 +783,7 @@ mod tests {
             assert!(cfg.connection_timeout.is_none());
             assert!(!cfg.output_to_log);
             assert_eq!(cfg.log_max_line_size, None);
-            assert_eq!(cfg.stats_cardinality_limit, None);
+            assert_eq!(cfg.stats_cardinality_limits, None);
 
             ddog_trace_exporter_config_free(cfg);
         }
@@ -1501,16 +1502,34 @@ mod tests {
     fn config_stats_cardinality_limit_test() {
         unsafe {
             // Null config → InvalidArgument
-            let error = ddog_trace_exporter_config_set_stats_cardinality_limit(None, 100);
+            let error = ddog_trace_exporter_config_set_stats_cardinality_limit(
+                None,
+                CardinalityLimitConfig {
+                    whole_key_limit: 100,
+                    ..Default::default()
+                },
+            );
             assert_eq!(error.as_ref().unwrap().code, ErrorCode::InvalidArgument);
             ddog_trace_exporter_error_free(error);
 
             // Valid config → value stored
             let mut config = Some(TraceExporterConfig::default());
-            let error =
-                ddog_trace_exporter_config_set_stats_cardinality_limit(config.as_mut(), 500);
+            let error = ddog_trace_exporter_config_set_stats_cardinality_limit(
+                config.as_mut(),
+                CardinalityLimitConfig {
+                    whole_key_limit: 500,
+                    ..Default::default()
+                },
+            );
             assert_eq!(error, None);
-            assert_eq!(config.unwrap().stats_cardinality_limit, Some(500));
+            assert_eq!(
+                config
+                    .unwrap()
+                    .stats_cardinality_limits
+                    .unwrap()
+                    .whole_key_limit,
+                500
+            );
         }
     }
 

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

@@ -255,7 +255,7 @@ pub struct OtlpStatsExporter<C: HttpClientCapability + SleepCapability> {
 impl<C: HttpClientCapability + SleepCapability> OtlpStatsExporter<C> {
     /// Flush the concentrator and export stats; returns `Ok(true)` if anything was sent.
     async fn send(&self, force_flush: bool, max_retries: u32) -> anyhow::Result<bool> {
-        let (buckets, _collapsed_count) = {
+        let buckets = {
             #[allow(clippy::unwrap_used)]
             let mut c = self.concentrator.lock().unwrap();
             c.flush_with_otlp_exact(SystemTime::now(), force_flush)

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

@@ -26,6 +26,7 @@ use libdd_dogstatsd_client::new;
 use libdd_shared_runtime::SharedRuntime;
 #[cfg(not(target_arch = "wasm32"))]
 use libdd_shared_runtime::{BlockingRuntime, ForkSafeRuntime};
+use libdd_trace_stats::span_concentrator::CardinalityLimitConfig;
 use libdd_trace_utils::trace_filter::TraceFilterer;
 use std::sync::Arc;
 use std::time::Duration;
@@ -75,7 +76,7 @@ pub struct TraceExporterBuilder<R: SharedRuntime> {
     peer_tags_aggregation: bool,
     compute_stats_by_span_kind: bool,
     peer_tags: Vec<String>,
-    stats_cardinality_limit: Option<usize>,
+    stats_cardinality_limits: Option<CardinalityLimitConfig>,
     #[cfg(feature = "stats-obfuscation")]
     client_side_stats_obfuscation_enabled: bool,
     #[cfg(feature = "telemetry")]
@@ -143,7 +144,7 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
             peer_tags_aggregation: false,
             compute_stats_by_span_kind: false,
             peer_tags: Vec::new(),
-            stats_cardinality_limit: None,
+            stats_cardinality_limits: None,
             #[cfg(feature = "stats-obfuscation")]
             client_side_stats_obfuscation_enabled: false,
             #[cfg(feature = "telemetry")]
@@ -333,8 +334,11 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
     /// This bounds memory usage when the trace population has very high cardinality.
     ///
     /// Has no effect unless stats computation is enabled.
-    pub fn set_stats_cardinality_limit(&mut self, cardinality_limit: usize) -> &mut Self {
-        self.stats_cardinality_limit = Some(cardinality_limit);
+    pub fn set_stats_cardinality_limit(
+        &mut self,
+        cardinality_limits: CardinalityLimitConfig,
+    ) -> &mut Self {
+        self.stats_cardinality_limits = Some(cardinality_limits);
         self
     }
 
@@ -820,7 +824,7 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
             common_stats_tags: vec![libdatadog_version],
             client_side_stats: StatsComputationConfig {
                 status: ArcSwap::new(stats.into()),
-                stats_cardinality_limit: self.stats_cardinality_limit,
+                stats_cardinality_limits: self.stats_cardinality_limits,
                 #[cfg(feature = "stats-obfuscation")]
                 obfuscation_config: Arc::new(ArcSwap::from_pointee(
                     StatsComputationObfuscationConfig::default(),

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

@@ -469,7 +469,7 @@ impl<
                     metadata: &self.metadata,
                     endpoint_url: &self.endpoint.url,
                     shared_runtime: &*self.shared_runtime,
-                    stats_cardinality_limit: self.client_side_stats.stats_cardinality_limit,
+                    stats_cardinality_limits: self.client_side_stats.stats_cardinality_limits,
                     dogstatsd: if self.health_metrics_enabled {
                         self.dogstatsd.clone()
                     } else {
@@ -875,8 +875,8 @@ impl<
                     && self.v1_active.swap(false, Ordering::Relaxed)
                 {
                     warn!(
-                            "V1 trace send returned 404; agent no longer advertises {V1_TRACES_ENDPOINT} — falling back to V0.4"
-                        );
+                        "V1 trace send returned 404; agent no longer advertises {V1_TRACES_ENDPOINT} — falling back to V0.4"
+                    );
                     self.info_response_observer.manual_trigger();
                 }
             }

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

@@ -7,6 +7,8 @@
 //! including starting/stopping stats workers, managing the span concentrator,
 //! and processing traces for stats collection.
 
+pub use libdd_trace_stats::span_concentrator::CardinalityLimitConfig;
+
 #[cfg(not(target_arch = "wasm32"))]
 use super::add_path;
 use super::TracerMetadata;
@@ -49,7 +51,7 @@ pub(crate) struct StatsContext<'a, R: SharedRuntime> {
     pub metadata: &'a TracerMetadata,
     pub endpoint_url: &'a http::Uri,
     pub shared_runtime: &'a R,
-    pub stats_cardinality_limit: Option<usize>,
+    pub stats_cardinality_limits: Option<CardinalityLimitConfig>,
     /// Optional DogStatsD client forwarded to the [`StatsExporter`].
     pub dogstatsd: Option<std::sync::Arc<libdd_dogstatsd_client::Client>>,
     /// Optional telemetry handle forwarded to the [`StatsExporter`].
@@ -77,7 +79,7 @@ pub(crate) enum StatsComputationStatus {
 #[cfg_attr(target_arch = "wasm32", allow(dead_code))]
 pub(crate) struct StatsComputationConfig {
     pub(crate) status: ArcSwap<StatsComputationStatus>,
-    pub(crate) stats_cardinality_limit: Option<usize>,
+    pub(crate) stats_cardinality_limits: Option<CardinalityLimitConfig>,
     #[cfg(feature = "stats-obfuscation")]
     pub(crate) obfuscation_config: SharedStatsComputationObfuscationConfig,
     /// Builder-level opt-in. When false, stats obfuscation stays off
@@ -107,7 +109,7 @@ fn is_obfuscation_active(agent_info: &AgentInfo) -> bool {
     agent_info
         .info
         .obfuscation_version
-        .is_some_and(|v| v >= 1 && v <= SUPPORTED_OBFUSCATION_VERSION)
+        .is_some_and(|v| v >= 1 && v == SUPPORTED_OBFUSCATION_VERSION)
 }
 
 #[cfg(not(target_arch = "wasm32"))]
@@ -142,7 +144,7 @@ pub(crate) fn start_stats_computation<
             std::time::SystemTime::now(),
             span_kinds,
             peer_tags,
-            ctx.stats_cardinality_limit,
+            ctx.stats_cardinality_limits,
             #[cfg(feature = "stats-obfuscation")]
             Some(client_side_stats.obfuscation_config.clone()),
         )));
@@ -170,8 +172,6 @@ fn create_and_start_stats_worker<
         Endpoint::from_url(add_path(ctx.endpoint_url, STATS_ENDPOINT)),
         capabilities.clone(),
         #[cfg(feature = "stats-obfuscation")]
-        client_side_stats.obfuscation_config.clone(),
-        #[cfg(feature = "stats-obfuscation")]
         SUPPORTED_OBFUSCATION_VERSION_STR,
         #[cfg(feature = "telemetry")]
         ctx.telemetry.clone(),
@@ -258,6 +258,7 @@ fn update_obfuscation_config(
     ) {
         let obfuscation_active =
             client_side_stats.obfuscation_enabled && is_obfuscation_active(agent_info);
+        // FIXME(APMSP-3720): there is more than this to obfuscation config
         let sql_obfuscation_mode = (|| {
             agent_info
                 .info

libdd-trace-stats/Cargo.toml

@@ -29,6 +29,7 @@ tokio = { version = "1.23", features = ["macros", "time"], default-features = fa
 tokio-util = "0.7.11"
 tracing = { version = "0.1", default-features = false }
 async-trait = "0.1.85"
+futures = { version = "0.3.32", default-features = false, features = ["alloc"] }
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 libdd-capabilities-impl = { version = "3.0.0", path = "../libdd-capabilities-impl", default-features = false }