fix!(codecs): Use '\0' delimiter as default stream decode framer (#20778

)

* fix(gelf): Use null delimiter as default streaming framer

* fix(gelf): compilation issues

* test(gelf): Improve gelf decoder test

* test(decoder): Fix gelf stream decoding test

* test(gelf): Improve gelf decoder test

* test(gelf): Fix test fail

* doc: add changelog

* test: simplify gelf stream default framing test

* doc: rewrite changelog

* style: cargo fmt

* docs: update changelog with PR suggestion

Co-authored-by: Jesse Szwedko <[email protected]>

* docs: change changelog example to yaml

* Update 20768-default_gelf_stream_framing.breaking.md

---------

Co-authored-by: Jesse Szwedko <[email protected]>

changelog.d/20768-default_gelf_stream_framing.breaking.md

@@ -0,0 +1,37 @@
+Now the GELF codec with stream-based sources uses null byte (`\0`) by default as messages delimiter instead of newline (`\n`) character. This better matches GELF server behavior.
+
+### Configuration changes
+
+In order to maintain the previous behavior, you must set the `framing.method` option to the `character_delimited` method and the `framing.character_delimited.delimiter` option to `\n` when using GELF codec with stream-based sources.
+
+### Example configuration change for socket source
+
+#### Previous
+
+```yaml
+sources:
+  my_source_id:
+    type: "socket"
+    address: "0.0.0.0:9000"
+    mode: "tcp"
+    decoding:
+      codec: "gelf"
+```
+
+#### Current
+
+```yaml
+sources:
+  my_source_id:
+    type: "socket"
+    address: "0.0.0.0:9000"
+    mode: "tcp"
+    decoding:
+      codec: "gelf"
+    framing:
+      method: "character_delimited"
+    character_delimited:
+      delimiter: "\n"
+```
+
+authors: jorgehermo9

lib/codecs/src/decoding/framing/character_delimited.rs

@@ -15,6 +15,12 @@ pub struct CharacterDelimitedDecoderConfig {
 }
 
 impl CharacterDelimitedDecoderConfig {
+    /// Creates a `CharacterDelimitedDecoderConfig` with the specified delimiter and default max length.
+    pub const fn new(delimiter: u8) -> Self {
+        Self {
+            character_delimited: CharacterDelimitedDecoderOptions::new(delimiter, None),
+        }
+    }
     /// Build the `CharacterDelimitedDecoder` from this configuration.
     pub const fn build(&self) -> CharacterDelimitedDecoder {
         if let Some(max_length) = self.character_delimited.max_length {
@@ -53,7 +59,7 @@ pub struct CharacterDelimitedDecoderOptions {
 
 impl CharacterDelimitedDecoderOptions {
     /// Create a `CharacterDelimitedDecoderOptions` with a delimiter and optional max_length.
-    pub fn new(delimiter: u8, max_length: Option<usize>) -> Self {
+    pub const fn new(delimiter: u8, max_length: Option<usize>) -> Self {
         Self {
             delimiter,
             max_length,

lib/codecs/src/decoding/mod.rs

@@ -334,14 +334,16 @@ impl DeserializerConfig {
             DeserializerConfig::Native => FramingConfig::LengthDelimited(Default::default()),
             DeserializerConfig::Bytes
             | DeserializerConfig::Json(_)
-            | DeserializerConfig::Gelf(_)
             | DeserializerConfig::NativeJson(_) => {
                 FramingConfig::NewlineDelimited(Default::default())
             }
             DeserializerConfig::Protobuf(_) => FramingConfig::Bytes,
             #[cfg(feature = "syslog")]
             DeserializerConfig::Syslog(_) => FramingConfig::NewlineDelimited(Default::default()),
             DeserializerConfig::Vrl(_) => FramingConfig::Bytes,
+            DeserializerConfig::Gelf(_) => {
+                FramingConfig::CharacterDelimited(CharacterDelimitedDecoderConfig::new(0))
+            }
         }
     }
 
@@ -467,3 +469,23 @@ impl format::Deserializer for Deserializer {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn gelf_stream_default_framing_is_null_delimited() {
+        let deserializer_config = DeserializerConfig::from(GelfDeserializerConfig::default());
+        let framing_config = deserializer_config.default_stream_framing();
+        matches!(
+            framing_config,
+            FramingConfig::CharacterDelimited(CharacterDelimitedDecoderConfig {
+                character_delimited: CharacterDelimitedDecoderOptions {
+                    delimiter: 0,
+                    max_length: None,
+                }
+            })
+        );
+    }
+}

lib/codecs/src/encoding/mod.rs

@@ -362,12 +362,14 @@ impl SerializerConfig {
                 FramingConfig::LengthDelimited(LengthDelimitedEncoderConfig::default())
             }
             SerializerConfig::Csv(_)
-            | SerializerConfig::Gelf
             | SerializerConfig::Json(_)
             | SerializerConfig::Logfmt
             | SerializerConfig::NativeJson
             | SerializerConfig::RawMessage
             | SerializerConfig::Text(_) => FramingConfig::NewlineDelimited,
+            SerializerConfig::Gelf => {
+                FramingConfig::CharacterDelimited(CharacterDelimitedEncoderConfig::new(0))
+            }
         }
     }