diff --git a/Cargo.lock b/Cargo.lock
index 2917cbba83..3fd5276f74 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1954,9 +1954,9 @@ checksum = "673464e1e314dd67a0fd9544abc99e8eb28d0c7e3b69b033bcff9b2d00b87333"
 
 [[package]]
 name = "futures"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876"
+checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -1969,9 +1969,9 @@ dependencies = [
 
 [[package]]
 name = "futures-channel"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10"
+checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d"
 dependencies = [
  "futures-core",
  "futures-sink",
@@ -1979,15 +1979,15 @@ dependencies = [
 
 [[package]]
 name = "futures-core"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e"
+checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d"
 
 [[package]]
 name = "futures-executor"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f"
+checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d"
 dependencies = [
  "futures-core",
  "futures-task",
@@ -1996,9 +1996,9 @@ dependencies = [
 
 [[package]]
 name = "futures-io"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6"
+checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718"
 
 [[package]]
 name = "futures-lite"
@@ -2015,9 +2015,9 @@ dependencies = [
 
 [[package]]
 name = "futures-macro"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650"
+checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2026,15 +2026,15 @@ dependencies = [
 
 [[package]]
 name = "futures-sink"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7"
+checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893"
 
 [[package]]
 name = "futures-task"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988"
+checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393"
 
 [[package]]
 name = "futures-timer"
@@ -2044,9 +2044,9 @@ checksum = "f288b0a4f20f9a56b5d1da57e2227c661b7b16168e2f72365f57b63326e29b24"
 
 [[package]]
 name = "futures-util"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81"
+checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -2056,7 +2056,6 @@ dependencies = [
  "futures-task",
  "memchr",
  "pin-project-lite",
- "pin-utils",
  "slab",
 ]
 
@@ -3495,6 +3494,7 @@ dependencies = [
  "arc-swap",
  "async-trait",
  "criterion",
+ "futures",
  "hashbrown 0.15.1",
  "http",
  "httpmock",
diff --git a/datadog-ipc/src/shm_stats.rs b/datadog-ipc/src/shm_stats.rs
index 6a7c513fda..87b55981da 100644
--- a/datadog-ipc/src/shm_stats.rs
+++ b/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,
+        }
     }
 }
 
diff --git a/datadog-sidecar/src/service/stats_flusher.rs b/datadog-sidecar/src/service/stats_flusher.rs
index 13332bbb82..ac144e669d 100644
--- a/datadog-sidecar/src/service/stats_flusher.rs
+++ b/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,
diff --git a/libdd-data-pipeline-ffi/cbindgen.toml b/libdd-data-pipeline-ffi/cbindgen.toml
index d3e36b4945..5e18df2f87 100644
--- a/libdd-data-pipeline-ffi/cbindgen.toml
+++ b/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"]
diff --git a/libdd-data-pipeline-ffi/src/trace_exporter.rs b/libdd-data-pipeline-ffi/src/trace_exporter.rs
index c3a68962a4..f7369386b7 100644
--- a/libdd-data-pipeline-ffi/src/trace_exporter.rs
+++ b/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
+            );
         }
     }
 
diff --git a/libdd-data-pipeline/results.xml b/libdd-data-pipeline/results.xml
deleted file mode 100644
index e69de29bb2..0000000000
diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs
index 46104b12f1..21c30c1785 100644
--- a/libdd-data-pipeline/src/otlp/metrics.rs
+++ b/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)
diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs
index eef1c9e3f9..628ab71eaf 100644
--- a/libdd-data-pipeline/src/trace_exporter/builder.rs
+++ b/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(),
diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs
index 08f69e8b99..f0a4e5c9a7 100644
--- a/libdd-data-pipeline/src/trace_exporter/mod.rs
+++ b/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();
                 }
             }
diff --git a/libdd-data-pipeline/src/trace_exporter/stats.rs b/libdd-data-pipeline/src/trace_exporter/stats.rs
index 8d7c7a47f1..4ccf3f0fb1 100644
--- a/libdd-data-pipeline/src/trace_exporter/stats.rs
+++ b/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
diff --git a/libdd-trace-stats/Cargo.toml b/libdd-trace-stats/Cargo.toml
index fd2cf45b78..bb87e1ee0f 100644
--- a/libdd-trace-stats/Cargo.toml
+++ b/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 }
diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs
index 5d17d54056..73873779fb 100644
--- a/libdd-trace-stats/src/span_concentrator/aggregation.rs
+++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs
@@ -5,14 +5,19 @@
 //! This includes the aggregation key to group spans together and the computation of stats from a
 //! span.
 
-use hashbrown::HashMap;
+use hashbrown::{hash_set::Entry, HashMap, HashSet};
 use libdd_trace_obfuscation::ip_address::quantize_peer_ip_addresses;
 use libdd_trace_protobuf::pb;
 use libdd_trace_utils::span::SpanText;
-use std::borrow::{Borrow, Cow};
+use std::{
+    borrow::{Borrow, Cow},
+    hash::{DefaultHasher, Hash, Hasher as _},
+};
 
 use crate::span_concentrator::StatSpan;
 
+use super::CardinalityLimitConfig;
+
 /// Sentinel value used for cardinality limiting.
 pub const TRACER_BLOCKED_VALUE: &str = "tracer_blocked_value";
 
@@ -299,7 +304,7 @@ impl OwnedAggregationKey {
                 is_synthetics_request: false,
                 is_trace_root: pb::Trilean::NotSet,
             },
-            peer_tags: vec![],
+            peer_tags: vec![(TRACER_BLOCKED_VALUE.to_owned(), "".to_owned())],
         }
     }
 }
@@ -428,22 +433,43 @@ pub(super) struct StatsBucket {
     start: u64,
     /// Maximum number of distinct aggregation keys this bucket will hold before collapsing new
     /// ones into the overflow sentinel key.
-    max_entries: usize,
+    cardinality_limits: CardinalityLimitConfig,
+    distinct_resources: HashSet<u64>,
+    distinct_http_endpoints: HashSet<u64>,
+    distinct_peer_tags: HashSet<u64>,
+    #[allow(
+        unused,
+        reason = "FIXME(SVLS-8787|github.com/DataDog/libdatadog/pull/2170): implement stats additional tags"
+    )]
+    distinct_additional_tags: HashSet<u64>,
     /// Number of spans collapsed into the overflow bucket due to cardinality limiting.
     collapsed_count: u64,
