feat(tag_cardinality_limit transform): add tracking_scope setting for per-metric vs global tag tracking

When metrics do not have an explicit `per_metric_limits` entry, their tag values
were always pooled into a single shared bucket. The new `tracking_scope` setting
lets users opt into per-metric tracking buckets instead, providing isolation at
the cost of higher memory.

Default is `global` (current behavior); `per_metric` gives every distinct
(namespace, name) its own bucket regardless of `per_metric_limits` membership.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

Author: ArunPiduguDD

changelog.d/tag_cardinality_limit_tracking_scope.enhancement.md

@@ -0,0 +1,12 @@
+The `tag_cardinality_limit` transform gained a new top-level `tracking_scope` setting
+that controls how tag tracking state is partitioned across metrics that do not have an
+explicit `per_metric_limits` entry:
+
+- `global` (default — preserves existing behavior): all such metrics share a single
+  tracking bucket, and the global `value_limit` caps the combined set of tag values
+  across them.
+- `per_metric`: every distinct metric (namespace, name) gets its own tracking bucket
+  regardless of whether it has a `per_metric_limits` entry, providing full isolation at
+  the cost of higher memory.
+
+authors: ArunPiduguDD

src/transforms/tag_cardinality_limit/config.rs

@@ -35,6 +35,12 @@ pub struct Config {
     #[serde(flatten)]
     pub global: Inner,
 
+    /// Controls how tag tracking is partitioned for metrics that don't have a
+    /// `per_metric_limits` entry.
+    #[configurable(derived)]
+    #[serde(default)]
+    pub tracking_scope: TrackingScope,
+
     /// Tag cardinality limits configuration per metric name.
     #[configurable(
         derived,
@@ -44,6 +50,25 @@ pub struct Config {
     pub per_metric_limits: HashMap<String, PerMetricConfig>,
 }
 
+/// Controls how tag tracking state is partitioned across metrics that do not
+/// have an explicit `per_metric_limits` entry.
+#[configurable_component]
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
+#[serde(rename_all = "snake_case")]
+pub enum TrackingScope {
+    /// All metrics without an explicit `per_metric_limits` entry share a single
+    /// tracking bucket. Tag values pool across metrics, and the global
+    /// `value_limit` caps the combined set. Lower memory but cross-metric
+    /// pollution.
+    #[default]
+    Global,
+
+    /// Every distinct metric (namespace, name) gets its own tracking bucket
+    /// regardless of whether it has a `per_metric_limits` entry. Each metric is
+    /// capped independently. Higher memory but full isolation.
+    PerMetric,
+}
+
 /// Configuration for the `tag_cardinality_limit` transform for a specific group of metrics.
 #[configurable_component]
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -150,6 +175,7 @@ impl GenerateConfig for Config {
                 limit_exceeded_action: default_limit_exceeded_action(),
                 internal_metrics: InternalMetricsConfig::default(),
             },
+            tracking_scope: TrackingScope::default(),
             per_metric_limits: HashMap::default(),
         })
         .unwrap()

src/transforms/tag_cardinality_limit/mod.rs

@@ -15,7 +15,9 @@ mod tag_value_set;
 #[cfg(test)]
 mod tests;
 
-pub use config::{BloomFilterConfig, Config, Inner, LimitExceededAction, Mode, PerMetricConfig};
+pub use config::{
+    BloomFilterConfig, Config, Inner, LimitExceededAction, Mode, PerMetricConfig, TrackingScope,
+};
 use tag_value_set::AcceptedTagValueSet;
 
 use crate::event::metric::TagValueSet;
@@ -125,14 +127,20 @@ impl TagCardinalityLimit {
         let metric = event.as_mut_metric();
         let metric_name = metric.name().to_string();
         let metric_namespace = metric.namespace().map(|n| n.to_string());
-        let has_per_metric_config = self.config.per_metric_limits.iter().any(|(name, config)| {
-            *name == metric_name
-                && (config.namespace.is_none() || config.namespace == metric_namespace)
-        });
-        let metric_key = if has_per_metric_config {
-            Some((metric_namespace, metric_name.clone()))
-        } else {
-            None
+        let metric_key = match self.config.tracking_scope {
+            TrackingScope::PerMetric => Some((metric_namespace, metric_name.clone())),
+            TrackingScope::Global => {
+                let has_per_metric_config =
+                    self.config.per_metric_limits.iter().any(|(name, config)| {
+                        *name == metric_name
+                            && (config.namespace.is_none() || config.namespace == metric_namespace)
+                    });
+                if has_per_metric_config {
+                    Some((metric_namespace, metric_name.clone()))
+                } else {
+                    None
+                }
+            }
         };
         if let Some(tags_map) = metric.tags_mut() {
             match self

src/transforms/tag_cardinality_limit/tests.rs

@@ -57,6 +57,7 @@ fn make_transform_hashset(
             mode: Mode::Exact,
             internal_metrics: InternalMetricsConfig::default(),
         },
+        tracking_scope: TrackingScope::default(),
         per_metric_limits: HashMap::new(),
     }
 }
@@ -71,6 +72,7 @@ fn make_transform_bloom(value_limit: usize, limit_exceeded_action: LimitExceeded
             }),
             internal_metrics: InternalMetricsConfig::default(),
         },
