enhancement(reduce transform): add internal events (#3812)

Closes #3401

Signed-off-by: Luke Steensen <[email protected]>

Author: lukesteensen

src/internal_events/mod.rs

@@ -33,6 +33,8 @@ mod lua;
 mod process;
 #[cfg(feature = "sources-prometheus")]
 mod prometheus;
+#[cfg(feature = "transforms-reduce")]
+mod reduce;
 #[cfg(feature = "transforms-regex_parser")]
 mod regex_parser;
 mod remap;
@@ -102,6 +104,8 @@ pub use self::lua::*;
 pub use self::process::*;
 #[cfg(feature = "sources-prometheus")]
 pub use self::prometheus::*;
+#[cfg(feature = "transforms-reduce")]
+pub(crate) use self::reduce::*;
 #[cfg(feature = "transforms-regex_parser")]
 pub(crate) use self::regex_parser::*;
 pub use self::remap::*;

src/internal_events/reduce.rs

@@ -0,0 +1,26 @@
+use super::InternalEvent;
+use metrics::counter;
+
+#[derive(Debug)]
+pub(crate) struct ReduceEventProcessed;
+
+impl InternalEvent for ReduceEventProcessed {
+    fn emit_metrics(&self) {
+        counter!("events_processed", 1,
+            "component_kind" => "transform",
+            "component_type" => "reduce",
+        );
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct ReduceStaleEventFlushed;
+
+impl InternalEvent for ReduceStaleEventFlushed {
+    fn emit_metrics(&self) {
+        counter!("stale_events_flushed", 1,
+            "component_kind" => "transform",
+            "component_type" => "reduce",
+        );
+    }
+}

src/transforms/reduce/mod.rs

@@ -4,6 +4,7 @@ use crate::{
     config::{DataType, TransformConfig, TransformContext, TransformDescription},
     event::discriminant::Discriminant,
     event::{Event, LogEvent},
+    internal_events::{ReduceEventProcessed, ReduceStaleEventFlushed},
 };
 use async_stream::stream;
 use futures::{
@@ -180,6 +181,7 @@ impl Reduce {
         }
         for k in &flush_discriminants {
             if let Some(t) = self.reduce_merge_states.remove(k) {
+                emit!(ReduceStaleEventFlushed);
                 output.push(Event::from(t.flush()));
             }
         }
@@ -230,6 +232,8 @@ impl Transform for Reduce {
             }
         }
 
+        emit!(ReduceEventProcessed);
+
         self.flush_into(output);
     }