From a2b2926852e84e512735a87ca22b5d522b4d1495 Mon Sep 17 00:00:00 2001 From: Harshvardhan Shrivastava Date: Sat, 8 Aug 2026 14:22:51 +0530 Subject: [PATCH 1/3] OBE-11235 - chunked_gelf framer: bound the reassembly map by default --- .../src/decoding/framing/chunked_gelf.rs | 237 +++++++++++++++++- 1 file changed, 224 insertions(+), 13 deletions(-) diff --git a/lib/codecs/src/decoding/framing/chunked_gelf.rs b/lib/codecs/src/decoding/framing/chunked_gelf.rs index f8fcc8da4..5c3172cfc 100644 --- a/lib/codecs/src/decoding/framing/chunked_gelf.rs +++ b/lib/codecs/src/decoding/framing/chunked_gelf.rs @@ -20,10 +20,33 @@ const GELF_MAGIC: &[u8] = &[0x1e, 0x0f]; const GELF_MAX_TOTAL_CHUNKS: u8 = 128; const DEFAULT_TIMEOUT_SECS: f64 = 5.0; +/// The number of incomplete messages tracked at once, by default. +/// +/// Each pending message costs roughly 10 KB regardless of how many payload bytes have +/// actually arrived, because `MessageState` holds a fixed `[Bytes; 128]` slot array. A +/// 12-byte header with no payload is enough to allocate one, so leaving this unbounded +/// lets an unauthenticated peer amplify its traffic by ~870x. +const DEFAULT_PENDING_MESSAGES_LIMIT: usize = 5_000; + +/// The maximum reassembled payload of a single chunked message, by default. +/// +/// Comfortably above any real GELF-over-UDP message: the protocol caps a message at 128 +/// chunks, and clients pick an MTU-sized chunk (Graylog's own go-gelf uses 1420 bytes, +/// for ~182 KB total), so this only rejects payloads that no conforming sender produces. +const DEFAULT_MAX_LENGTH_BYTES: usize = 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_length() -> Option { + Some(DEFAULT_MAX_LENGTH_BYTES) +} + /// Config used to build a `ChunkedGelfDecoder`. #[configurable_component] #[derive(Debug, Clone, Default, PartialEq, Eq)] @@ -57,22 +80,29 @@ pub struct ChunkedGelfDecoderOptions { pub timeout_secs: f64, /// 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")] + /// dropping chunks of *new* messages; chunks of messages already being reassembled are still + /// accepted, so reaching the limit does not corrupt in-flight messages. + /// + /// Defaults to 5000. Setting this to `null` restores the unbounded behavior, which matches + /// Graylog Server but lets any peer that can reach the socket exhaust memory: a pending message + /// costs roughly 10 KB no matter how few payload bytes have arrived, so a 12-byte header is + /// enough to allocate one. Do not disable this on an unauthenticated listener. + #[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, and the chunk that would exceed it is never buffered. /// /// 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")] + /// + /// Defaults to 1048576 (1 MiB). Setting this to `null` leaves the per-message memory unbounded. + #[serde(default = "default_max_length")] + #[derivative(Default(value = "default_max_length()"))] pub max_length: Option, /// Decompression configuration for GELF messages. @@ -377,9 +407,14 @@ impl ChunkedGelfDecoder { let mut state_lock = self.state.lock().expect("poisoned lock"); + // The limit bounds how many *distinct* messages are tracked, so it must only reject + // message ids that are not already being reassembled. Checking it before the lookup + // would drop follow-up chunks of in-flight messages whenever the map is full - i.e. + // precisely under the load the limit exists to survive - silently corrupting + // legitimate traffic instead of only shedding new attacker-chosen ids. if let Some(pending_messages_limit) = self.pending_messages_limit { ensure!( - state_lock.len() < pending_messages_limit, + state_lock.len() < pending_messages_limit || state_lock.contains_key(&message_id), PendingMessagesLimitReachedSnafu { message_id, sequence_number, @@ -428,10 +463,13 @@ impl ChunkedGelfDecoder { return Ok(None); } - message_state.add_chunk(sequence_number, chunk); - + // Checked before the chunk is stored rather than after. This is defensive only: `chunk` is + // a refcounted slice of the datagram, so storing it copies nothing and the entry is dropped + // on the error path either way - the two orders retain the same memory today. It matters if + // `add_chunk` ever starts copying. `length` is the size the message would reach, which is + // the same value the previous store-then-measure order reported. if let Some(max_length) = self.max_length { - let length = message_state.current_length(); + let length = message_state.current_length() + chunk.remaining(); if length > max_length { state_lock.remove(&message_id); return Err(ChunkedGelfDecoderError::MaxLengthExceed { @@ -443,6 +481,8 @@ impl ChunkedGelfDecoder { } } + message_state.add_chunk(sequence_number, chunk); + if let Some(message) = message_state.retrieve_message() { state_lock.remove(&message_id); Ok(Some(message)) @@ -486,8 +526,8 @@ impl Default for ChunkedGelfDecoder { fn default() -> Self { Self::new( DEFAULT_TIMEOUT_SECS, - None, - None, + default_pending_messages_limit(), + default_max_length(), ChunkedGelfDecompressionConfig::Auto, ) } @@ -1278,4 +1318,175 @@ mod tests { assert_eq!(detected_compression, ChunkedGelfDecompression::None); } + + // OBE-11235: the reassembly map is keyed solely on an attacker-chosen `message_id` and was + // unbounded by default. A 12-byte header with no payload allocates a ~10 KB `MessageState`, + // so an unauthenticated peer could amplify its traffic ~870x until the process was OOM-killed. + + /// A header-only chunk: enough to allocate reassembly state, no payload. + fn header_only_chunk(message_id: u64) -> BytesMut { + create_chunk(message_id, 0, GELF_MAX_TOTAL_CHUNKS, &"") + } + + #[tokio::test] + async fn defaults_bound_the_reassembly_state() { + let options = ChunkedGelfDecoderOptions::default(); + + assert_eq!( + options.pending_messages_limit, + Some(DEFAULT_PENDING_MESSAGES_LIMIT) + ); + assert_eq!(options.max_length, Some(DEFAULT_MAX_LENGTH_BYTES)); + + // The socket source reaches the decoder through `Default`, not through the options + // struct, so both paths have to carry the limits. + let decoder = ChunkedGelfDecoder::default(); + assert_eq!( + decoder.pending_messages_limit, + Some(DEFAULT_PENDING_MESSAGES_LIMIT) + ); + assert_eq!(decoder.max_length, Some(DEFAULT_MAX_LENGTH_BYTES)); + } + + #[tokio::test] + async fn pending_map_stops_growing_at_the_default_limit() { + let mut decoder = ChunkedGelfDecoder::default(); + let overshoot = 250; + + let mut rejected = 0; + for message_id in 0..(DEFAULT_PENDING_MESSAGES_LIMIT + overshoot) as u64 { + if decoder + .decode_eof(&mut header_only_chunk(message_id)) + .is_err() + { + rejected += 1; + } + } + + assert_eq!( + decoder.state.lock().unwrap().len(), + DEFAULT_PENDING_MESSAGES_LIMIT, + "the map must stop growing at the limit" + ); + assert_eq!( + rejected, overshoot, + "every id beyond the limit must be rejected" + ); + } + + #[rstest] + #[tokio::test] + async fn a_full_pending_map_still_accepts_chunks_of_tracked_messages( + two_chunks_message: ([BytesMut; 2], String), + ) { + // The regression the limit itself used to introduce: because the check ran before the + // map lookup, filling the map rejected follow-up chunks of messages already being + // reassembled - dropping legitimate traffic exactly when the limit started to bite. + let (mut chunks, expected_message) = two_chunks_message; + let limit = 4; + let mut decoder = ChunkedGelfDecoder { + pending_messages_limit: Some(limit), + ..Default::default() + }; + + // Start reassembling a real message, then fill the rest of the map with other ids. + assert!(decoder.decode_eof(&mut chunks[0]).unwrap().is_none()); + let tracked_id = *decoder.state.lock().unwrap().keys().next().unwrap(); + for message_id in 0..limit as u64 { + if message_id != tracked_id { + let _ = decoder.decode_eof(&mut header_only_chunk(message_id)); + } + } + assert_eq!(decoder.state.lock().unwrap().len(), limit, "map is full"); + + // A brand-new id is shed... + let error = decoder + .decode_eof(&mut header_only_chunk(u64::MAX)) + .unwrap_err(); + assert!(matches!( + downcast_framing_error(&error), + ChunkedGelfDecoderError::PendingMessagesLimitReached { .. } + )); + + // ...but the in-flight message still completes. + let frame = decoder + .decode_eof(&mut chunks[1]) + .expect("a chunk of an already-tracked message must not be rejected") + .expect("the message should be complete"); + assert_eq!(frame, expected_message); + } + + #[tokio::test] + async fn oversized_message_is_rejected_and_leaves_no_state() { + let message_id = 1u64; + let max_length = 32; + let mut decoder = ChunkedGelfDecoder { + max_length: Some(max_length), + ..Default::default() + }; + + let payload = "a".repeat(max_length + 1); + let error = decoder + .decode_eof(&mut create_chunk(message_id, 0, 2, &payload)) + .unwrap_err(); + + assert!(matches!( + downcast_framing_error(&error), + ChunkedGelfDecoderError::MaxLengthExceed { .. } + )); + assert!( + decoder.state.lock().unwrap().is_empty(), + "the over-budget message must not be left buffered" + ); + } + + #[tokio::test] + async fn message_over_the_default_max_length_is_rejected() { + let mut decoder = ChunkedGelfDecoder::default(); + + // Two chunks, so the message never completes and the length check is what stops it. + let payload = "a".repeat(DEFAULT_MAX_LENGTH_BYTES + 1); + let error = decoder + .decode_eof(&mut create_chunk(1, 0, 2, &payload)) + .unwrap_err(); + + assert!(matches!( + downcast_framing_error(&error), + ChunkedGelfDecoderError::MaxLengthExceed { .. } + )); + } + + #[tokio::test] + async fn message_exactly_at_max_length_is_accepted() { + let message_id = 1u64; + let max_length = 32; + let mut decoder = ChunkedGelfDecoder { + max_length: Some(max_length), + ..Default::default() + }; + + let payload = "a".repeat(max_length); + let frame = decoder + .decode_eof(&mut create_chunk(message_id, 0, 1, &payload)) + .expect("a message exactly at the limit must be accepted") + .expect("the single-chunk message should be complete"); + + assert_eq!(frame, payload); + } + + #[rstest] + #[tokio::test] + async fn ordinary_chunked_message_decodes_under_the_default_limits( + three_chunks_message: ([BytesMut; 3], String), + ) { + let (mut chunks, expected_message) = three_chunks_message; + let mut decoder = ChunkedGelfDecoder::default(); + + assert!(decoder.decode_eof(&mut chunks[0]).unwrap().is_none()); + assert!(decoder.decode_eof(&mut chunks[1]).unwrap().is_none()); + let frame = decoder.decode_eof(&mut chunks[2]).unwrap().unwrap(); + + assert_eq!(frame, expected_message); + assert!(decoder.state.lock().unwrap().is_empty()); + } } From df302244cdd36bb6df48e7315816ef5a653da67f Mon Sep 17 00:00:00 2001 From: Harshvardhan Shrivastava Date: Sat, 8 Aug 2026 14:23:24 +0530 Subject: [PATCH 2/3] OBE-11563 - docker_logs: bound the partial-event merge buffer --- lib/vector-core/src/event/merge_state.rs | 64 ++++++++- src/sources/docker_logs/mod.rs | 136 +++++++++++++----- src/sources/docker_logs/tests.rs | 168 +++++++++++++++++++++++ 3 files changed, 327 insertions(+), 41 deletions(-) diff --git a/lib/vector-core/src/event/merge_state.rs b/lib/vector-core/src/event/merge_state.rs index 6c1322d93..3449ffe92 100644 --- a/lib/vector-core/src/event/merge_state.rs +++ b/lib/vector-core/src/event/merge_state.rs @@ -1,29 +1,50 @@ -use super::LogEvent; +use super::{EstimatedJsonEncodedSizeOf, LogEvent}; /// Encapsulates the inductive events merging algorithm. /// -/// In the future, this might be extended by various counters (the number of -/// events that contributed to the current merge event for instance, or the -/// event size) to support circuit breaker logic. +/// Tracks the size of everything merged so far so callers can apply circuit-breaker logic: +/// a merge that is never terminated returns nothing to the pipeline, so bounded-channel +/// backpressure cannot engage and the accumulator is otherwise unbounded. #[derive(Debug)] pub struct LogEventMergeState { /// Intermediate event we merge into. intermediate_merged_event: LogEvent, + /// Running total of the sizes of every event folded in so far. + merged_bytes: usize, } impl LogEventMergeState { /// Initialize the algorithm with a first (partial) event. pub fn new(first_partial_event: LogEvent) -> Self { + let merged_bytes = first_partial_event.estimated_json_encoded_size_of().get(); Self { intermediate_merged_event: first_partial_event, + merged_bytes, } } /// Merge the incoming (partial) event in. pub fn merge_in_next_event(&mut self, incoming: LogEvent, fields: &[impl AsRef]) { + // Measured on the incoming event rather than the accumulator, so this stays O(incoming) + // and adds no term to the merge's existing cost. + self.merged_bytes = self + .merged_bytes + .saturating_add(incoming.estimated_json_encoded_size_of().get()); self.intermediate_merged_event.merge(incoming, fields); } + /// The total size of every event folded in so far. + pub const fn merged_bytes(&self) -> usize { + self.merged_bytes + } + + /// Take the event accumulated so far, abandoning the merge. + /// + /// Used to flush an over-budget merge downstream instead of growing it further. + pub fn into_merged_event(self) -> LogEvent { + self.intermediate_merged_event + } + /// Merge the final (non-partial) event in and return the resulting (merged) /// event. pub fn merge_in_final_event( @@ -61,4 +82,39 @@ mod test { b"hello world" ); } + + #[test] + fn merged_bytes_grows_with_every_event_folded_in() { + let fields = vec!["message".to_string()]; + + let mut state = LogEventMergeState::new(log_event_with_message("hel")); + let after_first = state.merged_bytes(); + assert!(after_first > 0, "the initial event must be accounted for"); + + state.merge_in_next_event(log_event_with_message("lo "), &fields); + let after_second = state.merged_bytes(); + assert!( + after_second > after_first, + "folding an event in must grow the running total" + ); + + // Growth tracks payload size, which is what a caller budgets against. + state.merge_in_next_event(log_event_with_message(&"x".repeat(1024)), &fields); + assert!(state.merged_bytes() >= after_second + 1024); + } + + #[test] + fn into_merged_event_returns_what_was_accumulated() { + let fields = vec!["message".to_string()]; + + let mut state = LogEventMergeState::new(log_event_with_message("hel")); + state.merge_in_next_event(log_event_with_message("lo"), &fields); + + let event = state.into_merged_event(); + + assert_eq!( + event.get("message").unwrap().coerce_to_bytes().as_ref(), + b"hello" + ); + } } diff --git a/src/sources/docker_logs/mod.rs b/src/sources/docker_logs/mod.rs index 634cf45a1..378623633 100644 --- a/src/sources/docker_logs/mod.rs +++ b/src/sources/docker_logs/mod.rs @@ -144,6 +144,19 @@ pub struct DockerLogsConfig { /// Enables automatic merging of partial events. auto_partial_merge: bool, + /// The maximum size, in bytes, that a single merged log line may reach before it is emitted. + /// + /// Only applies when `auto_partial_merge` is enabled. A container that writes without ever + /// emitting a newline would otherwise grow the per-stream merge buffer without limit: partial + /// frames are folded in and nothing is returned to the pipeline, so bounded-channel + /// backpressure never engages. On reaching this size the accumulated line is emitted (marked + /// with `partial_event_marker_field`, since the line continues) and merging restarts, so no + /// data is lost - an over-long line is split across several events rather than buffered. + #[serde(default = "default_max_merged_line_bytes")] + #[configurable(metadata(docs::type_unit = "bytes"))] + #[configurable(metadata(docs::human_name = "Max Merged Line Size"))] + max_merged_line_bytes: usize, + /// The amount of time to wait before retrying after an error. #[serde_as(as = "serde_with::DurationSeconds")] #[serde(default = "default_retry_backoff_secs")] @@ -177,6 +190,7 @@ impl Default for DockerLogsConfig { include_images: None, partial_event_marker_field: default_partial_event_marker_field(), auto_partial_merge: true, + max_merged_line_bytes: default_max_merged_line_bytes(), multiline: None, retry_backoff_secs: default_retry_backoff_secs(), log_namespace: None, @@ -188,6 +202,14 @@ fn default_partial_event_marker_field() -> Option { Some(event::PARTIAL.to_string()) } +/// Docker's json-file and local drivers split output into 16 KiB frames, so this allows a single +/// logical line to span 64 of them - far more than any real log line - while capping the per-stream +/// buffer. It also bounds the quadratic re-copy that merging performs: the cost of a merge cycle is +/// a function of this limit rather than of how long a container keeps writing. +const fn default_max_merged_line_bytes() -> usize { + 1024 * 1024 +} + const fn default_retry_backoff_secs() -> Duration { Duration::from_secs(2) } @@ -765,6 +787,7 @@ impl EventStreamBuilder { message, core.config.partial_event_marker_field.clone(), core.config.auto_partial_merge, + core.config.max_merged_line_bytes, &mut partial_event_merge_state, &bytes_received, self.log_namespace, @@ -979,6 +1002,7 @@ impl ContainerLogInfo { log_output: LogOutput, partial_event_marker_field: Option, auto_partial_merge: bool, + max_merged_line_bytes: usize, partial_event_merge_state: &mut Option, bytes_received: &Registered, log_namespace: LogNamespace, @@ -1167,55 +1191,93 @@ impl ContainerLogInfo { let log = if auto_partial_merge { // Partial event events merging logic. - // If event is partial, stash it and return `None`. + // If event is partial, stash it and return `None` - unless the line being + // accumulated has outgrown its budget, in which case emit what we have. if is_partial { // If we already have a partial event merge state, the current // message has to be merged into that existing state. // Otherwise, create a new partial event merge state with the // current message being the initial one. - if let Some(partial_event_merge_state) = partial_event_merge_state { - // Depending on the log namespace the actual contents of the log "message" will be - // found in either the root of the event ("."), or at the globally configured "message_key". - match log_namespace { - LogNamespace::Vector => { - partial_event_merge_state.merge_in_next_event(log, &["."]); - } - LogNamespace::Legacy => { - partial_event_merge_state.merge_in_next_event( - log, - &[log_schema() - .message_key() - .expect("global log_schema.message_key to be valid path") - .to_string()], - ); - } - } - } else { + let Some(merge_state) = partial_event_merge_state.as_mut() else { *partial_event_merge_state = Some(LogEventMergeState::new(log)); + return None; }; - return None; - }; - // This is not a partial event. If we have a partial event merge - // state from before, the current event must be a final event, that - // would give us a merged event we can return. - // Otherwise it's just a regular event that we return as-is. - match partial_event_merge_state.take() { // Depending on the log namespace the actual contents of the log "message" will be // found in either the root of the event ("."), or at the globally configured "message_key". - Some(partial_event_merge_state) => match log_namespace { + match log_namespace { LogNamespace::Vector => { - partial_event_merge_state.merge_in_final_event(log, &["."]) + merge_state.merge_in_next_event(log, &["."]); } - LogNamespace::Legacy => partial_event_merge_state.merge_in_final_event( - log, - &[log_schema() - .message_key() - .expect("global log_schema.message_key to be valid path") - .to_string()], - ), - }, - None => log, + LogNamespace::Legacy => { + merge_state.merge_in_next_event( + log, + &[log_schema() + .message_key() + .expect("global log_schema.message_key to be valid path") + .to_string()], + ); + } + } + + if merge_state.merged_bytes() <= max_merged_line_bytes { + return None; + } + + // Nothing is returned to the pipeline while a merge is in progress, so bounded- + // channel backpressure cannot engage: a container that never emits a newline + // would grow this buffer until the process is OOM-killed. Emit the accumulated + // line instead and start a fresh merge, so an over-long line is split across + // several events rather than buffered without limit. + warn!( + message = "Merged log line exceeded the maximum size. Emitting it and continuing the merge in a new event.", + max_merged_line_bytes = max_merged_line_bytes, + container_id = self.id.as_str(), + internal_log_rate_limit = true, + ); + + let mut merged = partial_event_merge_state + .take() + .expect("merge state was just observed to be present") + .into_merged_event(); + + // The line continues in the next event, so mark this one the same way an + // unmerged partial event would be marked. + if let Some(partial_event_marker_field) = partial_event_marker_field { + log_namespace.insert_source_metadata( + DockerLogsConfig::NAME, + &mut merged, + Some(LegacyKey::Overwrite(path!( + partial_event_marker_field.as_str() + ))), + path!(event::PARTIAL), + true, + ); + } + + merged + } else { + // This is not a partial event. If we have a partial event merge + // state from before, the current event must be a final event, that + // would give us a merged event we can return. + // Otherwise it's just a regular event that we return as-is. + match partial_event_merge_state.take() { + // Depending on the log namespace the actual contents of the log "message" will be + // found in either the root of the event ("."), or at the globally configured "message_key". + Some(partial_event_merge_state) => match log_namespace { + LogNamespace::Vector => { + partial_event_merge_state.merge_in_final_event(log, &["."]) + } + LogNamespace::Legacy => partial_event_merge_state.merge_in_final_event( + log, + &[log_schema() + .message_key() + .expect("global log_schema.message_key to be valid path") + .to_string()], + ), + }, + None => log, + } } } else { // If the event is partial, just set the partial event marker field. diff --git a/src/sources/docker_logs/tests.rs b/src/sources/docker_logs/tests.rs index 72b0dd75e..8d7e6f63b 100644 --- a/src/sources/docker_logs/tests.rs +++ b/src/sources/docker_logs/tests.rs @@ -23,6 +23,174 @@ fn exclude_self() { assert!(!source.exclude_self("a29d569bd46c")); } +// OBE-11563: a container that writes without ever emitting a newline folds every frame into a +// per-stream merge buffer and returns nothing to the pipeline, so bounded-channel backpressure +// never engages and the buffer grows until the process dies. +mod partial_merge_bounds { + use super::*; + + const LOG_TIMESTAMP: &str = "2026-08-08T12:00:00.000000000Z"; + + fn container_log_info() -> ContainerLogInfo { + let created = DateTime::parse_from_rfc3339("2026-08-08T11:00:00Z") + .unwrap() + .with_timezone(&Utc); + + ContainerLogInfo::new( + ContainerId::new("test-container-id".to_owned()), + ContainerMetadata { + labels: HashMap::new(), + name: "/test".into(), + name_str: "/test".to_owned(), + image: "test-image".into(), + created_at: created, + }, + created, + ) + } + + /// A frame as Docker delivers it: RFC3339 timestamp, a space, then the payload. Without a + /// trailing newline the source treats it as a partial line and merges it. + fn frame(payload: &str, terminated: bool) -> LogOutput { + let mut message = format!("{LOG_TIMESTAMP} {payload}"); + if terminated { + message.push('\n'); + } + LogOutput::StdOut { + message: Bytes::from(message), + } + } + + struct Harness { + info: ContainerLogInfo, + bytes_received: Registered, + merge_state: Option, + max_merged_line_bytes: usize, + } + + impl Harness { + fn new(max_merged_line_bytes: usize) -> Self { + Self { + info: container_log_info(), + bytes_received: register!(BytesReceived::from(Protocol::HTTP)), + merge_state: None, + max_merged_line_bytes, + } + } + + fn feed(&mut self, payload: &str, terminated: bool) -> Option { + self.info.new_event( + frame(payload, terminated), + Some(event::PARTIAL.to_string()), + true, + self.max_merged_line_bytes, + &mut self.merge_state, + &self.bytes_received, + LogNamespace::Legacy, + ) + } + + fn buffered_bytes(&self) -> usize { + self.merge_state + .as_ref() + .map_or(0, LogEventMergeState::merged_bytes) + } + } + + fn message_of(event: &LogEvent) -> String { + String::from_utf8_lossy(&event.get("message").unwrap().coerce_to_bytes()).into_owned() + } + + #[test] + fn unterminated_line_is_flushed_instead_of_buffered_forever() { + let max_merged_line_bytes = 4096; + let frame_payload = "a".repeat(512); + let mut harness = Harness::new(max_merged_line_bytes); + + let mut emitted = Vec::new(); + let mut high_water = 0; + for _ in 0..200 { + if let Some(event) = harness.feed(&frame_payload, false) { + emitted.push(event); + } + high_water = high_water.max(harness.buffered_bytes()); + } + + assert!( + !emitted.is_empty(), + "a line that never terminates must still be emitted" + ); + assert!( + high_water <= max_merged_line_bytes, + "the merge buffer reached {high_water} bytes against a {max_merged_line_bytes}-byte budget" + ); + } + + #[test] + fn flushed_events_carry_the_partial_marker_and_lose_no_data() { + let max_merged_line_bytes = 2048; + let frame_payload = "b".repeat(512); + let mut harness = Harness::new(max_merged_line_bytes); + + let mut emitted = Vec::new(); + for _ in 0..20 { + if let Some(event) = harness.feed(&frame_payload, false) { + emitted.push(event); + } + } + // Terminate the line so the tail is emitted too. + if let Some(event) = harness.feed(&frame_payload, true) { + emitted.push(event); + } + + assert!(emitted.len() > 1, "the line should have been split"); + + // Every event except the last continues into the next one, so each is marked partial. + for event in &emitted[..emitted.len() - 1] { + assert_eq!( + event.get(event::PARTIAL).map(|v| v == &Value::from(true)), + Some(true), + "a flushed fragment must be marked partial so consumers can rejoin it" + ); + } + + // No payload bytes were dropped by the split. + let recovered: usize = emitted.iter().map(|event| message_of(event).len()).sum(); + assert_eq!( + recovered, + frame_payload.len() * 21, + "splitting the line must not lose data" + ); + } + + #[test] + fn ordinary_multi_frame_line_still_merges_into_one_event() { + let mut harness = Harness::new(default_max_merged_line_bytes()); + + assert!(harness.feed("hel", false).is_none()); + assert!(harness.feed("lo ", false).is_none()); + let event = harness + .feed("world", true) + .expect("the terminated line should be emitted as a single event"); + + assert_eq!(message_of(&event), "hello world"); + assert!( + event.get(event::PARTIAL).is_none(), + "a line that fit within the budget is not partial" + ); + assert!(harness.merge_state.is_none(), "merge state should be reset"); + } + + #[test] + fn default_budget_is_finite() { + assert_eq!( + DockerLogsConfig::default().max_merged_line_bytes, + default_max_merged_line_bytes() + ); + assert!(default_max_merged_line_bytes() > 0); + } +} + #[cfg(all(test, feature = "docker-logs-integration-tests"))] mod integration_tests { use bollard::{ From c407673c16a674950b9583a266716ab27f558232 Mon Sep 17 00:00:00 2001 From: Harshvardhan Shrivastava Date: Sat, 8 Aug 2026 14:23:49 +0530 Subject: [PATCH 3/3] OBE-11555 - tcp source: release the request-limiter permit before the ack write --- src/sources/fluent/mod.rs | 107 +++++++++++++++++++++++++++++++- src/sources/util/net/tcp/mod.rs | 40 +++++++++++- 2 files changed, 143 insertions(+), 4 deletions(-) diff --git a/src/sources/fluent/mod.rs b/src/sources/fluent/mod.rs index 47afb99de..020f0d21a 100644 --- a/src/sources/fluent/mod.rs +++ b/src/sources/fluent/mod.rs @@ -524,6 +524,13 @@ impl Decoder for FluentEntryStreamDecoder { } } +/// Fluentd's `chunk` is a base64-encoded unique id - 24 bytes in practice. +/// +/// The ack echoes it back verbatim, so without a cap a client can make Vector write an +/// arbitrarily large ack to a socket it never reads from, parking that write indefinitely. +/// A single 1 MiB chunk id is enough to stall the very first `write_all`. +const MAX_CHUNK_ID_BYTES: usize = 256; + struct FluentAcker { chunks: Vec, } @@ -531,7 +538,24 @@ struct FluentAcker { impl FluentAcker { fn new(frames: &[FluentFrame]) -> Self { Self { - chunks: frames.iter().filter_map(|f| f.chunk.clone()).collect(), + chunks: frames + .iter() + .filter_map(|f| f.chunk.clone()) + .filter(|chunk| { + if chunk.len() > MAX_CHUNK_ID_BYTES { + tracing::warn!( + message = + "Fluent chunk id exceeds the maximum length; not acknowledging it.", + chunk_id_bytes = chunk.len(), + max_chunk_id_bytes = MAX_CHUNK_ID_BYTES, + internal_log_rate_limit = true, + ); + false + } else { + true + } + }) + .collect(), } } } @@ -890,6 +914,87 @@ mod tests { assert_eq!(output, expected); } + // OBE-11555: the ack echoes the client-supplied chunk id verbatim, so an oversized id makes + // Vector write a huge ack to a socket the peer never reads - parking the write, and (before + // the permit was scoped to end first) pinning a request-limiter permit for the whole source. + mod ack_size_bounds { + use super::*; + + fn frame_with_chunk(chunk: Option) -> FluentFrame { + FluentFrame { + events: SmallVec::new(), + chunk, + } + } + + fn ack_bytes_for(chunk: Option) -> Option { + FluentAcker::new(&[frame_with_chunk(chunk)]).build_ack(TcpSourceAck::Ack) + } + + #[test] + fn ordinary_chunk_id_is_acknowledged() { + // What a real fluent client sends: base64 of a uuid, 24 bytes. + let chunk = BASE64_STANDARD.encode(uuid::Uuid::new_v4().as_bytes()); + assert!(chunk.len() <= MAX_CHUNK_ID_BYTES); + + let ack = ack_bytes_for(Some(chunk.clone())).expect("a normal chunk id must be acked"); + assert!( + ack.windows(chunk.len()).any(|w| w == chunk.as_bytes()), + "the ack must echo the chunk id back" + ); + } + + #[test] + fn chunk_id_at_the_cap_is_still_acknowledged() { + let chunk = "a".repeat(MAX_CHUNK_ID_BYTES); + assert!( + ack_bytes_for(Some(chunk)).is_some(), + "a chunk id exactly at the cap must be acked" + ); + } + + #[test] + fn oversized_chunk_id_is_not_echoed() { + let chunk = "a".repeat(MAX_CHUNK_ID_BYTES + 1); + assert!( + ack_bytes_for(Some(chunk)).is_none(), + "an over-cap chunk id must not produce an ack" + ); + } + + #[test] + fn ack_size_stays_bounded_for_a_huge_chunk_id() { + // The measured attack: one 1 MiB chunk id stalls the very first write_all, because a + // single ack exceeds the combined send and receive buffers. + let chunk = "a".repeat(1024 * 1024); + + let ack_len = ack_bytes_for(Some(chunk)).map_or(0, |ack| ack.len()); + + assert!( + ack_len <= MAX_CHUNK_ID_BYTES + 64, + "a 1 MiB chunk id produced a {ack_len}-byte ack" + ); + } + + #[test] + fn oversized_chunk_id_does_not_suppress_a_valid_one() { + let good = BASE64_STANDARD.encode(uuid::Uuid::new_v4().as_bytes()); + let acker = FluentAcker::new(&[ + frame_with_chunk(Some("a".repeat(MAX_CHUNK_ID_BYTES + 1))), + frame_with_chunk(Some(good.clone())), + ]); + + let ack = acker + .build_ack(TcpSourceAck::Ack) + .expect("the well-formed chunk id must still be acked"); + assert!( + ack.windows(good.len()).any(|w| w == good.as_bytes()), + "the valid chunk id must be echoed even alongside an oversized one" + ); + assert!(ack.len() <= MAX_CHUNK_ID_BYTES + 64); + } + } + async fn check_acknowledgements( status: EventStatus, with_chunk: bool, diff --git a/src/sources/util/net/tcp/mod.rs b/src/sources/util/net/tcp/mod.rs index 13bb464ab..5672ad798 100644 --- a/src/sources/util/net/tcp/mod.rs +++ b/src/sources/util/net/tcp/mod.rs @@ -44,6 +44,13 @@ pub use vector_lib::net::*; pub const MAX_IN_FLIGHT_EVENTS_TARGET: usize = 100_000; +/// How long to wait for a peer to accept an acknowledgement before dropping the connection. +/// +/// `write_all` progresses only as the peer's TCP receive window opens, so a peer that simply +/// stops calling `recv()` parks the write - and with it the task, socket and fd - indefinitely. +/// Generous enough that a merely slow client is never dropped. +const ACK_WRITE_TIMEOUT: Duration = Duration::from_secs(30); + pub async fn try_bind_tcp_listener( addr: SocketListenAddr, mut listenfd: ListenFd, @@ -376,11 +383,38 @@ async fn handle_stream( } } }; + // The permit bounds in-flight *decoded events*, and that work is + // finished: the batch is in the pipeline and has been acknowledged. + // Holding it across the ack write lets a peer that stops reading pin + // it forever, because `write_all` makes progress only as the peer's + // TCP receive window opens. Permits are replenished (and grown) only + // on drop and one limiter is shared per source, so a couple of such + // connections freeze ingestion for every other client. + drop(permit.take()); + if let Some(ack_bytes) = acker.build_ack(ack){ let stream = reader.get_mut().get_mut(); - if let Err(error) = stream.write_all(&ack_bytes).await { - emit!(TcpSendAckError{ error }); - break; + // Releasing the permit stops the source-wide freeze but would + // still leak this task, its socket and its fd to a peer that + // never drains. Time the write out and treat expiry as fatal. + match tokio::time::timeout(ACK_WRITE_TIMEOUT, stream.write_all(&ack_bytes)).await { + Ok(Ok(())) => {} + Ok(Err(error)) => { + emit!(TcpSendAckError{ error }); + break; + } + Err(_) => { + emit!(TcpSendAckError { + error: io::Error::new( + io::ErrorKind::TimedOut, + format!( + "peer did not accept the acknowledgement within {}s", + ACK_WRITE_TIMEOUT.as_secs() + ), + ), + }); + break; + } } } if ack != TcpSourceAck::Ack {