From 0b2ca337e6d50223c213f45b9fea6795a023d97c Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Wed, 2 Sep 2026 20:10:06 -0600 Subject: [PATCH 1/6] fix(event proto): reject malformed internal protobuf instead of panicking Decode Vector's native event protobuf through `TryFrom` so a missing `oneof` or NaN float drops the record with error telemetry rather than crashing the source, tap, or disk-buffer reader. --- .../fallible_event_proto_decode.fix.md | 3 + lib/codecs/src/decoding/format/native.rs | 2 +- lib/vector-core/src/event/proto.rs | 421 +++++++++++++----- lib/vector-core/src/event/ser.rs | 46 +- .../src/event/test/serialization.rs | 60 +++ lib/vector-tap/src/runner.rs | 5 +- src/internal_events/grpc.rs | 33 ++ src/sinks/vector/mod.rs | 2 +- src/sources/vector/mod.rs | 35 +- 9 files changed, 463 insertions(+), 144 deletions(-) create mode 100644 changelog.d/fallible_event_proto_decode.fix.md diff --git a/changelog.d/fallible_event_proto_decode.fix.md b/changelog.d/fallible_event_proto_decode.fix.md new file mode 100644 index 0000000000000..0b2f4ad601a46 --- /dev/null +++ b/changelog.d/fallible_event_proto_decode.fix.md @@ -0,0 +1,3 @@ +Decoding Vector's native protobuf format (`decoding.codec = "native"`) and disk-buffer records no longer panics when an event variant is missing or unrecognized, or when a float field is `NaN`. Those payloads are rejected, dropped, and reported through existing decode/buffer error telemetry. A `NaN` float in event data or metadata rejects the entire record rather than rewriting the value. + +authors: bruceg diff --git a/lib/codecs/src/decoding/format/native.rs b/lib/codecs/src/decoding/format/native.rs index b26b390239de2..a0285a6788902 100644 --- a/lib/codecs/src/decoding/format/native.rs +++ b/lib/codecs/src/decoding/format/native.rs @@ -52,7 +52,7 @@ impl Deserializer for NativeDeserializer { if bytes.is_empty() { Ok(smallvec![]) } else { - let event_array = EventArray::from(proto::EventArray::decode(bytes)?); + let event_array = EventArray::try_from(proto::EventArray::decode(bytes)?)?; Ok(event_array.into_events().collect()) } } diff --git a/lib/vector-core/src/event/proto.rs b/lib/vector-core/src/event/proto.rs index 5261ff2917ef2..7289362e299e6 100644 --- a/lib/vector-core/src/event/proto.rs +++ b/lib/vector-core/src/event/proto.rs @@ -2,6 +2,7 @@ use std::{collections::BTreeMap, sync::Arc}; use chrono::TimeZone; use ordered_float::NotNan; +use snafu::Snafu; use uuid::Uuid; use super::{MetricTags, WithMetadata}; @@ -20,6 +21,28 @@ use super::EventFinalizers; use super::metadata::{Inner, default_schema_definition}; use super::{EventMetadata, array, metric::MetricSketch}; +/// Failure converting a structurally valid internal event protobuf into Vector's in-memory types. +/// +/// Distinct from a `prost` decode failure: the bytes parsed as protobuf, but a required event +/// variant was absent/unrecognized or a value could not be represented. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Snafu)] +pub enum EventProtoError { + #[snafu(display( + "event protobuf was structurally valid but an event or metric variant was absent or unrecognized; this often indicates a version mismatch" + ))] + UnrecognizedEventVariant, + #[snafu(display( + "event protobuf contained a NaN float, which cannot be represented in Vector's event model" + ))] + NanFloat, + #[snafu(display("event protobuf contained an invalid timestamp"))] + InvalidTimestamp, +} + +fn require_variant(value: Option) -> Result { + value.ok_or(EventProtoError::UnrecognizedEventVariant) +} + impl event_array::Events { // We can't use the standard `From` traits here because the actual // type of `LogArray` and `TraceArray` are the same. @@ -50,20 +73,31 @@ impl From for EventArray { } } -impl From for array::EventArray { - fn from(events: EventArray) -> Self { - let events = events.events.unwrap(); +impl TryFrom for array::EventArray { + type Error = EventProtoError; - match events { - event_array::Events::Logs(logs) => { - array::EventArray::Logs(logs.logs.into_iter().map(Into::into).collect()) - } - event_array::Events::Metrics(metrics) => { - array::EventArray::Metrics(metrics.metrics.into_iter().map(Into::into).collect()) - } - event_array::Events::Traces(traces) => { - array::EventArray::Traces(traces.traces.into_iter().map(Into::into).collect()) - } + fn try_from(events: EventArray) -> Result { + match require_variant(events.events)? { + event_array::Events::Logs(logs) => Ok(Self::Logs( + logs.logs + .into_iter() + .map(TryInto::try_into) + .collect::>()?, + )), + event_array::Events::Metrics(metrics) => Ok(Self::Metrics( + metrics + .metrics + .into_iter() + .map(TryInto::try_into) + .collect::>()?, + )), + event_array::Events::Traces(traces) => Ok(Self::Traces( + traces + .traces + .into_iter() + .map(TryInto::try_into) + .collect::>()?, + )), } } } @@ -92,62 +126,56 @@ impl From for Event { } } -impl From for super::LogEvent { +impl TryFrom for super::LogEvent { + type Error = EventProtoError; + #[allow(deprecated)] - fn from(log: Log) -> Self { - let metadata = log - .metadata_full - .map(Into::into) - .or_else(|| { - log.metadata - .and_then(decode_value) - .map(EventMetadata::default_with_value) - }) - .unwrap_or_default(); + fn try_from(log: Log) -> Result { + let metadata = decode_event_metadata(log.metadata_full, log.metadata)?; if let Some(value) = log.value { - Self::from_parts(decode_value(value).unwrap_or(VrlValue::Null), metadata) + Ok(Self::from_parts( + decode_value(value)?.unwrap_or(VrlValue::Null), + metadata, + )) } else { // This is for backwards compatibility. Only `value` should be set - let fields = log - .fields - .into_iter() - .filter_map(|(k, v)| decode_value(v).map(|value| (k.into(), value))) - .collect::(); + let mut fields = ObjectMap::new(); + for (k, v) in log.fields { + if let Some(value) = decode_value(v)? { + fields.insert(k.into(), value); + } + } - Self::from_map(fields, metadata) + Ok(Self::from_map(fields, metadata)) } } } -impl From for super::TraceEvent { - fn from(trace: Trace) -> Self { +impl TryFrom for super::TraceEvent { + type Error = EventProtoError; + + fn try_from(trace: Trace) -> Result { #[allow(deprecated)] - let metadata = trace - .metadata_full - .map(Into::into) - .or_else(|| { - trace - .metadata - .and_then(decode_value) - .map(EventMetadata::default_with_value) - }) - .unwrap_or_default(); + let metadata = decode_event_metadata(trace.metadata_full, trace.metadata)?; - let fields = trace - .fields - .into_iter() - .filter_map(|(k, v)| decode_value(v).map(|value| (k.into(), value))) - .collect::(); + let mut fields = ObjectMap::new(); + for (k, v) in trace.fields { + if let Some(value) = decode_value(v)? { + fields.insert(k.into(), value); + } + } - Self::from(super::LogEvent::from_map(fields, metadata)) + Ok(Self::from(super::LogEvent::from_map(fields, metadata))) } } -impl From for super::MetricValue { +impl TryFrom for super::MetricValue { + type Error = EventProtoError; + #[allow(deprecated)] - fn from(value: MetricValue) -> Self { - match value { + fn try_from(value: MetricValue) -> Result { + Ok(match value { MetricValue::Counter(counter) => Self::Counter { value: counter.value, }, @@ -196,18 +224,20 @@ impl From for super::MetricValue { count: summary.count, sum: summary.sum, }, - MetricValue::Sketch(sketch) => match sketch.sketch.unwrap() { + MetricValue::Sketch(sketch) => match require_variant(sketch.sketch)? { sketch::Sketch::AgentDdSketch(ddsketch) => Self::Sketch { sketch: ddsketch.into(), }, }, - } + }) } } -impl From for super::Metric { +impl TryFrom for super::Metric { + type Error = EventProtoError; + #[allow(deprecated)] - fn from(metric: Metric) -> Self { + fn try_from(metric: Metric) -> Result { let kind = match metric.kind() { metric::Kind::Incremental => super::MetricKind::Incremental, metric::Kind::Absolute => super::MetricKind::Absolute, @@ -217,14 +247,11 @@ impl From for super::Metric { let namespace = (!metric.namespace.is_empty()).then_some(metric.namespace); - // Sign can never be lost as ts.nanos is always non negative (per proto spec) - #[allow(clippy::cast_sign_loss)] - let timestamp = metric.timestamp.map(|ts| { - chrono::Utc - .timestamp_opt(ts.seconds, ts.nanos as u32) - .single() - .expect("invalid timestamp") - }); + let timestamp = metric + .timestamp + .as_ref() + .map(decode_timestamp) + .transpose()?; let mut tags = MetricTags( metric @@ -252,35 +279,26 @@ impl From for super::Metric { } let tags = (!tags.is_empty()).then_some(tags); - let value = super::MetricValue::from(metric.value.unwrap()); + let value = require_variant(metric.value)?.try_into()?; - let metadata = metric - .metadata_full - .map(Into::into) - .or_else(|| { - metric - .metadata - .and_then(decode_value) - .map(EventMetadata::default_with_value) - }) - .unwrap_or_default(); + let metadata = decode_event_metadata(metric.metadata_full, metric.metadata)?; - Self::new_with_metadata(name, kind, value, metadata) + Ok(Self::new_with_metadata(name, kind, value, metadata) .with_namespace(namespace) .with_tags(tags) .with_timestamp(timestamp) - .with_interval_ms(std::num::NonZeroU32::new(metric.interval_ms)) + .with_interval_ms(std::num::NonZeroU32::new(metric.interval_ms))) } } -impl From for super::Event { - fn from(proto: EventWrapper) -> Self { - let event = proto.event.unwrap(); +impl TryFrom for super::Event { + type Error = EventProtoError; - match event { - Event::Log(proto) => Self::Log(proto.into()), - Event::Metric(proto) => Self::Metric(proto.into()), - Event::Trace(proto) => Self::Trace(proto.into()), + fn try_from(proto: EventWrapper) -> Result { + match require_variant(proto.event)? { + Event::Log(proto) => Ok(Self::Log(proto.try_into()?)), + Event::Metric(proto) => Ok(Self::Metric(proto.try_into()?)), + Event::Trace(proto) => Ok(Self::Trace(proto.try_into()?)), } } } @@ -654,8 +672,10 @@ impl From for Metadata { } } -impl From for EventMetadata { - fn from(value: Metadata) -> Self { +impl TryFrom for EventMetadata { + type Error = EventProtoError; + + fn try_from(value: Metadata) -> Result { let Metadata { value: metadata_value, source_id, @@ -666,7 +686,10 @@ impl From for EventMetadata { source_event_id, } = value; - let metadata_value = metadata_value.and_then(decode_value); + let metadata_value = match metadata_value { + Some(value) => decode_value(value)?, + None => None, + }; let source_id = source_id.map(|s| Arc::new(s.into())); let upstream_id = upstream_id.map(|id| Arc::new(id.into())); let secrets = secrets.map(Into::into); @@ -687,7 +710,7 @@ impl From for EventMetadata { } }; - EventMetadata { + Ok(EventMetadata { inner: Arc::new(Inner { value: metadata_value .unwrap_or_else(|| vrl::value::Value::Object(ObjectMap::new())), @@ -702,48 +725,78 @@ impl From for EventMetadata { source_event_id, }), last_transform_timestamp: None, - } + }) } } -fn decode_value(input: Value) -> Option { +fn decode_event_metadata( + metadata_full: Option, + metadata: Option, +) -> Result { + if let Some(full) = metadata_full { + full.try_into() + } else if let Some(value) = metadata { + Ok(decode_value(value)? + .map(EventMetadata::default_with_value) + .unwrap_or_default()) + } else { + Ok(EventMetadata::default()) + } +} + +fn decode_timestamp( + ts: &prost_types::Timestamp, +) -> Result, EventProtoError> { + // Sign is never lost as ts.nanos is always non negative (per proto spec) + #[allow(clippy::cast_sign_loss)] + chrono::Utc + .timestamp_opt(ts.seconds, ts.nanos as u32) + .single() + .ok_or(EventProtoError::InvalidTimestamp) +} + +fn decode_value(input: Value) -> Result, EventProtoError> { match input.kind { - Some(value::Kind::RawBytes(data)) => Some(super::Value::Bytes(data)), - // Sign is never lost as ts.nanos is always non negative (per proto spec) - #[allow(clippy::cast_sign_loss)] - Some(value::Kind::Timestamp(ts)) => Some(super::Value::Timestamp( - chrono::Utc - .timestamp_opt(ts.seconds, ts.nanos as u32) - .single() - .expect("invalid timestamp"), - )), - Some(value::Kind::Integer(value)) => Some(super::Value::Integer(value)), - Some(value::Kind::Float(value)) => Some(super::Value::Float(NotNan::new(value).unwrap())), - Some(value::Kind::Boolean(value)) => Some(super::Value::Boolean(value)), + Some(value::Kind::RawBytes(data)) => Ok(Some(super::Value::Bytes(data))), + Some(value::Kind::Timestamp(ts)) => { + Ok(Some(super::Value::Timestamp(decode_timestamp(&ts)?))) + } + Some(value::Kind::Integer(value)) => Ok(Some(super::Value::Integer(value))), + Some(value::Kind::Float(value)) => { + let value = NotNan::new(value).map_err(|_| EventProtoError::NanFloat)?; + Ok(Some(super::Value::Float(value))) + } + Some(value::Kind::Boolean(value)) => Ok(Some(super::Value::Boolean(value))), Some(value::Kind::Map(map)) => decode_map(map.fields), Some(value::Kind::Array(array)) => decode_array(array.items), - Some(value::Kind::Null(_)) => Some(super::Value::Null), + Some(value::Kind::Null(_)) => Ok(Some(super::Value::Null)), None => { error!("Encoded event contains unknown value kind."); - None + Ok(None) } } } -fn decode_map(fields: BTreeMap) -> Option { - fields - .into_iter() - .map(|(key, value)| decode_value(value).map(|value| (key.into(), value))) - .collect::>() - .map(event::Value::Object) +fn decode_map(fields: BTreeMap) -> Result, EventProtoError> { + let mut map = ObjectMap::new(); + for (key, value) in fields { + let Some(decoded) = decode_value(value)? else { + return Ok(None); + }; + map.insert(key.into(), decoded); + } + Ok(Some(event::Value::Object(map))) } -fn decode_array(items: Vec) -> Option { - items - .into_iter() - .map(decode_value) - .collect::>>() - .map(super::Value::Array) +fn decode_array(items: Vec) -> Result, EventProtoError> { + let mut decoded_items = Vec::with_capacity(items.len()); + for item in items { + let Some(decoded) = decode_value(item)? else { + return Ok(None); + }; + decoded_items.push(decoded); + } + Ok(Some(super::Value::Array(decoded_items))) } fn encode_value(value: super::Value) -> Value { @@ -883,7 +936,9 @@ mod tests { let decoded = metrics .metrics .into_iter() - .map(crate::event::Metric::from) + .map(|metric| { + crate::event::Metric::try_from(metric).expect("legacy metric should decode") + }) .map(|metric| metric.value().clone()) .collect::>(); @@ -894,7 +949,8 @@ mod tests { fn decodes_pre_v27_single_valued_metric_tags() { let encoded = Metric::decode(PRE_V27_TAGS).unwrap(); - let decoded = crate::event::Metric::from(encoded); + let decoded = + crate::event::Metric::try_from(encoded).expect("legacy metric tags should decode"); assert_eq!(decoded.tag_value("service").as_deref(), Some("api")); } @@ -916,7 +972,8 @@ mod tests { ) .with_tags(Some(tags)); - let decoded = crate::event::Metric::from(Metric::from(event)); + let decoded = crate::event::Metric::try_from(Metric::from(event)) + .expect("encoded metric should decode"); let values = decoded .tags() .unwrap() @@ -932,9 +989,14 @@ mod tests { fn decodes_pre_v34_metadata_for_all_event_types() { let expected = VrlValue::from("legacy metadata"); - let log = crate::event::LogEvent::from(Log::decode(PRE_V34_LOG_METADATA).unwrap()); - let trace = crate::event::TraceEvent::from(Trace::decode(PRE_V34_TRACE_METADATA).unwrap()); - let metric = crate::event::Metric::from(Metric::decode(PRE_V34_METRIC_METADATA).unwrap()); + let log = crate::event::LogEvent::try_from(Log::decode(PRE_V34_LOG_METADATA).unwrap()) + .expect("legacy log metadata should decode"); + let trace = + crate::event::TraceEvent::try_from(Trace::decode(PRE_V34_TRACE_METADATA).unwrap()) + .expect("legacy trace metadata should decode"); + let metric = + crate::event::Metric::try_from(Metric::decode(PRE_V34_METRIC_METADATA).unwrap()) + .expect("legacy metric metadata should decode"); assert_eq!(log.metadata().value(), &expected); assert_eq!(trace.metadata().value(), &expected); @@ -943,9 +1005,122 @@ mod tests { #[test] fn decodes_pre_v41_metadata_without_source_event_id() { - let decoded = EventMetadata::from(Metadata::decode(PRE_V41_METADATA).unwrap()); + let decoded = EventMetadata::try_from(Metadata::decode(PRE_V41_METADATA).unwrap()) + .expect("legacy metadata should decode"); assert_eq!(decoded.source_event_id(), None); assert_eq!(decoded.source_type(), Some("legacy")); } + + #[test] + fn missing_event_array_variant_is_an_error() { + let proto = EventArray { events: None }; + assert_eq!( + array::EventArray::try_from(proto), + Err(EventProtoError::UnrecognizedEventVariant) + ); + } + + #[test] + fn missing_event_wrapper_variant_is_an_error() { + let proto = EventWrapper { event: None }; + assert_eq!( + crate::event::Event::try_from(proto), + Err(EventProtoError::UnrecognizedEventVariant) + ); + } + + #[test] + fn missing_metric_value_variant_is_an_error() { + let proto = Metric { + name: "requests".into(), + value: None, + ..Metric::default() + }; + assert_eq!( + crate::event::Metric::try_from(proto), + Err(EventProtoError::UnrecognizedEventVariant) + ); + } + + #[test] + fn missing_sketch_variant_is_an_error() { + let proto = Metric { + name: "requests".into(), + value: Some(MetricValue::Sketch(Sketch { sketch: None })), + ..Metric::default() + }; + assert_eq!( + crate::event::Metric::try_from(proto), + Err(EventProtoError::UnrecognizedEventVariant) + ); + } + + #[test] + fn nan_float_value_is_an_error() { + let value = Value { + kind: Some(value::Kind::Float(f64::NAN)), + }; + assert_eq!(decode_value(value), Err(EventProtoError::NanFloat)); + } + + #[test] + fn nan_float_in_event_data_rejects_the_record() { + let proto = EventWrapper { + event: Some(Event::Log(Log { + value: Some(Value { + kind: Some(value::Kind::Float(f64::NAN)), + }), + ..Log::default() + })), + }; + assert_eq!( + crate::event::Event::try_from(proto), + Err(EventProtoError::NanFloat) + ); + } + + #[test] + fn nan_float_in_metadata_rejects_the_record() { + let proto = EventWrapper { + event: Some(Event::Log(Log { + metadata_full: Some(Metadata { + value: Some(Value { + kind: Some(value::Kind::Float(f64::NAN)), + }), + ..Metadata::default() + }), + ..Log::default() + })), + }; + assert_eq!( + crate::event::Event::try_from(proto), + Err(EventProtoError::NanFloat) + ); + } + + #[test] + fn unknown_event_array_oneof_tag_is_unrecognized_variant() { + // Field 4 is not a member of `EventArray.events` (logs=1, metrics=2, traces=3). + // Tag = (4 << 3) | 2 (length-delimited). + let bytes = bytes::Bytes::from_static(&[34, 0]); + let proto = EventArray::decode(bytes).expect("unknown field is valid protobuf"); + assert!(proto.events.is_none()); + assert_eq!( + array::EventArray::try_from(proto), + Err(EventProtoError::UnrecognizedEventVariant) + ); + } + + #[test] + fn unknown_event_wrapper_oneof_tag_is_unrecognized_variant() { + // Field 4 is not a member of `EventWrapper.event` (log=1, metric=2, trace=3). + let bytes = bytes::Bytes::from_static(&[34, 0]); + let proto = EventWrapper::decode(bytes).expect("unknown field is valid protobuf"); + assert!(proto.event.is_none()); + assert_eq!( + crate::event::Event::try_from(proto), + Err(EventProtoError::UnrecognizedEventVariant) + ); + } } diff --git a/lib/vector-core/src/event/ser.rs b/lib/vector-core/src/event/ser.rs index beddb0abe1b01..1917515de5c18 100644 --- a/lib/vector-core/src/event/ser.rs +++ b/lib/vector-core/src/event/ser.rs @@ -158,11 +158,34 @@ pub enum EncodeError { #[derive(Debug, Snafu)] pub enum DecodeError { #[snafu(display( - "the provided buffer could not be decoded as a valid Protocol Buffers payload" + "the provided buffer could not be decoded as EventArray ({event_array}) or as EventWrapper ({event_wrapper})" ))] - InvalidProtobufPayload, + InvalidProtobufPayload { + event_array: prost::DecodeError, + event_wrapper: prost::DecodeError, + }, #[snafu(display("unsupported encoding metadata for this context"))] UnsupportedEncodingMetadata, + #[snafu(display( + "event protobuf was structurally valid but an event or metric variant was absent or unrecognized; this often indicates a version mismatch" + ))] + UnrecognizedEventVariant, + #[snafu(display( + "event protobuf contained a NaN float, which cannot be represented in Vector's event model" + ))] + NanFloat, + #[snafu(display("event protobuf contained an invalid timestamp"))] + InvalidTimestamp, +} + +impl From for DecodeError { + fn from(error: proto::EventProtoError) -> Self { + match error { + proto::EventProtoError::UnrecognizedEventVariant => Self::UnrecognizedEventVariant, + proto::EventProtoError::NanFloat => Self::NanFloat, + proto::EventProtoError::InvalidTimestamp => Self::InvalidTimestamp, + } + } } /// Flags for describing the encoding scheme used by our primary event types that flow through buffers. /// @@ -266,13 +289,18 @@ impl Encodable for EventArray { B: Buf + Clone, { if metadata.contains(EventEncodableMetadataFlags::DiskBufferV1CompatibilityMode) { - proto::EventArray::decode(buffer.clone()) - .map(Into::into) - .or_else(|_| { - proto::EventWrapper::decode(buffer) - .map(|pe| EventArray::from(Event::from(pe))) - .map_err(|_| DecodeError::InvalidProtobufPayload) - }) + match proto::EventArray::decode(buffer.clone()) { + Ok(array) => array.try_into().map_err(DecodeError::from), + Err(event_array) => match proto::EventWrapper::decode(buffer) { + Ok(wrapper) => Event::try_from(wrapper) + .map(EventArray::from) + .map_err(DecodeError::from), + Err(event_wrapper) => Err(DecodeError::InvalidProtobufPayload { + event_array, + event_wrapper, + }), + }, + } } else { Err(DecodeError::UnsupportedEncodingMetadata) } diff --git a/lib/vector-core/src/event/test/serialization.rs b/lib/vector-core/src/event/test/serialization.rs index bc875032a887c..a6dcdf75f86ab 100644 --- a/lib/vector-core/src/event/test/serialization.rs +++ b/lib/vector-core/src/event/test/serialization.rs @@ -778,3 +778,63 @@ fn check_value_nesting_cost_with_mixed_variants() { assert!(check_value_nesting_cost(&value, 0, 7).is_ok()); assert!(check_value_nesting_cost(&value, 0, 6).is_err()); } + +#[test] +fn truncated_protobuf_is_invalid_payload() { + let array = EventArray::Logs(vec![LogEvent::from("hello")]); + let mut buffer = BytesMut::with_capacity(64); + encode_value(array, &mut buffer); + assert!(buffer.len() > 1); + buffer.truncate(buffer.len() - 1); + + let error = EventArray::decode(EventArray::get_metadata(), buffer).unwrap_err(); + assert!( + matches!( + error, + crate::event::ser::DecodeError::InvalidProtobufPayload { .. } + ), + "truncated protobuf should be InvalidProtobufPayload, got {error:?}" + ); + let message = error.to_string(); + assert!( + message.contains("EventArray") && message.contains("EventWrapper"), + "invalid payload should report both decode attempts, got {message}" + ); +} + +#[test] +fn unknown_event_array_variant_is_not_invalid_protobuf() { + // Field 4 is not a member of `EventArray.events`. Prost keeps it as an unknown + // field and leaves the oneof unset, which must not be reported as corrupt protobuf. + let buffer = bytes::Bytes::from_static(&[34, 0]); + let error = EventArray::decode(EventArray::get_metadata(), buffer).unwrap_err(); + assert!( + matches!( + error, + crate::event::ser::DecodeError::UnrecognizedEventVariant + ), + "unknown oneof tag should be UnrecognizedEventVariant, got {error:?}" + ); +} + +#[test] +fn nan_float_is_rejected_by_encodable_decode() { + let proto_array = proto::EventArray { + events: Some(proto::event_array::Events::Logs(proto::LogArray { + logs: vec![proto::Log { + value: Some(proto::Value { + kind: Some(proto::value::Kind::Float(f64::NAN)), + }), + ..proto::Log::default() + }], + })), + }; + let mut buffer = BytesMut::with_capacity(64); + proto_array.encode(&mut buffer).unwrap(); + + let error = EventArray::decode(EventArray::get_metadata(), buffer).unwrap_err(); + assert!( + matches!(error, crate::event::ser::DecodeError::NanFloat), + "NaN float should be NanFloat, got {error:?}" + ); +} diff --git a/lib/vector-tap/src/runner.rs b/lib/vector-tap/src/runner.rs index 8c6ab815a8bf8..7e3b815e063fb 100644 --- a/lib/vector-tap/src/runner.rs +++ b/lib/vector-tap/src/runner.rs @@ -263,10 +263,11 @@ impl<'a> TapRunner<'a> { let core_event_wrapper = vector_core::event::proto::EventWrapper::decode(Bytes::from(bytes)) - .map_err(|e| format!("Failed to decode event: {}", e))?; + .map_err(|error| format!("Failed to decode event: {error}"))?; // Convert to vector-core Event (which has Serialize) - let event: Event = core_event_wrapper.into(); + let event = Event::try_from(core_event_wrapper) + .map_err(|error| format!("Failed to convert event: {error}"))?; // Serialize based on format match format { diff --git a/src/internal_events/grpc.rs b/src/internal_events/grpc.rs index 66da41e10b797..e9a32f31cc519 100644 --- a/src/internal_events/grpc.rs +++ b/src/internal_events/grpc.rs @@ -67,6 +67,39 @@ pub struct GrpcError { pub error: E, } +#[cfg(feature = "sources-vector")] +pub(crate) const EVENT_PROTO_DECODE_REASON: &str = "Failed to decode Vector protobuf event."; + +/// A structurally valid gRPC event protobuf could not be converted into a Vector event. +#[cfg(feature = "sources-vector")] +#[derive(Debug, NamedInternalEvent)] +pub struct GrpcEventDecodeError { + pub error: E, +} + +#[cfg(feature = "sources-vector")] +impl InternalEvent for GrpcEventDecodeError +where + E: std::fmt::Display, +{ + fn emit(self) { + error!( + message = EVENT_PROTO_DECODE_REASON, + error = %self.error, + error_code = "event_proto_decode", + error_type = error_type::PARSER_FAILED, + stage = error_stage::RECEIVING, + ); + counter!( + CounterName::ComponentErrorsTotal, + "error_code" => "event_proto_decode", + "error_type" => error_type::PARSER_FAILED, + "stage" => error_stage::RECEIVING, + ) + .increment(1); + } +} + impl InternalEvent for GrpcError where E: std::fmt::Display, diff --git a/src/sinks/vector/mod.rs b/src/sinks/vector/mod.rs index c5ff1bc7bf13e..e40908ed891dd 100644 --- a/src/sinks/vector/mod.rs +++ b/src/sinks/vector/mod.rs @@ -951,7 +951,7 @@ mod tests { let mut events = Vec::with_capacity(req.events.len()); for event in req.events { - let event: Event = event.into(); + let event = Event::try_from(event).expect("encoded test event should decode"); let string = event .as_log() .get_message() diff --git a/src/sources/vector/mod.rs b/src/sources/vector/mod.rs index d63972fd43bec..f31e13bf655ef 100644 --- a/src/sources/vector/mod.rs +++ b/src/sources/vector/mod.rs @@ -11,7 +11,9 @@ use vector_lib::{ config::LogNamespace, configurable::configurable_component, event::{BatchNotifier, BatchStatus, BatchStatusReceiver, Event}, - internal_event::{CountByteSize, InternalEventHandle as _}, + internal_event::{ + ComponentEventsDropped, CountByteSize, InternalEventHandle as _, UNINTENTIONAL, + }, }; use crate::{ @@ -20,7 +22,9 @@ use crate::{ DataType, GenerateConfig, Resource, SourceAcknowledgementsConfig, SourceConfig, SourceContext, SourceOutput, }, - internal_events::{EventsReceived, StreamClosedError}, + internal_events::{ + EVENT_PROTO_DECODE_REASON, EventsReceived, GrpcEventDecodeError, StreamClosedError, + }, proto::vector as proto, serde::bool_or_struct, sources::{ @@ -55,12 +59,27 @@ impl proto::Service for Service { &self, request: Request, ) -> Result, Status> { - let mut events: Vec = request - .into_inner() - .events - .into_iter() - .map(Event::from) - .collect(); + let request = request.into_inner(); + let mut events = Vec::with_capacity(request.events.len()); + let mut dropped = 0; + for wrapper in request.events { + match Event::try_from(wrapper) { + Ok(event) => events.push(event), + Err(error) => { + dropped += 1; + emit!(GrpcEventDecodeError { error }); + } + } + } + if dropped > 0 { + emit!(ComponentEventsDropped:: { + count: dropped, + reason: EVENT_PROTO_DECODE_REASON, + }); + } + if events.is_empty() { + return Ok(Response::new(proto::PushEventsResponse {})); + } let now = Utc::now(); for event in &mut events { From c175e01f8ada8271d32240ea4b788cd50b3b464c Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Thu, 3 Sep 2026 15:48:22 -0600 Subject: [PATCH 2/6] Improve and simplify error handling --- lib/vector-core/src/event/ser.rs | 21 ++----------------- .../src/event/test/serialization.rs | 11 ++++++++-- 2 files changed, 11 insertions(+), 21 deletions(-) diff --git a/lib/vector-core/src/event/ser.rs b/lib/vector-core/src/event/ser.rs index 1917515de5c18..6d722d31301d8 100644 --- a/lib/vector-core/src/event/ser.rs +++ b/lib/vector-core/src/event/ser.rs @@ -166,27 +166,10 @@ pub enum DecodeError { }, #[snafu(display("unsupported encoding metadata for this context"))] UnsupportedEncodingMetadata, - #[snafu(display( - "event protobuf was structurally valid but an event or metric variant was absent or unrecognized; this often indicates a version mismatch" - ))] - UnrecognizedEventVariant, - #[snafu(display( - "event protobuf contained a NaN float, which cannot be represented in Vector's event model" - ))] - NanFloat, - #[snafu(display("event protobuf contained an invalid timestamp"))] - InvalidTimestamp, + #[snafu(transparent)] + InvalidEvent { source: proto::EventProtoError }, } -impl From for DecodeError { - fn from(error: proto::EventProtoError) -> Self { - match error { - proto::EventProtoError::UnrecognizedEventVariant => Self::UnrecognizedEventVariant, - proto::EventProtoError::NanFloat => Self::NanFloat, - proto::EventProtoError::InvalidTimestamp => Self::InvalidTimestamp, - } - } -} /// Flags for describing the encoding scheme used by our primary event types that flow through buffers. /// /// # Stability diff --git a/lib/vector-core/src/event/test/serialization.rs b/lib/vector-core/src/event/test/serialization.rs index a6dcdf75f86ab..a8980444e751c 100644 --- a/lib/vector-core/src/event/test/serialization.rs +++ b/lib/vector-core/src/event/test/serialization.rs @@ -811,7 +811,9 @@ fn unknown_event_array_variant_is_not_invalid_protobuf() { assert!( matches!( error, - crate::event::ser::DecodeError::UnrecognizedEventVariant + crate::event::ser::DecodeError::InvalidEvent { + source: proto::EventProtoError::UnrecognizedEventVariant, + } ), "unknown oneof tag should be UnrecognizedEventVariant, got {error:?}" ); @@ -834,7 +836,12 @@ fn nan_float_is_rejected_by_encodable_decode() { let error = EventArray::decode(EventArray::get_metadata(), buffer).unwrap_err(); assert!( - matches!(error, crate::event::ser::DecodeError::NanFloat), + matches!( + error, + crate::event::ser::DecodeError::InvalidEvent { + source: proto::EventProtoError::NanFloat, + } + ), "NaN float should be NanFloat, got {error:?}" ); } From 836678c1c41fd53a9451b00e3d573f6835ce2ba7 Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Thu, 3 Sep 2026 15:52:09 -0600 Subject: [PATCH 3/6] fix missed use of `into` --- src/components/validation/runner/io.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/components/validation/runner/io.rs b/src/components/validation/runner/io.rs index 4768889e0fdb4..7d141fe1fe9e2 100644 --- a/src/components/validation/runner/io.rs +++ b/src/components/validation/runner/io.rs @@ -46,7 +46,9 @@ impl VectorService for EventForwardService { .into_inner() .events .into_iter() - .map(Event::from) + .map(|wrapper| { + Event::try_from(wrapper).expect("validation events are encoded by Vector") + }) .collect(); self.tx From ebed07830fc903d413f3f63f810e4794ad605920 Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Thu, 3 Sep 2026 15:58:10 -0600 Subject: [PATCH 4/6] Handle failible AgentDDSketch conversion --- .../fallible_event_proto_decode.fix.md | 2 +- lib/vector-core/src/event/proto.rs | 41 ++++++++++++++++--- 2 files changed, 36 insertions(+), 7 deletions(-) diff --git a/changelog.d/fallible_event_proto_decode.fix.md b/changelog.d/fallible_event_proto_decode.fix.md index 0b2f4ad601a46..17d752094d103 100644 --- a/changelog.d/fallible_event_proto_decode.fix.md +++ b/changelog.d/fallible_event_proto_decode.fix.md @@ -1,3 +1,3 @@ -Decoding Vector's native protobuf format (`decoding.codec = "native"`) and disk-buffer records no longer panics when an event variant is missing or unrecognized, or when a float field is `NaN`. Those payloads are rejected, dropped, and reported through existing decode/buffer error telemetry. A `NaN` float in event data or metadata rejects the entire record rather than rewriting the value. +Decoding Vector's native protobuf format (`decoding.codec = "native"`) and disk-buffer records no longer panics when an event variant is missing or unrecognized, when a float field is `NaN`, or when an AgentDDSketch has mismatched bin lists. Those payloads are rejected, dropped, and reported through existing decode/buffer error telemetry. A `NaN` float in event data or metadata rejects the entire record rather than rewriting the value. authors: bruceg diff --git a/lib/vector-core/src/event/proto.rs b/lib/vector-core/src/event/proto.rs index 7289362e299e6..05b8a5db1a6c7 100644 --- a/lib/vector-core/src/event/proto.rs +++ b/lib/vector-core/src/event/proto.rs @@ -37,6 +37,10 @@ pub enum EventProtoError { NanFloat, #[snafu(display("event protobuf contained an invalid timestamp"))] InvalidTimestamp, + #[snafu(display( + "event protobuf contained an AgentDDSketch whose k and n bin lists have different lengths" + ))] + MismatchedSketchBins, } fn require_variant(value: Option) -> Result { @@ -226,7 +230,7 @@ impl TryFrom for super::MetricValue { }, MetricValue::Sketch(sketch) => match require_variant(sketch.sketch)? { sketch::Sketch::AgentDdSketch(ddsketch) => Self::Sketch { - sketch: ddsketch.into(), + sketch: ddsketch.try_into()?, }, }, }) @@ -558,8 +562,10 @@ impl From for Sketch { } } -impl From for MetricSketch { - fn from(sketch: sketch::AgentDdSketch) -> Self { +impl TryFrom for MetricSketch { + type Error = EventProtoError; + + fn try_from(sketch: sketch::AgentDdSketch) -> Result { // These safe conversions are annoying because the Datadog Agent internally uses i16/u16, // but the proto definition uses i32/u32, so we have to jump through these hoops. let keys = sketch @@ -576,7 +582,7 @@ impl From for MetricSketch { .into_iter() .map(|n| n.try_into().unwrap_or(u16::MAX)) .collect::>(); - MetricSketch::AgentDDSketch( + Ok(MetricSketch::AgentDDSketch( AgentDDSketch::from_raw( sketch.count, sketch.min, @@ -586,8 +592,8 @@ impl From for MetricSketch { &keys, &counts, ) - .expect("keys/counts were unexpectedly mismatched"), - ) + .ok_or(EventProtoError::MismatchedSketchBins)?, + )) } } @@ -1056,6 +1062,29 @@ mod tests { ); } + #[test] + fn mismatched_sketch_bins_is_an_error() { + let proto = Metric { + name: "requests".into(), + value: Some(MetricValue::Sketch(Sketch { + sketch: Some(sketch::Sketch::AgentDdSketch(sketch::AgentDdSketch { + count: 1, + min: 0.0, + max: 1.0, + sum: 1.0, + avg: 1.0, + k: vec![1], + n: vec![1, 2], + })), + })), + ..Metric::default() + }; + assert_eq!( + crate::event::Metric::try_from(proto), + Err(EventProtoError::MismatchedSketchBins) + ); + } + #[test] fn nan_float_value_is_an_error() { let value = Value { From 665cfa029809b8df26515bd731806f3913214323 Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Thu, 3 Sep 2026 16:00:21 -0600 Subject: [PATCH 5/6] rework erroneous emission of ComponentEventsDropped --- src/internal_events/grpc.rs | 2 +- src/sources/vector/mod.rs | 20 +++----------------- 2 files changed, 4 insertions(+), 18 deletions(-) diff --git a/src/internal_events/grpc.rs b/src/internal_events/grpc.rs index e9a32f31cc519..a527b80c091ab 100644 --- a/src/internal_events/grpc.rs +++ b/src/internal_events/grpc.rs @@ -68,7 +68,7 @@ pub struct GrpcError { } #[cfg(feature = "sources-vector")] -pub(crate) const EVENT_PROTO_DECODE_REASON: &str = "Failed to decode Vector protobuf event."; +const EVENT_PROTO_DECODE_REASON: &str = "Failed to decode Vector protobuf event."; /// A structurally valid gRPC event protobuf could not be converted into a Vector event. #[cfg(feature = "sources-vector")] diff --git a/src/sources/vector/mod.rs b/src/sources/vector/mod.rs index f31e13bf655ef..53eada41e7264 100644 --- a/src/sources/vector/mod.rs +++ b/src/sources/vector/mod.rs @@ -11,9 +11,7 @@ use vector_lib::{ config::LogNamespace, configurable::configurable_component, event::{BatchNotifier, BatchStatus, BatchStatusReceiver, Event}, - internal_event::{ - ComponentEventsDropped, CountByteSize, InternalEventHandle as _, UNINTENTIONAL, - }, + internal_event::{CountByteSize, InternalEventHandle as _}, }; use crate::{ @@ -22,9 +20,7 @@ use crate::{ DataType, GenerateConfig, Resource, SourceAcknowledgementsConfig, SourceConfig, SourceContext, SourceOutput, }, - internal_events::{ - EVENT_PROTO_DECODE_REASON, EventsReceived, GrpcEventDecodeError, StreamClosedError, - }, + internal_events::{EventsReceived, GrpcEventDecodeError, StreamClosedError}, proto::vector as proto, serde::bool_or_struct, sources::{ @@ -61,22 +57,12 @@ impl proto::Service for Service { ) -> Result, Status> { let request = request.into_inner(); let mut events = Vec::with_capacity(request.events.len()); - let mut dropped = 0; for wrapper in request.events { match Event::try_from(wrapper) { Ok(event) => events.push(event), - Err(error) => { - dropped += 1; - emit!(GrpcEventDecodeError { error }); - } + Err(error) => emit!(GrpcEventDecodeError { error }), } } - if dropped > 0 { - emit!(ComponentEventsDropped:: { - count: dropped, - reason: EVENT_PROTO_DECODE_REASON, - }); - } if events.is_empty() { return Ok(Response::new(proto::PushEventsResponse {})); } From 93627bad575218d14070fbf41e8c521fe1524f20 Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Tue, 8 Sep 2026 16:51:10 -0600 Subject: [PATCH 6/6] Return non-retriable error on vector source decode errors --- src/internal_events/grpc.rs | 5 +---- src/sources/vector/mod.rs | 5 ++++- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/internal_events/grpc.rs b/src/internal_events/grpc.rs index a527b80c091ab..2009fb97937ce 100644 --- a/src/internal_events/grpc.rs +++ b/src/internal_events/grpc.rs @@ -67,9 +67,6 @@ pub struct GrpcError { pub error: E, } -#[cfg(feature = "sources-vector")] -const EVENT_PROTO_DECODE_REASON: &str = "Failed to decode Vector protobuf event."; - /// A structurally valid gRPC event protobuf could not be converted into a Vector event. #[cfg(feature = "sources-vector")] #[derive(Debug, NamedInternalEvent)] @@ -84,7 +81,7 @@ where { fn emit(self) { error!( - message = EVENT_PROTO_DECODE_REASON, + message = "Failed to decode Vector protobuf event.", error = %self.error, error_code = "event_proto_decode", error_type = error_type::PARSER_FAILED, diff --git a/src/sources/vector/mod.rs b/src/sources/vector/mod.rs index 53eada41e7264..16ecdc00bc134 100644 --- a/src/sources/vector/mod.rs +++ b/src/sources/vector/mod.rs @@ -60,7 +60,10 @@ impl proto::Service for Service { for wrapper in request.events { match Event::try_from(wrapper) { Ok(event) => events.push(event), - Err(error) => emit!(GrpcEventDecodeError { error }), + Err(error) => { + emit!(GrpcEventDecodeError { error }); + return Err(Status::invalid_argument(error.to_string())); + } } } if events.is_empty() {