Implement per tag control in tag cardinality processor

Author: ArunPiduguDD

changelog.d/tag_cardinality_limit_per_tag_and_exclude.enhancement.md

@@ -0,0 +1,16 @@
+The `tag_cardinality_limit` transform gained two new configuration capabilities:
+
+- **Per-tag overrides**: each entry in `per_metric_limits` now supports a `per_tag_limits` map
+  whose entries can override `value_limit` and `mode` for a specific tag key. When a per-tag
+  entry omits `value_limit`, it inherits the enclosing per-metric (or global) `value_limit`
+  rather than falling back to the default. `limit_exceeded_action` and `internal_metrics`
+  are always inherited from the enclosing per-metric (or global) configuration and are not
+  per-tag-overridable. Resolution order is per-tag → per-metric → global.
+- **Exclusion**: `mode: excluded` is now available as a third mode option in `per_metric_limits`
+  and `per_tag_limits` entries (not at the global level). When set, the metric or tag is opted
+  out of cardinality control — all tag values pass through and nothing is tracked. Other
+  tracking fields on the entry (`value_limit`) are ignored when `mode: excluded` is selected.
+  Per-metric exclusion is blanket: when a metric's `mode` is `excluded`, every tag on that
+  metric is excluded and any `per_tag_limits` overrides on it are ignored.
+
+authors: ArunPiduguDD

src/transforms/tag_cardinality_limit/config.rs

@@ -66,7 +66,7 @@ pub enum TrackingScope {
     PerMetric,
 }
 
-/// Configuration for the `tag_cardinality_limit` transform for a specific group of metrics.
+/// Configuration block used at the global level.
 #[configurable_component]
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub struct Inner {
@@ -86,7 +86,45 @@ pub struct Inner {
     pub internal_metrics: InternalMetricsConfig,
 }
 
-/// Controls the approach taken for tracking tag cardinality.
+/// Configuration block used at per-metric level. Same shape as the global configuration but
+/// with `OverrideMode`, which adds `excluded` for opting that metric out of cardinality
+/// control entirely.
+#[configurable_component]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct OverrideInner {
+    /// How many distinct values to accept for any given key. Ignored when `mode: excluded`.
+    #[serde(default = "default_value_limit")]
+    pub value_limit: usize,
+
+    #[configurable(derived)]
+    #[serde(default = "default_limit_exceeded_action")]
+    pub limit_exceeded_action: LimitExceededAction,
+
+    #[serde(flatten)]
+    pub mode: OverrideMode,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub internal_metrics: InternalMetricsConfig,
+}
+
+/// Configuration block used at the per-tag level. Same as `OverrideInner` minus
+/// `limit_exceeded_action` (inherited from the enclosing per-metric config) and
+/// `internal_metrics` (inherited from the enclosing per-metric or global config).
+#[configurable_component]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct PerTagInner {
+    /// How many distinct values to accept for this tag key. If unset, inherits
+    /// the `value_limit` from the enclosing per-metric (or global) configuration.
+    /// Ignored when `mode: excluded`.
+    #[serde(default)]
+    pub value_limit: Option<usize>,
+
+    #[serde(flatten)]
+    pub mode: OverrideMode,
+}
+
+/// Controls the approach taken for tracking tag cardinality at the global level.
 #[configurable_component]
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 #[serde(tag = "mode", rename_all = "snake_case", deny_unknown_fields)]
@@ -109,6 +147,41 @@ pub enum Mode {
     Probabilistic(BloomFilterConfig),
 }
 
+/// Controls the approach taken for tracking tag cardinality at the per-metric or per-tag level.
+/// Adds `excluded` to the global `Mode` variants.
+#[configurable_component]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[serde(tag = "mode", rename_all = "snake_case", deny_unknown_fields)]
+#[configurable(metadata(
+    docs::enum_tag_description = "Controls the approach taken for tracking tag cardinality."
+))]
+pub enum OverrideMode {
+    /// Tracks cardinality exactly. See `Mode::Exact` for details.
+    Exact,
+
+    /// Tracks cardinality probabilistically. See `Mode::Probabilistic` for details.
+    Probabilistic(BloomFilterConfig),
+
+    /// Skip cardinality tracking for this scope. All tag values pass through and nothing is
+    /// recorded. Other tracking fields on the entry (`value_limit`, `limit_exceeded_action`,
+    /// `internal_metrics`) are ignored when this is selected.
+    ///
+    /// Only valid in `per_metric_limits` and `per_tag_limits` entries; using it as the global
+    /// `mode` is a configuration error.
+    Excluded,
+}
+
+impl OverrideMode {
+    /// Returns the equivalent global `Mode` if this scope is tracked, or `None` if excluded.
+    pub const fn as_mode(&self) -> Option<Mode> {
+        match self {
+            OverrideMode::Exact => Some(Mode::Exact),
+            OverrideMode::Probabilistic(b) => Some(Mode::Probabilistic(*b)),
+            OverrideMode::Excluded => None,
+        }
+    }
+}
+
 /// Bloom filter configuration in probabilistic mode.
 #[configurable_component]
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -143,8 +216,29 @@ pub struct PerMetricConfig {
     #[serde(default)]
     pub namespace: Option<String>,
 
+    /// Per-tag-key overrides scoped to this metric.
+    ///
+    /// Each entry may override `value_limit` and `mode` for a specific tag key.
+    /// `limit_exceeded_action` and `internal_metrics` are always inherited from the enclosing
+    /// per-metric (or global) configuration and cannot be set per-tag.
+    /// Tags not listed here use the per-metric configuration.
+    #[configurable(
+        derived,
+        metadata(docs::additional_props_description = "An individual tag configuration.")
+    )]
+    #[serde(default)]
+    pub per_tag_limits: HashMap<String, PerTagConfig>,
+
+    #[serde(flatten)]
+    pub config: OverrideInner,
+}
+
+/// Tag cardinality limit configuration for a specific tag key, scoped under a per-metric override.
+#[configurable_component]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct PerTagConfig {
     #[serde(flatten)]
