diff --git a/lib/codecs/src/decoding/framing/chunked_gelf.rs b/lib/codecs/src/decoding/framing/chunked_gelf.rs index f8fcc8da4..516617619 100644 --- a/lib/codecs/src/decoding/framing/chunked_gelf.rs +++ b/lib/codecs/src/decoding/framing/chunked_gelf.rs @@ -19,11 +19,28 @@ use vector_config::configurable_component; const GELF_MAGIC: &[u8] = &[0x1e, 0x0f]; const GELF_MAX_TOTAL_CHUNKS: u8 = 128; const DEFAULT_TIMEOUT_SECS: f64 = 5.0; +/// Cap on concurrent incomplete messages, bounding the reassembly map. +/// Graylog Server itself has no such cap, so this is sized well above what a +/// legitimate sender holds in flight within the 5s reassembly window. +pub const DEFAULT_PENDING_MESSAGES_LIMIT: usize = 10_000; +/// Cap on one reassembled message. The protocol ceiling is 128 chunks +/// (`GELF_MAX_TOTAL_CHUNKS`) times the 65507-byte max UDP payload, so 8 MiB is +/// above anything the wire format can produce. Matches Graylog's own +/// `decompress_size_limit` default. +pub const DEFAULT_MAX_MESSAGE_LENGTH: usize = 8 * 1024 * 1024; const fn default_timeout_secs() -> f64 { DEFAULT_TIMEOUT_SECS } +const fn default_pending_messages_limit() -> Option { + Some(DEFAULT_PENDING_MESSAGES_LIMIT) +} + +const fn default_max_message_length() -> Option { + Some(DEFAULT_MAX_MESSAGE_LENGTH) +} + /// Config used to build a `ChunkedGelfDecoder`. #[configurable_component] #[derive(Debug, Clone, Default, PartialEq, Eq)] @@ -58,21 +75,22 @@ pub struct ChunkedGelfDecoderOptions { /// The maximum number of pending incomplete messages. If this limit is reached, the decoder starts /// dropping chunks of new messages, ensuring the memory usage of the decoder's state is bounded. - /// If this option is not set, the decoder does not limit the number of pending messages and the memory usage - /// of its messages buffer can grow unbounded. This matches Graylog Server's behavior. - #[serde(default, skip_serializing_if = "vector_core::serde::is_default")] + /// Defaults to 10000. Set explicitly to raise or lower it. + #[serde(default = "default_pending_messages_limit")] + #[derivative(Default(value = "default_pending_messages_limit()"))] pub pending_messages_limit: Option, /// The maximum length of a single GELF message, in bytes. Messages longer than this length will - /// be dropped. If this option is not set, the decoder does not limit the length of messages and - /// the per-message memory is unbounded. + /// be dropped. Defaults to 8 MiB, which is above the protocol's own ceiling of 128 chunks per + /// message. /// /// Note that a message can be composed of multiple chunks and this limit is applied to the whole /// message, not to individual chunks. /// /// This limit takes only into account the message's payload and the GELF header bytes are excluded from the calculation. /// The message's payload is the concatenation of all the chunks' payloads. - #[serde(default, skip_serializing_if = "vector_core::serde::is_default")] + #[serde(default = "default_max_message_length")] + #[derivative(Default(value = "default_max_message_length()"))] pub max_length: Option, /// Decompression configuration for GELF messages. @@ -486,8 +504,8 @@ impl Default for ChunkedGelfDecoder { fn default() -> Self { Self::new( DEFAULT_TIMEOUT_SECS, - None, - None, + default_pending_messages_limit(), + default_max_message_length(), ChunkedGelfDecompressionConfig::Auto, ) } @@ -1278,4 +1296,38 @@ mod tests { assert_eq!(detected_compression, ChunkedGelfDecompression::None); } + + #[tokio::test] + async fn defaults_are_finite_and_above_the_protocol_ceiling() { + let options = ChunkedGelfDecoderOptions::default(); + assert_eq!( + options.pending_messages_limit, + Some(DEFAULT_PENDING_MESSAGES_LIMIT) + ); + assert_eq!(options.max_length, Some(DEFAULT_MAX_MESSAGE_LENGTH)); + + // 128 chunks x the 65507-byte max UDP payload is the most the wire format can carry, + // so the default can never reject a well-formed message. + let protocol_ceiling = GELF_MAX_TOTAL_CHUNKS as usize * 65_507; + assert!(DEFAULT_MAX_MESSAGE_LENGTH >= protocol_ceiling); + } + + #[tokio::test] + async fn limits_are_per_message_and_do_not_kill_the_stream() { + // Both are per-message conditions; tearing down the connection would let one bad sender + // drop every other message multiplexed over it. + assert!(ChunkedGelfDecoderError::MaxLengthExceed { + message_id: 1, + sequence_number: 0, + length: 10, + max_length: 5, + } + .can_continue()); + assert!(ChunkedGelfDecoderError::PendingMessagesLimitReached { + message_id: 1, + sequence_number: 0, + pending_messages_limit: 1, + } + .can_continue()); + } }