lib/codecs: add pretty option to JsonSerializerConfig (#1821)

* add pretty option to JsonSerializerConfig

lib/codecs/src/encoding/config.rs

@@ -48,10 +48,9 @@ impl FramingConfig {
 
 /// Encoding configuration
 #[derive(Configurable, Clone, Debug, Deserialize, Serialize)]
-#[serde(deny_unknown_fields)]
 pub struct EncodingConfig {
  /// The encoding codec used to serialize the events before outputting.
- #[configurable(required)]
+ #[serde(flatten)]
  codec: SerializerConfig,
 
  #[serde(flatten)]
@@ -172,9 +171,38 @@ where
 
 #[cfg(test)]
 mod tests {
+ use event::log::parse_value_path;
+
  use super::*;
+ use crate::encoding::format::json::JsonSerializerConfig;
  use crate::encoding::TimestampFormat;
- use event::log::parse_value_path;
+
+ #[test]
+ fn json() {
+ for (input, want) in [
+ (
+ r#"codec: json"#,
+ SerializerConfig::Json(JsonSerializerConfig { pretty: false }),
+ ),
+ (
+ r#"
+ codec: json
+ pretty: false
+ "#,
+ SerializerConfig::Json(JsonSerializerConfig { pretty: false }),
+ ),
+ (
+ r#"
+ codec: json
+ pretty: true
+ "#,
+ SerializerConfig::Json(JsonSerializerConfig { pretty: true }),
+ ),
+ ] {
+ let got = serde_yaml::from_str::<SerializerConfig>(input).unwrap();
+ assert_eq!(got, want);
+ }
+ }
 
  #[test]
  fn deserialize() {
@@ -188,7 +216,7 @@ except_fields:
  - ignore_me
 timestamp_format: unix
 "##,
- SerializerConfig::Json,
+ SerializerConfig::Json(JsonSerializerConfig { pretty: false }),
  Transformer::new(
  Some(vec![parse_value_path("a.b[0]").unwrap()]),
  Some(vec![parse_value_path("ignore_me").unwrap()]),
@@ -241,7 +269,7 @@ encoding:
  only_fields: [ "a.b.c" ]
 "#,
  Some(FramingConfig::NewlineDelimited),
- SerializerConfig::Json,
+ SerializerConfig::Json(JsonSerializerConfig { pretty: false }),
  Transformer::new(Some(vec![parse_value_path("a.b.c").unwrap()]), None, None)
  .unwrap(),
  ),
@@ -252,7 +280,7 @@ encoding:
  only_fields: [ "a.b.c" ]
 "#,
  None,
- SerializerConfig::Json,
+ SerializerConfig::Json(JsonSerializerConfig { pretty: false }),
  Transformer::new(Some(vec![parse_value_path("a.b.c").unwrap()]), None, None)
  .unwrap(),
  ),

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

@@ -1,17 +1,32 @@
 use bytes::{BufMut, BytesMut};
+use configurable::Configurable;
 use event::Event;
+use serde::{Deserialize, Serialize};
 use tokio_util::codec::Encoder;
 
 use super::SerializeError;
+use crate::serde::skip_serializing_if_default;
+
+/// Config used to build a `JsonSerializer`
+#[derive(Clone, Configurable, Debug, Deserialize, Serialize)]
+#[cfg_attr(test, derive(PartialEq))]
+pub struct JsonSerializerConfig {
+ /// Whether to use pretty JSON formatting.
+ #[serde(default, skip_serializing_if = "skip_serializing_if_default")]
+ pub pretty: bool,
+}
 
 /// Serializer that converts an `Event` to bytes using the JSON format.
 #[derive(Clone, Debug)]
-pub struct JsonSerializer;
+pub struct JsonSerializer {
+ /// Whether to use pretty JSON formatting.
+ pub pretty: bool,
+}
 
 impl JsonSerializer {
  /// Creates a new `JsonSerializer`
- pub const fn new() -> Self {
- JsonSerializer
+ pub const fn new(pretty: bool) -> Self {
+ JsonSerializer { pretty }
  }
 }
 
@@ -21,10 +36,18 @@ impl Encoder<Event> for JsonSerializer {
  fn encode(&mut self, event: Event, dst: &mut BytesMut) -> Result<(), Self::Error> {
  let writer = dst.writer();
 
- match event {
- Event::Log(log) => serde_json::to_writer(writer, &log),
- Event::Metric(metric) => serde_json::to_writer(writer, &metric),
- Event::Trace(trace) => serde_json::to_writer(writer, &trace),
+ if self.pretty {
+ match event {
+ Event::Log(log) => serde_json::to_writer_pretty(writer, &log),
+ Event::Metric(metric) => serde_json::to_writer_pretty(writer, &metric),
+ Event::Trace(trace) => serde_json::to_writer_pretty(writer, &trace),
+ }
+ } else {
+ match event {
+ Event::Log(log) => serde_json::to_writer(writer, &log),
+ Event::Metric(metric) => serde_json::to_writer(writer, &metric),
+ Event::Trace(trace) => serde_json::to_writer(writer, &trace),
+ }
  }
  .map_err(Into::into)
  }
@@ -41,7 +64,7 @@ mod tests {
  let event = Event::from(fields!(
  "foo" => "bar"
  ));
- let mut serializer = JsonSerializer;
+ let mut serializer = JsonSerializer::new(false);
  let mut bytes = BytesMut::new();
 
  serializer.encode(event, &mut bytes).unwrap();

lib/codecs/src/encoding/mod.rs

@@ -18,16 +18,14 @@ use serde::{Deserialize, Serialize};
 
 pub use config::{EncodingConfig, EncodingConfigWithFraming, FramingConfig, SinkType};
 pub use encoder::{Encoder, EncodingError};
-pub use format::{
- json::JsonSerializer, logfmt::LogfmtSerializer, native_json::NativeJsonSerializer,
- text::TextSerializer,
-};
-pub use framing::{
- bytes::BytesEncoder,
- character::{CharacterDelimitedEncoder, CharacterDelimitedFramerConfig},
- length_delimited::LengthDelimitedEncoder,
- newline::NewlineDelimitedEncoder,
-};
+pub use format::json::{JsonSerializer, JsonSerializerConfig};
+pub use format::logfmt::LogfmtSerializer;
+pub use format::native_json::NativeJsonSerializer;
+pub use format::text::TextSerializer;
+pub use framing::bytes::BytesEncoder;
+pub use framing::character::{CharacterDelimitedEncoder, CharacterDelimitedFramerConfig};
+pub use framing::length_delimited::LengthDelimitedEncoder;
+pub use framing::newline::NewlineDelimitedEncoder;
 pub use transformer::{TimestampFormat, Transformer};
 
 use super::FramingError;
@@ -79,14 +77,18 @@ impl Display for SerializeError {
 
 /// Configuration for building a `Serializer`
 #[derive(Configurable, Clone, Debug, Deserialize, Serialize)]
-#[serde(rename_all = "snake_case")]
+#[serde(tag = "codec", rename_all = "snake_case")]
+#[cfg_attr(test, derive(PartialEq))]
 pub enum SerializerConfig {
  /// Configures the `JsonSerializer`
- Json,
+ Json(JsonSerializerConfig),
+
  /// Configures the `LogfmtSerializer`
  Logfmt,
+
  /// Configures the `NativeJsonSerializer`
  NativeJson,
+
  /// Configures the `TextSerializer`
  Text,
 }
@@ -95,7 +97,7 @@ impl SerializerConfig {
  /// Build the `Serializer` with this configuration.
  pub fn build(&self) -> Serializer {
  match self {
- SerializerConfig::Json => Serializer::Json(JsonSerializer::new()),
+ SerializerConfig::Json(config) => Serializer::Json(JsonSerializer::new(config.pretty)),
  SerializerConfig::Logfmt => Serializer::Logfmt(LogfmtSerializer::new()),
  SerializerConfig::NativeJson => Serializer::Native(NativeJsonSerializer::new()),
  SerializerConfig::Text => Serializer::Text(TextSerializer::new()),
@@ -108,10 +110,13 @@ impl SerializerConfig {
 pub enum Serializer {
  /// Uses a `JsonSerializer` for serialization.
  Json(JsonSerializer),
+
  /// Uses a `LogfmtSerializer` for serialization.
  Logfmt(LogfmtSerializer),
+
  /// Uses a `NativeJsonSerializer` for serialization.
  Native(NativeJsonSerializer),
+
  /// Uses a `TextSerializer` for serialization.
  Text(TextSerializer),
 }
@@ -178,3 +183,24 @@ impl tokio_util::codec::Encoder<()> for Framer {
  }
  }
 }
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn config_json() {
+ let config = SerializerConfig::Json(JsonSerializerConfig { pretty: false });
+ let json = serde_json::to_string_pretty(&config).unwrap();
+ let got = serde_json::from_str::<SerializerConfig>(&json).unwrap();
+ assert_eq!(config, got);
+ }
+
+ #[test]
+ fn config_yaml() {
+ let config = SerializerConfig::Json(JsonSerializerConfig { pretty: false });
+ let json = serde_yaml::to_string(&config).unwrap();
+ let got = serde_yaml::from_str::<SerializerConfig>(&json).unwrap();
+ assert_eq!(config, got);
+ }
+}

lib/framework/src/sink/util/encoding.rs

@@ -178,7 +178,7 @@ mod tests {
  Transformer::default(),
  codecs::Encoder::<Framer>::new(
  CharacterDelimitedEncoder::new(b',').into(),
- JsonSerializer.into(),
+ JsonSerializer { pretty: false }.into(),
  ),
  );
 

src/sinks/loki/sink.rs

@@ -424,7 +424,7 @@ mod tests {
  fn encoder_no_labels() {
  let mut encoder = EventEncoder {
  key_partitioner: KeyPartitioner::new(None),
- encoder: Encoder::<()>::new(JsonSerializer::new().into()),
+ encoder: Encoder::<()>::new(JsonSerializer::new(false).into()),
  transformer: Transformer::default(),
  labels: HashMap::default(),
  remove_label_fields: false,
@@ -460,7 +460,7 @@ mod tests {
 
  let mut encoder = EventEncoder {
  key_partitioner: KeyPartitioner::new(None),
- encoder: Encoder::<()>::new(JsonSerializer::new().into()),
+ encoder: Encoder::<()>::new(JsonSerializer::new(false).into()),
  transformer: Transformer::default(),
  labels,
  remove_label_fields: false,
@@ -485,7 +485,7 @@ mod tests {
  fn encoder_no_ts() {
  let mut encoder = EventEncoder {
  key_partitioner: KeyPartitioner::new(None),
- encoder: Encoder::<()>::new(JsonSerializer::new().into()),
+ encoder: Encoder::<()>::new(JsonSerializer::new(false).into()),
  transformer: Transformer::default(),
  labels: HashMap::default(),
  remove_label_fields: false,
@@ -514,7 +514,7 @@ mod tests {
 
  let mut encoder = EventEncoder {
  key_partitioner: KeyPartitioner::new(None),
- encoder: Encoder::<()>::new(JsonSerializer::new().into()),
+ encoder: Encoder::<()>::new(JsonSerializer::new(false).into()),
  transformer: Transformer::default(),
  labels,
  remove_label_fields: true,
@@ -533,7 +533,7 @@ mod tests {
  async fn filter_encoder_drop() {
  let mut encoder = EventEncoder {
  key_partitioner: KeyPartitioner::new(None),
- encoder: Encoder::<()>::new(JsonSerializer::new().into()),
+ encoder: Encoder::<()>::new(JsonSerializer::new(false).into()),
  transformer: Transformer::default(),
  labels: HashMap::default(),
  remove_label_fields: false,

tests/syslog.rs

@@ -4,7 +4,7 @@ use std::collections::HashMap;
 use std::fmt;
 use std::str::FromStr;
 
-use codecs::encoding::{FramingConfig, SerializerConfig};
+use codecs::encoding::{FramingConfig, JsonSerializerConfig, SerializerConfig};
 use codecs::EncodingConfigWithFraming;
 use framework::sink::util::tcp::TcpSinkConfig;
 use framework::testing::CountReceiver;
@@ -165,7 +165,7 @@ fn tcp_json_sink(address: String) -> Config {
  socket::Mode::Tcp(TcpSinkConfig::from_address(address)),
  EncodingConfigWithFraming::new(
  Some(FramingConfig::NewlineDelimited),
- SerializerConfig::Json,
+ SerializerConfig::Json(JsonSerializerConfig { pretty: false }),
  Default::default(),
  ),
  )