Repository: vectordotdev/vector (language: rust)

Issue:
GELF codec should support compression
### A note for the community

<!-- Please keep this note for the community -->
* Please vote on this issue by adding a 👍 [reaction](https://blog.github.com/2016-03-10-add-reactions-to-pull-requests-issues-and-comments/) to the original issue to help the community and maintainers prioritize this request
* If you are interested in working on this issue or have submitted a pull request, please leave a comment
<!-- Thank you for keeping this note for the community -->


### Use Cases

Currently, the GELF codec fails to decode compressed messages received via UDP. 

[GELF](https://go2docs.graylog.org/current/getting_in_log_data/gelf.html) supports compression:

> When using UDP as transport layer, GELF messages can be sent uncompressed or compressed with either GZIP or ZLIB.
> Graylog nodes automatically detect the compression type in the GELF magic byte header.
> Decide if you want to trade a bit more CPU load for saving network bandwidth. GZIP is the protocol default.



### Attempted Solutions

If I understand that correctly, decompression is just not  yet implemented.


### Proposal

The[ Graylog GELF decoder](https://github.com/Graylog2/graylog2-server/blob/6dab8975f5d80c90e6346f1fcab30af11bcadf19/graylog2-server/src/main/java/org/graylog2/inputs/codecs/gelf/GELFMessage.java#L108) decides whether it is gzip, zlib, or uncompressed by looking a the magic numbers in the first bytes, then decompress accordingly. It also detects Chunking (#13292) this way.

### References

#13292
#4868 

### Version

_No response_


Discussion / hints:
Another example in [Go from grafana](https://github.com/grafana/go-gelf/blob/1cbd6368efe24cfa12574d988b4f0653ed5123e2/gelf/reader.go#L90).

I should learn Rust  :thinking: 
I'm planning to implement this once my GELF chunking PR #20859 is merged.

Actually, I have decompression  kind of working, but wanted to wait until it's merged so the PR is not bloated with a lot of changes.

I'd appreciate if you could wait until #20859 is merged and I submit the decompression PR

@flyingflo
Thank you for your work @jorgehermo9! Take your time complete it.
Hi @flyingflo. My PR https://github.com/vectordotdev/vector/pull/20859 just got merged. I'll start working into the decompression part asap.



Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index b6a4bccabc306..baa3385573b5e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2225,6 +2225,7 @@ dependencies = [
  "csv-core",
  "derivative",
  "dyn-clone",
+ "flate2",
  "futures 0.3.31",
  "indoc",
  "influxdb-line-protocol",
@@ -3549,6 +3550,7 @@ dependencies = [
  "tempfile",
  "tokio",
  "tracing 0.1.40",
+ "vector-common",
  "vector-config",
  "vector-config-common",
  "vector-config-macros",
@@ -10968,6 +10970,7 @@ dependencies = [
  "typetag",
  "url",
  "uuid",
+ "vector-common",
  "vector-config",
  "vector-config-common",
  "vector-config-macros",
diff --git a/Cargo.toml b/Cargo.toml
index 299b5d2151cc0..3620d930541a8 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -137,6 +137,7 @@ cfg-if = { version = "1.0.0", default-features = false }
 chrono = { version = "0.4.38", default-features = false, features = ["clock", "serde"] }
 chrono-tz = { version = "0.10.0", default-features = false, features = ["serde"] }
 clap = { version = "4.5.21", default-features = false, features = ["derive", "error-context", "env", "help", "std", "string", "usage", "wrap_help"] }
+flate2 = { version = "1.0.35", default-features = false, features = ["default"] }
 futures = { version = "0.3.31", default-features = false, features = ["compat", "io-compat", "std"], package = "futures" }
 glob = { version = "0.3.1", default-features = false }
 indexmap = { version = "2.6.0", default-features = false, features = ["serde", "std"] }
@@ -179,6 +180,7 @@ dnsmsg-parser = { path = "lib/dnsmsg-parser", optional = true }
 fakedata = { path = "lib/fakedata", optional = true }
 portpicker = { path = "lib/portpicker" }
 tracing-limit = { path = "lib/tracing-limit" }
+vector-common = { path = "lib/vector-common", default-features = false}
 vector-lib.workspace = true
 vector-config.workspace = true
 vector-config-common.workspace = true
@@ -306,7 +308,7 @@ dyn-clone = { version = "1.0.17", default-features = false }
 encoding_rs = { version = "0.8.35", default-features = false, features = ["serde"] }
 enum_dispatch = { version = "0.3.13", default-features = false }
 exitcode = { version = "1.1.2", default-features = false }
-flate2 = { version = "1.0.35", default-features = false, features = ["default"] }
+flate2.workspace = true
 futures-util = { version = "0.3.29", default-features = false }
 glob.workspace = true
 governor = { version = "0.7.0", default-features = false, features = ["dashmap", "jitter", "std"], optional = true }
diff --git a/changelog.d/21153-chunked_gelf_decompression.feature.md b/changelog.d/21153-chunked_gelf_decompression.feature.md
new file mode 100644
index 0000000000000..388af4b86711a
--- /dev/null
+++ b/changelog.d/21153-chunked_gelf_decompression.feature.md
@@ -0,0 +1,3 @@
+Chunked gelf decoder now allows for payload decompression.
+
+authors: jorgehermo9
diff --git a/lib/codecs/Cargo.toml b/lib/codecs/Cargo.toml
index ece90dc2aafcb..f4ab9561f75a0 100644
--- a/lib/codecs/Cargo.toml
+++ b/lib/codecs/Cargo.toml
@@ -16,6 +16,7 @@ chrono.workspace = true
 csv-core = { version = "0.1.10", default-features = false }
 derivative = { version = "2", default-features = false }
 dyn-clone = { version = "1", default-features = false }
+flate2.workspace = true
 influxdb-line-protocol = { version = "2", default-features = false }
 lookup = { package = "vector-lookup", path = "../vector-lookup", default-features = false, features = ["test"] }
 memchr = { version = "2", default-features = false }
diff --git a/lib/codecs/src/decoding/framing/chunked_gelf.rs b/lib/codecs/src/decoding/framing/chunked_gelf.rs
index e29d304e972f4..bbddc457a3d3d 100644
--- a/lib/codecs/src/decoding/framing/chunked_gelf.rs
+++ b/lib/codecs/src/decoding/framing/chunked_gelf.rs
@@ -1,17 +1,19 @@
-use crate::{BytesDecoder, StreamDecodingError};
-
 use super::{BoxedFramingError, FramingError};
+use crate::{BytesDecoder, StreamDecodingError};
 use bytes::{Buf, Bytes, BytesMut};
 use derivative::Derivative;
-use snafu::{ensure, Snafu};
+use flate2::read::{MultiGzDecoder, ZlibDecoder};
+use snafu::{ensure, ResultExt, Snafu};
 use std::any::Any;
 use std::collections::HashMap;
+use std::io::Read;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tokio;
 use tokio::task::JoinHandle;
 use tokio_util::codec::Decoder;
-use tracing::{debug, warn};
+use tracing::{debug, trace, warn};
+use vector_common::constants::{GZIP_MAGIC, ZLIB_MAGIC};
 use vector_config::configurable_component;
 
 const GELF_MAGIC: &[u8] = &[0x1e, 0x0f];
@@ -38,6 +40,7 @@ impl ChunkedGelfDecoderConfig {
             self.chunked_gelf.timeout_secs,
             self.chunked_gelf.pending_messages_limit,
             self.chunked_gelf.max_length,
+            self.chunked_gelf.decompression,
         )
     }
 }
@@ -71,6 +74,37 @@ pub struct ChunkedGelfDecoderOptions {
     /// The message's payload is the concatenation of all the chunks' payloads.
     #[serde(default, skip_serializing_if = "vector_core::serde::is_default")]
     pub max_length: Option<usize>,
+
+    /// Decompression configuration for GELF messages.
+    #[serde(default, skip_serializing_if = "vector_core::serde::is_default")]
+    pub decompression: ChunkedGelfDecompressionConfig,
+}
+
+/// Decompression options for ChunkedGelfDecoder.
+#[configurable_component]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Derivative)]
+#[derivative(Default)]
+pub enum ChunkedGelfDecompressionConfig {
+    /// Automatically detect the decompression method based on the magic bytes of the message.
+    #[derivative(Default)]
+    Auto,
+    /// Use Gzip decompression.
+    Gzip,
+    /// Use Zlib decompression.
+    Zlib,
+    /// Do not decompress the message.
+    None,
+}
+
+impl ChunkedGelfDecompressionConfig {
+    pub fn get_decompression(&self, data: &Bytes) -> ChunkedGelfDecompression {
+        match self {
+            Self::Auto => ChunkedGelfDecompression::from_magic(data),
+            Self::Gzip => ChunkedGelfDecompression::Gzip,
+            Self::Zlib => ChunkedGelfDecompression::Zlib,
+            Self::None => ChunkedGelfDecompression::None,
+        }
+    }
 }
 
 #[derive(Debug)]
@@ -128,7 +162,72 @@ impl MessageState {
     }
 }
 
-#[derive(Debug, Snafu, PartialEq, Eq)]
+#[derive(Debug, PartialEq, Eq)]
+pub enum ChunkedGelfDecompression {
+    Gzip,
+    Zlib,
+    None,
+}
+
+impl ChunkedGelfDecompression {
+    pub fn from_magic(data: &Bytes) -> Self {
+        if data.starts_with(GZIP_MAGIC) {
+            trace!("Detected Gzip compression");
+            return Self::Gzip;
+        }
+
+        if data.starts_with(ZLIB_MAGIC) {
+            // Based on https://datatracker.ietf.org/doc/html/rfc1950#section-2.2
+            if let Some([first_byte, second_byte]) = data.get(0..2) {
+                if (*first_byte as u16 * 256 + *second_byte as u16) % 31 == 0 {
+                    trace!("Detected Zlib compression");
+                    return Self::Zlib;
+                }
+            };
+
+            warn!(
+                "Detected Zlib magic bytes but the header is invalid: {:?}",
+                data.get(0..2)
+            );
+        };
+
+        trace!("No compression detected",);
+        Self::None
+    }
+
+    pub fn decompress(&self, data: Bytes) -> Result<Bytes, ChunkedGelfDecompressionError> {
+        let decompressed = match self {
+            Self::Gzip => {
+                let mut decoder = MultiGzDecoder::new(data.reader());
+                let mut decompressed = Vec::new();
+                decoder
+                    .read_to_end(&mut decompressed)
+                    .context(GzipDecompressionSnafu)?;
+                Bytes::from(decompressed)
+            }
+            Self::Zlib => {
+                let mut decoder = ZlibDecoder::new(data.reader());
+                let mut decompressed = Vec::new();
+                decoder
+                    .read_to_end(&mut decompressed)
+                    .context(ZlibDecompressionSnafu)?;
+                Bytes::from(decompressed)
+            }
+            Self::None => data,
+        };
+        Ok(decompressed)
+    }
+}
+
+#[derive(Debug, Snafu)]
+pub enum ChunkedGelfDecompressionError {
+    #[snafu(display("Gzip decompression error: {source}"))]
+    GzipDecompression { source: std::io::Error },
+    #[snafu(display("Zlib decompression error: {source}"))]
+    ZlibDecompression { source: std::io::Error },
+}
+
+#[derive(Debug, Snafu)]
 pub enum ChunkedGelfDecoderError {
     #[snafu(display("Invalid chunk header with less than 10 bytes: 0x{header:0x}"))]
     InvalidChunkHeader { header: Bytes },
@@ -164,6 +263,10 @@ pub enum ChunkedGelfDecoderError {
         length: usize,
         max_length: usize,
     },
+    #[snafu(display("Error while decompressing message. {source}"))]
+    Decompression {
+        source: ChunkedGelfDecompressionError,
+    },
 }
 
 impl StreamDecodingError for ChunkedGelfDecoderError {
@@ -188,6 +291,7 @@ pub struct ChunkedGelfDecoder {
     // This limitation is due to the fact that the GELF format does not specify the length of the
     // message, so we have to read all the bytes from the message (datagram)
     bytes_decoder: BytesDecoder,
+    decompression_config: ChunkedGelfDecompressionConfig,
     state: Arc<Mutex<HashMap<u64, MessageState>>>,
     timeout: Duration,
     pending_messages_limit: Option<usize>,
@@ -200,9 +304,11 @@ impl ChunkedGelfDecoder {
         timeout_secs: f64,
         pending_messages_limit: Option<usize>,
         max_length: Option<usize>,
+        decompression_config: ChunkedGelfDecompressionConfig,
     ) -> Self {
         Self {
             bytes_decoder: BytesDecoder::new(),
+            decompression_config,
             state: Arc::new(Mutex::new(HashMap::new())),
             timeout: Duration::from_secs_f64(timeout_secs),
             pending_messages_limit,
@@ -339,18 +445,38 @@ impl ChunkedGelfDecoder {
         &mut self,
         mut src: Bytes,
     ) -> Result<Option<Bytes>, ChunkedGelfDecoderError> {
-        if src.starts_with(GELF_MAGIC) {
+        let message = if src.starts_with(GELF_MAGIC) {
+            trace!("Received a chunked GELF message based on the magic bytes");
             src.advance(2);
-            self.decode_chunk(src)
+            self.decode_chunk(src)?
         } else {
-            Ok(Some(src))
-        }
+            trace!(
+                "Received an unchunked GELF message. First two bytes of message: {:?}",
+                &src[0..2]
+            );
+            Some(src)
+        };
+
+        // We can have both chunked and unchunked messages that are compressed
+        message
+            .map(|message| {
+                self.decompression_config
+                    .get_decompression(&message)
+                    .decompress(message)
+                    .context(DecompressionSnafu)
+            })
+            .transpose()
     }
 }
 
 impl Default for ChunkedGelfDecoder {
     fn default() -> Self {
-        Self::new(DEFAULT_TIMEOUT_SECS, None, None)
+        Self::new(
+            DEFAULT_TIMEOUT_SECS,
+            None,
+            None,
+            ChunkedGelfDecompressionConfig::Auto,
+        )
     }
 }
 
@@ -388,22 +514,59 @@ mod tests {
 
     use super::*;
     use bytes::{BufMut, BytesMut};
+    use flate2::{write::GzEncoder, write::ZlibEncoder};
     use rand::{rngs::SmallRng, seq::SliceRandom, SeedableRng};
     use rstest::{fixture, rstest};
+    use std::fmt::Write as FmtWrite;
+    use std::io::Write as IoWrite;
     use tracing_test::traced_test;
 
+    pub enum Compression {
+        Gzip,
+        Zlib,
+    }
+
+    impl Compression {
+        pub fn compress(&self, payload: &impl AsRef<[u8]>) -> Bytes {
+            self.compress_with_level(payload, flate2::Compression::default())
+        }
+
+        pub fn compress_with_level(
+            &self,
+            payload: &impl AsRef<[u8]>,
+            level: flate2::Compression,
+        ) -> Bytes {
+            match self {
+                Compression::Gzip => {
+                    let mut encoder = GzEncoder::new(Vec::new(), level);
+                    encoder
+                        .write_all(payload.as_ref())
+                        .expect("failed to write to encoder");
+                    encoder.finish().expect("failed to finish encoder").into()
+                }
+                Compression::Zlib => {
+                    let mut encoder = ZlibEncoder::new(Vec::new(), level);
+                    encoder
+                        .write_all(payload.as_ref())
+                        .expect("failed to write to encoder");
+                    encoder.finish().expect("failed to finish encoder").into()
+                }
+            }
+        }
+    }
+
     fn create_chunk(
         message_id: u64,
         sequence_number: u8,
         total_chunks: u8,
-        payload: &str,
+        payload: &impl AsRef<[u8]>,
     ) -> BytesMut {
         let mut chunk = BytesMut::new();
         chunk.put_slice(GELF_MAGIC);
         chunk.put_u64(message_id);
         chunk.put_u8(sequence_number);
         chunk.put_u8(total_chunks);
-        chunk.extend_from_slice(payload.as_bytes());
+        chunk.extend_from_slice(payload.as_ref());
         chunk
     }
 
@@ -424,7 +587,7 @@ mod tests {
             message_id,
             first_sequence_number,
             total_chunks,
-            first_payload,
+            &first_payload,
         );
 
         let second_sequence_number = 1u8;
@@ -433,7 +596,7 @@ mod tests {
             message_id,
             second_sequence_number,
             total_chunks,
-            second_payload,
+            &second_payload,
         );
 
         (
@@ -453,7 +616,7 @@ mod tests {
             message_id,
             first_sequence_number,
             total_chunks,
-            first_payload,
+            &first_payload,
         );
 
         let second_sequence_number = 1u8;
@@ -462,7 +625,7 @@ mod tests {
             message_id,
             second_sequence_number,
             total_chunks,
-            second_payload,
+            &second_payload,
         );
 
         let third_sequence_number = 2u8;
@@ -471,7 +634,7 @@ mod tests {
             message_id,
             third_sequence_number,
             total_chunks,
-            third_payload,
+            &third_payload,
         );
 
         (
@@ -582,7 +745,7 @@ mod tests {
                 first_message_id,
                 sequence_number,
                 total_chunks,
-                first_payload,
+                &first_payload,
             )
         });
         let second_message_chunks = (0..total_chunks).map(|sequence_number| {
@@ -590,7 +753,7 @@ mod tests {
                 second_message_id,
                 sequence_number,
                 total_chunks,
-                second_payload,
+                &second_payload,
             )
         });
         let expected_first_message = first_payload.repeat(total_chunks as usize);
@@ -672,12 +835,10 @@ mod tests {
 
         let error = frame.unwrap_err();
         let downcasted_error = downcast_framing_error(&error);
-        assert_eq!(
-            *downcasted_error,
-            ChunkedGelfDecoderError::InvalidChunkHeader {
-                header: Bytes::from_static(&[0x12, 0x34])
-            }
-        );
+        assert!(matches!(
+            downcasted_error,
+            ChunkedGelfDecoderError::InvalidChunkHeader { .. }
+        ));
     }
 
     #[tokio::test]
@@ -686,20 +847,20 @@ mod tests {
         let sequence_number = 1u8;
         let invalid_total_chunks = GELF_MAX_TOTAL_CHUNKS + 1;
         let payload = "foo";
-        let mut chunk = create_chunk(message_id, sequence_number, invalid_total_chunks, payload);
+        let mut chunk = create_chunk(message_id, sequence_number, invalid_total_chunks, &payload);
         let mut decoder = ChunkedGelfDecoder::default();
 
         let frame = decoder.decode_eof(&mut chunk);
         let error = frame.unwrap_err();
         let downcasted_error = downcast_framing_error(&error);
-        assert_eq!(
-            *downcasted_error,
+        assert!(matches!(
+            downcasted_error,
             ChunkedGelfDecoderError::InvalidTotalChunks {
                 message_id: 1,
                 sequence_number: 1,
                 total_chunks: 129,
             }
-        );
+        ));
     }
 
     #[tokio::test]
@@ -708,20 +869,20 @@ mod tests {
         let total_chunks = 2u8;
         let invalid_sequence_number = total_chunks + 1;
         let payload = "foo";
-        let mut chunk = create_chunk(message_id, invalid_sequence_number, total_chunks, payload);
+        let mut chunk = create_chunk(message_id, invalid_sequence_number, total_chunks, &payload);
         let mut decoder = ChunkedGelfDecoder::default();
 
         let frame = decoder.decode_eof(&mut chunk);
         let error = frame.unwrap_err();
         let downcasted_error = downcast_framing_error(&error);
-        assert_eq!(
-            *downcasted_error,
+        assert!(matches!(
+            downcasted_error,
             ChunkedGelfDecoderError::InvalidSequenceNumber {
                 message_id: 1,
                 sequence_number: 3,
                 total_chunks: 2,
-            },
-        );
+            }
+        ));
     }
 
     #[rstest]
@@ -732,7 +893,10 @@ mod tests {
     ) {
         let (mut two_chunks, _) = two_chunks_message;
         let (mut three_chunks, _) = three_chunks_message;
-        let mut decoder = ChunkedGelfDecoder::new(DEFAULT_TIMEOUT_SECS, Some(1), None);
+        let mut decoder = ChunkedGelfDecoder {
+            pending_messages_limit: Some(1),
+            ..Default::default()
+        };
 
         let frame = decoder.decode_eof(&mut two_chunks[0]).unwrap();
         assert!(frame.is_none());
@@ -741,14 +905,14 @@ mod tests {
         let frame = decoder.decode_eof(&mut three_chunks[0]);
         let error = frame.unwrap_err();
         let downcasted_error = downcast_framing_error(&error);
-        assert_eq!(
-            *downcasted_error,
+        assert!(matches!(
+            downcasted_error,
             ChunkedGelfDecoderError::PendingMessagesLimitReached {
                 message_id: 2u64,
                 sequence_number: 0u8,
                 pending_messages_limit: 1,
             }
-        );
+        ));
         assert!(decoder.state.lock().unwrap().len() == 1);
     }
 
@@ -759,9 +923,9 @@ mod tests {
         let sequence_number = 0u8;
         let total_chunks = 2u8;
         let payload = "foo";
-        let mut first_chunk = create_chunk(message_id, sequence_number, total_chunks, payload);
+        let mut first_chunk = create_chunk(message_id, sequence_number, total_chunks, &payload);
         let mut second_chunk =
-            create_chunk(message_id, sequence_number + 1, total_chunks + 1, payload);
+            create_chunk(message_id, sequence_number + 1, total_chunks + 1, &payload);
         let mut decoder = ChunkedGelfDecoder::default();
 
         let frame = decoder.decode_eof(&mut first_chunk).unwrap();
@@ -770,37 +934,40 @@ mod tests {
         let frame = decoder.decode_eof(&mut second_chunk);
         let error = frame.unwrap_err();
         let downcasted_error = downcast_framing_error(&error);
-        assert_eq!(
-            *downcasted_error,
+        assert!(matches!(
+            downcasted_error,
             ChunkedGelfDecoderError::TotalChunksMismatch {
                 message_id: 1,
                 sequence_number: 1,
                 original_total_chunks: 2,
                 received_total_chunks: 3,
             }
-        );
+        ));
     }
 
     #[rstest]
     #[tokio::test]
     async fn decode_message_greater_than_max_length(two_chunks_message: ([BytesMut; 2], String)) {
         let (mut chunks, _) = two_chunks_message;
-        let mut decoder = ChunkedGelfDecoder::new(DEFAULT_TIMEOUT_SECS, None, Some(5));
+        let mut decoder = ChunkedGelfDecoder {
+            max_length: Some(5),
+            ..Default::default()
+        };
 
         let frame = decoder.decode_eof(&mut chunks[0]).unwrap();
         assert!(frame.is_none());
         let frame = decoder.decode_eof(&mut chunks[1]);
         let error = frame.unwrap_err();
         let downcasted_error = downcast_framing_error(&error);
-        assert_eq!(
-            *downcasted_error,
+        assert!(matches!(
+            downcasted_error,
             ChunkedGelfDecoderError::MaxLengthExceed {
                 message_id: 1,
                 sequence_number: 1,
                 length: 6,
                 max_length: 5,
             }
-        );
+        ));
         assert_eq!(decoder.state.lock().unwrap().len(), 0);
     }
 
@@ -818,4 +985,284 @@ mod tests {
         assert!(frame.is_none());
         assert!(logs_contain("Received a duplicate chunk. Ignoring it."));
     }
+
+    #[tokio::test]
+    #[rstest]
+    #[case::gzip(Compression::Gzip)]
+    #[case::zlib(Compression::Zlib)]
+    async fn decode_compressed_unchunked_message(#[case] compression: Compression) {
+        let payload = (0..100).fold(String::new(), |mut payload, n| {
+            write!(payload, "foo{n}").unwrap();
+            payload
+        });
+        let compressed_payload = compression.compress(&payload);
+        let mut decoder = ChunkedGelfDecoder::default();
+
+        let frame = decoder
+            .decode_eof(&mut compressed_payload.into())
+            .expect("decoding should not fail")
+            .expect("decoding should return a frame");
+
+        assert_eq!(frame, payload);
+    }
+
+    #[tokio::test]
+    #[rstest]
+    #[case::gzip(Compression::Gzip)]
+    #[case::zlib(Compression::Zlib)]
+    async fn decode_compressed_chunked_message(#[case] compression: Compression) {
+        let message_id = 1u64;
+        let max_chunk_size = 5;
+        let payload = (0..100).fold(String::new(), |mut payload, n| {
+            write!(payload, "foo{n}").unwrap();
+            payload
+        });
+        let compressed_payload = compression.compress(&payload);
+        let total_chunks = compressed_payload.len().div_ceil(max_chunk_size) as u8;
+        assert!(total_chunks < GELF_MAX_TOTAL_CHUNKS);
+        let mut chunks = compressed_payload
+            .chunks(max_chunk_size)
+            .enumerate()
+            .map(|(i, chunk)| create_chunk(message_id, i as u8, total_chunks, &chunk))
+            .collect::<Vec<_>>();
+        let (last_chunk, first_chunks) =
+            chunks.split_last_mut().expect("chunks should not be empty");
+        let mut decoder = ChunkedGelfDecoder::default();
+
+        for chunk in first_chunks {
+            let frame = decoder.decode_eof(chunk).expect("decoding should not fail");
+            assert!(frame.is_none());
+        }
+        let frame = decoder
+            .decode_eof(last_chunk)
+            .expect("decoding should not fail")
+            .expect("decoding should return a frame");
+
+        assert_eq!(frame, payload);
+    }
+
+    #[tokio::test]
+    async fn decode_malformed_gzip_message() {
+        let mut compressed_payload = BytesMut::new();
+        compressed_payload.extend(GZIP_MAGIC);
+        compressed_payload.extend(&[0x12, 0x34, 0x56, 0x78]);
+        let mut decoder = ChunkedGelfDecoder::default();
+
+        let error = decoder
+            .decode_eof(&mut compressed_payload)
+            .expect_err("decoding should fail");
+
+        let downcasted_error = downcast_framing_error(&error);
+        assert!(matches!(
+            downcasted_error,
+            ChunkedGelfDecoderError::Decompression {
+                source: ChunkedGelfDecompressionError::GzipDecompression { .. }
+            }
+        ));
+    }
+
+    #[tokio::test]
+    async fn decode_malformed_zlib_message() {
+        let mut compressed_payload = BytesMut::new();
+        compressed_payload.extend(ZLIB_MAGIC);
+        compressed_payload.extend(&[0x9c, 0x12, 0x34, 0x56]);
+        let mut decoder = ChunkedGelfDecoder::default();
+
+        let error = decoder
+            .decode_eof(&mut compressed_payload)
+            .expect_err("decoding should fail");
+
+        let downcasted_error = downcast_framing_error(&error);
+        assert!(matches!(
+            downcasted_error,
+            ChunkedGelfDecoderError::Decompression {
+                source: ChunkedGelfDecompressionError::ZlibDecompression { .. }
+            }
+        ));
+    }
+
+    #[tokio::test]
+    async fn decode_zlib_payload_with_zlib_decoder() {
+        let payload = "foo";
+        let compressed_payload = Compression::Zlib.compress(&payload);
+        let mut decoder = ChunkedGelfDecoder {
+            decompression_config: ChunkedGelfDecompressionConfig::Zlib,
+            ..Default::default()
+        };
+
+        let frame = decoder
+            .decode_eof(&mut compressed_payload.into())
+            .expect("decoding should not fail")
+            .expect("decoding should return a frame");
+
+        assert_eq!(frame, payload);
+    }
+
+    #[tokio::test]
+    async fn decode_gzip_payload_with_zlib_decoder() {
+        let payload = "foo";
+        let compressed_payload = Compression::Gzip.compress(&payload);
+        let mut decoder = ChunkedGelfDecoder {
+            decompression_config: ChunkedGelfDecompressionConfig::Zlib,
+            ..Default::default()
+        };
+
+        let error = decoder
+            .decode_eof(&mut compressed_payload.into())
+            .expect_err("decoding should fail");
+
+        let downcasted_error = downcast_framing_error(&error);
+        assert!(matches!(
+            downcasted_error,
+            ChunkedGelfDecoderError::Decompression {
+                source: ChunkedGelfDecompressionError::ZlibDecompression { .. }
+            }
+        ));
+    }
+
+    #[tokio::test]
+    async fn decode_uncompressed_payload_with_zlib_decoder() {
+        let payload = "foo";
+        let mut decoder = ChunkedGelfDecoder {
+            decompression_config: ChunkedGelfDecompressionConfig::Zlib,
+            ..Default::default()
+        };
+
+        let error = decoder
+            .decode_eof(&mut payload.into())
+            .expect_err("decoding should fail");
+
+        let downcasted_error = downcast_framing_error(&error);
+        assert!(matches!(
+            downcasted_error,
+            ChunkedGelfDecoderError::Decompression {
+                source: ChunkedGelfDecompressionError::ZlibDecompression { .. }
+            }
+        ));
+    }
+
+    #[tokio::test]
+    async fn decode_gzip_payload_with_gzip_decoder() {
+        let payload = "foo";
+        let compressed_payload = Compression::Gzip.compress(&payload);
+        let mut decoder = ChunkedGelfDecoder {
+            decompression_config: ChunkedGelfDecompressionConfig::Gzip,
+            ..Default::default()
+        };
+
+        let frame = decoder
+            .decode_eof(&mut compressed_payload.into())
+            .expect("decoding should not fail")
+            .expect("decoding should return a frame");
+
+        assert_eq!(frame, payload);
+    }
+
+    #[tokio::test]
+    async fn decode_zlib_payload_with_gzip_decoder() {
+        let payload = "foo";
+        let compressed_payload = Compression::Zlib.compress(&payload);
+        let mut decoder = ChunkedGelfDecoder {
+            decompression_config: ChunkedGelfDecompressionConfig::Gzip,
+            ..Default::default()
+        };
+
+        let error = decoder
+            .decode_eof(&mut compressed_payload.into())
+            .expect_err("decoding should fail");
+
+        let downcasted_error = downcast_framing_error(&error);
+        assert!(matches!(
+            downcasted_error,
+            ChunkedGelfDecoderError::Decompression {
+                source: ChunkedGelfDecompressionError::GzipDecompression { .. }
+            }
+        ));
+    }
+
+    #[tokio::test]
+    async fn decode_uncompressed_payload_with_gzip_decoder() {
+        let payload = "foo";
+        let mut decoder = ChunkedGelfDecoder {
+            decompression_config: ChunkedGelfDecompressionConfig::Gzip,
+            ..Default::default()
+        };
+
+        let error = decoder
+            .decode_eof(&mut payload.into())
+            .expect_err("decoding should fail");
+
+        let downcasted_error = downcast_framing_error(&error);
+        assert!(matches!(
+            downcasted_error,
+            ChunkedGelfDecoderError::Decompression {
+                source: ChunkedGelfDecompressionError::GzipDecompression { .. }
+            }
+        ));
+    }
+
+    #[tokio::test]
+    #[rstest]
+    #[case::gzip(Compression::Gzip)]
+    #[case::zlib(Compression::Zlib)]
+    async fn decode_compressed_payload_with_no_decompression_decoder(
+        #[case] compression: Compression,
+    ) {
+        let payload = "foo";
+        let compressed_payload = compression.compress(&payload);
+        let mut decoder = ChunkedGelfDecoder {
+            decompression_config: ChunkedGelfDecompressionConfig::None,
+            ..Default::default()
+        };
+
+        let frame = decoder
+            .decode_eof(&mut compressed_payload.clone().into())
+            .expect("decoding should not fail")
+            .expect("decoding should return a frame");
+
+        assert_eq!(frame, compressed_payload);
+    }
+
+    #[test]
+    fn detect_gzip_compression() {
+        let payload = "foo";
+
+        for level in 0..=9 {
+            let level = flate2::Compression::new(level);
+            let compressed_payload = Compression::Gzip.compress_with_level(&payload, level);
+            let actual = ChunkedGelfDecompression::from_magic(&compressed_payload);
+            assert_eq!(
+                actual,
+                ChunkedGelfDecompression::Gzip,
+                "Failed for level {}",
+                level.level()
+            );
+        }
+    }
+
+    #[test]
+    fn detect_zlib_compression() {
+        let payload = "foo";
+
+        for level in 0..=9 {
+            let level = flate2::Compression::new(level);
+            let compressed_payload = Compression::Zlib.compress_with_level(&payload, level);
+            let actual = ChunkedGelfDecompression::from_magic(&compressed_payload);
+            assert_eq!(
+                actual,
+                ChunkedGelfDecompression::Zlib,
+                "Failed for level {}",
+                level.level()
+            );
+        }
+    }
+
+    #[test]
+    fn detect_no_compression() {
+        let payload = "foo";
+
+        let detected_compression = ChunkedGelfDecompression::from_magic(&payload.into());
+
+        assert_eq!(detected_compression, ChunkedGelfDecompression::None);
+    }
 }
diff --git a/lib/file-source/Cargo.toml b/lib/file-source/Cargo.toml
index c942fda74bd91..284999113766d 100644
--- a/lib/file-source/Cargo.toml
+++ b/lib/file-source/Cargo.toml
@@ -14,6 +14,7 @@ winapi = { version = "0.3", features = ["winioctl"] }
 crc = "3.2.1"
 glob.workspace = true
 scan_fmt = "0.2.6"
+vector-common = { path = "../vector-common", default-features = false }
 vector-config = { path = "../vector-config", default-features = false }
 vector-config-common = { path = "../vector-config-common", default-features = false }
 vector-config-macros = { path = "../vector-config-macros", default-features = false }
diff --git a/lib/file-source/src/file_watcher/mod.rs b/lib/file-source/src/file_watcher/mod.rs
index 7ac50173ea2b7..80db2b9bc876c 100644
--- a/lib/file-source/src/file_watcher/mod.rs
+++ b/lib/file-source/src/file_watcher/mod.rs
@@ -9,6 +9,7 @@ use bytes::{Bytes, BytesMut};
 use chrono::{DateTime, Utc};
 use flate2::bufread::MultiGzDecoder;
 use tracing::debug;
+use vector_common::constants::GZIP_MAGIC;
 
 use crate::{
     buffer::read_until_with_max_size, metadata_ext::PortableFileExt, FilePosition, ReadFrom,
@@ -293,7 +294,7 @@ fn is_gzipped(r: &mut io::BufReader<fs::File>) -> io::Result<bool> {
     let header_bytes = r.fill_buf()?;
     // WARN: The paired `BufReader::consume` is not called intentionally. If we
     // do we'll chop a decent part of the potential gzip stream off.
-    Ok(header_bytes.starts_with(&[0x1f, 0x8b]))
+    Ok(header_bytes.starts_with(GZIP_MAGIC))
 }
 
 fn null_reader() -> impl BufRead {
diff --git a/lib/vector-common/src/constants.rs b/lib/vector-common/src/constants.rs
new file mode 100644
index 0000000000000..e463a19461874
--- /dev/null
+++ b/lib/vector-common/src/constants.rs
@@ -0,0 +1,2 @@
+pub const GZIP_MAGIC: &[u8] = &[0x1f, 0x8b];
+pub const ZLIB_MAGIC: &[u8] = &[0x78];
diff --git a/lib/vector-common/src/lib.rs b/lib/vector-common/src/lib.rs
index b795415c43267..3e6d179f13b95 100644
--- a/lib/vector-common/src/lib.rs
+++ b/lib/vector-common/src/lib.rs
@@ -22,6 +22,8 @@ pub mod json_size;
 
 pub mod config;
 
+pub mod constants;
+
 #[cfg(feature = "conversion")]
 pub use vrl::compiler::TimeZone;
 
diff --git a/src/sources/aws_kinesis_firehose/handlers.rs b/src/sources/aws_kinesis_firehose/handlers.rs
index d623dc297842f..cd424d97a336a 100644
--- a/src/sources/aws_kinesis_firehose/handlers.rs
+++ b/src/sources/aws_kinesis_firehose/handlers.rs
@@ -7,6 +7,7 @@ use flate2::read::MultiGzDecoder;
 use futures::StreamExt;
 use snafu::{ResultExt, Snafu};
 use tokio_util::codec::FramedRead;
+use vector_common::constants::GZIP_MAGIC;
 use vector_lib::codecs::StreamDecodingError;
 use vector_lib::lookup::{metadata_path, path, PathPrefix};
 use vector_lib::{
@@ -242,7 +243,7 @@ fn is_gzip(data: &[u8]) -> bool {
     // deflate algorithm.
     //
     // Reference: https://datatracker.ietf.org/doc/html/rfc1952 Section 2.3
-    data.len() >= 10 && &data[..3] == b"\x1f\x8b\x08"
+    data.starts_with(GZIP_MAGIC)
 }
 
 fn decode_gzip(data: &[u8]) -> std::io::Result<Bytes> {
diff --git a/website/cue/reference/components/sources/base/amqp.cue b/website/cue/reference/components/sources/base/amqp.cue
index 155fa0d536dd6..d0ebadf50da28 100644
--- a/website/cue/reference/components/sources/base/amqp.cue
+++ b/website/cue/reference/components/sources/base/amqp.cue
@@ -335,6 +335,19 @@ base: components: sources: amqp: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/aws_kinesis_firehose.cue b/website/cue/reference/components/sources/base/aws_kinesis_firehose.cue
index 4313099b9bc49..f00dc9c67bee1 100644
--- a/website/cue/reference/components/sources/base/aws_kinesis_firehose.cue
+++ b/website/cue/reference/components/sources/base/aws_kinesis_firehose.cue
@@ -333,6 +333,19 @@ base: components: sources: aws_kinesis_firehose: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/aws_s3.cue b/website/cue/reference/components/sources/base/aws_s3.cue
index 12d3322a216ce..807c2eaa2b1b1 100644
--- a/website/cue/reference/components/sources/base/aws_s3.cue
+++ b/website/cue/reference/components/sources/base/aws_s3.cue
@@ -433,6 +433,19 @@ base: components: sources: aws_s3: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/aws_sqs.cue b/website/cue/reference/components/sources/base/aws_sqs.cue
index 19db1cf8830c9..39def33c7888c 100644
--- a/website/cue/reference/components/sources/base/aws_sqs.cue
+++ b/website/cue/reference/components/sources/base/aws_sqs.cue
@@ -437,6 +437,19 @@ base: components: sources: aws_sqs: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/datadog_agent.cue b/website/cue/reference/components/sources/base/datadog_agent.cue
index 156aa609cd356..60bb9aa6e8b0c 100644
--- a/website/cue/reference/components/sources/base/datadog_agent.cue
+++ b/website/cue/reference/components/sources/base/datadog_agent.cue
@@ -330,6 +330,19 @@ base: components: sources: datadog_agent: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/demo_logs.cue b/website/cue/reference/components/sources/base/demo_logs.cue
index b7539615c41bf..b7af9828fb75e 100644
--- a/website/cue/reference/components/sources/base/demo_logs.cue
+++ b/website/cue/reference/components/sources/base/demo_logs.cue
@@ -326,6 +326,19 @@ base: components: sources: demo_logs: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/exec.cue b/website/cue/reference/components/sources/base/exec.cue
index b21f203dde81c..ad6297800984f 100644
--- a/website/cue/reference/components/sources/base/exec.cue
+++ b/website/cue/reference/components/sources/base/exec.cue
@@ -314,6 +314,19 @@ base: components: sources: exec: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/file_descriptor.cue b/website/cue/reference/components/sources/base/file_descriptor.cue
index f368cc4eedcc0..23124afa6a9c6 100644
--- a/website/cue/reference/components/sources/base/file_descriptor.cue
+++ b/website/cue/reference/components/sources/base/file_descriptor.cue
@@ -292,6 +292,19 @@ base: components: sources: file_descriptor: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/gcp_pubsub.cue b/website/cue/reference/components/sources/base/gcp_pubsub.cue
index 61187238c72f2..55b16cdfbca17 100644
--- a/website/cue/reference/components/sources/base/gcp_pubsub.cue
+++ b/website/cue/reference/components/sources/base/gcp_pubsub.cue
@@ -369,6 +369,19 @@ base: components: sources: gcp_pubsub: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/heroku_logs.cue b/website/cue/reference/components/sources/base/heroku_logs.cue
index 876b6f3a51dde..418f61da96d33 100644
--- a/website/cue/reference/components/sources/base/heroku_logs.cue
+++ b/website/cue/reference/components/sources/base/heroku_logs.cue
@@ -327,6 +327,19 @@ base: components: sources: heroku_logs: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/http.cue b/website/cue/reference/components/sources/base/http.cue
index e3edaf3fc03a3..d53ca9b9f93fd 100644
--- a/website/cue/reference/components/sources/base/http.cue
+++ b/website/cue/reference/components/sources/base/http.cue
@@ -342,6 +342,19 @@ base: components: sources: http: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/http_client.cue b/website/cue/reference/components/sources/base/http_client.cue
index 3e3b44cef3ec3..609949072fba6 100644
--- a/website/cue/reference/components/sources/base/http_client.cue
+++ b/website/cue/reference/components/sources/base/http_client.cue
@@ -330,6 +330,19 @@ base: components: sources: http_client: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/http_server.cue b/website/cue/reference/components/sources/base/http_server.cue
index abaafa5c7a7f8..543a97a42c96c 100644
--- a/website/cue/reference/components/sources/base/http_server.cue
+++ b/website/cue/reference/components/sources/base/http_server.cue
@@ -342,6 +342,19 @@ base: components: sources: http_server: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/kafka.cue b/website/cue/reference/components/sources/base/kafka.cue
index b5906e7d3aa3f..6e96d91f5fb74 100644
--- a/website/cue/reference/components/sources/base/kafka.cue
+++ b/website/cue/reference/components/sources/base/kafka.cue
@@ -363,6 +363,19 @@ base: components: sources: kafka: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/nats.cue b/website/cue/reference/components/sources/base/nats.cue
index 52b2cbc2b5dc2..40516215dec9d 100644
--- a/website/cue/reference/components/sources/base/nats.cue
+++ b/website/cue/reference/components/sources/base/nats.cue
@@ -382,6 +382,19 @@ base: components: sources: nats: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/pulsar.cue b/website/cue/reference/components/sources/base/pulsar.cue
index 1f411bbb831ab..14452ffe67932 100644
--- a/website/cue/reference/components/sources/base/pulsar.cue
+++ b/website/cue/reference/components/sources/base/pulsar.cue
@@ -393,6 +393,19 @@ base: components: sources: pulsar: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/redis.cue b/website/cue/reference/components/sources/base/redis.cue
index f93131f0be48f..d9f167d400733 100644
--- a/website/cue/reference/components/sources/base/redis.cue
+++ b/website/cue/reference/components/sources/base/redis.cue
@@ -300,6 +300,19 @@ base: components: sources: redis: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/socket.cue b/website/cue/reference/components/sources/base/socket.cue
index eca545eb86d65..dc338ced7cd66 100644
--- a/website/cue/reference/components/sources/base/socket.cue
+++ b/website/cue/reference/components/sources/base/socket.cue
@@ -302,6 +302,19 @@ base: components: sources: socket: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will
diff --git a/website/cue/reference/components/sources/base/stdin.cue b/website/cue/reference/components/sources/base/stdin.cue
index 6fcd46ddbfa9f..ded73f44e9095 100644
--- a/website/cue/reference/components/sources/base/stdin.cue
+++ b/website/cue/reference/components/sources/base/stdin.cue
@@ -285,6 +285,19 @@ base: components: sources: stdin: configuration: {
 				relevant_when: "method = \"chunked_gelf\""
 				required:      false
 				type: object: options: {
+					decompression: {
+						description: "Decompression configuration for GELF messages."
+						required:    false
+						type: string: {
+							default: "Auto"
+							enum: {
+								Auto: "Automatically detect the decompression method based on the magic bytes of the message."
+								Gzip: "Use Gzip decompression."
+								None: "Do not decompress the message."
+								Zlib: "Use Zlib decompression."
+							}
+						}
+					}
 					max_length: {
 						description: """
 																The maximum length of a single GELF message, in bytes. Messages longer than this length will


Test patch: