feat(internal_metrics source): Instrument few more components with metrics

src/internal_events/add_fields.rs

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

src/internal_events/aws_kinesis_streams.rs

@@ -0,0 +1,22 @@
+use super::InternalEvent;
+use metrics::counter;
+
+#[derive(Debug)]
+pub struct AwsKinesisStreamsEventSent {
+    pub byte_size: usize,
+}
+
+impl InternalEvent for AwsKinesisStreamsEventSent {
+    fn emit_metrics(&self) {
+        counter!(
+            "events_processed", 1,
+            "component_kind" => "sink",
+            "component_type" => "aws_kinesis_streams",
+        );
+        counter!(
+            "bytes_processed", self.byte_size as u64,
+            "component_kind" => "sink",
+            "component_type" => "aws_kinesis_streams",
+        );
+    }
+}

src/internal_events/json.rs

@@ -0,0 +1,41 @@
+use super::InternalEvent;
+use metrics::counter;
+use serde_json::Error;
+use string_cache::DefaultAtom as Atom;
+
+#[derive(Debug)]
+pub struct JsonEventProcessed;
+
+impl InternalEvent for JsonEventProcessed {
+    fn emit_metrics(&self) {
+        counter!("events_processed", 1,
+            "component_kind" => "transform",
+            "component_type" => "json_parser",
+        );
+    }
+}
+
+#[derive(Debug)]
+pub struct JsonFailedParse<'a> {
+    pub field: &'a Atom,
+    pub error: Error,
+}
+
+impl InternalEvent for JsonFailedParse<'_> {
+    fn emit_logs(&self) {
+        warn!(
+            message = "Event failed to parse as JSON",
+            field = %self.field,
+            %self.error,
+            rate_limit_secs = 30
+        )
+    }
+
+    fn emit_metrics(&self) {
+        counter!("processing_error", 1,
+            "component_kind" => "transform",
+            "component_type" => "json_parser",
+            "error_type" => "failed_parse",
+        );
+    }
+}

src/internal_events/mod.rs

@@ -1,25 +1,33 @@
+mod add_fields;
+mod aws_kinesis_streams;
 mod blackhole;
 mod elasticsearch;
 mod file;
+mod json;
 #[cfg(feature = "transforms-lua")]
 mod lua;
 #[cfg(feature = "sources-prometheus")]
 mod prometheus;
 mod regex;
+mod splunk_hec;
 mod syslog;
 mod tcp;
 mod udp;
 mod unix;
 mod vector;
 
+pub use self::add_fields::*;
+pub use self::aws_kinesis_streams::*;
 pub use self::blackhole::*;
 pub use self::elasticsearch::*;
 pub use self::file::*;
+pub use self::json::*;
 #[cfg(feature = "transforms-lua")]
 pub use self::lua::*;
 #[cfg(feature = "sources-prometheus")]
 pub use self::prometheus::*;
 pub use self::regex::*;
+pub use self::splunk_hec::*;
 pub use self::syslog::*;
 pub use self::tcp::*;
 pub use self::udp::*;

src/internal_events/splunk_hec.rs

@@ -0,0 +1,46 @@
+use super::InternalEvent;
+use metrics::counter;
+use serde_json::Error;
+
+#[derive(Debug)]
+pub struct SplunkEventSent {
+    pub byte_size: usize,
+}
+
+impl InternalEvent for SplunkEventSent {
+    fn emit_metrics(&self) {
+        counter!(
+            "events_processed", 1,
+            "component_kind" => "sink",
+            "component_type" => "splunk_hec",
+        );
+        counter!(
+            "bytes_processed", self.byte_size as u64,
+            "component_kind" => "sink",
+            "component_type" => "splunk_hec",
+        );
+    }
+}
+
+#[derive(Debug)]
+pub struct SplunkEventEncodeError {
+    pub error: Error,
+}
+
+impl InternalEvent for SplunkEventEncodeError {
+    fn emit_logs(&self) {
+        error!(
+            message = "Error encoding Splunk HEC event to json",
+            error = ?self.error,
+            rate_limit_secs = 30,
+        );
+    }
+
+    fn emit_metrics(&self) {
+        counter!(
+            "encode_errors", 1,
+            "component_kind" => "sink",
+            "component_type" => "splunk_hec",
+        );
+    }
+}

