enhancement(sample transform): Add group tag to discarded event metrics

changelog.d/sample_discarded_events_group_tag.enhancement.md

@@ -0,0 +1,3 @@
+Add `internal_metrics.include_group_tag` to the `sample` transform. When enabled, `component_discarded_events_total` includes a `group` tag showing which `group_by` group an event belonged to when it was discarded. This option is disabled by default because configurations with many group values can produce a large number of unique metric tags.
+
+authors: arun.pidugu

src/internal_events/sample.rs

@@ -4,13 +4,89 @@ use vector_lib::{
 };
 
 #[derive(Debug, NamedInternalEvent)]
-pub struct SampleEventDiscarded;
+pub struct SampleEventDiscarded {
+    pub group: Option<String>,
+    pub include_group_tag: bool,
+}
 
 impl InternalEvent for SampleEventDiscarded {
     fn emit(self) {
-        emit!(ComponentEventsDropped::<INTENTIONAL> {
+        let group = self
+            .include_group_tag
+            .then(|| self.group.unwrap_or_else(|| "None".to_string()));
+        ComponentEventsDropped::<INTENTIONAL> {
             count: 1,
-            reason: "Sample discarded."
-        })
+            reason: "Sample discarded.",
+        }
+        .emit_with_group(group);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use serial_test::serial;
+    use vector_lib::{event::MetricValue, internal_event::InternalEvent, metrics::Controller};
+
+    use super::SampleEventDiscarded;
+
+    fn discarded_events_counter(tags: &[(&str, &str)]) -> Option<f64> {
+        Controller::get()
+            .expect("metrics controller initialized")
+            .capture_metrics()
+            .into_iter()
+            .find(|metric| {
+                metric.name() == "component_discarded_events_total"
+                    && tags.iter().all(|(key, value)| {
+                        metric
+                            .tags()
+                            .is_some_and(|tags| tags.get(key) == Some(*value))
+                    })
+            })
+            .map(|metric| match metric.value() {
+                MetricValue::Counter { value } => *value,
+                other => panic!("expected counter, got {other:?}"),
+            })
+    }
+
+    #[test]
+    #[serial]
+    fn emits_component_discarded_events_with_group_tag() {
+        vector_lib::metrics::init_test();
+        for group in ["group-a", "group-b"] {
+            SampleEventDiscarded {
+                group: Some(group.to_string()),
+                include_group_tag: true,
+            }
+            .emit();
+        }
+        SampleEventDiscarded {
+            group: None,
+            include_group_tag: true,
+        }
+        .emit();
+
+        for group in ["group-a", "group-b", "None"] {
+            assert_eq!(
+                discarded_events_counter(&[("intentional", "true"), ("group", group)]),
+                Some(1.0)
+            );
+        }
+    }
+
+    #[test]
+    #[serial]
+    fn emits_component_discarded_events_without_group_tag_by_default() {
+        vector_lib::metrics::init_test();
+        SampleEventDiscarded {
+            group: None,
+            include_group_tag: false,
+        }
+        .emit();
+
+        assert_eq!(
+            discarded_events_counter(&[("intentional", "true")]),
+            Some(1.0)
+        );
+        assert_eq!(discarded_events_counter(&[("group", "group-a")]), None);
     }
 }

src/transforms/sample/config.rs

@@ -48,6 +48,23 @@ pub enum SampleError {
     InvalidKeyFieldDynamicCombination,
 }
 
+/// Configuration of internal metrics for the Sample transform.
+#[configurable_component]
+#[derive(Clone, Debug, Default)]
+#[serde(deny_unknown_fields)]
+pub struct SampleInternalMetricsConfig {
+    /// Whether or not to include the `group` tag on the `component_discarded_events_total`
+    /// internal metric.
+    ///
+    /// When enabled, adds a `group` tag containing the rendered `group_by` value.
+    /// Missing or unrenderable values use `None`.
+    ///
+    /// Note that this defaults to false because the `group` tag has potentially unbounded
+    /// cardinality. Only set this to true if you know that the number of unique groups is bounded.
+    #[serde(default)]
+    pub include_group_tag: bool,
+}
+
 /// Configuration for the `sample` transform.
 #[configurable_component(transform(
     "sample",
@@ -125,6 +142,10 @@ pub struct SampleConfig {
 
     /// A logical condition used to exclude events from sampling.
     pub exclude: Option<AnyCondition>,
+
+    /// Configuration of internal metrics for the Sample transform.
+    #[serde(default)]
+    pub internal_metrics: SampleInternalMetricsConfig,
 }
 
 impl SampleConfig {
@@ -173,6 +194,7 @@ impl GenerateConfig for SampleConfig {
             group_by: None,
             exclude: None::<AnyCondition>,
             sample_rate_key: default_sample_rate_key(),
+            internal_metrics: Default::default(),
         })
         .unwrap()
     }
@@ -212,7 +234,9 @@ impl TransformConfig for SampleConfig {
             )
         };
 
-        Ok(Transform::function(sample))
+        Ok(Transform::function(sample.with_include_group_tag(
+            self.internal_metrics.include_group_tag,
+        )))
     }
 
     fn input(&self) -> Input {
@@ -279,6 +303,22 @@ mod tests {
         crate::test_util::test_generate_config::<SampleConfig>();
     }
 
+    #[test]
+    fn internal_metrics_include_group_tag_defaults_to_false() {
+        let config =
+            serde_yaml::from_str::<SampleConfig>("ratio: 0.5\ninternal_metrics: {}\n").unwrap();
+        assert!(!config.internal_metrics.include_group_tag);
+    }
+
+    #[test]
+    fn internal_metrics_include_group_tag_can_be_enabled() {
+        let config = serde_yaml::from_str::<SampleConfig>(
+            "ratio: 0.5\ninternal_metrics:\n  include_group_tag: true\n",
+        )
+        .unwrap();
+        assert!(config.internal_metrics.include_group_tag);
+    }
+
     #[test]
     fn rejects_dynamic_ratio_only_configuration() {
         let config = SampleConfig {
@@ -290,6 +330,7 @@ mod tests {
             sample_rate_key: super::default_sample_rate_key(),
             group_by: None,
             exclude: None,
+            internal_metrics: Default::default(),
         };
 
         let err = config.sample_rate().unwrap_err();
@@ -307,6 +348,7 @@ mod tests {
             sample_rate_key: super::default_sample_rate_key(),
             group_by: None,
             exclude: None,
+            internal_metrics: Default::default(),
         };
 
         let err = config.sample_rate().unwrap_err();
@@ -324,6 +366,7 @@ mod tests {
             sample_rate_key: super::default_sample_rate_key(),
             group_by: None,
             exclude: None,
+            internal_metrics: Default::default(),
         };
 
         assert!(config.validate_structure().is_ok());
@@ -340,6 +383,7 @@ mod tests {
             sample_rate_key: super::default_sample_rate_key(),
             group_by: None,
             exclude: None,
+            internal_metrics: Default::default(),
         };
 
         let err = config.sample_rate().unwrap_err();
@@ -357,6 +401,7 @@ mod tests {
             sample_rate_key: super::default_sample_rate_key(),
             group_by: None,
             exclude: None,
+            internal_metrics: Default::default(),
         };
 
         let err = config.sample_rate().unwrap_err();

src/transforms/sample/tests.rs

@@ -33,6 +33,7 @@ async fn emits_internal_events() {
             group_by: None,
             exclude: None,
             sample_rate_key: default_sample_rate_key(),
+            internal_metrics: Default::default(),
         };
         let (tx, rx) = mpsc::channel(1);
         let (topology, mut out) = create_topology(ReceiverStream::new(rx), config).await;

src/transforms/sample/transform.rs

@@ -144,6 +144,7 @@ pub struct Sample {
     dynamic_event_counters: HashMap<Option<String>, u64>,
     exclude: Option<Condition>,
     sample_rate_key: OptionalValuePath,
+    include_group_tag: bool,
 }
 
 impl Sample {
@@ -201,9 +202,15 @@ impl Sample {
             dynamic_event_counters: HashMap::default(),
             exclude,
             sample_rate_key,
+            include_group_tag: false,
         }
     }
 
+    pub const fn with_include_group_tag(mut self, include_group_tag: bool) -> Self {
+        self.include_group_tag = include_group_tag;
+        self
+    }
+
     #[cfg(test)]
     pub fn ratio(&self) -> f64 {
         match &self.static_mode {
@@ -339,6 +346,10 @@ impl FunctionTransform for Sample {
         };
 
         let group_by_key = self.group_by_key(&event);
+        let discarded_group = self
+            .include_group_tag
+            .then(|| group_by_key.clone())
+            .flatten();
         let value = self.static_key_value(&event);
 
         let event_sample_mode = self.event_sample_mode(&event);
@@ -375,7 +386,10 @@ impl FunctionTransform for Sample {
             }
             output.push(event);
         } else {
-            emit!(SampleEventDiscarded);
+            emit!(SampleEventDiscarded {
+                group: discarded_group,
+                include_group_tag: self.include_group_tag,
+            });
         }
     }
 }

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

@@ -22,6 +22,24 @@ generated: components: transforms: sample: configuration: {
 			syntax: "template"
 		}
 	}
+	internal_metrics: {
+		description: "Configuration of internal metrics for the Sample transform."
+		required:    false
+		type: object: options: include_group_tag: {
+			description: """
+				Whether or not to include the `group` tag on the `component_discarded_events_total`
+				internal metric.
+
+				When enabled, adds a `group` tag containing the rendered `group_by` value.
+				Missing or unrenderable values use `None`.
+
+				Note that this defaults to false because the `group` tag has potentially unbounded
+				cardinality. Only set this to true if you know that the number of unique groups is bounded.
+				"""
+			required: false
+			type: bool: default: false
+		}
+	}
 	key_field: {
 		description: """
 			The name of the field whose value is hashed to determine if the event should be