feat(codecs): add varint length delimited framing for protobuf (#23352)

* feat(codecs): add varint length delimited framing for protobuf

This commit adds support for varint length-delimited framing for protobuf
sources and sinks in Vector. This addresses the use case where tools like
ClickHouse expect protobuf messages with varint length prefixes instead
of the standard 32-bit length prefixes.

- Add VarintLengthDelimitedEncoder for encoding varint length prefixes
- Add VarintLengthDelimited option to FramingConfig enums
- Update default protobuf framing to use varint instead of 32-bit length
- Add comprehensive tests for varint framing (7 tests, all passing)
- Update validation resources to handle new framing option

- Better compatibility with tools like ClickHouse
- Eliminates risk of protobuf messages being cut or skipped
- Properly handles zero-length messages
- Backward compatible with existing configurations

```yaml
sources:
  protobuf_source:
    type: socket
    decoding:
      codec: protobuf
      protobuf:
        desc_file: "path/to/protobuf.desc"
        message_type: "package.MessageType"
    framing:
      method: varint_length_delimited

sinks:
  protobuf_sink:
    type: socket
    encoding:
      codec: protobuf
      protobuf:
        desc_file: "path/to/protobuf.desc"
        message_type: "package.MessageType"
    framing:
      method: varint_length_delimited
```

- All varint framing tests pass (7/7)
- Vector compiles successfully
- Configuration validation works
- Default behavior updated correctly

Closes: [Issue number]

* Delete COMMIT_MESSAGE.md

* Delete PULL_REQUEST_SUMMARY.md

* Address Vector team feedback for varint framing

- Change usize to u64 for varint values and u32 to u8 for shift
- Fix test assertions to use Bytes::from() instead of string literals
- Move documentation example to config/examples/varint_framing_protobuf.yaml
- Add custom error types with Snafu for better error handling
- Implement proper FramingError trait for custom error types
- Fix validation resources to handle varint length delimited framing
- All tests passing (9 varint tests + 7 integration tests)

* Add TrailingData error variant and improve can_continue logic

* docs: add changelog fragment for protobuf varint framing default

* Address Vector team feedback: revert protobuf default to length_delimited framing

- Revert default from VarintLengthDelimitedEncoder to LengthDelimitedEncoder
- Keep varint framing as opt-in feature to avoid breaking changes
- Update documentation to reflect opt-in approach
- Fix formatting issues (trailing spaces, newlines)
- Rename changelog file to reflect feature addition rather than default change

* Fix formatting and clippy issues for protobuf varint framing

* fix example

* generate components docs

* fix example config

---------

Co-authored-by: Pavlos Rontidis <[email protected]>

Author: modev2301

changelog.d/protobuf_varint_framing_support.enhancement.md

@@ -0,0 +1,5 @@
+Added support for varint length delimited framing for protobuf, which is compatible with standard protobuf streaming implementations and tools like ClickHouse.
+
+Users can now opt-in to varint framing by explicitly specifying `framing.method: varint_length_delimited` in their configuration. The default remains length-delimited framing for backward compatibility.
+
+authors: modev2301

config/examples/varint_framing_protobuf.yaml

@@ -0,0 +1,56 @@
+# Example configuration demonstrating varint framing for protobuf
+# This is compatible with tools like ClickHouse that use protobuf with varint length prefixes
+
+sources:
+  protobuf_source:
+    type: socket
+    mode: tcp
+    address: "0.0.0.0:8080"
+    decoding:
+      codec: protobuf
+      protobuf:
+        desc_file: "path/to/your/protobuf.desc"
+        message_type: "your.package.MessageType"
+    framing:
+      method: varint_length_delimited
+
+  socket_source:
+    type: socket
+    mode: tcp
+    address: "0.0.0.0:8081"
+    decoding:
+      codec: protobuf
+      protobuf:
+        desc_file: "input.desc"
+        message_type: "input.Message"
+    framing:
+      method: varint_length_delimited
+
+sinks:
+  protobuf_sink:
+    inputs:
+      - protobuf_source
+    type: socket
+    mode: tcp
+    address: "localhost:9090"
+    encoding:
+      codec: protobuf
+      protobuf:
+        desc_file: "path/to/your/protobuf.desc"
+        message_type: "your.package.MessageType"
+    framing:
+      method: varint_length_delimited
+
+  socket_output:
+    inputs:
+      - socket_source
+    type: socket
+    mode: tcp
+    address: "localhost:9090"
+    encoding:
+      codec: protobuf
+      protobuf:
+        desc_file: "output.desc"
+        message_type: "output.Message"
+    framing:
+      method: varint_length_delimited

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

@@ -9,6 +9,7 @@ mod chunked_gelf;
 mod length_delimited;
 mod newline_delimited;
 mod octet_counting;
+mod varint_length_delimited;
 
 use std::{any::Any, fmt::Debug};
 
@@ -26,6 +27,9 @@ pub use octet_counting::{
     OctetCountingDecoder, OctetCountingDecoderConfig, OctetCountingDecoderOptions,
 };
 use tokio_util::codec::LinesCodecError;
+pub use varint_length_delimited::{
+    VarintLengthDelimitedDecoder, VarintLengthDelimitedDecoderConfig,
+};
 
 pub use self::bytes::{BytesDecoder, BytesDecoderConfig};
 use super::StreamDecodingError;

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

@@ -0,0 +1,229 @@
+use bytes::{Buf, Bytes, BytesMut};
+use derivative::Derivative;
+use snafu::Snafu;
+use tokio_util::codec::Decoder;
+use vector_config::configurable_component;
+
+use super::{BoxedFramingError, FramingError, StreamDecodingError};
+
+/// Errors that can occur during varint length delimited framing.
+#[derive(Debug, Snafu)]
+pub enum VarintFramingError {
+    #[snafu(display("Varint too large"))]
+    VarintOverflow,
+
+    #[snafu(display("Frame too large: {length} bytes (max: {max})"))]
+    FrameTooLarge { length: usize, max: usize },
+
+    #[snafu(display("Trailing data at EOF"))]
+    TrailingData,
+}
+
+impl StreamDecodingError for VarintFramingError {
+    fn can_continue(&self) -> bool {
+        match self {
+            // Varint overflow and frame too large are not recoverable
+            Self::VarintOverflow | Self::FrameTooLarge { .. } => false,
+            // Trailing data at EOF is not recoverable
+            Self::TrailingData => false,
+        }
+    }
+}
+
+impl FramingError for VarintFramingError {
+    fn as_any(&self) -> &dyn std::any::Any {
+        self as &dyn std::any::Any
+    }
+}
+
+/// Config used to build a `VarintLengthDelimitedDecoder`.
+#[configurable_component]
+#[derive(Debug, Clone, Derivative)]
+#[derivative(Default)]
+pub struct VarintLengthDelimitedDecoderConfig {
+    /// Maximum frame length
+    #[serde(default = "default_max_frame_length")]
+    pub max_frame_length: usize,
+}
+
+const fn default_max_frame_length() -> usize {
+    8 * 1_024 * 1_024
+}
+
+impl VarintLengthDelimitedDecoderConfig {
+    /// Build the `VarintLengthDelimitedDecoder` from this configuration.
+    pub fn build(&self) -> VarintLengthDelimitedDecoder {
+        VarintLengthDelimitedDecoder::new(self.max_frame_length)
+    }
+}
+
+/// A codec for handling bytes sequences whose length is encoded as a varint prefix.
+/// This is compatible with protobuf's length-delimited encoding.
+#[derive(Debug, Clone)]
+pub struct VarintLengthDelimitedDecoder {
+    max_frame_length: usize,
+}
+
+impl VarintLengthDelimitedDecoder {
+    /// Creates a new `VarintLengthDelimitedDecoder`.
+    pub fn new(max_frame_length: usize) -> Self {
+        Self { max_frame_length }
+    }
+
+    /// Decode a varint from the buffer
+    fn decode_varint(&self, buf: &mut BytesMut) -> Result<Option<u64>, BoxedFramingError> {
+        if buf.is_empty() {
+            return Ok(None);
+        }
+
+        let mut value: u64 = 0;
+        let mut shift: u8 = 0;
+        let mut bytes_read = 0;
+
+        for byte in buf.iter() {
+            bytes_read += 1;
+            let byte_value = (*byte & 0x7F) as u64;
+            value |= byte_value << shift;
+
+            if *byte & 0x80 == 0 {
+                // Last byte of varint
+                buf.advance(bytes_read);
+                return Ok(Some(value));
+            }
+
+            shift += 7;
+            if shift >= 64 {
+                return Err(VarintFramingError::VarintOverflow.into());
+            }
+        }
+
+        // Incomplete varint
+        Ok(None)
+    }
+}
+
+impl Default for VarintLengthDelimitedDecoder {
+    fn default() -> Self {
+        Self::new(default_max_frame_length())
+    }
+}
+
+impl Decoder for VarintLengthDelimitedDecoder {
+    type Item = Bytes;
+    type Error = BoxedFramingError;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        // First, try to decode the varint length
+        let length = match self.decode_varint(src)? {
+            Some(len) => len as usize,
+            None => return Ok(None), // Incomplete varint
+        };
+
+        // Check if the length is reasonable
+        if length > self.max_frame_length {
+            return Err(VarintFramingError::FrameTooLarge {
+                length,
+                max: self.max_frame_length,
+            }
+            .into());
+        }
+
+        // Check if we have enough data for the complete frame
+        if src.len() < length {
+            return Ok(None); // Incomplete frame
+        }
+
+        // Extract the frame
+        let frame = src.split_to(length).freeze();
+        Ok(Some(frame))
+    }
+
+    fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        if src.is_empty() {
+            Ok(None)
+        } else {
+            // Try to decode what we have, even if incomplete
+            match self.decode(src)? {
+                Some(frame) => Ok(Some(frame)),
+                None => {
+                    // If we have data but couldn't decode it, it's trailing data
+                    if !src.is_empty() {
+                        Err(VarintFramingError::TrailingData.into())
+                    } else {
+                        Ok(None)
+                    }
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn decode_single_byte_varint() {
+        let mut input = BytesMut::from(&[0x03, b'f', b'o', b'o'][..]);
+        let mut decoder = VarintLengthDelimitedDecoder::default();
+
+        assert_eq!(
+            decoder.decode(&mut input).unwrap().unwrap(),
+            Bytes::from("foo")
+        );
+        assert_eq!(decoder.decode(&mut input).unwrap(), None);
+    }
+
+    #[test]
+    fn decode_multi_byte_varint() {
+        // 300 in varint encoding: 0xAC 0x02
+        let mut input = BytesMut::from(&[0xAC, 0x02][..]);
+        // Add 300 bytes of data
+        input.extend_from_slice(&vec![b'x'; 300]);
+        let mut decoder = VarintLengthDelimitedDecoder::default();
+
+        let result = decoder.decode(&mut input).unwrap().unwrap();
+        assert_eq!(result.len(), 300);
+        assert_eq!(decoder.decode(&mut input).unwrap(), None);
+    }
+
+    #[test]
+    fn decode_incomplete_varint() {
+        let mut input = BytesMut::from(&[0x80][..]); // Incomplete varint
+        let mut decoder = VarintLengthDelimitedDecoder::default();
+
+        assert_eq!(decoder.decode(&mut input).unwrap(), None);
+    }
+
+    #[test]
+    fn decode_incomplete_frame() {
+        let mut input = BytesMut::from(&[0x05, b'f', b'o'][..]); // Length 5, but only 2 bytes
+        let mut decoder = VarintLengthDelimitedDecoder::default();
+
+        assert_eq!(decoder.decode(&mut input).unwrap(), None);
+    }
+
+    #[test]
+    fn decode_frame_too_large() {
+        let mut input =
+            BytesMut::from(&[0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x01][..]);
+        let mut decoder = VarintLengthDelimitedDecoder::new(1000);
+
+        assert!(decoder.decode(&mut input).is_err());
+    }
+
+    #[test]
+    fn decode_trailing_data_at_eof() {
+        let mut input = BytesMut::from(&[0x03, b'f', b'o', b'o', b'e', b'x', b't', b'r', b'a'][..]);
+        let mut decoder = VarintLengthDelimitedDecoder::default();
+
+        // First decode should succeed
+        assert_eq!(
+            decoder.decode(&mut input).unwrap().unwrap(),
+            Bytes::from("foo")
+        );
+
+        // Second decode should fail with trailing data
+        assert!(decoder.decode_eof(&mut input).is_err());
+    }
+}

lib/codecs/src/decoding/mod.rs

@@ -24,7 +24,7 @@ pub use framing::{
     ChunkedGelfDecoderConfig, ChunkedGelfDecoderOptions, FramingError, LengthDelimitedDecoder,
     LengthDelimitedDecoderConfig, NewlineDelimitedDecoder, NewlineDelimitedDecoderConfig,
     NewlineDelimitedDecoderOptions, OctetCountingDecoder, OctetCountingDecoderConfig,
-    OctetCountingDecoderOptions,
+    OctetCountingDecoderOptions, VarintLengthDelimitedDecoder, VarintLengthDelimitedDecoderConfig,
 };
 use smallvec::SmallVec;
 use std::fmt::Debug;
@@ -105,6 +105,10 @@ pub enum FramingConfig {
     ///
     /// [chunked_gelf]: https://go2docs.graylog.org/current/getting_in_log_data/gelf.html
     ChunkedGelf(ChunkedGelfDecoderConfig),
+
+    /// Byte frames which are prefixed by a varint indicating the length.
+    /// This is compatible with protobuf's length-delimited encoding.
+    VarintLengthDelimited(VarintLengthDelimitedDecoderConfig),
 }
 
 impl From<BytesDecoderConfig> for FramingConfig {
@@ -143,6 +147,12 @@ impl From<ChunkedGelfDecoderConfig> for FramingConfig {
     }
 }
 
+impl From<VarintLengthDelimitedDecoderConfig> for FramingConfig {
+    fn from(config: VarintLengthDelimitedDecoderConfig) -> Self {
+        Self::VarintLengthDelimited(config)
+    }
+}
+
 impl FramingConfig {
     /// Build the `Framer` from this configuration.
     pub fn build(&self) -> Framer {
@@ -153,6 +163,9 @@ impl FramingConfig {
             FramingConfig::NewlineDelimited(config) => Framer::NewlineDelimited(config.build()),
             FramingConfig::OctetCounting(config) => Framer::OctetCounting(config.build()),
             FramingConfig::ChunkedGelf(config) => Framer::ChunkedGelf(config.build()),
+            FramingConfig::VarintLengthDelimited(config) => {
+                Framer::VarintLengthDelimited(config.build())
+            }
         }
     }
 }
@@ -174,6 +187,8 @@ pub enum Framer {
     Boxed(BoxedFramer),
     /// Uses a `ChunkedGelfDecoder` for framing.
     ChunkedGelf(ChunkedGelfDecoder),
+    /// Uses a `VarintLengthDelimitedDecoder` for framing.
+    VarintLengthDelimited(VarintLengthDelimitedDecoder),
 }
 
 impl tokio_util::codec::Decoder for Framer {
@@ -189,6 +204,7 @@ impl tokio_util::codec::Decoder for Framer {
             Framer::OctetCounting(framer) => framer.decode(src),
             Framer::Boxed(framer) => framer.decode(src),
             Framer::ChunkedGelf(framer) => framer.decode(src),
+            Framer::VarintLengthDelimited(framer) => framer.decode(src),
         }
     }
 
@@ -201,6 +217,7 @@ impl tokio_util::codec::Decoder for Framer {
             Framer::OctetCounting(framer) => framer.decode_eof(src),
             Framer::Boxed(framer) => framer.decode_eof(src),
             Framer::ChunkedGelf(framer) => framer.decode_eof(src),
+            Framer::VarintLengthDelimited(framer) => framer.decode_eof(src),
         }
     }
 }