-    pub config: Inner,
+    pub config: PerTagInner,
 }
 
 const fn default_limit_exceeded_action() -> LimitExceededAction {

src/transforms/tag_cardinality_limit/mod.rs

@@ -16,14 +16,23 @@ mod tag_value_set;
 mod tests;
 
 pub use config::{
-    BloomFilterConfig, Config, Inner, LimitExceededAction, Mode, PerMetricConfig, TrackingScope,
+    BloomFilterConfig, Config, Inner, LimitExceededAction, Mode, OverrideInner, OverrideMode,
+    PerMetricConfig, PerTagConfig, PerTagInner, TrackingScope,
 };
 use tag_value_set::AcceptedTagValueSet;
 
 use crate::event::metric::TagValueSet;
 
 type MetricId = (Option<String>, String);
 
+/// Tag tracking settings for a single (metric, tag) pair.
+enum TagSettings {
+    /// The tag is excluded from cardinality control; pass values through unchanged.
+    Excluded,
+    /// The tag is tracked using these settings.
+    Tracked(Inner),
+}
+
 #[derive(Debug)]
 pub struct TagCardinalityLimit {
     config: Config,
@@ -38,19 +47,73 @@ impl TagCardinalityLimit {
         }
     }
 
-    fn get_config_for_metric(&self, metric_key: Option<&MetricId>) -> &Inner {
-        match metric_key {
-            Some(id) => self
-                .config
-                .per_metric_limits
-                .iter()
-                .find(|(name, config)| {
-                    **name == id.1 && (config.namespace.is_none() || config.namespace == id.0)
+    /// Resolve the configuration that applies to a specific (metric, tag) pair.
+    ///
+    /// Lookup chain (per field):
+    /// - `value_limit`, `mode`, `internal_metrics`: per-tag override → per-metric override → global.
+    /// - `limit_exceeded_action`: per-metric override → global. Per-tag entries always inherit
+    ///   the action from the enclosing per-metric (or global) config.
+    ///
+    /// Per-metric exclusion is blanket: if the matching per-metric entry has `mode: excluded`,
+    /// every tag on the metric is excluded and `per_tag_limits` is ignored.
+    fn get_config_for_metric_tag(
+        &self,
+        metric_key: Option<&MetricId>,
+        tag_key: &str,
+    ) -> TagSettings {
+        // No matching per-metric override → use the global config as-is.
+        let Some((metric_namespace, metric_name)) = metric_key else {
+            return TagSettings::Tracked(self.config.global);
+        };
+        let Some((_, per_metric)) = self.config.per_metric_limits.iter().find(|(name, cfg)| {
+            *name == metric_name && (cfg.namespace.is_none() || cfg.namespace == *metric_namespace)
+        }) else {
+            return TagSettings::Tracked(self.config.global);
+        };
+
+        // Per-metric exclusion is blanket — per-tag overrides do not apply.
+        let Some(metric_mode) = per_metric.config.mode.as_mode() else {
+            return TagSettings::Excluded;
+        };
+        let limit_exceeded_action = per_metric.config.limit_exceeded_action;
+
+        // Per-tag override may further exclude a specific tag, replace `mode`,
+        // or replace `value_limit` (unset `value_limit` inherits from the enclosing
+        // per-metric config).
+        if let Some(per_tag) = per_metric.per_tag_limits.get(tag_key) {
+            let Some(mode) = per_tag.config.mode.as_mode() else {
+                return TagSettings::Excluded;
+            };
+            return TagSettings::Tracked(Inner {
+                value_limit: per_tag
+                    .config
+                    .value_limit
+                    .unwrap_or(per_metric.config.value_limit),
+                limit_exceeded_action,
+                mode,
+                internal_metrics: per_metric.config.internal_metrics,
+            });
+        }
+        TagSettings::Tracked(Inner {
+            value_limit: per_metric.config.value_limit,
+            limit_exceeded_action,
+            mode: metric_mode,
+            internal_metrics: per_metric.config.internal_metrics,
+        })
+    }
+
+    /// Returns the `limit_exceeded_action` that applies to this metric. Decided once per event:
+    /// per-metric override if any, else global.
+    fn metric_action(&self, metric_key: Option<&MetricId>) -> LimitExceededAction {
+        if let Some(id) = metric_key
+            && let Some((_, pmc)) =
+                self.config.per_metric_limits.iter().find(|(name, c)| {
+                    **name == id.1 && (c.namespace.is_none() || c.namespace == id.0)
                 })
-                .map(|(_, c)| &c.config)
-                .unwrap_or(&self.config.global),
-            None => &self.config.global,
+        {
+            return pmc.config.limit_exceeded_action;
         }
+        self.config.global.limit_exceeded_action
     }
 
     /// Takes in key and a value corresponding to a tag on an incoming Metric
@@ -67,7 +130,10 @@ impl TagCardinalityLimit {
         key: &str,
         value: &TagValueSet,
     ) -> bool {
-        let config = *self.get_config_for_metric(metric_key);
+        let config = match self.get_config_for_metric_tag(metric_key, key) {
+            TagSettings::Excluded => return true,
+            TagSettings::Tracked(inner) => inner,
+        };
         let metric_accepted_tags = self.accepted_tags.entry(metric_key.cloned()).or_default();
         let tag_value_set = metric_accepted_tags
             .entry_ref(key)
@@ -102,20 +168,28 @@ impl TagCardinalityLimit {
         key: &str,
         value: &TagValueSet,
     ) -> bool {
+        let resolved = match self.get_config_for_metric_tag(metric_key, key) {
+            TagSettings::Excluded => return false,
+            TagSettings::Tracked(inner) => inner,
+        };
         self.accepted_tags
             .get(&metric_key.cloned())
             .and_then(|metric_accepted_tags| {
                 metric_accepted_tags.get(key).map(|value_set| {
-                    !value_set.contains(value)
-                        && value_set.len() >= self.get_config_for_metric(metric_key).value_limit
+                    !value_set.contains(value) && value_set.len() >= resolved.value_limit
                 })
             })
             .unwrap_or(false)
     }
 
-    /// Record a key and value corresponding to a tag on an incoming Metric.
+    /// Record an accepted tag value (mutation-only, no limit check). Used by the `DropEvent`
+    /// path's record pass after a mutation-free pre-check has confirmed every tag has room.
+    /// Excluded tags are skipped — no storage allocated.
     fn record_tag_value(&mut self, metric_key: Option<&MetricId>, key: &str, value: &TagValueSet) {
-        let config = *self.get_config_for_metric(metric_key);
+        let config = match self.get_config_for_metric_tag(metric_key, key) {
+            TagSettings::Excluded => return,
+            TagSettings::Tracked(inner) => inner,
+        };
         let metric_accepted_tags = self.accepted_tags.entry(metric_key.cloned()).or_default();
         metric_accepted_tags
             .entry_ref(key)
@@ -143,24 +217,24 @@ impl TagCardinalityLimit {
             }
         };
         if let Some(tags_map) = metric.tags_mut() {
-            match self
-                .get_config_for_metric(metric_key.as_ref())
-                .limit_exceeded_action
-            {
+            match self.metric_action(metric_key.as_ref()) {
                 LimitExceededAction::DropEvent => {
-                    // This needs to check all the tags, to ensure that the ordering of tag names
-                    // doesn't change the behavior of the check.
-
+                    // This needs to check all the tags, to ensure that the ordering of tag
+                    // names doesn't change the behavior of the check.
                     for (key, value) in tags_map.iter_sets() {
+                        let TagSettings::Tracked(resolved) =
+                            self.get_config_for_metric_tag(metric_key.as_ref(), key)
+                        else {
+                            continue; // excluded tags can never trigger DropEvent
+                        };
                         if self.tag_limit_exceeded(metric_key.as_ref(), key, value) {
-                            let config = self.get_config_for_metric(metric_key.as_ref());
+                            let include_extended_tags =
+                                resolved.internal_metrics.include_extended_tags;
                             emit!(TagCardinalityLimitRejectingEvent {
                                 metric_name: &metric_name,
                                 tag_key: key,
                                 tag_value: &value.to_string(),
-                                include_extended_tags: config
-                                    .internal_metrics
-                                    .include_extended_tags,
+                                include_extended_tags,
                             });
                             return None;
                         }
@@ -170,12 +244,17 @@ impl TagCardinalityLimit {
                     }
                 }
                 LimitExceededAction::DropTag => {
-                    let config = self.get_config_for_metric(metric_key.as_ref());
-                    let include_extended_tags = config.internal_metrics.include_extended_tags;
                     tags_map.retain(|key, value| {
                         if self.try_accept_tag(metric_key.as_ref(), key, value) {
                             true
                         } else {
+                            let include_extended_tags =
+                                match self.get_config_for_metric_tag(metric_key.as_ref(), key) {
+                                    TagSettings::Tracked(inner) => {
+                                        inner.internal_metrics.include_extended_tags
+                                    }
+                                    TagSettings::Excluded => false, // unreachable: excluded tags accept
+                                };
                             emit!(TagCardinalityLimitRejectingTag {
                                 metric_name: &metric_name,
                                 tag_key: key,