vector_core/event/
ser.rs

1use bytes::{Buf, BufMut};
2use enumflags2::{BitFlags, FromBitsError, bitflags};
3use prost::Message;
4use snafu::Snafu;
5use vector_buffers::{
6    Bufferable, EventCount,
7    encoding::{AsMetadata, Encodable},
8};
9use vector_common::internal_event::{self, ComponentEventsDropped, UNINTENTIONAL};
10use vrl::value::Value;
11
12use super::{Event, EventArray, EventStatus, proto};
13
14/// Per-level prost recursion frame cost of an [`Value::Object`].
15///
16/// Decoding an object level walks `Value → ValueMap → map_entry (synthetic) → Value`,
17/// adding three message-decode frames before reaching the child Value.
18pub(crate) const OBJECT_FRAME_COST: usize = 3;
19
20/// Per-level prost recursion frame cost of an [`Value::Array`].
21///
22/// Decoding an array level walks `Value → ValueArray → Value`, adding two message-decode
23/// frames before reaching the child Value.
24pub(crate) const ARRAY_FRAME_COST: usize = 2;
25
26/// Per-leaf prost recursion frame cost of a [`Value::Timestamp`].
27///
28/// Unlike other scalar variants, `Value::Timestamp` is encoded as a nested
29/// `google.protobuf.Timestamp` message, so decoding it consumes one additional frame
30/// beyond the enclosing `Value`. Without this cost, a timestamp leaf under 32 object
31/// levels would sneak past the gate at cost 96 and trip prost's recursion limit on
32/// decode at cost 97.
33pub(crate) const TIMESTAMP_FRAME_COST: usize = 1;
34
35/// Maximum prost recursion frame cost accepted for any arbitrary [`Value`].
36///
37/// Prost enforces a decode recursion limit of 100 (no limit on encode). Each nesting level
38/// consumes 3 frames for [`Value::Object`], 2 for [`Value::Array`], or 1 for a
39/// [`Value::Timestamp`] leaf, plus a fixed overhead for the proto wrappers outside the
40/// Value tree.
41///
42/// Some protobuf paths (`Log.fields` and `Trace.fields`) can carry 99 frames, but the
43/// `Log.value` and metadata paths are only safe through 96. We use that highest common
44/// safe limit for every value so validation does not depend on its event type, root type,
45/// or destination protobuf field.
46pub const MAX_VALUE_NESTING_FRAMES: usize = 96;
47
48/// Walks a [`Value`] tree accumulating prost recursion frame cost, returning
49/// `Err(over_budget_cost)` as soon as any branch exceeds `budget`.
50///
51/// Object levels weigh [`OBJECT_FRAME_COST`] frames each, array levels weigh
52/// [`ARRAY_FRAME_COST`], and timestamp leaves weigh [`TIMESTAMP_FRAME_COST`] (because
53/// they decode into a nested `google.protobuf.Timestamp` message); other scalar leaves
54/// are free. Performs an early-exit traversal so well-formed events incur a single
55/// descent of the deepest branch only.
56///
57/// # Errors
58///
59/// Returns `Err(actual_cost)` if any branch's cumulative frame cost exceeds `budget`.
60pub(crate) fn check_value_nesting_cost(
61    value: &Value,
62    accumulated: usize,
63    budget: usize,
64) -> Result<(), usize> {
65    let level_cost = match value {
66        Value::Object(_) => OBJECT_FRAME_COST,
67        Value::Array(_) => ARRAY_FRAME_COST,
68        Value::Timestamp(_) => TIMESTAMP_FRAME_COST,
69        _ => 0,
70    };
71    let next = accumulated + level_cost;
72    if next > budget {
73        return Err(next);
74    }
75    match value {
76        Value::Object(map) => {
77            for v in map.values() {
78                check_value_nesting_cost(v, next, budget)?;
79            }
80        }
81        Value::Array(arr) => {
82            for v in arr {
83                check_value_nesting_cost(v, next, budget)?;
84            }
85        }
86        _ => {}
87    }
88    Ok(())
89}
90
91/// Checks whether an event's nesting frame cost exceeds the safe limits for protobuf encoding.
92///
93/// Returns `Some((cost, budget))` identifying the path that violated its budget, or `None`
94/// if the event is within bounds.
95///
96/// Every arbitrary value is checked against [`MAX_VALUE_NESTING_FRAMES`].
97///
98/// For metrics, only metadata is checked since metric values have a fixed structure.
99pub fn event_exceeds_max_nesting_cost(event: &Event) -> Option<(usize, usize)> {
100    let check = |value: &Value| {
101        check_value_nesting_cost(value, 0, MAX_VALUE_NESTING_FRAMES)
102            .map_err(|cost| (cost, MAX_VALUE_NESTING_FRAMES))
103    };
104    match event {
105        Event::Log(log) => check(log.value())
106            .and_then(|()| check(log.metadata().value()))
107            .err(),
108        Event::Trace(trace) => check(trace.value())
109            .and_then(|()| check(trace.metadata().value()))
110            .err(),
111        Event::Metric(metric) => check(metric.metadata().value()).err(),
112    }
113}
114
115/// Checks all events in an `EventArray` for nesting cost violations.
116///
117/// Every arbitrary value is checked against [`MAX_VALUE_NESTING_FRAMES`]. For metrics,
118/// only metadata is checked since metric values have a fixed structure.
119fn check_event_array_nesting_cost(events: &EventArray) -> Result<(), EncodeError> {
120    let check = |value: &Value| {
121        check_value_nesting_cost(value, 0, MAX_VALUE_NESTING_FRAMES).map_err(|cost| {
122            EncodeError::NestingTooDeep {
123                cost,
124                budget: MAX_VALUE_NESTING_FRAMES,
125            }
126        })
127    };
128    match events {
129        EventArray::Logs(logs) => {
130            for log in logs {
131                check(log.value())?;
132                check(log.metadata().value())?;
133            }
134        }
135        EventArray::Traces(traces) => {
136            for trace in traces {
137                check(trace.value())?;
138                check(trace.metadata().value())?;
139            }
140        }
141        EventArray::Metrics(metrics) => {
142            for metric in metrics {
143                check(metric.metadata().value())?;
144            }
145        }
146    }
147    Ok(())
148}
149
150#[derive(Debug, Snafu)]
151pub enum EncodeError {
152    #[snafu(display("the provided buffer was too small to fully encode this item"))]
153    BufferTooSmall,
154    #[snafu(display("event nesting cost {cost} exceeds protobuf budget of {budget}"))]
155    NestingTooDeep { cost: usize, budget: usize },
156}
157
158#[derive(Debug, Snafu)]
159pub enum DecodeError {
160    #[snafu(display(
161        "the provided buffer could not be decoded as a valid Protocol Buffers payload"
162    ))]
163    InvalidProtobufPayload,
164    #[snafu(display("unsupported encoding metadata for this context"))]
165    UnsupportedEncodingMetadata,
166}
167/// Flags for describing the encoding scheme used by our primary event types that flow through buffers.
168///
169/// # Stability
170///
171/// This enumeration should never have any flags removed, only added.  This ensures that previously
172/// used flags cannot have their meaning changed/repurposed after-the-fact.
173#[bitflags]
174#[repr(u32)]
175#[derive(Copy, Clone, Debug, PartialEq, Eq)]
176pub enum EventEncodableMetadataFlags {
177    /// Chained encoding scheme that first tries to decode as `EventArray` and then as `Event`, as a
178    /// way to support gracefully migrating existing v1-based disk buffers to the new
179    /// `EventArray`-based architecture.
180    ///
181    /// All encoding uses the `EventArray` variant, however.
182    DiskBufferV1CompatibilityMode = 0b1,
183}
184
185#[derive(Clone, Copy, Debug, PartialEq, Eq)]
186pub struct EventEncodableMetadata(BitFlags<EventEncodableMetadataFlags>);
187
188impl EventEncodableMetadata {
189    fn contains(self, flag: EventEncodableMetadataFlags) -> bool {
190        self.0.contains(flag)
191    }
192}
193
194impl From<EventEncodableMetadataFlags> for EventEncodableMetadata {
195    fn from(flag: EventEncodableMetadataFlags) -> Self {
196        Self(BitFlags::from(flag))
197    }
198}
199
200impl From<BitFlags<EventEncodableMetadataFlags>> for EventEncodableMetadata {
201    fn from(flags: BitFlags<EventEncodableMetadataFlags>) -> Self {
202        Self(flags)
203    }
204}
205
206impl TryFrom<u32> for EventEncodableMetadata {
207    type Error = FromBitsError<EventEncodableMetadataFlags>;
208
209    fn try_from(value: u32) -> Result<Self, Self::Error> {
210        BitFlags::try_from(value).map(Self)
211    }
212}
213
214impl AsMetadata for EventEncodableMetadata {
215    fn into_u32(self) -> u32 {
216        self.0.bits()
217    }
218
219    fn from_u32(value: u32) -> Option<Self> {
220        EventEncodableMetadata::try_from(value).ok()
221    }
222}
223
224impl Encodable for EventArray {
225    type Metadata = EventEncodableMetadata;
226    type EncodeError = EncodeError;
227    type DecodeError = DecodeError;
228
229    fn get_metadata() -> Self::Metadata {
230        EventEncodableMetadataFlags::DiskBufferV1CompatibilityMode.into()
231    }
232
233    fn can_decode(metadata: Self::Metadata) -> bool {
234        metadata.contains(EventEncodableMetadataFlags::DiskBufferV1CompatibilityMode)
235    }
236
237    /// # Errors
238    ///
239    /// Returns `EncodeError::NestingTooDeep` if any contained event's value or metadata
240    /// exceeds [`MAX_VALUE_NESTING_FRAMES`]. This is **all-or-nothing**: a single
241    /// over-budget event fails the entire batch, because a partially-encoded
242    /// `EventArray` reaching disk would trip prost's recursion limit on decode and
243    /// corrupt the buffer.
244    ///
245    /// Callers that want graceful per-item drop with telemetry and
246    /// `EventStatus::Rejected` must run [`Bufferable::filter_unencodable`] first.
247    /// `SenderAdapter::send`/`try_send` already does this on the disk-v2 path, so the
248    /// `NestingTooDeep` arm is unreachable from any current production call site — it
249    /// is defense-in-depth for a future caller that bypasses `SenderAdapter`.
250    ///
251    /// Returns `EncodeError::BufferTooSmall` if the buffer cannot hold the encoded
252    /// output.
253    fn encode<B>(self, buffer: &mut B) -> Result<(), Self::EncodeError>
254    where
255        B: BufMut,
256    {
257        check_event_array_nesting_cost(&self)?;
258
259        proto::EventArray::from(self)
260            .encode(buffer)
261            .map_err(|_| EncodeError::BufferTooSmall)
262    }
263
264    fn decode<B>(metadata: Self::Metadata, buffer: B) -> Result<Self, Self::DecodeError>
265    where
266        B: Buf + Clone,
267    {
268        if metadata.contains(EventEncodableMetadataFlags::DiskBufferV1CompatibilityMode) {
269            proto::EventArray::decode(buffer.clone())
270                .map(Into::into)
271                .or_else(|_| {
272                    proto::EventWrapper::decode(buffer)
273                        .map(|pe| EventArray::from(Event::from(pe)))
274                        .map_err(|_| DecodeError::InvalidProtobufPayload)
275                })
276        } else {
277            Err(DecodeError::UnsupportedEncodingMetadata)
278        }
279    }
280}
281
282impl Bufferable for EventArray {
283    /// Reuses the same budget walk as the encode-time gate, so the routing decision and
284    /// the eventual encode can never disagree about what is persistable.
285    fn is_fully_encodable(&self) -> bool {
286        check_event_array_nesting_cost(self).is_ok()
287    }
288
289    fn filter_unencodable(self) -> Option<Self> {
290        let exceeds =
291            |value: &Value| check_value_nesting_cost(value, 0, MAX_VALUE_NESTING_FRAMES).is_err();
292        let mut dropped = 0;
293        let filtered = match self {
294            EventArray::Logs(mut logs) => {
295                logs.retain(|log| {
296                    let too_deep = exceeds(log.value()) || exceeds(log.metadata().value());
297                    if too_deep {
298                        log.metadata().update_status(EventStatus::Rejected);
299                        dropped += 1;
300                    }
301                    !too_deep
302                });
303                EventArray::Logs(logs)
304            }
305            EventArray::Traces(mut traces) => {
306                traces.retain(|trace| {
307                    let too_deep = exceeds(trace.value()) || exceeds(trace.metadata().value());
308                    if too_deep {
309                        trace.metadata().update_status(EventStatus::Rejected);
310                        dropped += 1;
311                    }
312                    !too_deep
313                });
314                EventArray::Traces(traces)
315            }
316            EventArray::Metrics(mut metrics) => {
317                metrics.retain(|metric| {
318                    let too_deep = exceeds(metric.metadata().value());
319                    if too_deep {
320                        metric.metadata().update_status(EventStatus::Rejected);
321                        dropped += 1;
322                    }
323                    !too_deep
324                });
325                EventArray::Metrics(metrics)
326            }
327        };
328        if dropped > 0 {
329            internal_event::emit(ComponentEventsDropped::<UNINTENTIONAL> {
330                count: dropped,
331                reason: "Event nesting cost exceeds maximum for protobuf encoding.",
332            });
333        }
334        (filtered.event_count() > 0).then_some(filtered)
335    }
336}
vector_core/event/ser.rs

vector_core/event/
ser.rs
