enhancement(dedupe transform): add internal events (#3809)

Closes #3393

Signed-off-by: Luke Steensen <[email protected]>
Co-authored-by: Jean Mertz <[email protected]>

src/internal_events/dedupe.rs

@@ -0,0 +1,36 @@
+use super::InternalEvent;
+use metrics::counter;
+
+#[derive(Debug)]
+pub(crate) struct DedupeEventProcessed;
+
+impl InternalEvent for DedupeEventProcessed {
+    fn emit_metrics(&self) {
+        counter!("events_processed", 1,
+            "component_kind" => "transform",
+            "component_type" => "dedupe",
+        );
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct DedupeEventDiscarded {
+    pub event: crate::Event,
+}
+
+impl InternalEvent for DedupeEventDiscarded {
+    fn emit_logs(&self) {
+        warn!(
+            message = "Encountered duplicate event; discarding.",
+            rate_limit_secs = 30
+        );
+        trace!(message = "Encountered duplicate event; discarding.", event = ?self.event);
+    }
+
+    fn emit_metrics(&self) {
+        counter!("duplicate_events_discarded", 1,
+            "component_kind" => "transform",
+            "component_type" => "dedupe",
+        );
+    }
+}

src/internal_events/mod.rs

@@ -10,6 +10,8 @@ mod blackhole;
 mod coercer;
 #[cfg(feature = "transforms-concat")]
 mod concat;
+#[cfg(feature = "transforms-dedupe")]
+mod dedupe;
 #[cfg(feature = "sources-docker")]
 mod docker;
 mod elasticsearch;
@@ -80,6 +82,8 @@ pub use self::blackhole::*;
 pub(crate) use self::coercer::*;
 #[cfg(feature = "transforms-concat")]
 pub use self::concat::*;
+#[cfg(feature = "transforms-dedupe")]
+pub(crate) use self::dedupe::*;
 #[cfg(feature = "sources-docker")]
 pub use self::docker::*;
 pub use self::elasticsearch::*;

src/transforms/dedupe.rs

@@ -2,6 +2,7 @@ use super::Transform;
 use crate::{
     config::{log_schema, DataType, TransformConfig, TransformContext, TransformDescription},
     event::{Event, Value},
+    internal_events::{DedupeEventDiscarded, DedupeEventProcessed},
 };
 use bytes::Bytes;
 use lru::LruCache;
@@ -182,13 +183,10 @@ fn build_cache_entry(event: &Event, fields: &FieldMatchConfig) -> CacheEntry {
 
 impl Transform for Dedupe {
     fn transform(&mut self, event: Event) -> Option<Event> {
+        emit!(DedupeEventProcessed);
         let cache_entry = build_cache_entry(&event, &self.config.fields);
         if self.cache.put(cache_entry, true).is_some() {
-            warn!(
-                message = "Encountered duplicate event; discarding",
-                rate_limit_secs = 30
-            );
-            trace!(message = "Encountered duplicate event; discarding", ?event);
+            emit!(DedupeEventDiscarded { event });
             None
         } else {
             Some(event)