src/sinks/aws_kinesis_streams.rs

@@ -1,6 +1,7 @@
 use crate::{
     dns::Resolver,
     event::{self, Event},
+    internal_events::AwsKinesisStreamsEventSent,
     region::RegionOrEndpoint,
     sinks::util::{
         encoding::{EncodingConfig, EncodingConfiguration},
@@ -251,6 +252,9 @@ fn encode_event(
 
     let data = Bytes::from(data);
 
+    emit!(AwsKinesisStreamsEventSent {
+        byte_size: data.len()
+    });
     Some(PutRecordsRequestEntry {
         data,
         partition_key,

src/sinks/splunk_hec.rs

@@ -1,6 +1,7 @@
 use crate::{
     dns::Resolver,
     event::{self, Event, LogEvent, Value},
+    internal_events::{SplunkEventEncodeError, SplunkEventSent},
     sinks::util::{
         encoding::{EncodingConfigWithDefault, EncodingConfiguration},
         http::{BatchedHttpSink, HttpClient, HttpSink},
@@ -159,9 +160,18 @@ impl HttpSink for HecSinkConfig {
             body["sourcetype"] = json!(sourcetype);
         }
 
-        serde_json::to_vec(&body)
-            .map_err(|e| error!("Error encoding json body: {}", e))
-            .ok()
+        match serde_json::to_vec(&body) {
+            Ok(value) => {
+                emit!(SplunkEventSent {
+                    byte_size: value.len()
+                });
+                Some(value)
+            }
+            Err(e) => {
+                emit!(SplunkEventEncodeError { error: e });
+                None
+            }
+        }
     }
 
     fn build_request(&self, events: Self::Output) -> http::Request<Vec<u8>> {

src/transforms/add_fields.rs

@@ -2,6 +2,7 @@ use super::Transform;
 use crate::serde::Fields;
 use crate::{
     event::{Event, Value},
+    internal_events::AddFieldsEventProcessed,
     template::Template,
     topology::config::{DataType, TransformConfig, TransformContext, TransformDescription},
 };
@@ -89,6 +90,8 @@ impl AddFields {
 
 impl Transform for AddFields {
     fn transform(&mut self, mut event: Event) -> Option<Event> {
+        emit!(AddFieldsEventProcessed);
+
         for (key, value_or_template) in self.fields.clone() {
             let value = match value_or_template {
                 TemplateOrValue::Template(v) => match v.render_string(&event) {

src/transforms/json_parser.rs

@@ -1,6 +1,7 @@
 use super::Transform;
 use crate::{
     event::{self, Event},
+    internal_events::{JsonEventProcessed, JsonFailedParse},
     topology::config::{DataType, TransformConfig, TransformContext, TransformDescription},
 };
 use serde::{Deserialize, Serialize};
@@ -74,16 +75,16 @@ impl Transform for JsonParser {
         let log = event.as_mut_log();
         let to_parse = log.get(&self.field).map(|s| s.as_bytes());
 
+        emit!(JsonEventProcessed);
+
         let parsed = to_parse
             .and_then(|to_parse| {
                 serde_json::from_slice::<Value>(to_parse.as_ref())
                     .map_err(|error| {
-                        warn!(
-                            message = "Event failed to parse as JSON",
-                            field = self.field.as_ref(),
-                            %error,
-                            rate_limit_secs = 30
-                        )
+                        emit!(JsonFailedParse {
+                            field: &self.field,
+                            error: error
+                        })
                     })
                     .ok()
             })