+        tracking_scope: TrackingScope::default(),
         per_metric_limits: HashMap::new(),
     }
 }
@@ -87,6 +89,7 @@ fn make_transform_hashset_with_per_metric_limits(
             mode: Mode::Exact,
             internal_metrics: InternalMetricsConfig::default(),
         },
+        tracking_scope: TrackingScope::default(),
         per_metric_limits,
     }
 }
@@ -105,6 +108,7 @@ fn make_transform_bloom_with_per_metric_limits(
             }),
             internal_metrics: InternalMetricsConfig::default(),
         },
+        tracking_scope: TrackingScope::default(),
         per_metric_limits,
     }
 }
@@ -595,3 +599,46 @@ async fn separate_value_limit_per_metric_name(config: Config) {
     })
     .await;
 }
+
+/// With `tracking_scope: per_metric`, two metrics without explicit `per_metric_limits` entries
+/// each get their own tracking bucket, so one hitting the limit does not affect the other.
+#[test]
+fn tracking_scope_per_metric_isolates_metrics() {
+    let mut config = make_transform_hashset(2, LimitExceededAction::DropEvent);
+    config.tracking_scope = TrackingScope::PerMetric;
+    let mut transform = TagCardinalityLimit::new(config);
+
+    // Fill metric_a's bucket to its limit (2 distinct values).
+    let a1 = make_metric_with_name(metric_tags!("tag" => "v1"), "metric_a");
+    let a2 = make_metric_with_name(metric_tags!("tag" => "v2"), "metric_a");
+    // metric_b should be tracked in its own bucket — not affected by metric_a.
+    let b1 = make_metric_with_name(metric_tags!("tag" => "v3"), "metric_b");
+    let b2 = make_metric_with_name(metric_tags!("tag" => "v4"), "metric_b");
+    // A 3rd unique value on metric_a should be rejected (its bucket is full).
+    let a3 = make_metric_with_name(metric_tags!("tag" => "v5"), "metric_a");
+
+    assert_eq!(transform.transform_one(a1.clone()), Some(a1));
+    assert_eq!(transform.transform_one(a2.clone()), Some(a2));
+    assert_eq!(transform.transform_one(b1.clone()), Some(b1));
+    assert_eq!(transform.transform_one(b2.clone()), Some(b2));
+    assert_eq!(transform.transform_one(a3), None);
+}
+
+/// With the default `tracking_scope: global`, metrics without explicit `per_metric_limits`
+/// entries share a single tracking bucket — so values from different metrics pool together.
+#[test]
+fn tracking_scope_global_pools_metrics() {
+    // Default `tracking_scope` is `Global`.
+    let config = make_transform_hashset(2, LimitExceededAction::DropEvent);
+    let mut transform = TagCardinalityLimit::new(config);
+
+    let a1 = make_metric_with_name(metric_tags!("tag" => "v1"), "metric_a");
+    // Different metric, but values pool into the shared bucket → 2/2 used.
+    let b1 = make_metric_with_name(metric_tags!("tag" => "v2"), "metric_b");
+    // 3rd unique value across the shared bucket → rejected.
+    let a2 = make_metric_with_name(metric_tags!("tag" => "v3"), "metric_a");
+
+    assert_eq!(transform.transform_one(a1.clone()), Some(a1));
+    assert_eq!(transform.transform_one(b1.clone()), Some(b1));
+    assert_eq!(transform.transform_one(a2), None);
+}

website/cue/reference/components/transforms/generated/tag_cardinality_limit.cue

@@ -139,6 +139,29 @@ generated: components: transforms: tag_cardinality_limit: configuration: {
 			}
 		}
 	}
+	tracking_scope: {
+		description: """
+			Controls how tag tracking is partitioned for metrics that don't have a
+			`per_metric_limits` entry.
+			"""
+		required: false
+		type: string: {
+			default: "global"
+			enum: {
+				global: """
+					All metrics without an explicit `per_metric_limits` entry share a single
+					tracking bucket. Tag values pool across metrics, and the global
+					`value_limit` caps the combined set. Lower memory but cross-metric
+					pollution.
+					"""
+				per_metric: """
+					Every distinct metric (namespace, name) gets its own tracking bucket
+					regardless of whether it has a `per_metric_limits` entry. Each metric is
+					capped independently. Higher memory but full isolation.
+					"""
+			}
+		}
+	}
 	value_limit: {
 		description: "How many distinct values to accept for any given key."
 		required:    false