Implement per tag control in tag cardinality processor

Author: ArunPiduguDD

changelog.d/exclude_cardinality_limit.enhancement.md

@@ -0,0 +1,5 @@
+The `tag_cardinality_limit` transform now supports an `exclude: true` option that opts a metric or
+tag entirely out of cardinality control — all tag values pass through and nothing is tracked.
+`exclude` works at the global, per-metric, and per-tag levels, and takes precedence over `mode`,
+`value_limit`, `limit_exceeded_action`, and `internal_metrics` when also set. At least one of
+`mode` or `exclude: true` must be present on every entry.

changelog.d/per_tag_cardinality_limits.enhancement.md

@@ -0,0 +1,4 @@
+The `tag_cardinality_limit` transform now supports per-tag-key configuration overrides nested
+under each entry in `per_metric_limits`. Each per-tag entry has the same fields as a per-metric
+configuration (`value_limit`, `limit_exceeded_action`, `mode`, `internal_metrics`), and replaces
+the per-metric configuration for that tag. Resolution order is per-tag → per-metric → global.

lib/vector-core/src/event/metric/tags.rs

@@ -777,4 +777,34 @@ mod tests {
             }
         }
     }
+
+    fn make_tags(pairs: &[(&str, &str)]) -> MetricTags {
+        pairs
+            .iter()
+            .map(|(k, v)| (k.to_string(), v.to_string()))
+            .collect()
+    }
+
+    #[test]
+    fn rename_with_replacement_basic() {
+        let mut tags = make_tags(&[("old", "v1")]);
+        assert!(tags.rename_with_replacement("old", "new".to_string()));
+        assert_eq!(tags.get("new"), Some("v1"));
+        assert!(!tags.contains_key("old"));
+    }
+
+    #[test]
+    fn rename_with_replacement_missing_old_returns_false() {
+        let mut tags = make_tags(&[("a", "1")]);
+        assert!(!tags.rename_with_replacement("missing", "b".to_string()));
+        assert!(tags.contains_key("a"));
+    }
+
+    #[test]
+    fn rename_with_replacement_overwrites_existing_destination() {
+        let mut tags = make_tags(&[("old", "v1"), ("new", "v2")]);
+        assert!(tags.rename_with_replacement("old", "new".to_string()));
+        assert_eq!(tags.get("new"), Some("v1"));
+        assert!(!tags.contains_key("old"));
+    }
 }

src/transforms/tag_cardinality_limit/config.rs

@@ -48,20 +48,53 @@ pub struct Config {
 #[configurable_component]
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub struct Inner {
-    /// How many distinct values to accept for any given key.
+    /// How many distinct values to accept for any given key. Ignored when `exclude` is `true`.
     #[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: Mode,
+    /// Cardinality tracking mode. Required unless `exclude` is `true`. When both `exclude` and
+    /// `mode` are set, `exclude` takes precedence and `mode` is ignored.
+    #[serde(flatten, default)]
+    pub mode: Option<Mode>,
 
     #[configurable(derived)]
     #[serde(default)]
     pub internal_metrics: InternalMetricsConfig,
+
+    /// Exclude this metric or tag from cardinality control entirely. When `true`, all tag values
+    /// pass through and nothing is tracked, regardless of any other fields on this entry.
+    #[serde(default)]
+    pub exclude: bool,
+}
+
+impl Inner {
+    /// Validate that at least one of `exclude` or `mode` is set.
+    pub(super) fn validate(&self, location: &str) -> crate::Result<()> {
+        if !self.exclude && self.mode.is_none() {
+            return Err(
+                format!("{location}: `mode` is required unless `exclude: true` is set").into(),
+            );
+        }
+        Ok(())
+    }
+}
+
+impl Config {
+    pub(super) fn validate(&self) -> crate::Result<()> {
+        self.global.validate("global")?;
+        for (name, pmc) in &self.per_metric_limits {
+            pmc.config.validate(&format!("per_metric_limits[{name}]"))?;
+            for (tag, ptc) in &pmc.per_tag_limits {
+                ptc.config
+                    .validate(&format!("per_metric_limits[{name}].per_tag_limits[{tag}]"))?;
+            }
+        }
+        Ok(())
+    }
 }
 
 /// Controls the approach taken for tracking tag cardinality.
@@ -121,6 +154,26 @@ pub struct PerMetricConfig {
     #[serde(default)]
     pub namespace: Option<String>,
 
+    /// Per-tag-key overrides scoped to this metric.
+    ///
+    /// Each entry has the same fields as a per-metric configuration. When a tag has an entry here,
+    /// that entry replaces the per-metric configuration for that tag. Tags not listed here use this
+    /// 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: Inner,
+}
+
+/// 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,
 }
