enhancement(splunk_hec sink, humio_logs sink, codecs): Integrate encoding::Encoder with splunk_hec/humio_logs sink (#12495)

* Integrate `encoding::Encoder` with `splunk_hec` sink
* Test that `encode_json` implementation matches `encode
* Rename `encode_json` -> `to_json_value`
* Make sink input type dependend on available codec input type

Signed-off-by: Pablo Sichert <[email protected]>

Author: pablosichert

lib/codecs/src/encoding/format/json.rs

@@ -40,6 +40,15 @@ impl JsonSerializer {
     pub const fn new() -> Self {
         Self
     }
+
+    /// Encode event and represent it as JSON value.
+    pub fn to_json_value(&self, event: Event) -> Result<serde_json::Value, serde_json::Error> {
+        match event {
+            Event::Log(log) => serde_json::to_value(&log),
+            Event::Metric(metric) => serde_json::to_value(&metric),
+            Event::Trace(trace) => serde_json::to_value(&trace),
+        }
+    }
 }
 
 impl Encoder<Event> for JsonSerializer {
@@ -76,4 +85,19 @@ mod tests {
 
         assert_eq!(bytes.freeze(), r#"{"foo":"bar"}"#);
     }
+
+    #[test]
+    fn serialize_equals_to_json_value() {
+        let event = Event::from(btreemap! {
+            "foo" => Value::from("bar")
+        });
+        let mut serializer = JsonSerializer::new();
+        let mut bytes = BytesMut::new();
+
+        serializer.encode(event.clone(), &mut bytes).unwrap();
+
+        let json = serializer.to_json_value(event).unwrap();
+
+        assert_eq!(bytes.freeze(), serde_json::to_string(&json).unwrap());
+    }
 }

lib/codecs/src/encoding/format/native_json.rs

@@ -28,6 +28,18 @@ impl NativeJsonSerializerConfig {
 #[derive(Debug, Clone)]
 pub struct NativeJsonSerializer;
 
+impl NativeJsonSerializer {
+    /// Creates a new `NativeJsonSerializer`.
+    pub const fn new() -> Self {
+        Self
+    }
+
+    /// Encode event and represent it as native JSON value.
+    pub fn to_json_value(&self, event: Event) -> Result<serde_json::Value, serde_json::Error> {
+        serde_json::to_value(&event)
+    }
+}
+
 impl Encoder<Event> for NativeJsonSerializer {
     type Error = vector_core::Error;
 
@@ -36,3 +48,40 @@ impl Encoder<Event> for NativeJsonSerializer {
         serde_json::to_writer(writer, &event).map_err(Into::into)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use bytes::BytesMut;
+    use vector_common::btreemap;
+    use vector_core::event::Value;
+
+    use super::*;
+
+    #[test]
+    fn serialize_json() {
+        let event = Event::from(btreemap! {
+            "foo" => Value::from("bar")
+        });
+        let mut serializer = NativeJsonSerializer::new();
+        let mut bytes = BytesMut::new();
+
+        serializer.encode(event, &mut bytes).unwrap();
+
+        assert_eq!(bytes.freeze(), r#"{"log":{"foo":"bar"}}"#);
+    }
+
+    #[test]
+    fn serialize_equals_to_json_value() {
+        let event = Event::from(btreemap! {
+            "foo" => Value::from("bar")
+        });
+        let mut serializer = NativeJsonSerializer::new();
+        let mut bytes = BytesMut::new();
+
+        serializer.encode(event.clone(), &mut bytes).unwrap();
+
+        let json = serializer.to_json_value(event).unwrap();
+
+        assert_eq!(bytes.freeze(), serde_json::to_string(&json).unwrap());
+    }
+}

lib/codecs/src/encoding/mod.rs

@@ -290,6 +290,38 @@ pub enum Serializer {
     Text(TextSerializer),
 }
 
+impl Serializer {
+    /// Check if the serializer supports encoding to JSON via `Serializer::to_json_value`.
+    pub fn supports_json(&self) -> bool {
+        match self {
+            Serializer::Json(_) | Serializer::NativeJson(_) => true,
+            Serializer::Logfmt(_)
+            | Serializer::Text(_)
+            | Serializer::Native(_)
+            | Serializer::RawMessage(_) => false,
+        }
+    }
+
+    /// Encode event and represent it as JSON value.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the serializer does not support encoding to JSON. Call `Serializer::supports_json`
+    /// if you need to determine the capability to encode to JSON at runtime.
+    pub fn to_json_value(&self, event: Event) -> Result<serde_json::Value, serde_json::Error> {
+        match self {
+            Serializer::Json(serializer) => serializer.to_json_value(event),
+            Serializer::NativeJson(serializer) => serializer.to_json_value(event),
+            Serializer::Logfmt(_)
+            | Serializer::Text(_)
+            | Serializer::Native(_)
+            | Serializer::RawMessage(_) => {
+                panic!("Serializer does not support JSON")
+            }
+        }
+    }
+}
+
 impl From<JsonSerializer> for Serializer {
     fn from(serializer: JsonSerializer) -> Self {
         Self::Json(serializer)

src/sinks/humio/logs.rs

@@ -1,6 +1,6 @@
 use serde::{Deserialize, Serialize};
 
-use super::{host_key, Encoding};
+use super::host_key;
 use crate::{
     config::{
         AcknowledgementsConfig, GenerateConfig, Input, SinkConfig, SinkContext, SinkDescription,
@@ -11,9 +11,12 @@ use crate::{
                 acknowledgements::HecClientAcknowledgementsConfig, timestamp_key,
                 SplunkHecDefaultBatchSettings,
             },
-            logs::config::HecLogsSinkConfig,
+            logs::config::{HecEncoding, HecEncodingMigrator, HecLogsSinkConfig},
+        },
+        util::{
+            encoding::{EncodingConfig, EncodingConfigAdapter},
+            BatchConfig, Compression, TowerRequestConfig,
         },
-        util::{encoding::EncodingConfig, BatchConfig, Compression, TowerRequestConfig},
         Healthcheck, VectorSink,
     },
     template::Template,
@@ -23,13 +26,14 @@ use crate::{
 const HOST: &str = "https://cloud.humio.com";
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
 pub struct HumioLogsConfig {
     pub(super) token: String,
     // Deprecated name
     #[serde(alias = "host")]
     pub(super) endpoint: Option<String>,
     pub(super) source: Option<Template>,
-    pub(super) encoding: EncodingConfig<Encoding>,
+    pub(super) encoding: EncodingConfigAdapter<EncodingConfig<HecEncoding>, HecEncodingMigrator>,
     pub(super) event_type: Option<Template>,
     #[serde(default = "host_key")]
     pub(super) host_key: String,
@@ -70,7 +74,7 @@ impl GenerateConfig for HumioLogsConfig {
             token: "${HUMIO_TOKEN}".to_owned(),
             endpoint: None,
             source: None,
-            encoding: Encoding::Json.into(),
+            encoding: EncodingConfig::from(HecEncoding::Json).into(),
             event_type: None,
             indexed_fields: vec![],
             index: None,
@@ -95,7 +99,7 @@ impl SinkConfig for HumioLogsConfig {
     }
 
     fn input(&self) -> Input {
-        Input::log()
+        Input::new(self.encoding.config().input_type())
     }
 
     fn sink_type(&self) -> &'static str {
@@ -120,7 +124,7 @@ impl HumioLogsConfig {
             sourcetype: self.event_type.clone(),
             source: self.source.clone(),
             timestamp_nanos_key: self.timestamp_nanos_key.clone(),
-            encoding: self.encoding.clone().into_encoding(),
+            encoding: self.encoding.clone(),
             compression: self.compression,
             batch: self.batch,
             request: self.request,
@@ -302,7 +306,7 @@ mod integration_tests {
             token: token.to_string(),
             endpoint: Some(humio_address()),
             source: None,
-            encoding: Encoding::Json.into(),
+            encoding: EncodingConfig::from(HecEncoding::Json).into(),
             event_type: None,
             host_key: log_schema().host_key().to_string(),
             indexed_fields: vec![],

src/sinks/humio/metrics.rs

@@ -5,16 +5,22 @@ use indoc::indoc;
 use serde::{Deserialize, Serialize};
 use vector_core::{sink::StreamSink, transform::Transform};
 
-use super::{host_key, logs::HumioLogsConfig, Encoding};
+use super::{host_key, logs::HumioLogsConfig};
 use crate::{
     config::{
         AcknowledgementsConfig, GenerateConfig, Input, SinkConfig, SinkContext, SinkDescription,
         TransformConfig, TransformContext,
     },
     event::{Event, EventArray, EventContainer},
     sinks::{
-        splunk_hec::common::SplunkHecDefaultBatchSettings,
-        util::{encoding::EncodingConfig, BatchConfig, Compression, TowerRequestConfig},
+        splunk_hec::{
+            common::SplunkHecDefaultBatchSettings,
+            logs::config::{HecEncoding, HecEncodingMigrator},
+        },
+        util::{
+            encoding::{EncodingConfig, EncodingConfigAdapter},
+            BatchConfig, Compression, TowerRequestConfig,
+        },
         Healthcheck, VectorSink,
     },
     template::Template,
@@ -23,6 +29,7 @@ use crate::{
 };
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
 struct HumioMetricsConfig {
     #[serde(flatten)]
     transform: MetricToLogConfig,
@@ -31,7 +38,7 @@ struct HumioMetricsConfig {
     #[serde(alias = "host")]
     pub(in crate::sinks::humio) endpoint: Option<String>,
     source: Option<Template>,
-    encoding: EncodingConfig<Encoding>,
+    encoding: EncodingConfigAdapter<EncodingConfig<HecEncoding>, HecEncodingMigrator>,
     event_type: Option<Template>,
     #[serde(default = "host_key")]
     host_key: String,

src/sinks/humio/mod.rs

@@ -1,26 +1,6 @@
 pub mod logs;
 pub mod metrics;
 
-use serde::{Deserialize, Serialize};
-
-use crate::sinks::splunk_hec;
-
-#[derive(Deserialize, Serialize, Debug, Eq, PartialEq, Clone)]
-#[serde(rename_all = "snake_case")]
-enum Encoding {
-    Json,
-    Text,
-}
-
-impl From<Encoding> for splunk_hec::logs::encoder::HecLogsEncoder {
-    fn from(v: Encoding) -> Self {
-        match v {
-            Encoding::Json => splunk_hec::logs::encoder::HecLogsEncoder::Json,
-            Encoding::Text => splunk_hec::logs::encoder::HecLogsEncoder::Text,
-        }
-    }
-}
-
 fn host_key() -> String {
     crate::config::log_schema().host_key().to_string()
 }

src/sinks/splunk_hec/logs/config.rs

@@ -1,12 +1,14 @@
 use std::sync::Arc;
 
+use codecs::{encoding::SerializerConfig, JsonSerializerConfig, TextSerializerConfig};
 use futures_util::FutureExt;
 use serde::{Deserialize, Serialize};
 use tower::ServiceBuilder;
 use vector_core::sink::VectorSink;
 
 use super::{encoder::HecLogsEncoder, request_builder::HecLogsRequestBuilder, sink::HecLogsSink};
 use crate::{
+    codecs::Encoder,
     config::{AcknowledgementsConfig, GenerateConfig, Input, SinkConfig, SinkContext},
     http::HttpClient,
     sinks::{
@@ -17,15 +19,37 @@ use crate::{
             timestamp_key, SplunkHecDefaultBatchSettings,
         },
         util::{
-            encoding::EncodingConfig, http::HttpRetryLogic, BatchConfig, Compression,
-            ServiceBuilderExt, TowerRequestConfig,
+            encoding::{EncodingConfig, EncodingConfigAdapter, EncodingConfigMigrator},
+            http::HttpRetryLogic,
+            BatchConfig, Compression, ServiceBuilderExt, TowerRequestConfig,
         },
         Healthcheck,
     },
     template::Template,
     tls::TlsConfig,
 };
 
+#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum HecEncoding {
+    Json,
+    Text,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct HecEncodingMigrator;
+
+impl EncodingConfigMigrator for HecEncodingMigrator {
+    type Codec = HecEncoding;
+
+    fn migrate(codec: &Self::Codec) -> SerializerConfig {
+        match codec {
+            HecEncoding::Text => TextSerializerConfig::new().into(),
+            HecEncoding::Json => JsonSerializerConfig::new().into(),
+        }
+    }
+}
+
 #[derive(Deserialize, Serialize, Debug, Clone)]
 #[serde(deny_unknown_fields)]
 pub struct HecLogsSinkConfig {
@@ -40,7 +64,7 @@ pub struct HecLogsSinkConfig {
     pub index: Option<Template>,
     pub sourcetype: Option<Template>,
     pub source: Option<Template>,
-    pub encoding: EncodingConfig<HecLogsEncoder>,
+    pub encoding: EncodingConfigAdapter<EncodingConfig<HecEncoding>, HecEncodingMigrator>,
     #[serde(default)]
     pub compression: Compression,
     #[serde(default)]
@@ -66,7 +90,7 @@ impl GenerateConfig for HecLogsSinkConfig {
             index: None,
             sourcetype: None,
             source: None,
-            encoding: HecLogsEncoder::Text.into(),
+            encoding: EncodingConfig::from(HecEncoding::Text).into(),
             compression: Compression::default(),
             batch: BatchConfig::default(),
             request: TowerRequestConfig::default(),
@@ -96,7 +120,7 @@ impl SinkConfig for HecLogsSinkConfig {
     }
 
     fn input(&self) -> Input {
-        Input::log()
+        Input::new(self.encoding.config().input_type())
     }
 
     fn sink_type(&self) -> &'static str {
@@ -120,8 +144,15 @@ impl HecLogsSinkConfig {
             None
         };
 
+        let transformer = self.encoding.transformer();
+        let serializer = self.encoding.clone().encoding();
+        let encoder = Encoder::<()>::new(serializer);
+        let encoder = HecLogsEncoder {
+            transformer,
+            encoder,
+        };
         let request_builder = HecLogsRequestBuilder {
-            encoding: self.encoding.clone(),
+            encoder,
             compression: self.compression,
         };
 
@@ -167,7 +198,6 @@ impl HecLogsSinkConfig {
 
 // Add a compatibility alias to avoid breaking existing configs
 #[derive(Clone, Debug, Deserialize, Serialize)]
-#[serde(deny_unknown_fields)]
 struct HecSinkCompatConfig {
     #[serde(flatten)]
     config: HecLogsSinkConfig,