codecs/decoding/format/
protobuf.rs

1use std::path::PathBuf;
2
3use bytes::Bytes;
4use chrono::Utc;
5use derivative::Derivative;
6use prost_reflect::{DynamicMessage, MessageDescriptor};
7use smallvec::{SmallVec, smallvec};
8use vector_config::configurable_component;
9use vector_core::{
10    config::{DataType, LogNamespace, log_schema},
11    event::{Event, LogEvent},
12    schema,
13};
14use vrl::{
15    protobuf::{
16        descriptor::{get_message_descriptor, get_message_descriptor_from_bytes},
17        parse::{Options, proto_to_value},
18    },
19    value::{Kind, Value},
20};
21
22use super::Deserializer;
23
24/// Config used to build a `ProtobufDeserializer`.
25#[configurable_component]
26#[derive(Debug, Clone, Default)]
27pub struct ProtobufDeserializerConfig {
28    /// Protobuf-specific decoding options.
29    #[serde(default, skip_serializing_if = "vector_core::serde::is_default")]
30    pub protobuf: ProtobufDeserializerOptions,
31}
32
33impl ProtobufDeserializerConfig {
34    /// Build the `ProtobufDeserializer` from this configuration.
35    pub fn build(&self) -> vector_common::Result<ProtobufDeserializer> {
36        ProtobufDeserializer::try_from(self)
37    }
38
39    /// Return the type of event build by this deserializer.
40    pub fn output_type(&self) -> DataType {
41        DataType::Log
42    }
43
44    /// The schema produced by the deserializer.
45    pub fn schema_definition(&self, log_namespace: LogNamespace) -> schema::Definition {
46        match log_namespace {
47            LogNamespace::Legacy => {
48                let mut definition =
49                    schema::Definition::empty_legacy_namespace().unknown_fields(Kind::any());
50
51                if let Some(timestamp_key) = log_schema().timestamp_key() {
52                    definition = definition.try_with_field(
53                        timestamp_key,
54                        // The protobuf decoder will try to insert a new `timestamp`-type value into the
55                        // "timestamp_key" field, but only if that field doesn't already exist.
56                        Kind::any().or_timestamp(),
57                        Some("timestamp"),
58                    );
59                }
60                definition
61            }
62            LogNamespace::Vector => {
63                schema::Definition::new_with_default_metadata(Kind::any(), [log_namespace])
64            }
65        }
66    }
67}
68
69/// Protobuf-specific decoding options.
70#[configurable_component]
71#[derive(Debug, Clone, PartialEq, Eq, Derivative)]
72#[derivative(Default)]
73pub struct ProtobufDeserializerOptions {
74    /// The path to the protobuf descriptor set file.
75    ///
76    /// This file is the output of `protoc -I <include path> -o <desc output path> <proto>`.
77    ///
78    /// For more information, see [How Buf images work](https://buf.build/docs/reference/images/#how-buf-images-work).
79    pub desc_file: PathBuf,
80
81    /// The name of the message type to use for serializing.
82    #[configurable(metadata(docs::examples = "package.Message"))]
83    pub message_type: String,
84
85    /// Use JSON field names (camelCase) instead of protobuf field names (snake_case).
86    ///
87    /// When enabled, the deserializer will output fields using their JSON names as defined
88    /// in the `.proto` file (for example, `jobDescription` instead of `job_description`).
89    ///
90    /// This is useful when working with data that needs to be converted to JSON or
91    /// when interfacing with systems that use JSON naming conventions.
92    #[serde(default, skip_serializing_if = "vector_core::serde::is_default")]
93    pub use_json_names: bool,
94}
95
96/// Deserializer that builds `Event`s from a byte frame containing protobuf.
97#[derive(Debug, Clone)]
98pub struct ProtobufDeserializer {
99    message_descriptor: MessageDescriptor,
100    options: Options,
101}
102
103impl ProtobufDeserializer {
104    /// Creates a new `ProtobufDeserializer`.
105    pub fn new(message_descriptor: MessageDescriptor) -> Self {
106        Self {
107            message_descriptor,
108            options: Default::default(),
109        }
110    }
111
112    /// Creates a new deserializer instance using the descriptor bytes directly.
113    pub fn new_from_bytes(
114        desc_bytes: &[u8],
115        message_type: &str,
116        options: Options,
117    ) -> vector_common::Result<Self> {
118        let message_descriptor = get_message_descriptor_from_bytes(desc_bytes, message_type)?;
119        Ok(Self {
120            message_descriptor,
121            options,
122        })
123    }
124}
125
126fn extract_vrl_value(
127    bytes: Bytes,
128    message_descriptor: &MessageDescriptor,
129    options: &Options,
130) -> vector_common::Result<Value> {
131    let dynamic_message = DynamicMessage::decode(message_descriptor.clone(), bytes)
132        .map_err(|error| format!("Error parsing protobuf: {error:?}"))?;
133
134    Ok(proto_to_value(
135        &prost_reflect::Value::Message(dynamic_message),
136        None,
137        options,
138    )?)
139}
140
141impl Deserializer for ProtobufDeserializer {
142    fn parse(
143        &self,
144        bytes: Bytes,
145        log_namespace: LogNamespace,
146    ) -> vector_common::Result<SmallVec<[Event; 1]>> {
147        let vrl_value = extract_vrl_value(bytes, &self.message_descriptor, &self.options)?;
148        let mut event = Event::Log(LogEvent::from(vrl_value));
149
150        let event = match log_namespace {
151            LogNamespace::Vector => event,
152            LogNamespace::Legacy => {
153                let timestamp = Utc::now();
154                if let Some(timestamp_key) = log_schema().timestamp_key_target_path() {
155                    let log = event.as_mut_log();
156                    if !log.contains(timestamp_key) {
157                        log.insert(timestamp_key, timestamp);
158                    }
159                }
160                event
161            }
162        };
163
164        Ok(smallvec![event])
165    }
166}
167
168impl TryFrom<&ProtobufDeserializerConfig> for ProtobufDeserializer {
169    type Error = vector_common::Error;
170    fn try_from(config: &ProtobufDeserializerConfig) -> vector_common::Result<Self> {
171        let message_descriptor =
172            get_message_descriptor(&config.protobuf.desc_file, &config.protobuf.message_type)?;
173        Ok(Self {
174            message_descriptor,
175            options: Options {
176                use_json_names: config.protobuf.use_json_names,
177            },
178        })
179    }
180}
181
182#[cfg(test)]
183mod tests {
184    // TODO: add test for bad file path & invalid message_type
185
186    use std::{env, fs, path::PathBuf};
187
188    use vector_core::config::log_schema;
189    use vrl::event_path;
190
191    use super::*;
192
193    fn test_data_dir() -> PathBuf {
194        PathBuf::from(env::var_os("CARGO_MANIFEST_DIR").unwrap()).join("tests/data/protobuf")
195    }
196
197    fn parse_and_validate(
198        protobuf_bin_message: String,
199        protobuf_desc_path: PathBuf,
200        message_type: &str,
201        validate_log: fn(&LogEvent),
202    ) {
203        let input = Bytes::from(protobuf_bin_message);
204        let message_descriptor = get_message_descriptor(&protobuf_desc_path, message_type).unwrap();
205        let deserializer = ProtobufDeserializer::new(message_descriptor);
206
207        for namespace in [LogNamespace::Legacy, LogNamespace::Vector] {
208            let events = deserializer.parse(input.clone(), namespace).unwrap();
209            let mut events = events.into_iter();
210
211            {
212                let event = events.next().unwrap();
213                let log = event.as_log();
214                validate_log(log);
215                assert_eq!(
216                    log.get(log_schema().timestamp_key_target_path().unwrap())
217                        .is_some(),
218                    namespace == LogNamespace::Legacy
219                );
220            }
221
222            assert_eq!(events.next(), None);
223        }
224    }
225
226    #[test]
227    fn deserialize_protobuf() {
228        let protobuf_bin_message_path = test_data_dir().join("pbs/person_someone.pb");
229        let protobuf_desc_path = test_data_dir().join("protos/test_protobuf.desc");
230        let message_type = "test_protobuf.Person";
231        let validate_log = |log: &LogEvent| {
232            assert_eq!(log["name"], "someone".into());
233            assert_eq!(
234                log["phones"].as_array().unwrap()[0].as_object().unwrap()["number"]
235                    .as_str()
236                    .unwrap(),
237                "123456"
238            );
239        };
240
241        parse_and_validate(
242            fs::read_to_string(protobuf_bin_message_path).unwrap(),
243            protobuf_desc_path,
244            message_type,
245            validate_log,
246        );
247    }
248
249    #[test]
250    fn deserialize_protobuf3() {
251        let protobuf_bin_message_path = test_data_dir().join("pbs/person_someone3.pb");
252        let protobuf_desc_path = test_data_dir().join("protos/test_protobuf3.desc");
253        let message_type = "test_protobuf3.Person";
254        let validate_log = |log: &LogEvent| {
255            assert_eq!(log["name"], "someone".into());
256            assert_eq!(
257                log["phones"].as_array().unwrap()[0].as_object().unwrap()["number"]
258                    .as_str()
259                    .unwrap(),
260                "1234"
261            );
262            assert_eq!(
263                log["data"].as_object().unwrap()["data_phone"],
264                "HOME".into()
265            );
266        };
267
268        parse_and_validate(
269            fs::read_to_string(protobuf_bin_message_path).unwrap(),
270            protobuf_desc_path,
271            message_type,
272            validate_log,
273        );
274    }
275
276    #[test]
277    fn deserialize_empty_buffer() {
278        let protobuf_bin_message = "".to_string();
279        let protobuf_desc_path = test_data_dir().join("protos/test_protobuf.desc");
280        let message_type = "test_protobuf.Person";
281        let validate_log = |log: &LogEvent| {
282            // No field will be set.
283            assert!(!log.contains(event_path!("name")));
284            assert!(!log.contains(event_path!("id")));
285            assert!(!log.contains(event_path!("email")));
286            assert!(!log.contains(event_path!("phones")));
287        };
288
289        parse_and_validate(
290            protobuf_bin_message,
291            protobuf_desc_path,
292            message_type,
293            validate_log,
294        );
295    }
296
297    #[test]
298    fn deserialize_error_invalid_protobuf() {
299        let input = Bytes::from("{ foo");
300        let message_descriptor = get_message_descriptor(
301            &test_data_dir().join("protos/test_protobuf.desc"),
302            "test_protobuf.Person",
303        )
304        .unwrap();
305        let deserializer = ProtobufDeserializer::new(message_descriptor);
306
307        for namespace in [LogNamespace::Legacy, LogNamespace::Vector] {
308            assert!(deserializer.parse(input.clone(), namespace).is_err());
309        }
310    }
311}
codecs/decoding/format/protobuf.rs

codecs/decoding/format/
protobuf.rs
