diff --git a/changelog.d/protobuf_nesting_depth_limit.fix.md b/changelog.d/protobuf_nesting_depth_limit.fix.md new file mode 100644 index 0000000000000..e9b7dae534acf --- /dev/null +++ b/changelog.d/protobuf_nesting_depth_limit.fix.md @@ -0,0 +1,3 @@ +Fixed unrecoverable disk buffer corruption and vector-to-vector retry loops caused by event data or metadata that protobuf could encode but prost could not decode. Vector now drops only protobuf-unsafe nested payloads before disk buffer or `vector` sink gRPC encoding, while preserving nested shapes that prost can safely decode. + +authors: connoryy diff --git a/lib/vector-buffers/benches/common.rs b/lib/vector-buffers/benches/common.rs index 816e32ceb9f39..d36f75e7b0a2b 100644 --- a/lib/vector-buffers/benches/common.rs +++ b/lib/vector-buffers/benches/common.rs @@ -6,7 +6,7 @@ use metrics_util::{debugging::DebuggingRecorder, layers::Layer}; use tracing::Span; use tracing_subscriber::prelude::__tracing_subscriber_SubscriberExt; use vector_buffers::{ - BufferType, EventCount, + BufferType, Bufferable, EventCount, encoding::FixedEncodable, topology::{ builder::TopologyBuilder, @@ -55,6 +55,8 @@ impl EventCount for Message { } } +impl Bufferable for Message {} + impl Finalizable for Message { fn take_finalizers(&mut self) -> EventFinalizers { Default::default() // This benchmark doesn't need finalization diff --git a/lib/vector-buffers/examples/buffer_perf.rs b/lib/vector-buffers/examples/buffer_perf.rs index 08f4a5b95d636..38aff2d132e35 100644 --- a/lib/vector-buffers/examples/buffer_perf.rs +++ b/lib/vector-buffers/examples/buffer_perf.rs @@ -69,6 +69,8 @@ impl EventCount for VariableMessage { } } +impl Bufferable for VariableMessage {} + impl Finalizable for VariableMessage { fn take_finalizers(&mut self) -> EventFinalizers { std::mem::take(&mut self.finalizers) diff --git a/lib/vector-buffers/src/lib.rs b/lib/vector-buffers/src/lib.rs index 5f6ade35f9c78..1d9774ead6190 100644 --- a/lib/vector-buffers/src/lib.rs +++ b/lib/vector-buffers/src/lib.rs @@ -103,10 +103,45 @@ impl InMemoryBufferable for T where /// An item that can be buffered. /// /// This supertrait serves as the base trait for any item that can be pushed into a buffer. -pub trait Bufferable: InMemoryBufferable + Encodable {} - -// Blanket implementation for anything that is already bufferable. -impl Bufferable for T where T: InMemoryBufferable + Encodable {} +pub trait Bufferable: InMemoryBufferable + Encodable { + /// Drops any sub-items that cannot be persisted by the calling backend (e.g. due to + /// format-imposed nesting depth limits), reporting them as dropped via the appropriate + /// telemetry. Returns `None` if nothing remains worth writing. + /// + /// # Who calls this + /// + /// Only persistent backends with wire-format constraints invoke this — today that's + /// the disk-v2 sender (`SenderAdapter::send`/`try_send`). In-memory channels skip it + /// entirely because they hold the in-memory representation and have no nesting-limit + /// risk. A new backend with similar constraints should call this in the same place + /// and surface the resulting `FilterDrops` to `BufferSender` so that buffer-usage + /// instrumentation stays consistent with what actually lands in the buffer. + /// + /// # Default behaviour + /// + /// The default returns `Some(self)` if the item carries any events, and `None` if + /// it is already empty. This means an item that arrives empty (`event_count() == 0`) + /// is silently *not* persisted — preserving the pre-existing + /// "don't write empty records to disk" behaviour the call site used to enforce. + /// Types whose owners want empty items to be persisted must override this. + /// + /// # Skipping this call + /// + /// If a persistent backend writes an item without first calling `filter_unencodable`, + /// any sub-item that exceeds the format's limits will surface as a hard + /// [`Encodable::encode`] error and the *entire* item is rejected — including any + /// sibling sub-items that would otherwise have encoded fine. The filter is the only + /// path that produces graceful per-item drop with telemetry and a `Rejected` event + /// status; the encode-level check exists purely as defense-in-depth to ensure a + /// corrupt record cannot reach disk if a future caller forgets to filter. + fn filter_unencodable(self) -> Option { + if self.event_count() > 0 { + Some(self) + } else { + None + } + } +} /// Hook for observing items as they are sent into a `BufferSender`. pub trait BufferInstrumentation: Send + Sync + 'static { diff --git a/lib/vector-buffers/src/test/messages.rs b/lib/vector-buffers/src/test/messages.rs index 0eb57b195dd37..dae87511d159b 100644 --- a/lib/vector-buffers/src/test/messages.rs +++ b/lib/vector-buffers/src/test/messages.rs @@ -7,7 +7,11 @@ use vector_common::{ finalization::{AddBatchNotifier, BatchNotifier, EventFinalizer, EventFinalizers, Finalizable}, }; -use crate::{EventCount, encoding::FixedEncodable}; +use crate::{Bufferable, EventCount, encoding::FixedEncodable}; + +impl Bufferable for SizedRecord {} +impl Bufferable for UndecodableRecord {} +impl Bufferable for MultiEventRecord {} macro_rules! message_wrapper { ($id:ident: $ty:ty, $event_count:expr) => { diff --git a/lib/vector-buffers/src/topology/channel/sender.rs b/lib/vector-buffers/src/topology/channel/sender.rs index c3eb22c3c952e..9c45b0de3c3c0 100644 --- a/lib/vector-buffers/src/topology/channel/sender.rs +++ b/lib/vector-buffers/src/topology/channel/sender.rs @@ -44,14 +44,26 @@ where match self { Self::InMemory(tx) => tx.send(item).await.map_err(Into::into), Self::DiskV2(writer) => { + let pre_count = item.event_count() as u64; + let pre_size = item.size_of() as u64; let mut writer = writer.lock().await; + let Some(item) = item.filter_unencodable() else { + // The whole item was filtered out (e.g. every sub-item over the + // protobuf nesting budget). Report the drop directly via the + // ledger's usage handle so it shows up in the disk-v2 stage's + // `received` / `dropped` metrics — `BufferSender` does not carry + // its own handle for backends that `provides_instrumentation()`. + writer.track_dropped(pre_count, pre_size); + return Ok(()); + }; + if item.event_count() as u64 != pre_count { + let dropped_events = pre_count - item.event_count() as u64; + let dropped_bytes = pre_size.saturating_sub(item.size_of() as u64); + writer.track_dropped(dropped_events, dropped_bytes); + } + writer.write_record(item).await.map(|_| ()).map_err(|e| { - // TODO: Could some errors be handled and not be unrecoverable? Right now, - // encoding should theoretically be recoverable -- encoded value was too big, or - // error during encoding -- but the traits don't allow for recovering the - // original event value because we have to consume it to do the encoding... but - // that might not always be the case. error!("Disk buffer writer has encountered an unrecoverable error."); e.into() @@ -62,19 +74,57 @@ where pub(crate) async fn try_send(&mut self, item: T) -> crate::Result> { match self { - Self::InMemory(tx) => tx + Self::InMemory(tx) => Ok(tx .try_send(item) - .map(|()| None) - .or_else(|e| Ok(Some(e.into_inner()))), + .err() + .map(super::limited_queue::TrySendError::into_inner)), Self::DiskV2(writer) => { let mut writer = writer.lock().await; + // If the disk buffer is already at its size limit, hand the item off + // to the caller unfiltered. The caller forwards it to the overflow + // stage in `WhenFull::Overflow` mode, and the overflow stage may be + // an in-memory buffer with no wire-format constraint — filtering + // here would needlessly drop sub-items that the overflow could + // accept. Holding the writer lock makes the check race-free against + // other writers (only writers grow the buffer; readers only shrink). + if writer.is_buffer_full() { + return Ok(Some(item)); + } + + // KNOWN LIMITATION (accepted; tracked as a follow-up): past the + // steady-state-full check above, over-budget sub-items are filtered + // and dropped here even in `WhenFull::Overflow`, so a non-protobuf + // overflow stage (e.g. in-memory) never gets the chance to accept + // them. This surfaces two ways: + // 1. the item is partially over-budget and `try_write_record` + // below then rejects the *remainder* for fullness — the + // overflow receives the item minus the already-dropped events; + // 2. the item is fully over-budget — `filter_unencodable` returns + // `None` and the whole item is dropped before any capacity + // check, so nothing overflows. + // Routing unencodable items by `WhenFull` (drop in Block/DropNewest, + // overflow otherwise) is a `BufferSender`-level policy decision, + // whereas filtering lives here in the backend; reconciling the two + // is deferred. The window is narrow and atypical: it requires a + // disk-v2 stage in `Overflow` mode (disk is normally the terminal + // Block stage), a non-protobuf overflow target, an over-budget + // event (>32 nesting levels), and a downstream egress that could + // actually deliver it. In other topologies these events are dropped + // a stage later regardless. + let pre_count = item.event_count() as u64; + let pre_size = item.size_of() as u64; + let Some(item) = item.filter_unencodable() else { + writer.track_dropped(pre_count, pre_size); + return Ok(None); + }; + if item.event_count() as u64 != pre_count { + let dropped_events = pre_count - item.event_count() as u64; + let dropped_bytes = pre_size.saturating_sub(item.size_of() as u64); + writer.track_dropped(dropped_events, dropped_bytes); + } + writer.try_write_record(item).await.map_err(|e| { - // TODO: Could some errors be handled and not be unrecoverable? Right now, - // encoding should theoretically be recoverable -- encoded value was too big, or - // error during encoding -- but the traits don't allow for recovering the - // original event value because we have to consume it to do the encoding... but - // that might not always be the case. error!("Disk buffer writer has encountered an unrecoverable error."); e.into() @@ -245,6 +295,11 @@ impl BufferSender { } } + // Backend filter drops are accounted directly through the backend's own + // usage handle (e.g. disk-v2's ledger), so they show up in the buffer + // stage's `received` / `dropped` metrics even when the `BufferSender` + // does not carry instrumentation. This block only reports fullness-driven + // drops captured via `was_dropped`. if let Some(instrumentation) = self.usage_instrumentation.as_ref() && let Some((item_count, item_size)) = item_sizing && was_dropped diff --git a/lib/vector-buffers/src/topology/test_util.rs b/lib/vector-buffers/src/topology/test_util.rs index 684aecc977764..65e9baa6c4636 100644 --- a/lib/vector-buffers/src/topology/test_util.rs +++ b/lib/vector-buffers/src/topology/test_util.rs @@ -120,6 +120,8 @@ impl EventCount for Sample { } } +impl Bufferable for Sample {} + #[derive(Debug)] #[allow(dead_code)] // The inner _is_ read by the `Debug` impl, but that's ignored pub struct BasicError(pub(crate) String); diff --git a/lib/vector-buffers/src/variants/disk_v2/ledger.rs b/lib/vector-buffers/src/variants/disk_v2/ledger.rs index e63fc5fb16973..b76c69e02166d 100644 --- a/lib/vector-buffers/src/variants/disk_v2/ledger.rs +++ b/lib/vector-buffers/src/variants/disk_v2/ledger.rs @@ -389,6 +389,20 @@ where .increment_received_event_count_and_byte_size(event_count, record_size); } + /// Tracks events that arrived at the buffer but were rejected before being + /// persisted (e.g. `Bufferable::filter_unencodable` dropping over-budget + /// sub-items). Bumps both `received` and the unintentional-`dropped` counter + /// on the usage handle so `buffer_size = received - sent - dropped` stays + /// consistent and operators can see the rejection in buffer-usage metrics. + /// `total_buffer_size` is intentionally left alone — these events never + /// reached disk. + pub fn track_dropped(&self, event_count: u64, byte_size: u64) { + self.usage_handle + .increment_received_event_count_and_byte_size(event_count, byte_size); + self.usage_handle + .increment_dropped_event_count_and_byte_size(event_count, byte_size, false); + } + /// Tracks the statistics of multiple successful reads. pub fn track_reads(&self, event_count: u64, total_record_size: u64) { self.decrement_total_buffer_size(total_record_size); diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/filter_metrics.rs b/lib/vector-buffers/src/variants/disk_v2/tests/filter_metrics.rs new file mode 100644 index 0000000000000..cc37762d0f9d6 --- /dev/null +++ b/lib/vector-buffers/src/variants/disk_v2/tests/filter_metrics.rs @@ -0,0 +1,180 @@ +//! Buffer-usage accounting around `Bufferable::filter_unencodable`. +//! +//! When the disk-v2 sender drops sub-items because they exceed protobuf nesting +//! limits, those drops must show up as unintentional buffer drops so that +//! `buffer_size_*` (received minus left) stays consistent with what is actually +//! queued on disk. Without that, a single rejected event makes the buffer report +//! one queued event forever. + +use std::{error, fmt}; + +use bytes::{Buf, BufMut}; +use vector_common::{ + byte_size_of::ByteSizeOf, + finalization::{AddBatchNotifier, BatchNotifier}, +}; + +use super::create_default_buffer_v2_with_usage; +use crate::{ + Bufferable, EventCount, WhenFull, + encoding::FixedEncodable, + test::{install_tracing_helpers, with_temp_dir}, + topology::channel::{BufferSender, SenderAdapter}, +}; + +/// A bufferable carrying a self-declared `event_count` of `events`, whose +/// `filter_unencodable` shrinks it to `post_filter` events (or drops it entirely +/// when `post_filter == 0`). Lets the test pin "before vs after filter" sizing +/// without needing the full `EventArray` machinery. +#[derive(Clone, Debug, PartialEq, Eq)] +struct FilterableBatch { + events: u32, + post_filter: u32, +} + +impl AddBatchNotifier for FilterableBatch { + fn add_batch_notifier(&mut self, batch: BatchNotifier) { + drop(batch); + } +} +impl ByteSizeOf for FilterableBatch { + fn allocated_bytes(&self) -> usize { + 0 + } +} +impl EventCount for FilterableBatch { + fn event_count(&self) -> usize { + self.events as usize + } +} + +#[derive(Debug)] +struct CodecError; +impl fmt::Display for CodecError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{self:?}") + } +} +impl error::Error for CodecError {} + +impl FixedEncodable for FilterableBatch { + type EncodeError = CodecError; + type DecodeError = CodecError; + fn encode(self, buf: &mut B) -> Result<(), Self::EncodeError> { + if buf.remaining_mut() < 8 { + return Err(CodecError); + } + buf.put_u32(self.events); + buf.put_u32(self.post_filter); + Ok(()) + } + fn decode(mut buf: B) -> Result { + Ok(FilterableBatch { + events: buf.get_u32(), + post_filter: buf.get_u32(), + }) + } + fn encoded_size(&self) -> Option { + Some(8) + } +} + +impl Bufferable for FilterableBatch { + fn filter_unencodable(self) -> Option { + if self.post_filter == 0 { + None + } else { + Some(FilterableBatch { + events: self.post_filter, + post_filter: self.post_filter, + }) + } + } +} + +/// A partial-filter drop on a disk-v2 send must show up as an unintentional buffer +/// drop, so `buffer_size_*` stays consistent with what actually landed on disk. +/// +/// Note: we deliberately do NOT attach `with_usage_instrumentation` to the +/// `BufferSender`. In production, `TopologyBuilder::build` skips that call for +/// disk-v2 because the stage `provides_instrumentation()` itself via the ledger. +/// This test reflects that production wiring: filter-drop accounting goes +/// through `Ledger::track_dropped`, and `usage` (returned by the helper) IS the +/// ledger's handle. +#[tokio::test] +async fn filter_drops_are_reported_as_unintentional_buffer_drops() { + let _a = install_tracing_helpers(); + + with_temp_dir(|dir| { + let data_dir = dir.to_path_buf(); + + async move { + let (writer, _reader, _ledger, usage) = + create_default_buffer_v2_with_usage::<_, FilterableBatch>(data_dir).await; + let mut sender = BufferSender::new(SenderAdapter::from(writer), WhenFull::Block); + + // 10 events arrive, filter keeps 3. Flush after each send so the + // ledger's `track_write` actually reaches the usage handle (it only + // fires when buffered writes are flushed to disk). + sender + .send( + FilterableBatch { + events: 10, + post_filter: 3, + }, + None, + ) + .await + .expect("send should succeed"); + sender.flush().await.expect("flush should succeed"); + + let snapshot = usage.snapshot(); + assert_eq!( + snapshot.received_event_count, 10, + "received counts both the 7 filter-dropped events (via track_dropped) \ + and the 3 events flushed to disk (via track_write)", + ); + assert_eq!( + snapshot.dropped_event_count, 7, + "filter drops show up under the disk-v2 stage's unintentional dropped count \ + so buffer_size stays consistent (received - sent - dropped = 3 queued)", + ); + assert_eq!( + snapshot.dropped_event_count_intentional, 0, + "no buffer-fullness drops here", + ); + + // 5 events arrive, filter drops them all (nothing reaches disk). + sender + .send( + FilterableBatch { + events: 5, + post_filter: 0, + }, + None, + ) + .await + .expect("send should succeed"); + + let snapshot = usage.snapshot(); + assert_eq!( + snapshot.received_event_count, 15, + "fully-filtered item still bumps received via track_dropped", + ); + assert_eq!( + snapshot.dropped_event_count, 12, + "all 5 events from the fully-filtered item are reported as unintentional drops", + ); + } + }) + .await; +} + +// Note: A regression test that exercises the "full disk hands item to overflow +// unfiltered" path is not included here because reliably driving the disk-v2 +// writer's `is_buffer_full()` to `true` under the minimum-size config takes +// careful tuning of record/buffer sizes (the writer's `can_write_record` check +// generally short-circuits writes *before* `total_buffer_size` reaches +// `max_buffer_size`). The fix in `SenderAdapter::try_send` is a single +// `is_buffer_full()` short-circuit before the filter runs; the existing +// disk-v2 tests cover the full-buffer behaviour at the writer level. diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/known_errors.rs b/lib/vector-buffers/src/variants/disk_v2/tests/known_errors.rs index 9f5ec073c00a7..89da8124c3ff2 100644 --- a/lib/vector-buffers/src/variants/disk_v2/tests/known_errors.rs +++ b/lib/vector-buffers/src/variants/disk_v2/tests/known_errors.rs @@ -19,8 +19,9 @@ use vector_common::{ use super::{create_buffer_v2_with_max_data_file_size, create_default_buffer_v2}; use crate::{ - EventCount, assert_buffer_size, assert_enough_bytes_written, assert_file_does_not_exist_async, - assert_file_exists_async, assert_reader_writer_v2_file_positions, await_timeout, + Bufferable, EventCount, assert_buffer_size, assert_enough_bytes_written, + assert_file_does_not_exist_async, assert_file_exists_async, + assert_reader_writer_v2_file_positions, await_timeout, encoding::{AsMetadata, Encodable}, test::{SizedRecord, UndecodableRecord, acknowledge, install_tracing_helpers, with_temp_dir}, variants::disk_v2::{ReaderError, backed_archive::BackedArchive, record::Record}, @@ -744,6 +745,8 @@ async fn reader_throws_error_when_record_is_undecodable_via_metadata() { } } + impl Bufferable for ControllableRecord {} + with_temp_dir(|dir| { let data_dir = dir.to_path_buf(); diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/mod.rs b/lib/vector-buffers/src/variants/disk_v2/tests/mod.rs index fa8ac8b11ad23..7d260e3c856aa 100644 --- a/lib/vector-buffers/src/variants/disk_v2/tests/mod.rs +++ b/lib/vector-buffers/src/variants/disk_v2/tests/mod.rs @@ -24,6 +24,7 @@ type FilesystemUnderTest = ProductionFilesystem; mod acknowledgements; mod basic; +mod filter_metrics; mod initialization; mod invariants; mod known_errors; diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs b/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs index 914ffa2759de8..632cded82216f 100644 --- a/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs +++ b/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs @@ -7,11 +7,13 @@ use vector_common::{ }; use crate::{ - EventCount, + Bufferable, EventCount, encoding::FixedEncodable, variants::disk_v2::{record::RECORD_HEADER_LEN, tests::align16}, }; +impl Bufferable for Record {} + #[derive(Debug)] pub struct EncodeError; diff --git a/lib/vector-buffers/src/variants/disk_v2/writer.rs b/lib/vector-buffers/src/variants/disk_v2/writer.rs index f12ce6ca94dae..a0eaa59e61654 100644 --- a/lib/vector-buffers/src/variants/disk_v2/writer.rs +++ b/lib/vector-buffers/src/variants/disk_v2/writer.rs @@ -990,12 +990,28 @@ where Ok(()) } - fn is_buffer_full(&self) -> bool { + /// Returns whether the buffer is currently at or above its configured size limit. + /// + /// Exposed so callers can decide what to do with an item *before* it is encoded — + /// notably, an over-budget `EventArray` headed for a `WhenFull::Overflow` topology + /// should be handed to the overflow stage unfiltered rather than have its sub-items + /// pruned for a write that will never happen. + pub(crate) fn is_buffer_full(&self) -> bool { let total_buffer_size = self.ledger.get_total_buffer_size() + self.unflushed_bytes; let max_buffer_size = self.config.max_buffer_size; total_buffer_size >= max_buffer_size } + /// Records sub-items that arrived at the buffer but were dropped before + /// reaching disk (e.g. `Bufferable::filter_unencodable` rejecting events + /// the protobuf decoder cannot handle). Delegates to the ledger's usage + /// handle so the rejection shows up under the disk-v2 stage's + /// `received` / `dropped` metrics in production, where the + /// `BufferSender` does not carry its own usage instrumentation. + pub(crate) fn track_dropped(&self, event_count: u64, byte_size: u64) { + self.ledger.track_dropped(event_count, byte_size); + } + /// Ensures this writer is ready to attempt writer the next record. #[instrument(skip(self), level = "debug")] async fn ensure_ready_for_write(&mut self) -> io::Result<()> { diff --git a/lib/vector-core/src/event/mod.rs b/lib/vector-core/src/event/mod.rs index 9d72301e85804..a77d4d453b44b 100644 --- a/lib/vector-core/src/event/mod.rs +++ b/lib/vector-core/src/event/mod.rs @@ -10,6 +10,9 @@ pub use log_event::LogEvent; pub use metadata::{DatadogMetricOriginMetadata, EventMetadata, WithMetadata}; pub use metric::{Metric, MetricKind, MetricTags, MetricValue, StatisticKind}; pub use r#ref::{EventMutRef, EventRef}; +pub use ser::{ + MAX_METADATA_VALUE_NESTING_FRAMES, MAX_VALUE_NESTING_FRAMES, event_exceeds_max_nesting_cost, +}; use serde::{Deserialize, Serialize}; pub use trace::TraceEvent; use vector_buffers::EventCount; diff --git a/lib/vector-core/src/event/ser.rs b/lib/vector-core/src/event/ser.rs index a456ccf69368f..f8ef2ee5c4017 100644 --- a/lib/vector-core/src/event/ser.rs +++ b/lib/vector-core/src/event/ser.rs @@ -2,14 +2,185 @@ use bytes::{Buf, BufMut}; use enumflags2::{BitFlags, FromBitsError, bitflags}; use prost::Message; use snafu::Snafu; -use vector_buffers::encoding::{AsMetadata, Encodable}; +use vector_buffers::{ + Bufferable, EventCount, + encoding::{AsMetadata, Encodable}, +}; +use vector_common::internal_event::{self, ComponentEventsDropped, UNINTENTIONAL}; +use vrl::value::Value; -use super::{Event, EventArray, proto}; +use super::{Event, EventArray, EventStatus, proto}; + +/// Per-level prost recursion frame cost of an [`Value::Object`]. +/// +/// Decoding an object level walks `Value → ValueMap → map_entry (synthetic) → Value`, +/// adding three message-decode frames before reaching the child Value. +pub(crate) const OBJECT_FRAME_COST: usize = 3; + +/// Per-level prost recursion frame cost of an [`Value::Array`]. +/// +/// Decoding an array level walks `Value → ValueArray → Value`, adding two message-decode +/// frames before reaching the child Value. +pub(crate) const ARRAY_FRAME_COST: usize = 2; + +/// Per-leaf prost recursion frame cost of a [`Value::Timestamp`]. +/// +/// Unlike other scalar variants, `Value::Timestamp` is encoded as a nested +/// `google.protobuf.Timestamp` message, so decoding it consumes one additional frame +/// beyond the enclosing `Value`. Without this cost a timestamp leaf at the deepest +/// allowed branch (event-data object depth 33 or metadata object depth 32) sneaks past +/// the gate and trips prost's recursion limit on decode. +pub(crate) const TIMESTAMP_FRAME_COST: usize = 1; + +/// Maximum prost recursion frame cost for event data values (`Log.fields`, `Trace.fields`). +/// +/// Prost enforces a decode recursion limit of 100 (no limit on encode). Each nesting level +/// consumes 3 frames for [`Value::Object`], 2 for [`Value::Array`], or 1 for a +/// [`Value::Timestamp`] leaf, plus a fixed overhead for the proto wrappers outside the +/// Value tree. The event data path (`EventArray` → `*Array` → Event → fields) has fewer +/// wrappers than the metadata path, allowing a higher frame budget. +/// +/// Object-only depth 33 (cost 99) roundtrips; depth 34 (cost 102) fails decode. Array-only +/// nesting is correspondingly looser: depth 49 (cost 98) is the highest that fits. A +/// `Value::Timestamp` leaf added at depth 33 raises the cost to 100 and fails decode. +pub const MAX_VALUE_NESTING_FRAMES: usize = 99; + +/// Maximum prost recursion frame cost for event metadata values (via `metadata_full`). +/// +/// The metadata path (`EventArray` → `*Array` → Event → `Metadata` → Value) has one more +/// proto wrapper message than the event data path due to the `Metadata` message, reducing +/// the safe budget by 3 frames. +/// +/// Object-only depth 32 (cost 96) roundtrips; depth 33 (cost 99) fails decode. A +/// `Value::Timestamp` leaf added at depth 32 raises the cost to 97 and fails decode. +pub const MAX_METADATA_VALUE_NESTING_FRAMES: usize = 96; + +/// Walks a [`Value`] tree accumulating prost recursion frame cost, returning +/// `Err(over_budget_cost)` as soon as any branch exceeds `budget`. +/// +/// Object levels weigh [`OBJECT_FRAME_COST`] frames each, array levels weigh +/// [`ARRAY_FRAME_COST`], and timestamp leaves weigh [`TIMESTAMP_FRAME_COST`] (because +/// they decode into a nested `google.protobuf.Timestamp` message); other scalar leaves +/// are free. Performs an early-exit traversal so well-formed events incur a single +/// descent of the deepest branch only. +/// +/// # Errors +/// +/// Returns `Err(actual_cost)` if any branch's cumulative frame cost exceeds `budget`. +pub(crate) fn check_value_nesting_cost( + value: &Value, + accumulated: usize, + budget: usize, +) -> Result<(), usize> { + let level_cost = match value { + Value::Object(_) => OBJECT_FRAME_COST, + Value::Array(_) => ARRAY_FRAME_COST, + Value::Timestamp(_) => TIMESTAMP_FRAME_COST, + _ => 0, + }; + let next = accumulated + level_cost; + if next > budget { + return Err(next); + } + match value { + Value::Object(map) => { + for v in map.values() { + check_value_nesting_cost(v, next, budget)?; + } + } + Value::Array(arr) => { + for v in arr { + check_value_nesting_cost(v, next, budget)?; + } + } + _ => {} + } + Ok(()) +} + +/// Checks whether an event's nesting frame cost exceeds the safe limits for protobuf encoding. +/// +/// Returns `Some((cost, budget))` identifying the path that violated its budget, or `None` +/// if the event is within bounds. +/// +/// Event data values (Log.fields, Trace.fields) are checked against +/// [`MAX_VALUE_NESTING_FRAMES`], while metadata values are checked against the stricter +/// [`MAX_METADATA_VALUE_NESTING_FRAMES`] because the `Metadata` proto message adds an +/// extra wrapper layer. +/// +/// For metrics, only metadata is checked since metric values have a fixed structure. +pub fn event_exceeds_max_nesting_cost(event: &Event) -> Option<(usize, usize)> { + match event { + Event::Log(log) => check_value_nesting_cost(log.value(), 0, MAX_VALUE_NESTING_FRAMES) + .map_err(|cost| (cost, MAX_VALUE_NESTING_FRAMES)) + .and_then(|()| { + check_value_nesting_cost( + log.metadata().value(), + 0, + MAX_METADATA_VALUE_NESTING_FRAMES, + ) + .map_err(|cost| (cost, MAX_METADATA_VALUE_NESTING_FRAMES)) + }) + .err(), + Event::Trace(trace) => check_value_nesting_cost(trace.value(), 0, MAX_VALUE_NESTING_FRAMES) + .map_err(|cost| (cost, MAX_VALUE_NESTING_FRAMES)) + .and_then(|()| { + check_value_nesting_cost( + trace.metadata().value(), + 0, + MAX_METADATA_VALUE_NESTING_FRAMES, + ) + .map_err(|cost| (cost, MAX_METADATA_VALUE_NESTING_FRAMES)) + }) + .err(), + Event::Metric(metric) => check_value_nesting_cost( + metric.metadata().value(), + 0, + MAX_METADATA_VALUE_NESTING_FRAMES, + ) + .map_err(|cost| (cost, MAX_METADATA_VALUE_NESTING_FRAMES)) + .err(), + } +} + +/// Checks all events in an `EventArray` for nesting cost violations. +/// +/// Event data is checked against [`MAX_VALUE_NESTING_FRAMES`] and metadata against +/// [`MAX_METADATA_VALUE_NESTING_FRAMES`]. For metrics, only metadata is checked since +/// metric values have a fixed structure. +fn check_event_array_nesting_cost(events: &EventArray) -> Result<(), EncodeError> { + let check = |value: &Value, budget: usize| { + check_value_nesting_cost(value, 0, budget) + .map_err(|cost| EncodeError::NestingTooDeep { cost, budget }) + }; + match events { + EventArray::Logs(logs) => { + for log in logs { + check(log.value(), MAX_VALUE_NESTING_FRAMES)?; + check(log.metadata().value(), MAX_METADATA_VALUE_NESTING_FRAMES)?; + } + } + EventArray::Traces(traces) => { + for trace in traces { + check(trace.value(), MAX_VALUE_NESTING_FRAMES)?; + check(trace.metadata().value(), MAX_METADATA_VALUE_NESTING_FRAMES)?; + } + } + EventArray::Metrics(metrics) => { + for metric in metrics { + check(metric.metadata().value(), MAX_METADATA_VALUE_NESTING_FRAMES)?; + } + } + } + Ok(()) +} #[derive(Debug, Snafu)] pub enum EncodeError { #[snafu(display("the provided buffer was too small to fully encode this item"))] BufferTooSmall, + #[snafu(display("event nesting cost {cost} exceeds protobuf budget of {budget}"))] + NestingTooDeep { cost: usize, budget: usize }, } #[derive(Debug, Snafu)] @@ -91,10 +262,29 @@ impl Encodable for EventArray { metadata.contains(EventEncodableMetadataFlags::DiskBufferV1CompatibilityMode) } + /// # Errors + /// + /// Returns `EncodeError::NestingTooDeep` if any contained event's value or metadata + /// exceeds the per-path frame budget ([`MAX_VALUE_NESTING_FRAMES`] / + /// [`MAX_METADATA_VALUE_NESTING_FRAMES`]). This is **all-or-nothing**: a single + /// over-budget event fails the entire batch, because a partially-encoded + /// `EventArray` reaching disk would trip prost's recursion limit on decode and + /// corrupt the buffer. + /// + /// Callers that want graceful per-item drop with telemetry and + /// `EventStatus::Rejected` must run [`Bufferable::filter_unencodable`] first. + /// `SenderAdapter::send`/`try_send` already does this on the disk-v2 path, so the + /// `NestingTooDeep` arm is unreachable from any current production call site — it + /// is defense-in-depth for a future caller that bypasses `SenderAdapter`. + /// + /// Returns `EncodeError::BufferTooSmall` if the buffer cannot hold the encoded + /// output. fn encode(self, buffer: &mut B) -> Result<(), Self::EncodeError> where B: BufMut, { + check_event_array_nesting_cost(&self)?; + proto::EventArray::from(self) .encode(buffer) .map_err(|_| EncodeError::BufferTooSmall) @@ -117,3 +307,56 @@ impl Encodable for EventArray { } } } + +impl Bufferable for EventArray { + fn filter_unencodable(self) -> Option { + let exceeds = + |value: &Value, budget: usize| check_value_nesting_cost(value, 0, budget).is_err(); + let mut dropped = 0; + let filtered = match self { + EventArray::Logs(mut logs) => { + logs.retain(|log| { + let too_deep = exceeds(log.value(), MAX_VALUE_NESTING_FRAMES) + || exceeds(log.metadata().value(), MAX_METADATA_VALUE_NESTING_FRAMES); + if too_deep { + log.metadata().update_status(EventStatus::Rejected); + dropped += 1; + } + !too_deep + }); + EventArray::Logs(logs) + } + EventArray::Traces(mut traces) => { + traces.retain(|trace| { + let too_deep = exceeds(trace.value(), MAX_VALUE_NESTING_FRAMES) + || exceeds(trace.metadata().value(), MAX_METADATA_VALUE_NESTING_FRAMES); + if too_deep { + trace.metadata().update_status(EventStatus::Rejected); + dropped += 1; + } + !too_deep + }); + EventArray::Traces(traces) + } + EventArray::Metrics(mut metrics) => { + metrics.retain(|metric| { + let too_deep = + exceeds(metric.metadata().value(), MAX_METADATA_VALUE_NESTING_FRAMES); + if too_deep { + metric.metadata().update_status(EventStatus::Rejected); + dropped += 1; + } + !too_deep + }); + EventArray::Metrics(metrics) + } + }; + if dropped > 0 { + internal_event::emit(ComponentEventsDropped:: { + count: dropped, + reason: "Event nesting cost exceeds maximum for protobuf encoding.", + }); + } + (filtered.event_count() > 0).then_some(filtered) + } +} diff --git a/lib/vector-core/src/event/test/serialization.rs b/lib/vector-core/src/event/test/serialization.rs index 75304e3960d54..3cedfc5c1051b 100644 --- a/lib/vector-core/src/event/test/serialization.rs +++ b/lib/vector-core/src/event/test/serialization.rs @@ -1,4 +1,6 @@ use bytes::{Buf, BufMut, BytesMut}; +use chrono::TimeZone; +use prost::Message; use quickcheck::{QuickCheck, TestResult}; use regex::Regex; use similar_asserts::assert_eq; @@ -6,6 +8,12 @@ use vector_buffers::encoding::Encodable; use super::*; use crate::config::log_schema; +use crate::event::event_exceeds_max_nesting_cost; +use crate::event::ser::{ + ARRAY_FRAME_COST, MAX_METADATA_VALUE_NESTING_FRAMES, MAX_VALUE_NESTING_FRAMES, + OBJECT_FRAME_COST, TIMESTAMP_FRAME_COST, check_value_nesting_cost, +}; +use vector_buffers::Bufferable; fn encode_value(value: T, buffer: &mut B) { value.encode(buffer).expect("encoding should not fail"); @@ -96,3 +104,684 @@ fn type_serialization() { assert_eq!(map["bool"], json!(true)); assert_eq!(map["string"], json!("thisisastring")); } + +// --------------------------------------------------------------------------- +// Nesting validation tests +// --------------------------------------------------------------------------- +// +// Prost enforces a decode recursion limit of 100 (no limit on encode). Each nesting +// level consumes a path-dependent number of prost recursion frames: +// +// - `Value::Object` level: Value + ValueMap + map_entry = 3 frames +// - `Value::Array` level: Value + ValueArray = 2 frames +// +// Each encoding path has a fixed proto-wrapper overhead before the Value tree starts: +// +// - Event data path (Log.fields, Trace.fields): frame budget MAX_VALUE_NESTING_FRAMES (99) +// - Metadata path (metadata_full): frame budget MAX_METADATA_VALUE_NESTING_FRAMES (96) +// +// The `per_path_boundaries` test verifies both budgets empirically via prost roundtrip. +// +// The saturated-event tests create events with ALL Value-carrying fields at their +// respective max frame cost simultaneously. The proto conversion code populates every +// field (including deprecated ones like Log.metadata), so a single roundtrip per event +// type covers every proto path automatically. + +/// Maximum number of object-only nesting levels that fit the event-data frame budget. +const MAX_OBJECT_DEPTH_VALUE: usize = MAX_VALUE_NESTING_FRAMES / OBJECT_FRAME_COST; + +/// Maximum number of object-only nesting levels that fit the metadata frame budget. +const MAX_OBJECT_DEPTH_METADATA: usize = MAX_METADATA_VALUE_NESTING_FRAMES / OBJECT_FRAME_COST; + +/// Maximum number of array-only nesting levels that fit the event-data frame budget. +const MAX_ARRAY_DEPTH_VALUE: usize = MAX_VALUE_NESTING_FRAMES / ARRAY_FRAME_COST; + +/// Maximum number of array-only nesting levels that fit the metadata frame budget. +const MAX_ARRAY_DEPTH_METADATA: usize = MAX_METADATA_VALUE_NESTING_FRAMES / ARRAY_FRAME_COST; + +/// Creates a Value with the specified number of nested Object wrapping levels. +/// +/// Returns a Value that is `wrapping_levels` nested Objects deep, with a string leaf. +fn create_nested_value(wrapping_levels: usize) -> Value { + let mut value = Value::from("innermost"); + for _ in 0..wrapping_levels { + let mut map = ObjectMap::new(); + map.insert("nested".into(), value); + value = Value::Object(map); + } + value +} + +/// Creates a Value with the specified number of nested Array wrapping levels. +fn create_nested_array(wrapping_levels: usize) -> Value { + let mut value = Value::from("innermost"); + for _ in 0..wrapping_levels { + value = Value::Array(vec![value]); + } + value +} + +/// Creates a Value with the specified number of nested Object wrapping levels around +/// the supplied leaf. Used to probe leaf-specific frame costs (e.g. `Value::Timestamp`). +fn create_nested_value_with_leaf(wrapping_levels: usize, leaf: Value) -> Value { + let mut value = leaf; + for _ in 0..wrapping_levels { + let mut map = ObjectMap::new(); + map.insert("nested".into(), value); + value = Value::Object(map); + } + value +} + +/// A fixed [`Value::Timestamp`] for use as a leaf in nesting tests. +fn ts_leaf() -> Value { + Value::Timestamp( + chrono::Utc + .timestamp_opt(1_700_000_000, 0) + .single() + .unwrap(), + ) +} + +/// Create a [`LogEvent`] with event data at `value_depth` and metadata at `metadata_depth`. +fn create_saturated_log(value_depth: usize, metadata_depth: usize) -> LogEvent { + let mut event = LogEvent::default(); + event.insert("data", create_nested_value(value_depth - 1)); + *event.metadata_mut().value_mut() = create_nested_value(metadata_depth); + event +} + +/// Create a [`TraceEvent`] with event data at `value_depth` and metadata at `metadata_depth`. +fn create_saturated_trace(value_depth: usize, metadata_depth: usize) -> TraceEvent { + let mut trace = TraceEvent::default(); + trace.insert("data", create_nested_value(value_depth - 1)); + *trace.metadata_mut().value_mut() = create_nested_value(metadata_depth); + trace +} + +/// Create a Metric with metadata at `metadata_depth`. +/// (Metric values have fixed structure — only metadata carries arbitrary Values.) +fn create_saturated_metric(metadata_depth: usize) -> Metric { + let mut metric = Metric::new( + "test", + MetricKind::Incremental, + MetricValue::Counter { value: 1.0 }, + ); + *metric.metadata_mut().value_mut() = create_nested_value(metadata_depth); + metric +} + +/// Build all three `EventArray` variants with each field at its respective max depth. +fn saturated_event_arrays( + value_depth: usize, + metadata_depth: usize, +) -> Vec<(&'static str, EventArray)> { + vec![ + ( + "Log", + EventArray::Logs(LogArray::from(vec![create_saturated_log( + value_depth, + metadata_depth, + )])), + ), + ( + "Trace", + EventArray::Traces(TraceArray::from(vec![create_saturated_trace( + value_depth, + metadata_depth, + )])), + ), + ( + "Metric", + EventArray::Metrics(MetricArray::from(vec![create_saturated_metric( + metadata_depth, + )])), + ), + ] +} + +/// Build all three Event variants for `EventWrapper` encoding. +fn saturated_events(value_depth: usize, metadata_depth: usize) -> Vec<(&'static str, Event)> { + vec![ + ( + "Log", + Event::Log(create_saturated_log(value_depth, metadata_depth)), + ), + ( + "Trace", + Event::Trace(create_saturated_trace(value_depth, metadata_depth)), + ), + ( + "Metric", + Event::Metric(create_saturated_metric(metadata_depth)), + ), + ] +} + +/// Verify the frame budgets are exactly right: all event types roundtrip at the +/// max object-only depth, and at least one fails prost decode when either budget +/// is exceeded. +#[test] +fn max_nesting_budgets_are_correct() { + let max_val = MAX_OBJECT_DEPTH_VALUE; + let max_meta = MAX_OBJECT_DEPTH_METADATA; + + // --- Both budgets at max must roundtrip for all event types --- + + for (name, array) in saturated_event_arrays(max_val, max_meta) { + let proto_array = proto::EventArray::from(array); + let mut buf = BytesMut::with_capacity(65536); + proto_array.encode(&mut buf).unwrap(); + assert!( + proto::EventArray::decode(buf.freeze()).is_ok(), + "EventArray decode FAILED for {name} at value depth {max_val}, metadata depth {max_meta}.", + ); + } + + for (name, event) in saturated_events(max_val, max_meta) { + let wrapper = proto::EventWrapper::from(event); + let mut buf = BytesMut::with_capacity(65536); + wrapper.encode(&mut buf).unwrap(); + assert!( + proto::EventWrapper::decode(buf.freeze()).is_ok(), + "EventWrapper decode FAILED for {name} at value depth {max_val}, metadata depth {max_meta}.", + ); + } + + // --- Exceeding either budget must fail for at least one event type --- + + // Exceed value budget + let any_fails = saturated_event_arrays(max_val + 1, max_meta) + .into_iter() + .any(|(_, array)| { + let proto_array = proto::EventArray::from(array); + let mut buf = BytesMut::with_capacity(65536); + proto_array.encode(&mut buf).unwrap(); + proto::EventArray::decode(buf.freeze()).is_err() + }); + assert!( + any_fails, + "No path failed at object value depth {}. MAX_VALUE_NESTING_FRAMES could be raised.", + max_val + 1 + ); + + // Exceed metadata budget + let any_fails = saturated_event_arrays(max_val, max_meta + 1) + .into_iter() + .any(|(_, array)| { + let proto_array = proto::EventArray::from(array); + let mut buf = BytesMut::with_capacity(65536); + proto_array.encode(&mut buf).unwrap(); + proto::EventArray::decode(buf.freeze()).is_err() + }); + assert!( + any_fails, + "No path failed at object metadata depth {}. MAX_METADATA_VALUE_NESTING_FRAMES could be raised.", + max_meta + 1 + ); +} + +/// Verify the nesting gate accepts all event types at the max object-only depth. +#[test] +fn nesting_gate_accepts_all_types_at_max_depth() { + for (name, array) in saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE, MAX_OBJECT_DEPTH_METADATA) { + let mut buf = BytesMut::with_capacity(65536); + assert!( + array.encode(&mut buf).is_ok(), + "nesting gate rejected {name} at max object depths", + ); + } +} + +/// Verify the nesting gate rejects when either object-only budget is exceeded. +#[test] +fn nesting_gate_rejects_above_max_depth() { + // Exceed value budget (Log and Trace have event data; Metric does not) + for (name, array) in + saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE + 1, MAX_OBJECT_DEPTH_METADATA) + { + // Metric has no event data field, so it won't be rejected here + if name == "Metric" { + continue; + } + let mut buf = BytesMut::with_capacity(65536); + assert!( + matches!( + array.encode(&mut buf), + Err(super::super::ser::EncodeError::NestingTooDeep { .. }) + ), + "nesting gate should reject {name} at object value depth {}", + MAX_OBJECT_DEPTH_VALUE + 1, + ); + } + + // Exceed metadata budget + for (name, array) in + saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE, MAX_OBJECT_DEPTH_METADATA + 1) + { + let mut buf = BytesMut::with_capacity(65536); + assert!( + matches!( + array.encode(&mut buf), + Err(super::super::ser::EncodeError::NestingTooDeep { .. }) + ), + "nesting gate should reject {name} at object metadata depth {}", + MAX_OBJECT_DEPTH_METADATA + 1, + ); + } +} + +/// Verify the per-path prost boundaries match the budgets for both object-only and +/// array-only nesting. +/// +/// Object-only `Log.fields`: depth 33 succeeds, 34 fails. +/// Object-only `metadata_full`: depth 32 succeeds, 33 fails. +/// Array-only `Log.fields`: depth 49 succeeds, 50 fails. +/// Array-only `metadata_full`: depth 48 succeeds, 49 fails. +#[test] +fn per_path_boundaries() { + let roundtrip_value = |value: Value| -> bool { + let mut event = LogEvent::default(); + event.insert("data", value); + let array = EventArray::Logs(LogArray::from(vec![event])); + let proto_array = proto::EventArray::from(array); + let mut buf = BytesMut::with_capacity(65536); + proto_array.encode(&mut buf).unwrap(); + proto::EventArray::decode(buf.freeze()).is_ok() + }; + + let roundtrip_metadata = |value: Value| -> bool { + let mut event = LogEvent::from("flat"); + *event.metadata_mut().value_mut() = value; + let array = EventArray::Logs(LogArray::from(vec![event])); + let proto_array = proto::EventArray::from(array); + let mut buf = BytesMut::with_capacity(65536); + proto_array.encode(&mut buf).unwrap(); + proto::EventArray::decode(buf.freeze()).is_ok() + }; + + // Object-only Log.fields: the "data" key contributes one level on top of the inner + // nested value, so we subtract one when building the value. + assert!( + roundtrip_value(create_nested_value(MAX_OBJECT_DEPTH_VALUE - 1)), + "Log.fields should succeed at object depth {MAX_OBJECT_DEPTH_VALUE}" + ); + assert!( + !roundtrip_value(create_nested_value(MAX_OBJECT_DEPTH_VALUE)), + "Log.fields should fail at object depth {}", + MAX_OBJECT_DEPTH_VALUE + 1 + ); + + // Object-only metadata_full: metadata Value is the root, no key on top. + assert!( + roundtrip_metadata(create_nested_value(MAX_OBJECT_DEPTH_METADATA)), + "metadata_full should succeed at object depth {MAX_OBJECT_DEPTH_METADATA}" + ); + assert!( + !roundtrip_metadata(create_nested_value(MAX_OBJECT_DEPTH_METADATA + 1)), + "metadata_full should fail at object depth {}", + MAX_OBJECT_DEPTH_METADATA + 1 + ); + + // Array-only Log.fields: array contributes 2 frames per level, so it fits more levels. + assert!( + roundtrip_value(create_nested_array(MAX_ARRAY_DEPTH_VALUE - 1)), + "Log.fields should succeed at array depth {MAX_ARRAY_DEPTH_VALUE}" + ); + assert!( + !roundtrip_value(create_nested_array(MAX_ARRAY_DEPTH_VALUE)), + "Log.fields should fail at array depth {}", + MAX_ARRAY_DEPTH_VALUE + 1 + ); + + // Array-only metadata_full + assert!( + roundtrip_metadata(create_nested_array(MAX_ARRAY_DEPTH_METADATA)), + "metadata_full should succeed at array depth {MAX_ARRAY_DEPTH_METADATA}" + ); + assert!( + !roundtrip_metadata(create_nested_array(MAX_ARRAY_DEPTH_METADATA + 1)), + "metadata_full should fail at array depth {}", + MAX_ARRAY_DEPTH_METADATA + 1 + ); +} + +/// Verify that array-only nesting deeper than the object-only cap (33) is accepted by +/// the gate — this is the regression that the frame-cost check addresses. Previously a +/// uniform depth-33 cap dropped array-only events that prost would happily roundtrip. +#[test] +fn nesting_gate_accepts_deep_array_nesting() { + // An array depth 40 = 80 frames, comfortably under the 99-frame value budget but well + // over the 33-depth limit the old uniform check would have applied. + let mut event = LogEvent::default(); + event.insert("data", create_nested_array(40)); + let array = EventArray::Logs(LogArray::from(vec![event])); + let mut buf = BytesMut::with_capacity(65536); + assert!( + array.encode(&mut buf).is_ok(), + "nesting gate should accept array-only nesting at depth 40", + ); +} + +/// Verify the gate correctly accounts for mixed array/object nesting via the per-variant +/// frame weights. Uses the metadata path because it has no outer wrapping object, making +/// the arithmetic match the inserted Value's cost directly. +#[test] +fn nesting_gate_handles_mixed_array_object_nesting() { + // Alternating levels (innermost-Array, then Object, then Array, ...). For N levels, + // cost = ceil(N/2)*ARRAY_FRAME_COST + floor(N/2)*OBJECT_FRAME_COST. + let build_alternating = |total_levels: usize| -> Value { + let mut value = Value::from("leaf"); + for i in 0..total_levels { + if i.is_multiple_of(2) { + value = Value::Array(vec![value]); + } else { + let mut map = ObjectMap::new(); + map.insert("k".into(), value); + value = Value::Object(map); + } + } + value + }; + + // 38 alternating levels: 19 array (cost 38) + 19 object (cost 57) = 95 frames. + // Under the metadata budget of 96. Fits. + let mut event = LogEvent::from("flat"); + *event.metadata_mut().value_mut() = build_alternating(38); + let array = EventArray::Logs(LogArray::from(vec![event])); + let mut buf = BytesMut::with_capacity(65536); + assert!( + array.encode(&mut buf).is_ok(), + "nesting gate should accept 38 alternating metadata levels (cost 95)", + ); + + // 39 alternating levels: 20 array (cost 40) + 19 object (cost 57) = 97 frames. + // Over the metadata budget of 96. Fails. + let mut event = LogEvent::from("flat"); + *event.metadata_mut().value_mut() = build_alternating(39); + let array = EventArray::Logs(LogArray::from(vec![event])); + let mut buf = BytesMut::with_capacity(65536); + assert!( + matches!( + array.encode(&mut buf), + Err(super::super::ser::EncodeError::NestingTooDeep { .. }) + ), + "nesting gate should reject 39 alternating metadata levels (cost 97)", + ); +} + +/// Verify the gate rejects a `Value::Timestamp` leaf sitting at the deepest object +/// position the budget would otherwise allow, and that the underlying proto roundtrip +/// would in fact fail there — confirming the timestamp leaf is not free. +#[test] +fn nesting_gate_rejects_timestamp_leaf_at_max_object_depth() { + let roundtrip_log = |value: Value| -> bool { + let mut event = LogEvent::default(); + event.insert("data", value); + let array = EventArray::Logs(LogArray::from(vec![event])); + let proto_array = proto::EventArray::from(array); + let mut buf = BytesMut::with_capacity(65536); + proto_array.encode(&mut buf).unwrap(); + proto::EventArray::decode(buf.freeze()).is_ok() + }; + let roundtrip_metadata = |value: Value| -> bool { + let mut event = LogEvent::from("flat"); + *event.metadata_mut().value_mut() = value; + let array = EventArray::Logs(LogArray::from(vec![event])); + let proto_array = proto::EventArray::from(array); + let mut buf = BytesMut::with_capacity(65536); + proto_array.encode(&mut buf).unwrap(); + proto::EventArray::decode(buf.freeze()).is_ok() + }; + + // Event data: at object depth 33, a Bytes leaf decodes but a Timestamp leaf does not, + // because the Timestamp message consumes one more recursion frame. + let event_data_ts = create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE - 1, ts_leaf()); + assert!( + !roundtrip_log(event_data_ts.clone()), + "depth {MAX_OBJECT_DEPTH_VALUE} with Timestamp leaf is expected to fail prost decode" + ); + + let mut event = LogEvent::default(); + event.insert("data", event_data_ts); + let array = EventArray::Logs(LogArray::from(vec![event])); + let mut buf = BytesMut::with_capacity(65536); + assert!( + matches!( + array.encode(&mut buf), + Err(super::super::ser::EncodeError::NestingTooDeep { .. }) + ), + "gate should reject event-data Timestamp leaf at object depth {MAX_OBJECT_DEPTH_VALUE}", + ); + + // Metadata: same boundary, one shallower. + let metadata_ts = create_nested_value_with_leaf(MAX_OBJECT_DEPTH_METADATA, ts_leaf()); + assert!( + !roundtrip_metadata(metadata_ts.clone()), + "metadata depth {MAX_OBJECT_DEPTH_METADATA} with Timestamp leaf is expected to fail prost decode" + ); + + let mut event = LogEvent::from("flat"); + *event.metadata_mut().value_mut() = metadata_ts; + let array = EventArray::Logs(LogArray::from(vec![event])); + let mut buf = BytesMut::with_capacity(65536); + assert!( + matches!( + array.encode(&mut buf), + Err(super::super::ser::EncodeError::NestingTooDeep { .. }) + ), + "gate should reject metadata Timestamp leaf at object depth {MAX_OBJECT_DEPTH_METADATA}", + ); +} + +/// Verify the gate still admits Timestamp leaves one level shallower than the boundary +/// — they cost exactly one frame, no more — and that those payloads roundtrip cleanly +/// through prost. +#[test] +fn nesting_gate_accepts_timestamp_leaf_below_max_object_depth() { + // Event data: depth (max-1) Object + Timestamp leaf = (max-1)*3 + 1 frames. + let mut event = LogEvent::default(); + event.insert( + "data", + create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE - 2, ts_leaf()), + ); + let array = EventArray::Logs(LogArray::from(vec![event])); + let mut buf = BytesMut::with_capacity(65536); + assert!( + array.encode(&mut buf).is_ok(), + "gate should accept event-data Timestamp leaf at object depth {}", + MAX_OBJECT_DEPTH_VALUE - 1, + ); + assert!( + proto::EventArray::decode(buf.freeze()).is_ok(), + "prost should decode event-data Timestamp leaf at object depth {}", + MAX_OBJECT_DEPTH_VALUE - 1, + ); + + // Metadata: one shallower. + let mut event = LogEvent::from("flat"); + *event.metadata_mut().value_mut() = + create_nested_value_with_leaf(MAX_OBJECT_DEPTH_METADATA - 1, ts_leaf()); + let array = EventArray::Logs(LogArray::from(vec![event])); + let mut buf = BytesMut::with_capacity(65536); + assert!( + array.encode(&mut buf).is_ok(), + "gate should accept metadata Timestamp leaf at object depth {}", + MAX_OBJECT_DEPTH_METADATA - 1, + ); + assert!( + proto::EventArray::decode(buf.freeze()).is_ok(), + "prost should decode metadata Timestamp leaf at object depth {}", + MAX_OBJECT_DEPTH_METADATA - 1, + ); +} + +/// Verify `filter_unencodable` keeps the valid events and drops only the over-budget +/// ones, returning a smaller `EventArray` rather than failing the whole batch. +#[test] +fn filter_unencodable_drops_only_over_budget_events() { + let good = || { + let mut event = LogEvent::default(); + event.insert("data", "ok"); + event + }; + let bad = || { + let mut event = LogEvent::default(); + event.insert("data", create_nested_value(MAX_OBJECT_DEPTH_VALUE)); + event + }; + + let logs = vec![good(), bad(), good(), bad(), good()]; + let array = EventArray::Logs(LogArray::from(logs)); + + let filtered = array + .filter_unencodable() + .expect("3 good events should survive filtering"); + assert_eq!(filtered.event_count(), 3, "only good events should remain"); + + let EventArray::Logs(surviving) = filtered else { + panic!("variant should be preserved"); + }; + for log in &surviving { + assert_eq!( + log.value().get("data").and_then(|v| v.as_bytes()), + Some(&bytes::Bytes::from_static(b"ok")), + "only good events should remain", + ); + } +} + +/// Verify that the public per-event entry point used by both the native codec and the +/// vector sink charges `Value::Timestamp` for one frame, just like the buffer gate. +/// Without this, a depth-33 object chain ending in a timestamp would pass the codec +/// check and fail prost decode on the receiving end. +#[test] +fn event_exceeds_max_nesting_cost_charges_timestamp_leaf() { + let log_at_max_with_ts = { + let mut event = LogEvent::default(); + event.insert( + "data", + create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE - 1, ts_leaf()), + ); + Event::Log(event) + }; + assert!( + event_exceeds_max_nesting_cost(&log_at_max_with_ts).is_some(), + "depth {MAX_OBJECT_DEPTH_VALUE} log with Timestamp leaf must be rejected", + ); + + let trace_at_max_with_ts = { + let mut trace = TraceEvent::default(); + trace.insert( + "data", + create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE - 1, ts_leaf()), + ); + Event::Trace(trace) + }; + assert!( + event_exceeds_max_nesting_cost(&trace_at_max_with_ts).is_some(), + "depth {MAX_OBJECT_DEPTH_VALUE} trace with Timestamp leaf must be rejected", + ); + + let metric_at_max_with_ts = { + let mut metric = Metric::new( + "test", + MetricKind::Incremental, + MetricValue::Counter { value: 1.0 }, + ); + *metric.metadata_mut().value_mut() = + create_nested_value_with_leaf(MAX_OBJECT_DEPTH_METADATA, ts_leaf()); + Event::Metric(metric) + }; + assert!( + event_exceeds_max_nesting_cost(&metric_at_max_with_ts).is_some(), + "metric with metadata-Timestamp leaf at depth {MAX_OBJECT_DEPTH_METADATA} must be rejected", + ); + + // And one shallower stays under the budget. + let log_below_max_with_ts = { + let mut event = LogEvent::default(); + event.insert( + "data", + create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE - 2, ts_leaf()), + ); + Event::Log(event) + }; + assert!( + event_exceeds_max_nesting_cost(&log_below_max_with_ts).is_none(), + "depth {} log with Timestamp leaf must be accepted", + MAX_OBJECT_DEPTH_VALUE - 1, + ); +} + +/// Unit-level check that `check_value_nesting_cost` charges `TIMESTAMP_FRAME_COST` +/// for a `Value::Timestamp` leaf, independent of nesting. +#[test] +fn check_value_nesting_cost_charges_timestamp_leaf() { + let ts = ts_leaf(); + assert!(check_value_nesting_cost(&ts, 0, TIMESTAMP_FRAME_COST).is_ok()); + assert!(check_value_nesting_cost(&ts, 0, TIMESTAMP_FRAME_COST - 1).is_err()); + + // A single object level containing a timestamp leaf: OBJECT_FRAME_COST + TIMESTAMP_FRAME_COST. + let mut map = ObjectMap::new(); + map.insert("ts".into(), ts); + let nested = Value::Object(map); + let expected = OBJECT_FRAME_COST + TIMESTAMP_FRAME_COST; + assert!(check_value_nesting_cost(&nested, 0, expected).is_ok()); + assert!(check_value_nesting_cost(&nested, 0, expected - 1).is_err()); +} + +/// Verify flat events pass without issues. +#[test] +fn nesting_gate_accepts_flat_events() { + let mut log = LogEvent::from("hello world"); + log.insert("foo", "bar"); + let events = EventArray::Logs(LogArray::from(vec![log])); + let mut buf = BytesMut::with_capacity(1024); + assert!(events.encode(&mut buf).is_ok()); + + let mut trace = TraceEvent::default(); + trace.insert("foo", "bar"); + let events = EventArray::Traces(TraceArray::from(vec![trace])); + let mut buf = BytesMut::with_capacity(1024); + assert!(events.encode(&mut buf).is_ok()); + + let metric = Metric::new( + "test_counter", + MetricKind::Incremental, + MetricValue::Counter { value: 1.0 }, + ); + let events = EventArray::Metrics(MetricArray::from(vec![metric])); + let mut buf = BytesMut::with_capacity(1024); + assert!(events.encode(&mut buf).is_ok()); +} + +#[test] +fn check_value_nesting_cost_with_configurable_budget() { + // Five nested objects: 5 levels × 3 frames per object = 15 frame cost. + let mut value = Value::from("leaf"); + for _ in 0..5 { + let mut map = ObjectMap::new(); + map.insert("n".into(), value); + value = Value::Object(map); + } + + assert!(check_value_nesting_cost(&value, 0, 15).is_ok()); + assert!(check_value_nesting_cost(&value, 0, 14).is_err()); + assert!(check_value_nesting_cost(&value, 0, 30).is_ok()); + + let flat = Value::from("hello"); + assert!(check_value_nesting_cost(&flat, 0, 0).is_ok()); +} + +#[test] +fn check_value_nesting_cost_with_mixed_variants() { + // Outer array (2) → inner object (3) → inner array (2) → leaf = 7 frame cost. + let inner = Value::Array(vec![Value::from("leaf")]); + let mut map = ObjectMap::new(); + map.insert("arr".into(), inner); + let value = Value::Array(vec![Value::Object(map)]); + + assert!(check_value_nesting_cost(&value, 0, 7).is_ok()); + assert!(check_value_nesting_cost(&value, 0, 6).is_err()); +} diff --git a/src/sinks/vector/sink.rs b/src/sinks/vector/sink.rs index bac25450c615f..0f821c6dabbd4 100644 --- a/src/sinks/vector/sink.rs +++ b/src/sinks/vector/sink.rs @@ -7,6 +7,8 @@ use tower::Service; use vector_lib::{ ByteSizeOf, EstimatedJsonEncodedSizeOf, config::telemetry, + event::event_exceeds_max_nesting_cost, + internal_event::{ComponentEventsDropped, UNINTENTIONAL}, request_metadata::GroupedCountByteSize, stream::{BatcherSettings, DriverResponse, batcher::data::BatchReduce}, }; @@ -60,6 +62,34 @@ where { async fn run_inner(self: Box, input: BoxStream<'_, Event>) -> Result<(), ()> { input + .filter_map(|event| { + std::future::ready( + if let Some((cost, budget)) = event_exceeds_max_nesting_cost(&event) { + let reason = format!( + "Event nesting cost {cost} exceeds protobuf budget of {budget}." + ); + emit!(ComponentEventsDropped:: { + count: 1, + reason: &reason, + }); + match event { + Event::Log(log) => log + .metadata() + .update_status(vector_lib::event::EventStatus::Rejected), + Event::Metric(metric) => metric + .metadata() + .update_status(vector_lib::event::EventStatus::Rejected), + Event::Trace(trace) => trace + .metadata() + .update_status(vector_lib::event::EventStatus::Rejected), + } + + None + } else { + Some(event) + }, + ) + }) .map(|mut event| { let mut byte_size = telemetry().create_request_count_byte_size(); byte_size.add_event(&event, event.estimated_json_encoded_size_of()); @@ -119,3 +149,104 @@ where self.run_inner(input).await } } + +#[cfg(test)] +mod tests { + use bytes::BytesMut; + use prost::Message; + use vector_lib::event::{ + Event, LogEvent, MAX_METADATA_VALUE_NESTING_FRAMES, MAX_VALUE_NESTING_FRAMES, ObjectMap, + Value, event_exceeds_max_nesting_cost, + }; + + use super::EventWrapper; + use crate::proto::vector as proto_vector; + + fn build_nested_value(wrapping_levels: usize) -> Value { + let mut v = Value::from("leaf"); + for _ in 0..wrapping_levels { + let mut m = ObjectMap::new(); + m.insert("nested".into(), v); + v = Value::Object(m); + } + v + } + + /// Empirical check: an event sitting *exactly* at the value budget accepted by + /// `event_exceeds_max_nesting_cost` must roundtrip through the vector sink's + /// actual wire shape — `PushEventsRequest { events: [EventWrapper] }` — and + /// not fail decode at the receiver. If this test fails, the value budget is + /// too high for the gRPC path and needs to be reduced for the outer request + /// wrapper. + #[test] + fn push_events_request_decode_at_value_budget() { + // 32 nested objects under "data" key → 33 effective object levels in + // `log.value()` (one outer Object from the inserted key), cost = 99 = + // MAX_VALUE_NESTING_FRAMES. + let mut log = LogEvent::default(); + log.insert("data", build_nested_value(32)); + let event = Event::Log(log); + assert!( + event_exceeds_max_nesting_cost(&event).is_none(), + "test setup invariant: event must sit exactly at the value budget \ + (cost {MAX_VALUE_NESTING_FRAMES})", + ); + + let request = proto_vector::PushEventsRequest { + events: vec![EventWrapper::from(event)], + }; + + let mut buf = BytesMut::with_capacity(65536); + request.encode(&mut buf).expect("encode should succeed"); + + proto_vector::PushEventsRequest::decode(buf.freeze()) + .expect("PushEventsRequest decode should succeed at the accepted value budget"); + } + + /// Boundary check: one step past the value budget must fail decode through + /// the gRPC wire shape. Together with the at-budget test above this pins + /// `MAX_VALUE_NESTING_FRAMES` as the tight boundary for the vector-sink + /// path, identical to the disk-buffer / native-codec `EventArray` path. + #[test] + fn push_events_request_decode_one_past_value_budget_fails() { + // 33 nested objects under "data" → 34 effective object levels, cost 102. + let mut log = LogEvent::default(); + log.insert("data", build_nested_value(33)); + let event = Event::Log(log); + + let request = proto_vector::PushEventsRequest { + events: vec![EventWrapper::from(event)], + }; + + let mut buf = BytesMut::with_capacity(65536); + request.encode(&mut buf).expect("encode should succeed"); + + assert!( + proto_vector::PushEventsRequest::decode(buf.freeze()).is_err(), + "PushEventsRequest decode must fail one step past the value budget; \ + if this changes, the gate is no longer tight", + ); + } + + #[test] + fn push_events_request_decode_at_metadata_budget() { + let mut log = LogEvent::from("flat"); + *log.metadata_mut().value_mut() = build_nested_value(32); + let event = Event::Log(log); + assert!( + event_exceeds_max_nesting_cost(&event).is_none(), + "test setup invariant: metadata must sit exactly at the metadata \ + budget (cost {MAX_METADATA_VALUE_NESTING_FRAMES})", + ); + + let request = proto_vector::PushEventsRequest { + events: vec![EventWrapper::from(event)], + }; + + let mut buf = BytesMut::with_capacity(65536); + request.encode(&mut buf).expect("encode should succeed"); + + proto_vector::PushEventsRequest::decode(buf.freeze()) + .expect("PushEventsRequest decode should succeed at the accepted metadata budget"); + } +}