@@ -145,10 +198,11 @@ impl GenerateConfig for Config {
     fn generate_config() -> toml::Value {
         toml::Value::try_from(Self {
             global: Inner {
-                mode: Mode::Exact,
+                mode: Some(Mode::Exact),
                 value_limit: default_value_limit(),
                 limit_exceeded_action: default_limit_exceeded_action(),
                 internal_metrics: InternalMetricsConfig::default(),
+                exclude: false,
             },
             per_metric_limits: HashMap::default(),
         })
@@ -160,6 +214,7 @@ impl GenerateConfig for Config {
 #[typetag::serde(name = "tag_cardinality_limit")]
 impl TransformConfig for Config {
     async fn build(&self, _context: &TransformContext) -> crate::Result<Transform> {
+        self.validate()?;
         Ok(Transform::event_task(TagCardinalityLimit::new(
             self.clone(),
         )))

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, PerTagConfig,
+};
 use tag_value_set::AcceptedTagValueSet;
 
 use crate::event::metric::TagValueSet;
@@ -36,19 +38,22 @@ 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-tag override → per-metric override → global.
+    fn resolve_tag(&self, metric_key: Option<&MetricId>, tag_key: &str) -> &Inner {
+        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,
+        {
+            if let Some(t) = pmc.per_tag_limits.get(tag_key) {
+                return &t.config;
+            }
+            return &pmc.config;
         }
+        &self.config.global
     }
 
     /// Takes in key and a value corresponding to a tag on an incoming Metric
@@ -65,11 +70,15 @@ impl TagCardinalityLimit {
         key: &str,
         value: &TagValueSet,
     ) -> bool {
-        let config = *self.get_config_for_metric(metric_key);
+        let config = *self.resolve_tag(metric_key, key);
+        if config.exclude {
+            return true;
+        }
+        let mode = config.mode.expect("validated at build time");
         let metric_accepted_tags = self.accepted_tags.entry(metric_key.cloned()).or_default();
         let tag_value_set = metric_accepted_tags
             .entry_ref(key)
-            .or_insert_with(|| AcceptedTagValueSet::new(config.value_limit, &config.mode));
+            .or_insert_with(|| AcceptedTagValueSet::new(config.value_limit, &mode));
 
         if tag_value_set.contains(value) {
             // Tag value has already been accepted, nothing more to do.
@@ -100,27 +109,20 @@ impl TagCardinalityLimit {
         key: &str,
         value: &TagValueSet,
     ) -> bool {
+        let resolved = self.resolve_tag(metric_key, key);
+        if resolved.exclude {
+            return false;
+        }
         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.
-    fn record_tag_value(&mut self, metric_key: Option<&MetricId>, key: &str, value: &TagValueSet) {
-        let config = *self.get_config_for_metric(metric_key);
-        let metric_accepted_tags = self.accepted_tags.entry(metric_key.cloned()).or_default();
-        metric_accepted_tags
-            .entry_ref(key)
-            .or_insert_with(|| AcceptedTagValueSet::new(config.value_limit, &config.mode))
-            .insert(value.clone());
-    }
-
     pub fn transform_one(&mut self, mut event: Event) -> Option<Event> {
         let metric = event.as_mut_metric();
         let metric_name = metric.name().to_string();
@@ -135,50 +137,44 @@ impl TagCardinalityLimit {
             None
         };
         if let Some(tags_map) = metric.tags_mut() {
-            match self
-                .get_config_for_metric(metric_key.as_ref())
-                .limit_exceeded_action
-            {
-                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.
-
-                    for (key, value) in tags_map.iter_sets() {
-                        if self.tag_limit_exceeded(metric_key.as_ref(), key, value) {
-                            let config = self.get_config_for_metric(metric_key.as_ref());
-                            emit!(TagCardinalityLimitRejectingEvent {
-                                metric_name: &metric_name,
-                                tag_key: key,
-                                tag_value: &value.to_string(),
-                                include_extended_tags: config
-                                    .internal_metrics
-                                    .include_extended_tags,
-                            });
-                            return None;
-                        }
-                    }
-                    for (key, value) in tags_map.iter_sets() {
-                        self.record_tag_value(metric_key.as_ref(), key, value);
-                    }
-                }
-                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 {
-                            emit!(TagCardinalityLimitRejectingTag {
-                                metric_name: &metric_name,
-                                tag_key: key,
-                                tag_value: &value.to_string(),
-                                include_extended_tags,
-                            });
-                            false
-                        }
+            // Pre-check pass: tags whose resolved action is `DropEvent` must be checked before
+            // any mutation, so the drop decision is order-independent across tag iteration.
+            for (key, value) in tags_map.iter_sets() {
+                let resolved = self.resolve_tag(metric_key.as_ref(), key);
+                if resolved.limit_exceeded_action == LimitExceededAction::DropEvent
+                    && self.tag_limit_exceeded(metric_key.as_ref(), key, value)
+                {
+                    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,
                     });
+                    return None;
                 }
             }
+
+            // Apply pass: accept each tag. Tags whose resolved action is `DropEvent` are
+            // guaranteed to pass after the pre-check; a rejection here means the resolved
+            // action is `DropTag`.
+            tags_map.retain(|key, value| {
+                if self.try_accept_tag(metric_key.as_ref(), key, value) {
+                    true
+                } else {
+                    let include_extended_tags = self
+                        .resolve_tag(metric_key.as_ref(), key)
+                        .internal_metrics
+                        .include_extended_tags;
+                    emit!(TagCardinalityLimitRejectingTag {
+                        metric_name: &metric_name,
+                        tag_key: key,
+                        tag_value: &value.to_string(),
+                        include_extended_tags,
+                    });
+                    false
+                }
+            });
         }
         Some(event)
     }