vector/codecs/encoding/
transformer.rs

1#![deny(missing_docs)]
2
3use core::fmt::Debug;
4use std::collections::BTreeMap;
5
6use chrono::{DateTime, Utc};
7use ordered_float::NotNan;
8use serde::{Deserialize, Deserializer};
9use vector_lib::{
10    configurable::configurable_component,
11    event::{LogEvent, MaybeAsLogMut},
12    lookup::{PathPrefix, event_path, lookup_v2::ConfigValuePath},
13    schema::meaning,
14};
15use vrl::{path::OwnedValuePath, value::Value};
16
17use crate::{event::Event, serde::is_default};
18
19/// Transformations to prepare an event for serialization.
20#[configurable_component(no_deser)]
21#[derive(Clone, Debug, Default, PartialEq, Eq)]
22pub struct Transformer {
23    /// List of fields that are included in the encoded event.
24    #[serde(default, skip_serializing_if = "is_default")]
25    only_fields: Option<Vec<ConfigValuePath>>,
26
27    /// List of fields that are excluded from the encoded event.
28    #[serde(default, skip_serializing_if = "is_default")]
29    except_fields: Option<Vec<ConfigValuePath>>,
30
31    /// Format used for timestamp fields.
32    #[serde(default, skip_serializing_if = "is_default")]
33    timestamp_format: Option<TimestampFormat>,
34}
35
36impl<'de> Deserialize<'de> for Transformer {
37    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
38    where
39        D: Deserializer<'de>,
40    {
41        #[derive(Deserialize)]
42        #[serde(deny_unknown_fields)]
43        struct TransformerInner {
44            #[serde(default)]
45            only_fields: Option<Vec<OwnedValuePath>>,
46            #[serde(default)]
47            except_fields: Option<Vec<OwnedValuePath>>,
48            #[serde(default)]
49            timestamp_format: Option<TimestampFormat>,
50        }
51
52        let inner: TransformerInner = Deserialize::deserialize(deserializer)?;
53        Self::new(
54            inner
55                .only_fields
56                .map(|v| v.iter().map(|p| ConfigValuePath(p.clone())).collect()),
57            inner
58                .except_fields
59                .map(|v| v.iter().map(|p| ConfigValuePath(p.clone())).collect()),
60            inner.timestamp_format,
61        )
62        .map_err(serde::de::Error::custom)
63    }
64}
65
66impl Transformer {
67    /// Creates a new `Transformer`.
68    ///
69    /// Returns `Err` if `only_fields` and `except_fields` fail validation, i.e. are not mutually
70    /// exclusive.
71    pub fn new(
72        only_fields: Option<Vec<ConfigValuePath>>,
73        except_fields: Option<Vec<ConfigValuePath>>,
74        timestamp_format: Option<TimestampFormat>,
75    ) -> Result<Self, crate::Error> {
76        Self::validate_fields(only_fields.as_ref(), except_fields.as_ref())?;
77
78        Ok(Self {
79            only_fields,
80            except_fields,
81            timestamp_format,
82        })
83    }
84
85    /// Get the `Transformer`'s `only_fields`.
86    #[cfg(test)]
87    pub const fn only_fields(&self) -> &Option<Vec<ConfigValuePath>> {
88        &self.only_fields
89    }
90
91    /// Get the `Transformer`'s `except_fields`.
92    pub const fn except_fields(&self) -> &Option<Vec<ConfigValuePath>> {
93        &self.except_fields
94    }
95
96    /// Get the `Transformer`'s `timestamp_format`.
97    pub const fn timestamp_format(&self) -> &Option<TimestampFormat> {
98        &self.timestamp_format
99    }
100
101    /// Check if `except_fields` and `only_fields` items are mutually exclusive.
102    ///
103    /// If an error is returned, the entire encoding configuration should be considered inoperable.
104    fn validate_fields(
105        only_fields: Option<&Vec<ConfigValuePath>>,
106        except_fields: Option<&Vec<ConfigValuePath>>,
107    ) -> crate::Result<()> {
108        if let (Some(only_fields), Some(except_fields)) = (only_fields, except_fields)
109            && except_fields
110                .iter()
111                .any(|f| only_fields.iter().any(|v| v == f))
112        {
113            return Err("`except_fields` and `only_fields` should be mutually exclusive.".into());
114        }
115        Ok(())
116    }
117
118    /// Prepare an event for serialization by the given transformation rules.
119    pub fn transform(&self, event: &mut Event) {
120        // Rules are currently applied to logs only.
121        if let Some(log) = event.maybe_as_log_mut() {
122            // Ordering in here should not matter.
123            self.apply_except_fields(log);
124            self.apply_only_fields(log);
125            self.apply_timestamp_format(log);
126        }
127    }
128
129    fn apply_only_fields(&self, log: &mut LogEvent) {
130        if let Some(only_fields) = self.only_fields.as_ref() {
131            let mut old_value = std::mem::replace(log.value_mut(), Value::Object(BTreeMap::new()));
132
133            for field in only_fields {
134                if let Some(value) = old_value.remove(field, true) {
135                    log.insert((PathPrefix::Event, field), value);
136                }
137            }
138
139            // We may need the service field to apply tags to emitted metrics after the log message has been pruned. If there
140            // is a service meaning, we move this value to `dropped_fields` in the metadata.
141            // If the field is still in the new log message after pruning it will have been removed from `old_value` above.
142            let service_path = log
143                .metadata()
144                .schema_definition()
145                .meaning_path(meaning::SERVICE);
146            if let Some(service_path) = service_path {
147                let mut new_log = LogEvent::from(old_value);
148                if let Some(service) = new_log.remove(service_path) {
149                    log.metadata_mut()
150                        .add_dropped_field(meaning::SERVICE.into(), service);
151                }
152            }
153        }
154    }
155
156    fn apply_except_fields(&self, log: &mut LogEvent) {
157        if let Some(except_fields) = self.except_fields.as_ref() {
158            for field in except_fields {
159                let value_path = &field.0;
160                let value = log.remove((PathPrefix::Event, value_path));
161
162                let service_path = log
163                    .metadata()
164                    .schema_definition()
165                    .meaning_path(meaning::SERVICE);
166                // If we are removing the service field we need to store this in a `dropped_fields` list as we may need to
167                // refer to this later when emitting metrics.
168                if let (Some(v), Some(service_path)) = (value, service_path)
169                    && service_path.path == *value_path
170                {
171                    log.metadata_mut()
172                        .add_dropped_field(meaning::SERVICE.into(), v);
173                }
174            }
175        }
176    }
177
178    fn format_timestamps<F, T>(&self, log: &mut LogEvent, extract: F)
179    where
180        F: Fn(&DateTime<Utc>) -> T,
181        T: Into<Value>,
182    {
183        if log.value().is_object() {
184            let mut unix_timestamps = Vec::new();
185            for (k, v) in log.all_event_fields().expect("must be an object") {
186                if let Value::Timestamp(ts) = v {
187                    unix_timestamps.push((k.clone(), extract(ts).into()));
188                }
189            }
190            for (k, v) in unix_timestamps {
191                log.parse_path_and_insert(k, v).unwrap();
192            }
193        } else {
194            // root is not an object
195            let timestamp = if let Value::Timestamp(ts) = log.value() {
196                Some(extract(ts))
197            } else {
198                None
199            };
200            if let Some(ts) = timestamp {
201                log.insert(event_path!(), ts.into());
202            }
203        }
204    }
205
206    fn apply_timestamp_format(&self, log: &mut LogEvent) {
207        if let Some(timestamp_format) = self.timestamp_format.as_ref() {
208            match timestamp_format {
209                TimestampFormat::Unix => self.format_timestamps(log, |ts| ts.timestamp()),
210                TimestampFormat::UnixMs => self.format_timestamps(log, |ts| ts.timestamp_millis()),
211                TimestampFormat::UnixUs => self.format_timestamps(log, |ts| ts.timestamp_micros()),
212                TimestampFormat::UnixNs => self.format_timestamps(log, |ts| {
213                    ts.timestamp_nanos_opt().expect("Timestamp out of range")
214                }),
215                TimestampFormat::UnixFloat => self.format_timestamps(log, |ts| {
216                    NotNan::new(ts.timestamp_micros() as f64 / 1e6).unwrap()
217                }),
218                // RFC3339 is the default serialization of a timestamp.
219                TimestampFormat::Rfc3339 => (),
220            }
221        }
222    }
223
224    /// Set the `except_fields` value.
225    ///
226    /// Returns `Err` if the new `except_fields` fail validation, i.e. are not mutually exclusive
227    /// with `only_fields`.
228    #[cfg(test)]
229    pub fn set_except_fields(
230        &mut self,
231        except_fields: Option<Vec<ConfigValuePath>>,
232    ) -> crate::Result<()> {
233        Self::validate_fields(self.only_fields.as_ref(), except_fields.as_ref())?;
234        self.except_fields = except_fields;
235        Ok(())
236    }
237}
238
239#[configurable_component]
240#[derive(Clone, Copy, Debug, Eq, PartialEq)]
241#[serde(rename_all = "snake_case")]
242/// The format in which a timestamp should be represented.
243pub enum TimestampFormat {
244    /// Represent the timestamp as a Unix timestamp.
245    Unix,
246
247    /// Represent the timestamp as a RFC 3339 timestamp.
248    Rfc3339,
249
250    /// Represent the timestamp as a Unix timestamp in milliseconds.
251    UnixMs,
252
253    /// Represent the timestamp as a Unix timestamp in microseconds
254    UnixUs,
255
256    /// Represent the timestamp as a Unix timestamp in nanoseconds.
257    UnixNs,
258
259    /// Represent the timestamp as a Unix timestamp in floating point.
260    UnixFloat,
261}
262
263#[cfg(test)]
264mod tests {
265    use std::{collections::BTreeMap, sync::Arc};
266
267    use indoc::indoc;
268    use vector_lib::{
269        btreemap,
270        config::{LogNamespace, log_schema},
271        lookup::path::parse_target_path,
272    };
273    use vrl::value::Kind;
274
275    use super::*;
276    use crate::config::schema;
277
278    #[test]
279    fn serialize() {
280        let string =
281            r#"{"only_fields":["a.b[0]"],"except_fields":["ignore_me"],"timestamp_format":"unix"}"#;
282
283        let transformer = serde_json::from_str::<Transformer>(string).unwrap();
284
285        let serialized = serde_json::to_string(&transformer).unwrap();
286
287        assert_eq!(string, serialized);
288    }
289
290    #[test]
291    fn serialize_empty() {
292        let string = "{}";
293
294        let transformer = serde_json::from_str::<Transformer>(string).unwrap();
295
296        let serialized = serde_json::to_string(&transformer).unwrap();
297
298        assert_eq!(string, serialized);
299    }
300
301    #[test]
302    fn deserialize_and_transform_except() {
303        let transformer: Transformer =
304            toml::from_str(r#"except_fields = ["a.b.c", "b", "c[0].y", "d.z", "e"]"#).unwrap();
305        let mut log = LogEvent::default();
306        {
307            log.insert("a", 1);
308            log.insert("a.b", 1);
309            log.insert("a.b.c", 1);
310            log.insert("a.b.d", 1);
311            log.insert("b[0]", 1);
312            log.insert("b[1].x", 1);
313            log.insert("c[0].x", 1);
314            log.insert("c[0].y", 1);
315            log.insert("d.z", 1);
316            log.insert("e.a", 1);
317            log.insert("e.b", 1);
318        }
319        let mut event = Event::from(log);
320        transformer.transform(&mut event);
321        assert!(!event.as_mut_log().contains("a.b.c"));
322        assert!(!event.as_mut_log().contains("b"));
323        assert!(!event.as_mut_log().contains("b[1].x"));
324        assert!(!event.as_mut_log().contains("c[0].y"));
325        assert!(!event.as_mut_log().contains("d.z"));
326        assert!(!event.as_mut_log().contains("e.a"));
327
328        assert!(event.as_mut_log().contains("a.b.d"));
329        assert!(event.as_mut_log().contains("c[0].x"));
330    }
331
332    #[test]
333    fn deserialize_and_transform_only() {
334        let transformer: Transformer =
335            toml::from_str(r#"only_fields = ["a.b.c", "b", "c[0].y", "\"g.z\""]"#).unwrap();
336        let mut log = LogEvent::default();
337        {
338            log.insert("a", 1);
339            log.insert("a.b", 1);
340            log.insert("a.b.c", 1);
341            log.insert("a.b.d", 1);
342            log.insert("b[0]", 1);
343            log.insert("b[1].x", 1);
344            log.insert("c[0].x", 1);
345            log.insert("c[0].y", 1);
346            log.insert("d.y", 1);
347            log.insert("d.z", 1);
348            log.insert("e[0]", 1);
349            log.insert("e[1]", 1);
350            log.insert("\"f.z\"", 1);
351            log.insert("\"g.z\"", 1);
352            log.insert("h", BTreeMap::new());
353            log.insert("i", Vec::<Value>::new());
354        }
355        let mut event = Event::from(log);
356        transformer.transform(&mut event);
357        assert!(event.as_mut_log().contains("a.b.c"));
358        assert!(event.as_mut_log().contains("b"));
359        assert!(event.as_mut_log().contains("b[1].x"));
360        assert!(event.as_mut_log().contains("c[0].y"));
361        assert!(event.as_mut_log().contains("\"g.z\""));
362
363        assert!(!event.as_mut_log().contains("a.b.d"));
364        assert!(!event.as_mut_log().contains("c[0].x"));
365        assert!(!event.as_mut_log().contains("d"));
366        assert!(!event.as_mut_log().contains("e"));
367        assert!(!event.as_mut_log().contains("f"));
368        assert!(!event.as_mut_log().contains("h"));
369        assert!(!event.as_mut_log().contains("i"));
370    }
371
372    #[test]
373    fn deserialize_and_transform_timestamp() {
374        let mut base = Event::Log(LogEvent::from("Demo"));
375        let timestamp = base
376            .as_mut_log()
377            .get((PathPrefix::Event, log_schema().timestamp_key().unwrap()))
378            .unwrap()
379            .clone();
380        let timestamp = timestamp.as_timestamp().unwrap();
381        base.as_mut_log()
382            .insert("another", Value::Timestamp(*timestamp));
383
384        let cases = [
385            ("unix", Value::from(timestamp.timestamp())),
386            ("unix_ms", Value::from(timestamp.timestamp_millis())),
387            ("unix_us", Value::from(timestamp.timestamp_micros())),
388            (
389                "unix_ns",
390                Value::from(timestamp.timestamp_nanos_opt().unwrap()),
391            ),
392            (
393                "unix_float",
394                Value::from(timestamp.timestamp_micros() as f64 / 1e6),
395            ),
396        ];
397        for (fmt, expected) in cases {
398            let config: String = format!(r#"timestamp_format = "{fmt}""#);
399            let transformer: Transformer = toml::from_str(&config).unwrap();
400            let mut event = base.clone();
401            transformer.transform(&mut event);
402            let log = event.as_mut_log();
403
404            for actual in [
405                // original key
406                log.get((PathPrefix::Event, log_schema().timestamp_key().unwrap()))
407                    .unwrap(),
408                // second key
409                log.get("another").unwrap(),
410            ] {
411                // type matches
412                assert_eq!(expected.kind_str(), actual.kind_str());
413                // value matches
414                assert_eq!(&expected, actual);
415            }
416        }
417    }
418
419    #[test]
420    fn exclusivity_violation() {
421        let config: std::result::Result<Transformer, _> = toml::from_str(indoc! {r#"
422            except_fields = ["Doop"]
423            only_fields = ["Doop"]
424        "#});
425        assert!(config.is_err())
426    }
427
428    #[test]
429    fn deny_unknown_fields() {
430        // We're only checking this explicitly because of our custom deserializer arrangement to
431        // make it possible to throw the exclusivity error during deserialization, to ensure that we
432        // enforce this on the top-level `Transformer` type even though it has to be applied at the
433        // intermediate deserialization stage, on `TransformerInner`.
434        let config: std::result::Result<Transformer, _> = toml::from_str(indoc! {r#"
435            onlyfields = ["Doop"]
436        "#});
437        assert!(config.is_err())
438    }
439
440    #[test]
441    fn only_fields_with_service() {
442        let transformer: Transformer = toml::from_str(r#"only_fields = ["message"]"#).unwrap();
443        let mut log = LogEvent::default();
444        {
445            log.insert("message", 1);
446            log.insert("thing.service", "carrot");
447        }
448
449        let schema = schema::Definition::new_with_default_metadata(
450            Kind::object(btreemap! {
451                "thing" => Kind::object(btreemap! {
452                    "service" => Kind::bytes(),
453                })
454            }),
455            [LogNamespace::Vector],
456        );
457
458        let schema = schema.with_meaning(parse_target_path("thing.service").unwrap(), "service");
459
460        let mut event = Event::from(log);
461
462        event
463            .metadata_mut()
464            .set_schema_definition(&Arc::new(schema));
465
466        transformer.transform(&mut event);
467        assert!(event.as_mut_log().contains("message"));
468
469        // Event no longer contains the service field.
470        assert!(!event.as_mut_log().contains("thing.service"));
471
472        // But we can still get the service by meaning.
473        assert_eq!(
474            &Value::from("carrot"),
475            event.as_log().get_by_meaning("service").unwrap()
476        );
477    }
478
479    #[test]
480    fn except_fields_with_service() {
481        let transformer: Transformer =
482            toml::from_str(r#"except_fields = ["thing.service"]"#).unwrap();
483        let mut log = LogEvent::default();
484        {
485            log.insert("message", 1);
486            log.insert("thing.service", "carrot");
487        }
488
489        let schema = schema::Definition::new_with_default_metadata(
490            Kind::object(btreemap! {
491                "thing" => Kind::object(btreemap! {
492                    "service" => Kind::bytes(),
493                })
494            }),
495            [LogNamespace::Vector],
496        );
497
498        let schema = schema.with_meaning(parse_target_path("thing.service").unwrap(), "service");
499
500        let mut event = Event::from(log);
501
502        event
503            .metadata_mut()
504            .set_schema_definition(&Arc::new(schema));
505
506        transformer.transform(&mut event);
507        assert!(event.as_mut_log().contains("message"));
508
509        // Event no longer contains the service field.
510        assert!(!event.as_mut_log().contains("thing.service"));
511
512        // But we can still get the service by meaning.
513        assert_eq!(
514            &Value::from("carrot"),
515            event.as_log().get_by_meaning("service").unwrap()
516        );
517    }
518}
vector/codecs/encoding/transformer.rs

vector/codecs/encoding/
transformer.rs