+    /// Indicates if stats obfuscated in this bucket. This is set once at creation and stays
+    /// constant per bucket
+    #[cfg(feature = "stats-obfuscation")]
+    pub(super) obfuscated: bool,
 }
 
 impl StatsBucket {
     /// Return a new StatsBucket starting at `start_timestamp`.
     ///
-    /// `max_entries` is the maximum number of distinct aggregation keys the bucket will hold.
-    /// Once the limit is reached, new distinct keys are collapsed into the overflow sentinel key.
-    pub(super) fn new(start_timestamp: u64, max_entries: usize) -> Self {
+    /// `cardinality_limits` are the values for whole-key and per-field cardinality limits
+    pub(super) fn new(
+        start_timestamp: u64,
+        cardinality_limits: CardinalityLimitConfig,
+        #[cfg(feature = "stats-obfuscation")] obfuscation_enabled: bool,
+    ) -> Self {
         Self {
             data: HashMap::new(),
             start: start_timestamp,
-            max_entries,
+            cardinality_limits,
             collapsed_count: 0,
+            #[cfg(feature = "stats-obfuscation")]
+            obfuscated: obfuscation_enabled,
+            distinct_resources: HashSet::new(),
+            distinct_http_endpoints: HashSet::new(),
+            distinct_peer_tags: HashSet::new(),
+            distinct_additional_tags: HashSet::new(),
         }
     }
 
@@ -457,12 +483,18 @@ impl StatsBucket {
     /// instead merged into the overflow sentinel key.
     pub(super) fn insert(
         &mut self,
-        key: BorrowedAggregationKey<'_>,
+        mut key: BorrowedAggregationKey<'_>,
         duration: i64,
         is_error: bool,
         is_top_level: bool,
     ) {
-        if self.data.len() >= self.max_entries && !self.data.contains_key(&key) {
+        // Per field cardinality limiting
+        self.collapse_key_fields_cardinality(&mut key);
+
+        // Whole-key cardinality limiting
+        if self.data.len() >= self.cardinality_limits.whole_key_limit
+            && !self.data.contains_key(&key)
+        {
             self.collapsed_count += 1;
             self.data
                 .entry(OwnedAggregationKey::overflow_key())
@@ -470,12 +502,52 @@ impl StatsBucket {
                 .insert(duration, is_error, is_top_level);
             return;
         }
+
         self.data
             .entry_ref(&key)
             .or_default()
             .insert(duration, is_error, is_top_level);
     }
 
+    /// Collapse an aggregation key fields following the bucket's `CardinalityLimitConfig`
+    fn collapse_key_fields_cardinality(&mut self, key: &mut BorrowedAggregationKey<'_>) {
+        fn hash(input: &impl Hash) -> u64 {
+            let mut hasher = DefaultHasher::new();
+            input.hash(&mut hasher);
+            hasher.finish()
+        }
+
+        let resource_hash = hash(&key.fixed.resource_name);
+        let resources_count = self.distinct_resources.len();
+        if let Entry::Vacant(slot) = self.distinct_resources.entry(resource_hash) {
+            if resources_count >= self.cardinality_limits.resource_limit {
+                key.fixed.resource_name = TRACER_BLOCKED_VALUE;
+            } else {
+                slot.insert();
+            }
+        }
+
+        let http_endpoint_hash = hash(&key.fixed.http_endpoint);
+        let http_endpoints_count = self.distinct_http_endpoints.len();
+        if let Entry::Vacant(slot) = self.distinct_http_endpoints.entry(http_endpoint_hash) {
+            if http_endpoints_count >= self.cardinality_limits.http_endpoint_limit {
+                key.fixed.http_endpoint = TRACER_BLOCKED_VALUE;
+            } else {
+                slot.insert();
+            }
+        }
+
+        let peer_tags_hash = hash(&key.peer_tags);
+        let peer_tags_count = self.distinct_peer_tags.len();
+        if let Entry::Vacant(slot) = self.distinct_peer_tags.entry(peer_tags_hash) {
+            if peer_tags_count >= self.cardinality_limits.peer_tags_limit {
+                key.peer_tags = vec![(TRACER_BLOCKED_VALUE, Cow::Borrowed(""))];
+            } else {
+                slot.insert();
+            }
+        }
+    }
+
     /// Consume the bucket and return a ClientStatsBucket containing the bucket stats.
     /// `bucket_duration` is the size of buckets for the concentrator containing the bucket.
     pub(super) fn flush(self, bucket_duration: u64) -> pb::ClientStatsBucket {
@@ -540,7 +612,13 @@ fn encode_grouped_stats(key: OwnedAggregationKey, group: GroupedStats) -> pb::Cl
         peer_tags: key
             .peer_tags
             .into_iter()
-            .map(|(k, v)| format!("{k}:{v}"))
+            .map(|(k, v)| {
+                if v.is_empty() {
+                    k.to_string()
+                } else {
+                    format!("{k}:{v}")
+                }
+            })
             .collect(),
         is_trace_root: f.is_trace_root.into(),
         http_method: f.http_method,
diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs
index f1beb5eab9..6f62b5e53c 100644
--- a/libdd-trace-stats/src/span_concentrator/mod.rs
+++ b/libdd-trace-stats/src/span_concentrator/mod.rs
@@ -3,7 +3,7 @@
 //! This module implements the SpanConcentrator used to aggregate spans into stats
 use std::collections::HashMap;
 use std::time::{self, Duration, SystemTime};
-use tracing::debug;
+use tracing::{debug, warn};
 
 use libdd_trace_protobuf::pb;
 
@@ -16,18 +16,42 @@ pub use aggregation::{FixedAggregationKey, OtlpExactCell, OtlpExactGroup, OtlpSt
 pub mod stat_span;
 pub use stat_span::StatSpan;
 
+/// Result of flushing a concentrator.
+///
+/// Obfuscated and un-obfuscated buckets are kept separate because they must be sent in distinct
+/// stats payloads: only the obfuscated payload carries the `datadog-obfuscation-version` header.
+pub struct FlushResult {
+    /// Buckets whose resource names were obfuscated client-side.
+    pub obfuscated_buckets: Vec<pb::ClientStatsBucket>,
+    /// Buckets whose resource names were left as-is.
+    pub unobfuscated_buckets: Vec<pb::ClientStatsBucket>,
+    /// Total number of spans that were collapsed into the overflow sentinel bucket due to
+    /// cardinality limiting across all flushed time buckets.
+    pub collapsed_spans: u64,
+}
+
+impl FlushResult {
+    /// All flushed buckets regardless of obfuscation.
+    #[cfg(test)]
+    pub fn all_buckets(self) -> Vec<pb::ClientStatsBucket> {
+        let mut buckets = self.obfuscated_buckets;
+        buckets.extend(self.unobfuscated_buckets);
+        buckets
+    }
+}
+
 /// Concentrators that can provide raw time buckets for export implement this trait.
 ///
 /// `StatsExporter` is generic over `C: FlushableConcentrator` so it can work with
 /// both the in-process [`SpanConcentrator`] and the SHM-backed `ShmSpanConcentrator`.
 pub trait FlushableConcentrator {
-    /// Flush time buckets and return them together with the number of spans that were
-    /// collapsed into the overflow sentinel bucket due to cardinality limiting.
-    fn flush_buckets(&mut self, force: bool) -> (Vec<pb::ClientStatsBucket>, u64);
+    /// Flush time buckets and return them together with flush metadata. If `force` is true, flush
+    /// all buckets. See [`FlushResult`] for the returned data.
+    fn flush_buckets(&mut self, force: bool) -> FlushResult;
 }
 
 impl FlushableConcentrator for SpanConcentrator {
-    fn flush_buckets(&mut self, force: bool) -> (Vec<pb::ClientStatsBucket>, u64) {
+    fn flush_buckets(&mut self, force: bool) -> FlushResult {
         self.flush(SystemTime::now(), force)
     }
 }
@@ -68,13 +92,41 @@ pub struct StatsComputationObfuscationConfig {
 pub type SharedStatsComputationObfuscationConfig =
     std::sync::Arc<arc_swap::ArcSwap<StatsComputationObfuscationConfig>>;
 
-/// Default maximum number of distinct aggregation keys per time bucket.
-///
-/// 7 168 is the limit to exactly saturate hashbrown's internal table at its maximum load factor of
-/// 7/8. Any higher limit would immediately force a doubling of the table capacity, wasting
-/// half the allocated slots for a modest increase in cardinality. To avoid future changes going
-/// over this limit (e.g. adding extra overflow buckets) we set a slightly lower limit.
-pub const DEFAULT_MAX_ENTRIES_PER_BUCKET: usize = 7_000;
+/// Config to override the default stats cardinality limit values
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(C)]
+pub struct CardinalityLimitConfig {
+    /// The whole-key cardinality limit (defaults to 7000)
+    pub whole_key_limit: usize,
+    /// The per-field cardinality limit for the Resource field (defaults to 1024)
+    pub resource_limit: usize,
+    /// The per-field cardinality limit for the HttpEndpoint field (defaults to 512)
+    pub http_endpoint_limit: usize,
+    /// The per-field cardinality limit for the PeerTags field (defaults to 512)
+    pub peer_tags_limit: usize,
+    /// The per-field cardinality limit for the AdditionalTags field (defaults to 100)
+    pub additional_tags_limit: usize,
+}
+
+impl Default for CardinalityLimitConfig {
+    fn default() -> Self {
+        Self {
+            // Default maximum number of distinct aggregation keys per time bucket.
+            //
+            // 7 168 is the limit to exactly saturate hashbrown's internal table at its maximum
+            // load factor of 7/8. Any higher limit would immediately force a doubling
+            // of the table capacity, wasting half the allocated slots for a modest
+            // increase in cardinality. To avoid future changes going over this limit
+            // (e.g. adding extra overflow buckets) we set a slightly lower limit.
+            whole_key_limit: 7_000,
+            // Other defaults from the spec
+            resource_limit: 1_024,
+            http_endpoint_limit: 512,
+            peer_tags_limit: 512,
+            additional_tags_limit: 100,
+        }
+    }
+}
 
 /// SpanConcentrator compute stats on span aggregated by time and span attributes
 ///
@@ -106,8 +158,8 @@ pub struct SpanConcentrator {
     oldest_timestamp: u64,
     /// bufferLen is the number stats bucket we keep when flushing.
     buffer_len: usize,
-    /// Maximum number of distinct aggregation keys per bucket.
-    max_entries_per_bucket: usize,
+    /// Config values for whole-key and per-field cardinality limits
+    cardinality_limits: CardinalityLimitConfig,
     /// span.kind fields eligible for stats computation
     span_kinds_stats_computed: Vec<String>,
     /// keys for supplementary tags that describe peer.service entities
@@ -122,19 +174,44 @@ impl SpanConcentrator {
     /// - `now` the current system time, used to define the oldest bucket
     /// - `span_kinds_stats_computed` list of span kinds eligible for stats computation
     /// - `peer_tags_keys` list of keys considered as peer tags for aggregation
-    /// - `override_max_entries_per_bucket` maximum distinct aggregation keys per time bucket before
-    ///   cardinality limiting applies. Pass `None` to use [`DEFAULT_MAX_ENTRIES_PER_BUCKET`].
+    /// - `override_cardinality_limits` config values for whole-key and per-field cardinality limi.
+    ///   Pass `None` to use defaults (see [`CardinalityLimitConfig`]).
     /// - `obfuscation_config` optional and updatable config for resource key obfuscation
     pub fn new(
         bucket_size: Duration,
         now: SystemTime,
         span_kinds_stats_computed: Vec<String>,
         peer_tag_keys: Vec<String>,
-        override_max_entries_per_bucket: Option<usize>,
+        override_cardinality_limits: Option<CardinalityLimitConfig>,
         #[cfg(feature = "stats-obfuscation")] obfuscation_config: Option<
             SharedStatsComputationObfuscationConfig,
         >,
     ) -> SpanConcentrator {
+        if let Some(cardinality_limit_config) = override_cardinality_limits.as_ref() {
+            if cardinality_limit_config.whole_key_limit == 0
+                || cardinality_limit_config.http_endpoint_limit == 0
+                || cardinality_limit_config.peer_tags_limit == 0
+                || cardinality_limit_config.additional_tags_limit == 0
+            {
+                warn!(
+                    ?cardinality_limit_config,
+                    "Stats cardinality limit is misconfigured: cardinality limits must not be 0 otherwise all the stats get collapsed!"
+                );
+            }
+            if cardinality_limit_config.whole_key_limit <= cardinality_limit_config.resource_limit
+                || cardinality_limit_config.whole_key_limit
+                    <= cardinality_limit_config.http_endpoint_limit
+                || cardinality_limit_config.whole_key_limit
+                    <= cardinality_limit_config.peer_tags_limit
+                || cardinality_limit_config.whole_key_limit
+                    <= cardinality_limit_config.additional_tags_limit
+            {
+                warn!(
+                    ?cardinality_limit_config,
+                    "Stats cardinality limit is misconfigured: per-field limits must be lower than whole-key limit otherwise they have no effect and you will get over-collapsed stats!"
+                );
+            }
+        }
         SpanConcentrator {
             bucket_size: bucket_size.as_nanos() as u64,
             buckets: HashMap::new(),
@@ -143,8 +220,7 @@ impl SpanConcentrator {
                 bucket_size.as_nanos() as u64,
             ),
             buffer_len: 2,
-            max_entries_per_bucket: override_max_entries_per_bucket
-                .unwrap_or(DEFAULT_MAX_ENTRIES_PER_BUCKET),
+            cardinality_limits: override_cardinality_limits.unwrap_or_default(),
             span_kinds_stats_computed,
             peer_tag_keys,
             #[cfg(feature = "stats-obfuscation")]
@@ -190,7 +266,23 @@ impl SpanConcentrator {
         if bucket_timestamp < self.oldest_timestamp {
             bucket_timestamp = self.oldest_timestamp;
         }
-        let obfuscated_resource = self.compute_obfuscated_span(span);
+
+        let target_bucket = self.buckets.entry(bucket_timestamp).or_insert_with(|| {
+            StatsBucket::new(
+                bucket_timestamp,
+                self.cardinality_limits,
+                #[cfg(feature = "stats-obfuscation")]
+                self.obfuscation_config.load().enabled,
+            )
+        });
+        #[cfg(feature = "stats-obfuscation")]
+        let obfuscated_resource = if target_bucket.obfuscated {
+            Self::compute_obfuscated_span(self.obfuscation_config.load().sql_obfuscation_mode, span)
+        } else {
+            None
+        };
+        #[cfg(not(feature = "stats-obfuscation"))]
+        let obfuscated_resource: Option<String> = None;
         let agg_key = match obfuscated_resource.as_deref() {
             Some(res) => BorrowedAggregationKey::from_obfuscated_span(
                 res,
@@ -199,65 +291,70 @@ impl SpanConcentrator {
             ),
             None => BorrowedAggregationKey::from_span(span, self.peer_tag_keys.as_slice()),
         };
-        self.buckets
-            .entry(bucket_timestamp)
-            .or_insert_with(|| StatsBucket::new(bucket_timestamp, self.max_entries_per_bucket))
-            .insert(
-                agg_key,
-                span.duration(),
-                span.is_error(),
-                span.has_top_level(),
-            );
+        target_bucket.insert(
+            agg_key,
+            span.duration(),
+            span.is_error(),
+            span.has_top_level(),
+        );
     }
 
+    #[cfg(feature = "stats-obfuscation")]
     fn compute_obfuscated_span<'a>(
-        &self,
-        #[allow(unused)] span: &'a impl StatSpan<'a>,
+        sql_obfuscation_mode: libdd_trace_obfuscation::sql::SqlObfuscationMode,
+        span: &'a impl StatSpan<'a>,
     ) -> Option<String> {
-        #[cfg(feature = "stats-obfuscation")]
-        if self.obfuscation_config.load().enabled {
-            let dbms_hint: Option<&str> = span.get_meta("db.type");
-            return libdd_trace_obfuscation::obfuscate::obfuscate_resource_for_stats(
-                span.r#type(),
-                span.resource(),
-                dbms_hint,
-                self.obfuscation_config.load().sql_obfuscation_mode,
-            );
-        }
-        None
+        let dbms_hint: Option<&str> = span.get_meta("db.type");
+        libdd_trace_obfuscation::obfuscate::obfuscate_resource_for_stats(
+            span.r#type(),
+            span.resource(),
+            dbms_hint,
+            sql_obfuscation_mode,
+        )
     }
 
     /// Flush all stats bucket except for the `buffer_len` most recent. If `force` is true, flush
     /// all buckets.
     ///
-    /// Returns a tuple of `(buckets, collapsed_spans)` where `collapsed_spans` is the total number
-    /// of spans that were collapsed into the overflow sentinel bucket due to cardinality limiting
-    /// across all flushed time buckets.
-    pub fn flush(&mut self, now: SystemTime, force: bool) -> (Vec<pb::ClientStatsBucket>, u64) {
-        self.drain_due_buckets(now, force, StatsBucket::flush)
+    /// Obfuscated and un-obfuscated buckets are returned separately, see [`FlushResult`].
+    pub fn flush(&mut self, now: SystemTime, force: bool) -> FlushResult {
+        let (buckets, collapsed_spans) = self.drain_due_buckets(now, force, StatsBucket::flush);
+        let mut obfuscated_buckets = Vec::new();
+        let mut unobfuscated_buckets = Vec::new();
+        for (obfuscated, bucket) in buckets {
+            if obfuscated {
+                obfuscated_buckets.push(bucket);
+            } else {
+                unobfuscated_buckets.push(bucket);
+            }
+        }
+        FlushResult {
+            obfuscated_buckets,
+            unobfuscated_buckets,
+            collapsed_spans,
+        }
     }
 
     /// Like [`Self::flush`], but also emits exact per-cell scalars alongside each bucket for the
     /// OTLP trace-metrics path. The protobuf bucket inside each [`OtlpStatsBucket`] is identical
     /// to what [`Self::flush`] would produce, so the /v0.6/stats agent path is unaffected.
-    ///
-    /// Returns a tuple of `(buckets, collapsed_spans)` where `collapsed_spans` is the total number
-    /// of spans that were collapsed into the overflow sentinel bucket due to cardinality limiting
-    /// across all flushed time buckets.
-    pub fn flush_with_otlp_exact(
-        &mut self,
-        now: SystemTime,
-        force: bool,
-    ) -> (Vec<OtlpStatsBucket>, u64) {
-        self.drain_due_buckets(now, force, StatsBucket::flush_with_otlp_exact)
+    pub fn flush_with_otlp_exact(&mut self, now: SystemTime, force: bool) -> Vec<OtlpStatsBucket> {
+        let (buckets, _) = self.drain_due_buckets(now, force, StatsBucket::flush_with_otlp_exact);
+        buckets.into_iter().map(|(_, bucket)| bucket).collect()
     }
 
+    /// Drain the buckets that are due for flushing, encoding each with `encode`.
+    ///
+    /// Returns a tuple `(buckets, collapsed_spans)` where each encoded bucket is paired with a
+    /// boolean indicating whether it was obfuscated client-side (always `false` when the
+    /// `stats-obfuscation` feature is disabled), and `collapsed_spans` is the total number of
+    /// spans collapsed into the overflow sentinel bucket due to cardinality limiting.
     fn drain_due_buckets<T>(
         &mut self,
         now: SystemTime,
         force: bool,
         encode: impl Fn(StatsBucket, u64) -> T,
-    ) -> (Vec<T>, u64) {
+    ) -> (Vec<(bool, T)>, u64) {
         // TODO: Wait for HashMap::extract_if to be stabilized to avoid a full drain
         let now_timestamp = system_time_to_unix_duration(now).as_nanos() as u64;
         let buckets: Vec<(u64, StatsBucket)> = self.buckets.drain().collect();
@@ -278,17 +375,26 @@ impl SpanConcentrator {
                 // if the tracer stops while the latest buckets aren't old enough to be flushed.
                 // The "force" boolean skips the delay and flushes all buckets, typically on
                 // shutdown.
-                if !force && timestamp > (now_timestamp - self.buffer_len as u64 * self.bucket_size)
-                {
+                let keep = !force
+                    && timestamp > (now_timestamp - self.buffer_len as u64 * self.bucket_size);
+                if keep {
                     self.buckets.insert(timestamp, bucket);
                     return None;
                 }
                 total_collapsed += bucket.collapsed_count();
-                Some(encode(bucket, self.bucket_size))
+                #[cfg(feature = "stats-obfuscation")]
+                let obfuscated = bucket.obfuscated;
+                #[cfg(not(feature = "stats-obfuscation"))]
+                let obfuscated = false;
+                Some((obfuscated, encode(bucket, self.bucket_size)))
             })
             .collect();
         if total_collapsed > 0 {
-            debug!(max_entries_per_bucket = self.max_entries_per_bucket, total_collapsed, "Client-side stats values have been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_CARDINALITY_LIMIT");
+            debug!(
+                max_entries_per_bucket = self.cardinality_limits.whole_key_limit,
+                total_collapsed,
+                "Client-side stats values have been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_CARDINALITY_LIMIT"
+            );
         }
         (buckets_pb, total_collapsed)
     }
diff --git a/libdd-trace-stats/src/span_concentrator/tests.rs b/libdd-trace-stats/src/span_concentrator/tests.rs
index 3fd088e09b..938ef8e8b9 100644
--- a/libdd-trace-stats/src/span_concentrator/tests.rs
+++ b/libdd-trace-stats/src/span_concentrator/tests.rs
@@ -126,12 +126,12 @@ fn test_concentrator_oldest_timestamp_cold() {
 
     // Assert we didn't insert spans in older buckets
     for _ in 0..concentrator.buffer_len {
-        let (stats, _) = concentrator.flush(flushtime, false);
+        let stats = concentrator.flush(flushtime, false).all_buckets();
         assert_eq!(stats.len(), 0, "We should get 0 time buckets");
         flushtime += Duration::from_nanos(concentrator.bucket_size);
     }
 
-    let (stats, _) = concentrator.flush(flushtime, false);
+    let stats = concentrator.flush(flushtime, false).all_buckets();
 
     assert_eq!(stats.len(), 1, "We should get exactly one time bucket");
 
@@ -188,12 +188,12 @@ fn test_concentrator_oldest_timestamp_hot() {
     }
 
     for _ in 0..(concentrator.buffer_len - 1) {
-        let (stats, _) = concentrator.flush(flushtime, false);
+        let stats = concentrator.flush(flushtime, false).all_buckets();
         assert!(stats.is_empty(), "We should get 0 time buckets");
         flushtime += Duration::from_nanos(concentrator.bucket_size);
     }
 
-    let (stats, _) = concentrator.flush(flushtime, false);
+    let stats = concentrator.flush(flushtime, false).all_buckets();
     assert_eq!(stats.len(), 1, "We should get exactly one time bucket");
 
     // First oldest bucket aggregates, it should have it all except the
@@ -213,7 +213,7 @@ fn test_concentrator_oldest_timestamp_hot() {
     assert_counts_equal(expected, stats.first().unwrap().stats.clone());
 
     flushtime += Duration::from_nanos(concentrator.bucket_size);
-    let (stats, _) = concentrator.flush(flushtime, false);
+    let stats = concentrator.flush(flushtime, false).all_buckets();
     assert_eq!(stats.len(), 1, "We should get exactly one time bucket");
 
     // Stats of the last four spans.
@@ -278,7 +278,7 @@ fn test_concentrator_stats_totals() {
     let mut flushtime = now;
 
     for _ in 0..=concentrator.buffer_len {
-        let (stats, _) = concentrator.flush(flushtime, false);
+        let stats = concentrator.flush(flushtime, false).all_buckets();
         if stats.is_empty() {
             continue;
         }
@@ -528,7 +528,7 @@ fn test_concentrator_stats_counts() {
     let mut flushtime = now;
 
     for _ in 0..=concentrator.buffer_len + 2 {
-        let (stats, _) = concentrator.flush(flushtime, false);
+        let stats = concentrator.flush(flushtime, false).all_buckets();
         let expected_flushed_timestamps = align_timestamp(
             system_time_to_unix_duration(flushtime).as_nanos() as u64,
             concentrator.bucket_size,
@@ -553,7 +553,7 @@ fn test_concentrator_stats_counts() {
             stats.first().unwrap().stats.clone(),
         );
 
-        let (stats, _) = concentrator.flush(flushtime, false);
+        let stats = concentrator.flush(flushtime, false).all_buckets();
         assert_eq!(
             stats.len(),
             0,
@@ -658,10 +658,12 @@ fn test_span_should_be_included_in_stats() {
         },
     ];
 
-    let (stats, _) = concentrator.flush(
-        now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64),
-        false,
-    );
+    let stats = concentrator
+        .flush(
+            now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64),
+            false,
+        )
+        .all_buckets();
     assert_counts_equal(
         expected,
         stats
@@ -696,10 +698,12 @@ fn test_ignore_partial_spans() {
         concentrator.add_span(span);
     }
 
-    let (stats, _) = concentrator.flush(
-        now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64),
-        false,
-    );
+    let stats = concentrator
+        .flush(
+            now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64),
+            false,
+        )
+        .all_buckets();
     assert_eq!(0, stats.len());
 }
 
@@ -727,10 +731,10 @@ fn test_force_flush() {
     let flushtime = now - Duration::from_secs(3600);
 
     // Bucket should not be flushed without force flush
-    let (stats, _) = concentrator.flush(flushtime, false);
+    let stats = concentrator.flush(flushtime, false).all_buckets();
     assert_eq!(0, stats.len());
 
-    let (stats, _) = concentrator.flush(flushtime, true);
+    let stats = concentrator.flush(flushtime, true).all_buckets();
     assert_eq!(1, stats.len());
 }
 
@@ -900,7 +904,9 @@ fn test_peer_tags_aggregation() {
         },
     ];
 
-    let (stats_with_peer_tags, _) = concentrator_with_peer_tags.flush(flushtime, false);
+    let stats_with_peer_tags = concentrator_with_peer_tags
+        .flush(flushtime, false)
+        .all_buckets();
     assert_counts_equal(
         expected_with_peer_tags,
         stats_with_peer_tags
@@ -910,7 +916,9 @@ fn test_peer_tags_aggregation() {
             .clone(),
     );
 
-    let (stats_without_peer_tags, _) = concentrator_without_peer_tags.flush(flushtime, false);
+    let stats_without_peer_tags = concentrator_without_peer_tags
+        .flush(flushtime, false)
+        .all_buckets();
     assert_counts_equal(
         expected_without_peer_tags,
         stats_without_peer_tags
@@ -1023,7 +1031,9 @@ fn test_peer_tags_quantization_aggregation() {
         ..Default::default()
     }];
 
-    let (stats_with_peer_tags, _) = concentrator_with_peer_tags.flush(flushtime, false);
+    let stats_with_peer_tags = concentrator_with_peer_tags
+        .flush(flushtime, false)
+        .all_buckets();
     assert_counts_equal(
         expected_with_peer_tags,
         stats_with_peer_tags
@@ -1193,7 +1203,7 @@ fn test_base_service_peer_tag() {
         },
     ];
 
-    let (stats, _) = concentrator.flush(flushtime, false);
+    let stats = concentrator.flush(flushtime, false).all_buckets();
     assert_counts_equal(
         expected,
         stats
@@ -1497,7 +1507,7 @@ fn test_pb_span() {
     // Flush and get stats
     let flushtime =
         now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64);
-    let (stats, _) = concentrator.flush(flushtime, false);
+    let stats = concentrator.flush(flushtime, false).all_buckets();
 
     assert_eq!(stats.len(), 1, "Should get exactly one time bucket");
     let bucket = &stats[0];
@@ -1610,7 +1620,7 @@ fn test_flush_with_otlp_exact_per_cell_scalars() {
         concentrator.add_span(s);
     }
 
-    let flushed = concentrator.flush_with_otlp_exact(now, true).0;
+    let flushed = concentrator.flush_with_otlp_exact(now, true);
     assert_eq!(flushed.len(), 1);
     let b = &flushed[0];
     assert_eq!(b.exact.len(), 1);
@@ -1637,14 +1647,14 @@ fn test_flush_with_otlp_exact_per_cell_scalars() {
 }
 
 /// Build a minimal concentrator with a tiny `max_entries_per_bucket` for cardinality tests.
-fn make_cardinality_concentrator(max_entries: usize) -> SpanConcentrator {
+fn make_cardinality_concentrator(cardinality_limits: CardinalityLimitConfig) -> SpanConcentrator {
     let now = SystemTime::now();
     SpanConcentrator::new(
         Duration::from_nanos(BUCKET_SIZE),
         now,
         get_span_kinds(),
-        vec![],
-        Some(max_entries),
+        vec!["peer.hostname".to_owned()],
+        Some(cardinality_limits),
         #[cfg(feature = "stats-obfuscation")]
         None,
     )
@@ -1653,10 +1663,13 @@ fn make_cardinality_concentrator(max_entries: usize) -> SpanConcentrator {
 /// When the limit is 3 and we insert 5 distinct-resource spans, only 3 normal keys plus one
 /// overflow key must appear in the flushed stats. Total hits must equal 5.
 #[test]
-fn test_cardinality_limit_collapse() {
+fn test_whole_key_cardinality_limit_collapse() {
     let now = SystemTime::now();
     let limit: usize = 3;
-    let mut concentrator = make_cardinality_concentrator(limit);
+    let mut concentrator = make_cardinality_concentrator(CardinalityLimitConfig {
+        whole_key_limit: limit,
+        ..Default::default()
+    });
 
     // Insert limit + 2 distinct-resource root spans all in the same time bucket.
     let resources: Vec<String> = (0..limit + 2).map(|i| format!("resource-{i}")).collect();
@@ -1676,7 +1689,7 @@ fn test_cardinality_limit_collapse() {
         concentrator.add_span(&span);
     }
 
-    let (buckets, _) = concentrator.flush(SystemTime::now(), true);
+    let buckets = concentrator.flush(SystemTime::now(), true).all_buckets();
     assert!(!buckets.is_empty(), "should get at least one time bucket");
 
     let stats = &buckets[0].stats;
@@ -1709,12 +1722,185 @@ fn test_cardinality_limit_collapse() {
     );
 }
 
+/// When the `http_endpoint` cardinality limit is 3 and we insert 5 distinct spans differing only on
+/// their `http_endpoint`, only 3 normal keys plus one overflow key must appear in the flushed
+/// stats.
+///
+/// But then inserting spans differing on another field, it should not be collapsed.
+/// So total hits must equal 6.
+#[test]
+fn test_per_key_cardinality_limit_collapse_http_endpoint() {
+    let now = SystemTime::now();
+    let limit: usize = 3;
+    let mut concentrator = make_cardinality_concentrator(CardinalityLimitConfig {
+        http_endpoint_limit: limit,
+        ..Default::default()
+    });
+
+    // Insert limit + 2 distinct `http_endpoint` root spans all in the same time bucket.
+    let http_endpoints: Vec<String> = (0..limit + 2).map(|i| format!("endpoint-{i}")).collect();
+    for (i, http_endpoint) in http_endpoints.iter().enumerate() {
+        let meta = [("http.endpoint", http_endpoint.as_str())];
+        let span = get_test_span_with_meta(
+            now,
+            i as u64 + 1,
+            0,
+            100,
+            2,
+            "svc",
+            "resource",
+            0,
+            &meta,
+            &[("_dd.measured", 1.0)],
+        );
+        concentrator.add_span(&span);
+    }
+    // Insert a distinct `resource` root span, this one won't get collapsed
+    {
+        let meta = [("http.endpoint", "endpoint-0")];
+        let span = get_test_span_with_meta(
+            now,
+            limit as u64 + 3,
+            0,
+            100,
+            2,
+            "svc",
+            "different-resource",
+            0,
+            &meta,
+            &[("_dd.measured", 1.0)],
+        );
+        concentrator.add_span(&span);
+    }
+
+    let buckets = concentrator.flush(SystemTime::now(), true).all_buckets();
+    assert!(!buckets.is_empty(), "should get at least one time bucket");
+
+    let stats = &buckets[0].stats;
+
+    // Exactly limit normal keys + 1 overflow key + the distinct `resource` span
+    assert_eq!(
+        stats.len(),
+        limit + 2,
+        "expected {limit} normal groups + 1 overflow key + 1 for the distinct `resource` span, got {}",
+        stats.len()
+    );
+
+    // Total hits must be preserved.
+    let total_hits: u64 = stats.iter().map(|g| g.hits).sum();
+    assert_eq!(
+        total_hits,
+        (limit + 3) as u64,
+        "total hits must equal the number of inserted spans"
+    );
+
+    // No overflow group, identified by the sentinel resource.
+    let overflow_groups: Vec<_> = stats
+        .iter()
+        .filter(|g| g.resource == TRACER_BLOCKED_VALUE)
+        .collect();
+    assert_eq!(
+        overflow_groups.len(),
+        0,
+        "expected no overflow group, given whole key cardinality limit was not reached"
+    );
+    let http_overflow_groups: Vec<_> = stats
+        .iter()
+        .filter(|g| g.http_endpoint == TRACER_BLOCKED_VALUE)
+        .collect();
+    assert_eq!(
+        http_overflow_groups.len(),
+        1,
+        "expected exactly one overflow key for the http_endpoint field"
+    );
+}
+
+/// When whole-key cardinality limit is reached, check that per-key fields are collapsed before
+/// falling back to whole-key
+#[test]
+fn test_per_key_cardinality_limit_collapse_before_whole_key() {
+    let now = SystemTime::now();
+    let peer_tags_limit = 3;
+    let whole_key_limit = peer_tags_limit + 1;
+    let mut concentrator = make_cardinality_concentrator(CardinalityLimitConfig {
+        whole_key_limit,
+        peer_tags_limit,
+        ..Default::default()
+    });
+
+    // Insert limit + 2 distinct `peer.hostname` root spans all in the same time bucket.
+    let inserted_spans = peer_tags_limit + 2;
+    let peer_tag_values: Vec<String> = (0..inserted_spans).map(|i| format!("peer-{i}")).collect();
+    for (i, peer_tag_value) in peer_tag_values.iter().enumerate() {
+        let meta = [
+            ("peer.hostname", peer_tag_value.as_str()),
+            ("span.kind", "client"),
+        ];
+        let span = get_test_span_with_meta(
+            now,
+            i as u64 + 1,
+            0,
+            100,
+            2,
+            "svc",
+            "resource",
+            0,
+            &meta,
+            &[("_dd.measured", 1.0)],
+        );
+        concentrator.add_span(&span);
+    }
+
+    let buckets = concentrator.flush(SystemTime::now(), true).all_buckets();
+    assert!(!buckets.is_empty(), "should get at least one time bucket");
+
+    let stats = &buckets[0].stats;
+
+    // Exactly peer_tags_limit normal keys + 1 overflow key
+    assert_eq!(
+        stats.len(),
+        peer_tags_limit + 1,
+        "expected {peer_tags_limit} normal groups + 1 overflow key, got {}",
+        stats.len()
+    );
+
+    // Total hits must be preserved.
+    let total_hits: u64 = stats.iter().map(|g| g.hits).sum();
+    assert_eq!(
+        total_hits, inserted_spans as u64,
+        "total hits must equal the number of inserted spans"
+    );
+
+    // No overflow group, identified by the sentinel resource.
+    let overflow_groups: Vec<_> = stats
+        .iter()
+        .filter(|g| g.resource == TRACER_BLOCKED_VALUE)
+        .collect();
+    assert_eq!(
+        overflow_groups.len(),
+        0,
+        "expected no overflow group: whole key cardinality limit was not reached because per-field cardinality limit collapsed keys before overflowing whole key"
+    );
+    let peer_tag_overflow_groups: Vec<_> = stats
+        .iter()
+        .filter(|g| g.peer_tags == [TRACER_BLOCKED_VALUE])
+        .collect();
+    assert_eq!(
+        peer_tag_overflow_groups.len(),
+        1,
+        "expected exactly one overflow key for the peer_tags field"
+    );
+}
+
 /// The overflow bucket must correctly aggregate the hits from overflow spans.
 #[test]
 fn test_overflow_bucket_counts() {
     let now = SystemTime::now();
     let limit: usize = 1;
-    let mut concentrator = make_cardinality_concentrator(limit);
+    let mut concentrator = make_cardinality_concentrator(CardinalityLimitConfig {
+        whole_key_limit: limit,
+        ..Default::default()
+    });
 
     // First span fills the sole slot; the next 4 spans all have distinct keys → all overflow.
     for i in 0..5usize {
@@ -1734,7 +1920,7 @@ fn test_overflow_bucket_counts() {
         concentrator.add_span(&span);
     }
 
-    let (buckets, _) = concentrator.flush(SystemTime::now(), true);
+    let buckets = concentrator.flush(SystemTime::now(), true).all_buckets();
     assert!(!buckets.is_empty());
     let stats = &buckets[0].stats;
 
@@ -1763,7 +1949,10 @@ fn test_overflow_bucket_counts() {
 fn test_no_collapse_within_limit() {
     let now = SystemTime::now();
     let limit: usize = 10;
-    let mut concentrator = make_cardinality_concentrator(limit);
+    let mut concentrator = make_cardinality_concentrator(CardinalityLimitConfig {
+        whole_key_limit: limit,
+        ..Default::default()
+    });
 
     // Insert exactly `limit` distinct-resource spans — no overflow expected.
     for i in 0..limit {
@@ -1783,7 +1972,7 @@ fn test_no_collapse_within_limit() {
         concentrator.add_span(&span);
     }
 
-    let (buckets, _) = concentrator.flush(SystemTime::now(), true);
+    let buckets = concentrator.flush(SystemTime::now(), true).all_buckets();
     assert!(!buckets.is_empty());
     let stats = &buckets[0].stats;
 
@@ -1804,7 +1993,10 @@ fn test_no_collapse_within_limit() {
 fn test_overflow_bucket_key_sentinel_values() {
     let now = SystemTime::now();
     let limit: usize = 1;
-    let mut concentrator = make_cardinality_concentrator(limit);
+    let mut concentrator = make_cardinality_concentrator(CardinalityLimitConfig {
+        whole_key_limit: limit,
+        ..Default::default()
+    });
 
     // First span occupies the only slot; second one overflows.
     let first = get_test_span_with_meta(
@@ -1835,7 +2027,7 @@ fn test_overflow_bucket_key_sentinel_values() {
     concentrator.add_span(&first);
     concentrator.add_span(&second);
 
-    let (buckets, _) = concentrator.flush(SystemTime::now(), true);
+    let buckets = concentrator.flush(SystemTime::now(), true).all_buckets();
     assert!(!buckets.is_empty());
     let stats = &buckets[0].stats;
 
@@ -1886,7 +2078,7 @@ fn test_overflow_bucket_key_sentinel_values() {
         overflow.is_trace_root, 0,
         "is_trace_root must be NOT_SET (0)"
     );
-    assert!(overflow.peer_tags.is_empty(), "peer_tags must be empty");
+    assert_eq!(overflow.peer_tags, [TRACER_BLOCKED_VALUE]);
 
     // The normal group must be unaffected.
     let normal = stats
diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs
index c771fd5b2b..302edf75cc 100644
--- a/libdd-trace-stats/src/stats_exporter.rs
+++ b/libdd-trace-stats/src/stats_exporter.rs
@@ -9,12 +9,12 @@ use std::{
     time,
 };
 
-#[cfg(feature = "stats-obfuscation")]
-use crate::span_concentrator::SharedStatsComputationObfuscationConfig;
 use crate::span_concentrator::{FlushableConcentrator, SpanConcentrator};
 use async_trait::async_trait;
+use futures::stream::FuturesUnordered;
+use futures::StreamExt as _;
 use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability};
-use libdd_common::{tag, Endpoint};
+use libdd_common::Endpoint;
 use libdd_shared_runtime::Worker;
 use libdd_trace_protobuf::pb;
 use libdd_trace_utils::send_with_retry::{send_with_retry, RetryStrategy};
@@ -96,8 +96,6 @@ pub struct StatsExporter<
     sequence_id: AtomicU64,
     capabilities: Cap,
     #[cfg(feature = "stats-obfuscation")]
-    obfuscation_config: SharedStatsComputationObfuscationConfig,
-    #[cfg(feature = "stats-obfuscation")]
     supported_obfuscation_version: &'static str,
     /// Optional telemetry handle and context key.
     #[cfg(feature = "telemetry")]
@@ -127,8 +125,6 @@ impl<Cap: HttpClientCapability + SleepCapability, Con: FlushableConcentrator>
         meta: StatsMetadata,
         endpoint: Endpoint,
         capabilities: Cap,
-        #[cfg(feature = "stats-obfuscation")]
-        obfuscation_config: SharedStatsComputationObfuscationConfig,
         #[cfg(feature = "stats-obfuscation")] supported_obfuscation_version: &'static str,
         #[cfg(feature = "telemetry")] telemetry: Option<
             libdd_telemetry::worker::TelemetryWorkerHandle,
@@ -139,7 +135,7 @@ impl<Cap: HttpClientCapability + SleepCapability, Con: FlushableConcentrator>
         let telemetry = telemetry.map(|handle| {
             let key = handle.register_metric_context(
                 COLLAPSED_SPANS_TELEMETRY_METRIC.to_string(),
-                vec![tag!("collapsed_spans", "whole_key")],
+                vec![libdd_common::tag!("collapsed_spans", "whole_key")],
                 libdd_telemetry::data::metrics::MetricType::Count,
                 true,
                 libdd_telemetry::data::metrics::MetricNamespace::Tracers,
@@ -154,8 +150,6 @@ impl<Cap: HttpClientCapability + SleepCapability, Con: FlushableConcentrator>
             sequence_id: AtomicU64::new(0),
             capabilities,
             #[cfg(feature = "stats-obfuscation")]
-            obfuscation_config,
-            #[cfg(feature = "stats-obfuscation")]
             supported_obfuscation_version,
             #[cfg(feature = "telemetry")]
             telemetry,
@@ -181,37 +175,68 @@ impl<Cap: HttpClientCapability + SleepCapability, Con: FlushableConcentrator>
     /// case stats cannot be flushed since the concentrator might be corrupted.
     /// Returns `Ok(true)` if stats were sent, `Ok(false)` if the concentrator had nothing to send.
     pub async fn send(&self, force_flush: bool) -> anyhow::Result<bool> {
-        let (payload, collapsed_spans) = self.flush(force_flush);
+        let flush = {
+            #[allow(clippy::unwrap_used)]
+            let mut concentrator = self.concentrator.lock().unwrap();
+            concentrator.flush_buckets(force_flush)
+        };
 
-        if collapsed_spans > 0 {
+        if flush.collapsed_spans > 0 {
             #[cfg(feature = "telemetry")]
             if let Some((handle, key)) = &self.telemetry {
-                let _ = handle.add_point(collapsed_spans as f64, key, vec![]);
+                let _ = handle.add_point(flush.collapsed_spans as f64, key, vec![]);
             }
             #[cfg(feature = "dogstatsd")]
             if let Some(client) = &self.dogstatsd {
                 client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count(
                     COLLAPSED_SPANS_HEALTH_METRIC,
-                    collapsed_spans as i64,
-                    [tag!("collapsed_spans", "whole_key")].iter(),
+                    flush.collapsed_spans as i64,
+                    [libdd_common::tag!("collapsed_spans", "whole_key")].iter(),
                 )]);
             }
         }
 
-        if payload.stats.is_empty() {
-            return Ok(false);
+        let futures = FuturesUnordered::new();
+
+        if !flush.obfuscated_buckets.is_empty() {
+            futures.push(self.send_payload(flush.obfuscated_buckets, true));
+        }
+
+        if !flush.unobfuscated_buckets.is_empty() {
+            futures.push(self.send_payload(flush.unobfuscated_buckets, false));
         }
+
+        let sent_stats = !futures.is_empty();
+
+        futures
+            .collect::<Vec<anyhow::Result<()>>>()
+            .await
+            .into_iter()
+            .collect::<anyhow::Result<()>>()?;
+
+        Ok(sent_stats)
+    }
+
+    /// Encode the given buckets into a stats payload and send it to the agent.
+    ///
+    /// `obfuscated` indicates whether the buckets were obfuscated client-side, in which case the
+    /// `datadog-obfuscation-version` header is added.
+    async fn send_payload(
+        &self,
+        buckets: Vec<pb::ClientStatsBucket>,
+        #[cfg_attr(not(feature = "stats-obfuscation"), allow(unused))] obfuscated: bool,
+    ) -> anyhow::Result<()> {
+        let sequence = self.sequence_id.fetch_add(1, Ordering::Relaxed);
+        let payload = encode_stats_payload(&self.meta, sequence, buckets);
         let body = rmp_serde::encode::to_vec_named(&payload)?;
 
         let mut headers: http::HeaderMap = TracerHeaderTags::from(&self.meta).into();
-
         headers.insert(
             http::header::CONTENT_TYPE,
             libdd_common::header::APPLICATION_MSGPACK,
         );
-
         #[cfg(feature = "stats-obfuscation")]
-        if self.obfuscation_config.load().enabled {
+        if obfuscated {
             headers.insert(
                 http::HeaderName::from_static("datadog-obfuscation-version"),
                 http::HeaderValue::from_static(self.supported_obfuscation_version),
@@ -228,31 +253,13 @@ impl<Cap: HttpClientCapability + SleepCapability, Con: FlushableConcentrator>
         .await;
 
         match result {
-            Ok(_) => Ok(true),
+            Ok(_) => Ok(()),
             Err(err) => {
-                error!(?err, "Error with the StateExporter when sending stats");
+                error!(?err, "Error with the StatsExporter when sending stats");
                 anyhow::bail!("Failed to send stats: {err}");
             }
         }
     }
-
-    /// Flush stats from the concentrator into a payload
-    ///
-    /// # Arguments
-    /// - `force_flush` if true, triggers a force flush on the concentrator causing all buckets to
-    ///   be flushed regardless of their age.
-    ///
-    /// # Panic
-    /// Will panic if another thread panicked while holding the concentrator lock in which
-    /// case stats cannot be flushed since the concentrator might be corrupted.
-    fn flush(&self, force_flush: bool) -> (pb::ClientStatsPayload, u64) {
-        let sequence = self.sequence_id.fetch_add(1, Ordering::Relaxed);
-        #[allow(clippy::unwrap_used)]
-        let (buckets, collapsed_spans) =
-            self.concentrator.lock().unwrap().flush_buckets(force_flush);
-        let payload = encode_stats_payload(&self.meta, sequence, buckets);
-        (payload, collapsed_spans)
-    }
 }
 
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
@@ -316,6 +323,7 @@ pub fn stats_url_from_agent_url(agent_url: &str) -> anyhow::Result<http::Uri> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::span_concentrator::CardinalityLimitConfig;
     #[cfg(feature = "stats-obfuscation")]
     use crate::span_concentrator::StatsComputationObfuscationConfig;
     use httpmock::prelude::*;
@@ -353,6 +361,17 @@ mod tests {
     }
 
     fn get_test_concentrator() -> SpanConcentrator {
+        get_test_concentrator_with_obfuscation_config(
+            #[cfg(feature = "stats-obfuscation")]
+            None,
+        )
+    }
+
+    fn get_test_concentrator_with_obfuscation_config(
+        #[cfg(feature = "stats-obfuscation")] obfuscation_config: Option<
+            crate::span_concentrator::SharedStatsComputationObfuscationConfig,
+        >,
+    ) -> SpanConcentrator {
         let mut concentrator = SpanConcentrator::new(
             BUCKETS_DURATION,
             // Make sure the oldest bucket will be flushed on next send
@@ -361,7 +380,7 @@ mod tests {
             vec![],
             None,
             #[cfg(feature = "stats-obfuscation")]
-            None,
+            obfuscation_config,
         );
         let mut trace = vec![];
 
@@ -404,8 +423,6 @@ mod tests {
             Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()),
             NativeCapabilities::new_client(),
             #[cfg(feature = "stats-obfuscation")]
-            StatsComputationObfuscationConfig::disabled(),
-            #[cfg(feature = "stats-obfuscation")]
             "1",
             #[cfg(feature = "telemetry")]
             None,
@@ -439,8 +456,6 @@ mod tests {
             Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()),
             NativeCapabilities::new_client(),
             #[cfg(feature = "stats-obfuscation")]
-            StatsComputationObfuscationConfig::disabled(),
-            #[cfg(feature = "stats-obfuscation")]
             "1",
             #[cfg(feature = "telemetry")]
             None,
@@ -482,8 +497,6 @@ mod tests {
             Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()),
             caps.clone(),
             #[cfg(feature = "stats-obfuscation")]
-            StatsComputationObfuscationConfig::disabled(),
-            #[cfg(feature = "stats-obfuscation")]
             "1",
             #[cfg(feature = "telemetry")]
             None,
@@ -531,8 +544,6 @@ mod tests {
             Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()),
             caps.clone(),
             #[cfg(feature = "stats-obfuscation")]
-            StatsComputationObfuscationConfig::disabled(),
-            #[cfg(feature = "stats-obfuscation")]
             "1",
             #[cfg(feature = "telemetry")]
             None,
@@ -596,18 +607,20 @@ mod tests {
             })
             .await;
 
+        let concentrator = get_test_concentrator_with_obfuscation_config(Some(Arc::new(
+            ArcSwap::from_pointee(StatsComputationObfuscationConfig {
+                enabled: true,
+                ..Default::default()
+            }),
+        )));
+
         let stats_exporter = StatsExporter::new(
             BUCKETS_DURATION,
-            Arc::new(Mutex::new(get_test_concentrator())),
+            Arc::new(Mutex::new(concentrator)),
             get_test_metadata(),
             Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()),
             NativeCapabilities::new_client(),
             #[cfg(feature = "stats-obfuscation")]
-            Arc::new(ArcSwap::from_pointee(StatsComputationObfuscationConfig {
-                enabled: true,
-                ..Default::default()
-            })),
-            #[cfg(feature = "stats-obfuscation")]
             "1",
             #[cfg(feature = "telemetry")]
             None,
@@ -631,7 +644,10 @@ mod tests {
             SystemTime::now(),
             vec![],
             vec![],
-            Some(1), // max 1 distinct key → second span collapses
+            Some(CardinalityLimitConfig {
+                whole_key_limit: 1, // max 1 distinct key → second span collapses
+                ..Default::default()
+            }),
             #[cfg(feature = "stats-obfuscation")]
             None,
         );
@@ -703,8 +719,6 @@ mod tests {
             Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()),
             NativeCapabilities::new_client(),
             #[cfg(feature = "stats-obfuscation")]
-            StatsComputationObfuscationConfig::disabled(),
-            #[cfg(feature = "stats-obfuscation")]
             "1",
             #[cfg(feature = "telemetry")]
             None,
@@ -754,8 +768,6 @@ mod tests {
             Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()),
             NativeCapabilities::new_client(),
             #[cfg(feature = "stats-obfuscation")]
-            StatsComputationObfuscationConfig::disabled(),
-            #[cfg(feature = "stats-obfuscation")]
             "1",
             #[cfg(feature = "telemetry")]
             None,
@@ -806,8 +818,6 @@ mod tests {
             Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()),
             NativeCapabilities::new_client(),
             #[cfg(feature = "stats-obfuscation")]
-            StatsComputationObfuscationConfig::disabled(),
-            #[cfg(feature = "stats-obfuscation")]
             "1",
             #[cfg(feature = "telemetry")]
             Some(handle),