vector_core/event/
log_event.rs

1use std::{
2    collections::HashMap,
3    convert::{TryFrom, TryInto},
4    fmt::Debug,
5    iter::FromIterator,
6    mem::size_of,
7    num::NonZeroUsize,
8    sync::{Arc, LazyLock},
9};
10
11use bytes::Bytes;
12use chrono::Utc;
13use crossbeam_utils::atomic::AtomicCell;
14use lookup::{PathPrefix, lookup_v2::TargetPath, metadata_path, path};
15use serde::{Deserialize, Serialize, Serializer};
16use vector_common::{
17    EventDataEq,
18    byte_size_of::ByteSizeOf,
19    internal_event::{OptionalTag, TaggedEventsSent},
20    json_size::{JsonSize, NonZeroJsonSize},
21    request_metadata::GetEventCountTags,
22};
23use vrl::{
24    event_path, owned_value_path,
25    path::{OwnedTargetPath, PathParseError, parse_target_path},
26};
27
28use super::{
29    EventFinalizers, Finalizable, KeyString, MergeFinalizable, ObjectMap, Value,
30    estimated_json_encoded_size_of::EstimatedJsonEncodedSizeOf,
31    finalization::{BatchNotifier, EventFinalizer},
32    metadata::EventMetadata,
33    util,
34};
35use crate::{
36    config::{LogNamespace, log_schema, telemetry},
37    event::{
38        MaybeAsLogMut,
39        util::log::{all_fields, all_fields_skip_array_elements, all_metadata_fields},
40    },
41};
42
43static VECTOR_SOURCE_TYPE_PATH: LazyLock<Option<OwnedTargetPath>> = LazyLock::new(|| {
44    Some(OwnedTargetPath::metadata(owned_value_path!(
45        "vector",
46        "source_type"
47    )))
48});
49
50#[derive(Debug, Deserialize)]
51struct Inner {
52    #[serde(flatten)]
53    fields: Value,
54
55    #[serde(skip)]
56    size_cache: AtomicCell<Option<NonZeroUsize>>,
57
58    #[serde(skip)]
59    json_encoded_size_cache: AtomicCell<Option<NonZeroJsonSize>>,
60}
61
62impl Inner {
63    fn invalidate(&self) {
64        self.size_cache.store(None);
65        self.json_encoded_size_cache.store(None);
66    }
67
68    fn as_value(&self) -> &Value {
69        &self.fields
70    }
71}
72
73impl ByteSizeOf for Inner {
74    fn size_of(&self) -> usize {
75        self.size_cache
76            .load()
77            .unwrap_or_else(|| {
78                let size = size_of::<Self>() + self.allocated_bytes();
79                // The size of self will always be non-zero, and
80                // adding the allocated bytes cannot make it overflow
81                // since `usize` has a range the same as pointer
82                // space. Hence, the expect below cannot fail.
83                let size = NonZeroUsize::new(size).expect("Size cannot be zero");
84                self.size_cache.store(Some(size));
85                size
86            })
87            .into()
88    }
89
90    fn allocated_bytes(&self) -> usize {
91        self.fields.allocated_bytes()
92    }
93}
94
95impl EstimatedJsonEncodedSizeOf for Inner {
96    fn estimated_json_encoded_size_of(&self) -> JsonSize {
97        self.json_encoded_size_cache
98            .load()
99            .unwrap_or_else(|| {
100                let size = self.fields.estimated_json_encoded_size_of();
101                let size = NonZeroJsonSize::new(size).expect("Size cannot be zero");
102
103                self.json_encoded_size_cache.store(Some(size));
104                size
105            })
106            .into()
107    }
108}
109
110impl Clone for Inner {
111    fn clone(&self) -> Self {
112        Self {
113            fields: self.fields.clone(),
114            // This clone is only ever used in combination with
115            // `Arc::make_mut`, so don't bother fetching the size
116            // cache to copy it since it will be invalidated anyways.
117            size_cache: None.into(),
118
119            // This clone is only ever used in combination with
120            // `Arc::make_mut`, so don't bother fetching the size
121            // cache to copy it since it will be invalidated anyways.
122            json_encoded_size_cache: None.into(),
123        }
124    }
125}
126
127impl Default for Inner {
128    fn default() -> Self {
129        Self {
130            // **IMPORTANT:** Due to numerous legacy reasons this **must** be a Map variant.
131            fields: Value::Object(Default::default()),
132            size_cache: Default::default(),
133            json_encoded_size_cache: Default::default(),
134        }
135    }
136}
137
138impl From<Value> for Inner {
139    fn from(fields: Value) -> Self {
140        Self {
141            fields,
142            size_cache: Default::default(),
143            json_encoded_size_cache: Default::default(),
144        }
145    }
146}
147
148impl PartialEq for Inner {
149    fn eq(&self, other: &Self) -> bool {
150        self.fields.eq(&other.fields)
151    }
152}
153
154#[derive(Clone, Debug, Default, Deserialize, PartialEq)]
155pub struct LogEvent {
156    #[serde(flatten)]
157    inner: Arc<Inner>,
158
159    #[serde(skip)]
160    metadata: EventMetadata,
161}
162
163impl LogEvent {
164    /// This used to be the implementation for `LogEvent::from(&'str)`, but this is now only
165    /// valid for `LogNamespace::Legacy`
166    pub fn from_str_legacy(msg: impl Into<String>) -> Self {
167        let mut log = LogEvent::default();
168        log.maybe_insert(log_schema().message_key_target_path(), msg.into());
169
170        if let Some(timestamp_key) = log_schema().timestamp_key_target_path() {
171            log.insert(timestamp_key, Utc::now());
172        }
173
174        log
175    }
176
177    /// This used to be the implementation for `LogEvent::from(Bytes)`, but this is now only
178    /// valid for `LogNamespace::Legacy`
179    pub fn from_bytes_legacy(msg: &Bytes) -> Self {
180        Self::from_str_legacy(String::from_utf8_lossy(msg.as_ref()).to_string())
181    }
182
183    pub fn value(&self) -> &Value {
184        self.inner.as_ref().as_value()
185    }
186
187    pub fn value_mut(&mut self) -> &mut Value {
188        let result = Arc::make_mut(&mut self.inner);
189        // We MUST invalidate the inner size cache when making a
190        // mutable copy, since the _next_ action will modify the data.
191        result.invalidate();
192        &mut result.fields
193    }
194
195    pub fn metadata(&self) -> &EventMetadata {
196        &self.metadata
197    }
198
199    pub fn metadata_mut(&mut self) -> &mut EventMetadata {
200        &mut self.metadata
201    }
202
203    /// This detects the log namespace used at runtime by checking for the existence
204    /// of the read-only "vector" metadata, which only exists (and is required to exist)
205    /// with the `Vector` log namespace.
206    pub fn namespace(&self) -> LogNamespace {
207        if self.contains((PathPrefix::Metadata, path!("vector"))) {
208            LogNamespace::Vector
209        } else {
210            LogNamespace::Legacy
211        }
212    }
213}
214
215impl ByteSizeOf for LogEvent {
216    fn allocated_bytes(&self) -> usize {
217        self.inner.size_of() + self.metadata.allocated_bytes()
218    }
219}
220
221impl Finalizable for LogEvent {
222    fn take_finalizers(&mut self) -> EventFinalizers {
223        self.metadata.take_finalizers()
224    }
225}
226
227impl MergeFinalizable for LogEvent {
228    fn merge_finalizers(&mut self, finalizers: EventFinalizers) {
229        self.metadata.merge_finalizers(finalizers);
230    }
231}
232
233impl EstimatedJsonEncodedSizeOf for LogEvent {
234    fn estimated_json_encoded_size_of(&self) -> JsonSize {
235        self.inner.estimated_json_encoded_size_of()
236    }
237}
238
239impl GetEventCountTags for LogEvent {
240    fn get_tags(&self) -> TaggedEventsSent {
241        let source = if telemetry().tags().emit_source {
242            self.metadata().source_id().cloned().into()
243        } else {
244            OptionalTag::Ignored
245        };
246
247        let service = if telemetry().tags().emit_service {
248            self.get_by_meaning("service")
249                .map(|value| value.to_string_lossy().to_string())
250                .into()
251        } else {
252            OptionalTag::Ignored
253        };
254
255        TaggedEventsSent { source, service }
256    }
257}
258
259impl LogEvent {
260    #[must_use]
261    pub fn new_with_metadata(metadata: EventMetadata) -> Self {
262        Self {
263            inner: Default::default(),
264            metadata,
265        }
266    }
267
268    ///  Create a `LogEvent` from a `Value` and `EventMetadata`
269    pub fn from_parts(value: Value, metadata: EventMetadata) -> Self {
270        Self {
271            inner: Arc::new(value.into()),
272            metadata,
273        }
274    }
275
276    ///  Create a `LogEvent` from an `ObjectMap` and `EventMetadata`
277    pub fn from_map(map: ObjectMap, metadata: EventMetadata) -> Self {
278        let inner = Arc::new(Inner::from(Value::Object(map)));
279        Self { inner, metadata }
280    }
281
282    /// Convert a `LogEvent` into a tuple of its components
283    pub fn into_parts(mut self) -> (Value, EventMetadata) {
284        self.value_mut();
285
286        let value = Arc::try_unwrap(self.inner)
287            .unwrap_or_else(|_| unreachable!("inner fields already cloned after owning"))
288            .fields;
289        let metadata = self.metadata;
290        (value, metadata)
291    }
292
293    #[must_use]
294    pub fn with_batch_notifier(mut self, batch: &BatchNotifier) -> Self {
295        self.metadata = self.metadata.with_batch_notifier(batch);
296        self
297    }
298
299    #[must_use]
300    pub fn with_batch_notifier_option(mut self, batch: &Option<BatchNotifier>) -> Self {
301        self.metadata = self.metadata.with_batch_notifier_option(batch);
302        self
303    }
304
305    pub fn add_finalizer(&mut self, finalizer: EventFinalizer) {
306        self.metadata.add_finalizer(finalizer);
307    }
308
309    /// Parse the specified `path` and if there are no parsing errors, attempt to get a reference to a value.
310    /// # Errors
311    /// Will return an error if path parsing failed.
312    pub fn parse_path_and_get_value(
313        &self,
314        path: impl AsRef<str>,
315    ) -> Result<Option<&Value>, PathParseError> {
316        parse_target_path(path.as_ref()).map(|path| self.get(&path))
317    }
318
319    #[allow(clippy::needless_pass_by_value)] // TargetPath is always a reference
320    pub fn get<'a>(&self, key: impl TargetPath<'a>) -> Option<&Value> {
321        match key.prefix() {
322            PathPrefix::Event => self.inner.fields.get(key.value_path()),
323            PathPrefix::Metadata => self.metadata.value().get(key.value_path()),
324        }
325    }
326
327    /// Retrieves the value of a field based on it's meaning.
328    /// This will first check if the value has previously been dropped. It is worth being
329    /// aware that if the field has been dropped and then somehow re-added, we still fetch
330    /// the dropped value here.
331    pub fn get_by_meaning(&self, meaning: impl AsRef<str>) -> Option<&Value> {
332        self.metadata().dropped_field(&meaning).or_else(|| {
333            self.metadata()
334                .schema_definition()
335                .meaning_path(meaning.as_ref())
336                .and_then(|path| self.get(path))
337        })
338    }
339
340    /// Retrieves the mutable value of a field based on it's meaning.
341    /// Note that this does _not_ check the dropped fields, unlike `get_by_meaning`, since the
342    /// purpose of the mutable reference is to be able to modify the value and modifying the dropped
343    /// fields has no effect on the resulting event.
344    pub fn get_mut_by_meaning(&mut self, meaning: impl AsRef<str>) -> Option<&mut Value> {
345        Arc::clone(self.metadata.schema_definition())
346            .meaning_path(meaning.as_ref())
347            .and_then(|path| self.get_mut(path))
348    }
349
350    /// Retrieves the target path of a field based on the specified `meaning`.
351    pub fn find_key_by_meaning(&self, meaning: impl AsRef<str>) -> Option<&OwnedTargetPath> {
352        self.metadata()
353            .schema_definition()
354            .meaning_path(meaning.as_ref())
355    }
356
357    #[allow(clippy::needless_pass_by_value)] // TargetPath is always a reference
358    pub fn get_mut<'a>(&mut self, path: impl TargetPath<'a>) -> Option<&mut Value> {
359        match path.prefix() {
360            PathPrefix::Event => self.value_mut().get_mut(path.value_path()),
361            PathPrefix::Metadata => self.metadata.value_mut().get_mut(path.value_path()),
362        }
363    }
364
365    #[allow(clippy::needless_pass_by_value)] // TargetPath is always a reference
366    pub fn contains<'a>(&self, path: impl TargetPath<'a>) -> bool {
367        match path.prefix() {
368            PathPrefix::Event => self.value().contains(path.value_path()),
369            PathPrefix::Metadata => self.metadata.value().contains(path.value_path()),
370        }
371    }
372
373    /// Parse the specified `path` and if there are no parsing errors, attempt to insert the specified `value`.
374    ///
375    /// # Errors
376    /// Will return an error if path parsing failed.
377    pub fn parse_path_and_insert(
378        &mut self,
379        path: impl AsRef<str>,
380        value: impl Into<Value>,
381    ) -> Result<Option<Value>, PathParseError> {
382        let target_path = parse_target_path(path.as_ref())?;
383        Ok(self.insert(&target_path, value))
384    }
385
386    #[allow(clippy::needless_pass_by_value)] // TargetPath is always a reference
387    pub fn insert<'a>(
388        &mut self,
389        path: impl TargetPath<'a>,
390        value: impl Into<Value>,
391    ) -> Option<Value> {
392        match path.prefix() {
393            PathPrefix::Event => self.value_mut().insert(path.value_path(), value.into()),
394            PathPrefix::Metadata => self
395                .metadata
396                .value_mut()
397                .insert(path.value_path(), value.into()),
398        }
399    }
400
401    pub fn maybe_insert<'a>(&mut self, path: Option<impl TargetPath<'a>>, value: impl Into<Value>) {
402        if let Some(path) = path {
403            self.insert(path, value);
404        }
405    }
406
407    // deprecated - using this means the schema is unknown
408    pub fn try_insert<'a>(&mut self, path: impl TargetPath<'a>, value: impl Into<Value>) {
409        if !self.contains(path.clone()) {
410            self.insert(path, value);
411        }
412    }
413
414    /// Rename a key
415    ///
416    /// If `to_key` already exists in the structure its value will be overwritten.
417    pub fn rename_key<'a>(&mut self, from: impl TargetPath<'a>, to: impl TargetPath<'a>) {
418        if let Some(val) = self.remove(from) {
419            self.insert(to, val);
420        }
421    }
422
423    pub fn remove<'a>(&mut self, path: impl TargetPath<'a>) -> Option<Value> {
424        self.remove_prune(path, false)
425    }
426
427    #[allow(clippy::needless_pass_by_value)] // TargetPath is always a reference
428    pub fn remove_prune<'a>(&mut self, path: impl TargetPath<'a>, prune: bool) -> Option<Value> {
429        match path.prefix() {
430            PathPrefix::Event => self.value_mut().remove(path.value_path(), prune),
431            PathPrefix::Metadata => self.metadata.value_mut().remove(path.value_path(), prune),
432        }
433    }
434
435    pub fn keys(&self) -> Option<impl Iterator<Item = KeyString> + '_> {
436        match &self.inner.fields {
437            Value::Object(map) => Some(util::log::keys(map)),
438            _ => None,
439        }
440    }
441
442    /// If the event root value is a map, build and return an iterator to event field and value pairs.
443    /// TODO: Ideally this should return target paths to be consistent with other `LogEvent` methods.
444    pub fn all_event_fields(
445        &self,
446    ) -> Option<impl Iterator<Item = (KeyString, &Value)> + Serialize> {
447        self.as_map().map(all_fields)
448    }
449
450    /// Similar to [`LogEvent::all_event_fields`], but doesn't traverse individual array elements.
451    pub fn all_event_fields_skip_array_elements(
452        &self,
453    ) -> Option<impl Iterator<Item = (KeyString, &Value)> + Serialize> {
454        self.as_map().map(all_fields_skip_array_elements)
455    }
456
457    /// If the metadata root value is a map, build and return an iterator to metadata field and value pairs.
458    /// TODO: Ideally this should return target paths to be consistent with other `LogEvent` methods.
459    pub fn all_metadata_fields(
460        &self,
461    ) -> Option<impl Iterator<Item = (KeyString, &Value)> + Serialize> {
462        match self.metadata.value() {
463            Value::Object(metadata_map) => Some(all_metadata_fields(metadata_map)),
464            _ => None,
465        }
466    }
467
468    /// Returns an iterator of all fields if the value is an Object. Otherwise, a single field is
469    /// returned with a "message" key. Field names that are could be interpreted as alternate paths
470    /// (i.e. containing periods, square brackets, etc) are quoted.
471    pub fn convert_to_fields(&self) -> impl Iterator<Item = (KeyString, &Value)> + Serialize {
472        if let Some(map) = self.as_map() {
473            util::log::all_fields(map)
474        } else {
475            util::log::all_fields_non_object_root(self.value())
476        }
477    }
478
479    /// Returns an iterator of all fields if the value is an Object. Otherwise, a single field is
480    /// returned with a "message" key. Field names are not quoted.
481    pub fn convert_to_fields_unquoted(
482        &self,
483    ) -> impl Iterator<Item = (KeyString, &Value)> + Serialize {
484        if let Some(map) = self.as_map() {
485            util::log::all_fields_unquoted(map)
486        } else {
487            util::log::all_fields_non_object_root(self.value())
488        }
489    }
490
491    pub fn is_empty_object(&self) -> bool {
492        if let Some(map) = self.as_map() {
493            map.is_empty()
494        } else {
495            false
496        }
497    }
498
499    pub fn as_map(&self) -> Option<&ObjectMap> {
500        match self.value() {
501            Value::Object(map) => Some(map),
502            _ => None,
503        }
504    }
505
506    pub fn as_map_mut(&mut self) -> Option<&mut ObjectMap> {
507        match self.value_mut() {
508            Value::Object(map) => Some(map),
509            _ => None,
510        }
511    }
512
513    /// Merge all fields specified at `fields` from `incoming` to `current`.
514    /// Note that `fields` containing dots and other special characters will be treated as a single segment.
515    pub fn merge(&mut self, mut incoming: LogEvent, fields: &[impl AsRef<str>]) {
516        for field in fields {
517            let field_path = event_path!(field.as_ref());
518            let Some(incoming_val) = incoming.remove(field_path) else {
519                continue;
520            };
521            match self.get_mut(field_path) {
522                None => {
523                    self.insert(field_path, incoming_val);
524                }
525                Some(current_val) => current_val.merge(incoming_val),
526            }
527        }
528        self.metadata.merge(incoming.metadata);
529    }
530}
531
532/// Log Namespace utility methods. These can only be used when an event has a
533/// valid schema definition set (which should be on every event in transforms and sinks).
534impl LogEvent {
535    /// Fetches the "message" path of the event. This is either from the "message" semantic meaning (Vector namespace)
536    /// or from the message key set on the "Global Log Schema" (Legacy namespace).
537    pub fn message_path(&self) -> Option<&OwnedTargetPath> {
538        match self.namespace() {
539            LogNamespace::Vector => self.find_key_by_meaning("message"),
540            LogNamespace::Legacy => log_schema().message_key_target_path(),
541        }
542    }
543
544    /// Fetches the "timestamp" path of the event. This is either from the "timestamp" semantic meaning (Vector namespace)
545    /// or from the timestamp key set on the "Global Log Schema" (Legacy namespace).
546    pub fn timestamp_path(&self) -> Option<&OwnedTargetPath> {
547        match self.namespace() {
548            LogNamespace::Vector => self.find_key_by_meaning("timestamp"),
549            LogNamespace::Legacy => log_schema().timestamp_key_target_path(),
550        }
551    }
552
553    /// Fetches the `host` path of the event. This is either from the "host" semantic meaning (Vector namespace)
554    /// or from the host key set on the "Global Log Schema" (Legacy namespace).
555    pub fn host_path(&self) -> Option<&OwnedTargetPath> {
556        match self.namespace() {
557            LogNamespace::Vector => self.find_key_by_meaning("host"),
558            LogNamespace::Legacy => log_schema().host_key_target_path(),
559        }
560    }
561
562    /// Fetches the `source_type` path of the event. This is either from the `source_type` Vector metadata field (Vector namespace)
563    /// or from the `source_type` key set on the "Global Log Schema" (Legacy namespace).
564    pub fn source_type_path(&self) -> Option<&OwnedTargetPath> {
565        match self.namespace() {
566            LogNamespace::Vector => VECTOR_SOURCE_TYPE_PATH.as_ref(),
567            LogNamespace::Legacy => log_schema().source_type_key_target_path(),
568        }
569    }
570
571    /// Fetches the `message` of the event. This is either from the "message" semantic meaning (Vector namespace)
572    /// or from the message key set on the "Global Log Schema" (Legacy namespace).
573    pub fn get_message(&self) -> Option<&Value> {
574        match self.namespace() {
575            LogNamespace::Vector => self.get_by_meaning("message"),
576            LogNamespace::Legacy => log_schema()
577                .message_key_target_path()
578                .and_then(|key| self.get(key)),
579        }
580    }
581
582    /// Fetches the `timestamp` of the event. This is either from the "timestamp" semantic meaning (Vector namespace)
583    /// or from the timestamp key set on the "Global Log Schema" (Legacy namespace).
584    pub fn get_timestamp(&self) -> Option<&Value> {
585        match self.namespace() {
586            LogNamespace::Vector => self.get_by_meaning("timestamp"),
587            LogNamespace::Legacy => log_schema()
588                .timestamp_key_target_path()
589                .and_then(|key| self.get(key)),
590        }
591    }
592
593    /// Removes the `timestamp` from the event. This is either from the "timestamp" semantic meaning (Vector namespace)
594    /// or from the timestamp key set on the "Global Log Schema" (Legacy namespace).
595    pub fn remove_timestamp(&mut self) -> Option<Value> {
596        self.timestamp_path()
597            .cloned()
598            .and_then(|key| self.remove(&key))
599    }
600
601    /// Fetches the `host` of the event. This is either from the "host" semantic meaning (Vector namespace)
602    /// or from the host key set on the "Global Log Schema" (Legacy namespace).
603    pub fn get_host(&self) -> Option<&Value> {
604        match self.namespace() {
605            LogNamespace::Vector => self.get_by_meaning("host"),
606            LogNamespace::Legacy => log_schema()
607                .host_key_target_path()
608                .and_then(|key| self.get(key)),
609        }
610    }
611
612    /// Fetches the `source_type` of the event. This is either from the `source_type` Vector metadata field (Vector namespace)
613    /// or from the `source_type` key set on the "Global Log Schema" (Legacy namespace).
614    pub fn get_source_type(&self) -> Option<&Value> {
615        match self.namespace() {
616            LogNamespace::Vector => self.get(metadata_path!("vector", "source_type")),
617            LogNamespace::Legacy => log_schema()
618                .source_type_key_target_path()
619                .and_then(|key| self.get(key)),
620        }
621    }
622}
623
624impl MaybeAsLogMut for LogEvent {
625    fn maybe_as_log_mut(&mut self) -> Option<&mut LogEvent> {
626        Some(self)
627    }
628}
629
630impl EventDataEq for LogEvent {
631    fn event_data_eq(&self, other: &Self) -> bool {
632        self.inner.fields == other.inner.fields && self.metadata.event_data_eq(&other.metadata)
633    }
634}
635
636#[cfg(any(test, feature = "test"))]
637mod test_utils {
638    use super::{Bytes, LogEvent, Utc, log_schema};
639
640    // these rely on the global log schema, which is no longer supported when using the
641    // "LogNamespace::Vector" namespace.
642    // The tests that rely on this are testing the "Legacy" log namespace. As these
643    // tests are updated, they should be migrated away from using these implementations
644    // to make it more clear which namespace is being used
645
646    impl From<Bytes> for LogEvent {
647        fn from(message: Bytes) -> Self {
648            let mut log = LogEvent::default();
649            log.maybe_insert(log_schema().message_key_target_path(), message);
650            if let Some(timestamp_key) = log_schema().timestamp_key_target_path() {
651                log.insert(timestamp_key, Utc::now());
652            }
653            log
654        }
655    }
656
657    impl From<&str> for LogEvent {
658        fn from(message: &str) -> Self {
659            message.to_owned().into()
660        }
661    }
662
663    impl From<String> for LogEvent {
664        fn from(message: String) -> Self {
665            Bytes::from(message).into()
666        }
667    }
668}
669
670impl From<Value> for LogEvent {
671    fn from(value: Value) -> Self {
672        Self::from_parts(value, EventMetadata::default())
673    }
674}
675
676impl From<ObjectMap> for LogEvent {
677    fn from(map: ObjectMap) -> Self {
678        Self::from_parts(Value::Object(map), EventMetadata::default())
679    }
680}
681
682impl From<HashMap<KeyString, Value>> for LogEvent {
683    fn from(map: HashMap<KeyString, Value>) -> Self {
684        Self::from_parts(
685            Value::Object(map.into_iter().collect::<ObjectMap>()),
686            EventMetadata::default(),
687        )
688    }
689}
690
691impl TryFrom<serde_json::Value> for LogEvent {
692    type Error = crate::Error;
693
694    fn try_from(map: serde_json::Value) -> Result<Self, Self::Error> {
695        match map {
696            serde_json::Value::Object(fields) => Ok(LogEvent::from(
697                fields
698                    .into_iter()
699                    .map(|(k, v)| (k.into(), v.into()))
700                    .collect::<ObjectMap>(),
701            )),
702            _ => Err(crate::Error::from(
703                "Attempted to convert non-Object JSON into a LogEvent.",
704            )),
705        }
706    }
707}
708
709impl TryInto<serde_json::Value> for LogEvent {
710    type Error = crate::Error;
711
712    fn try_into(self) -> Result<serde_json::Value, Self::Error> {
713        Ok(serde_json::to_value(&self.inner.fields)?)
714    }
715}
716
717#[cfg(any(test, feature = "test"))]
718impl<T> std::ops::Index<T> for LogEvent
719where
720    T: AsRef<str>,
721{
722    type Output = Value;
723
724    fn index(&self, key: T) -> &Value {
725        self.parse_path_and_get_value(key.as_ref())
726            .ok()
727            .flatten()
728            .unwrap_or_else(|| panic!("Key is not found: {:?}", key.as_ref()))
729    }
730}
731
732impl<K, V> Extend<(K, V)> for LogEvent
733where
734    K: AsRef<str>,
735    V: Into<Value>,
736{
737    fn extend<I: IntoIterator<Item = (K, V)>>(&mut self, iter: I) {
738        for (k, v) in iter {
739            if let Ok(path) = parse_target_path(k.as_ref()) {
740                self.insert(&path, v.into());
741            }
742        }
743    }
744}
745
746// Allow converting any kind of appropriate key/value iterator directly into a LogEvent.
747impl<K: AsRef<str>, V: Into<Value>> FromIterator<(K, V)> for LogEvent {
748    fn from_iter<T: IntoIterator<Item = (K, V)>>(iter: T) -> Self {
749        let mut log_event = Self::default();
750        log_event.extend(iter);
751        log_event
752    }
753}
754
755impl Serialize for LogEvent {
756    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
757    where
758        S: Serializer,
759    {
760        self.value().serialize(serializer)
761    }
762}
763
764// Tracing owned target paths used for tracing to log event conversions.
765struct TracingTargetPaths {
766    pub(crate) timestamp: OwnedTargetPath,
767    pub(crate) kind: OwnedTargetPath,
768    pub(crate) module_path: OwnedTargetPath,
769    pub(crate) level: OwnedTargetPath,
770    pub(crate) target: OwnedTargetPath,
771}
772
773/// Lazily initialized singleton.
774static TRACING_TARGET_PATHS: LazyLock<TracingTargetPaths> = LazyLock::new(|| TracingTargetPaths {
775    timestamp: OwnedTargetPath::event(owned_value_path!("timestamp")),
776    kind: OwnedTargetPath::event(owned_value_path!("metadata", "kind")),
777    level: OwnedTargetPath::event(owned_value_path!("metadata", "level")),
778    module_path: OwnedTargetPath::event(owned_value_path!("metadata", "module_path")),
779    target: OwnedTargetPath::event(owned_value_path!("metadata", "target")),
780});
781
782impl From<&tracing::Event<'_>> for LogEvent {
783    fn from(event: &tracing::Event<'_>) -> Self {
784        let now = chrono::Utc::now();
785        let mut maker = LogEvent::default();
786        event.record(&mut maker);
787
788        let mut log = maker;
789        log.insert(&TRACING_TARGET_PATHS.timestamp, now);
790
791        let meta = event.metadata();
792        log.insert(
793            &TRACING_TARGET_PATHS.kind,
794            if meta.is_event() {
795                Value::Bytes("event".to_string().into())
796            } else if meta.is_span() {
797                Value::Bytes("span".to_string().into())
798            } else {
799                Value::Null
800            },
801        );
802        log.insert(&TRACING_TARGET_PATHS.level, meta.level().to_string());
803        log.insert(
804            &TRACING_TARGET_PATHS.module_path,
805            meta.module_path()
806                .map_or(Value::Null, |mp| Value::Bytes(mp.to_string().into())),
807        );
808        log.insert(&TRACING_TARGET_PATHS.target, meta.target().to_string());
809        log
810    }
811}
812
813/// Note that `tracing::field::Field` containing dots and other special characters will be treated as a single segment.
814impl tracing::field::Visit for LogEvent {
815    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
816        self.insert(event_path!(field.name()), value.to_string());
817    }
818
819    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn Debug) {
820        self.insert(event_path!(field.name()), format!("{value:?}"));
821    }
822
823    fn record_i64(&mut self, field: &tracing::field::Field, value: i64) {
824        self.insert(event_path!(field.name()), value);
825    }
826
827    fn record_u64(&mut self, field: &tracing::field::Field, value: u64) {
828        let field_path = event_path!(field.name());
829        let converted: Result<i64, _> = value.try_into();
830        match converted {
831            Ok(value) => self.insert(field_path, value),
832            Err(_) => self.insert(field_path, value.to_string()),
833        };
834    }
835
836    fn record_bool(&mut self, field: &tracing::field::Field, value: bool) {
837        self.insert(event_path!(field.name()), value);
838    }
839}
840
841#[cfg(test)]
842mod test {
843    use lookup::event_path;
844    use uuid::Version;
845    use vrl::{btreemap, value};
846
847    use super::*;
848    use crate::test_util::open_fixture;
849
850    // The following two tests assert that renaming a key has no effect if the
851    // keys are equivalent, whether the key exists in the log or not.
852    #[test]
853    fn rename_key_flat_equiv_exists() {
854        let value = value!({
855            one: 1,
856            two: 2
857        });
858
859        let mut base = LogEvent::from_parts(value.clone(), EventMetadata::default());
860        base.rename_key(event_path!("one"), event_path!("one"));
861        let (actual_fields, _) = base.into_parts();
862
863        assert_eq!(value, actual_fields);
864    }
865    #[test]
866    fn rename_key_flat_equiv_not_exists() {
867        let value = value!({
868            one: 1,
869            two: 2
870        });
871
872        let mut base = LogEvent::from_parts(value.clone(), EventMetadata::default());
873        base.rename_key(event_path!("three"), event_path!("three"));
874        let (actual_fields, _) = base.into_parts();
875
876        assert_eq!(value, actual_fields);
877    }
878    // Assert that renaming a key has no effect if the key does not originally
879    // exist in the log, when the to -> from keys are not identical.
880    #[test]
881    fn rename_key_flat_not_exists() {
882        let value = value!({
883            one: 1,
884            two: 2
885        });
886
887        let mut base = LogEvent::from_parts(value.clone(), EventMetadata::default());
888        base.rename_key(event_path!("three"), event_path!("four"));
889        let (actual_fields, _) = base.into_parts();
890
891        assert_eq!(value, actual_fields);
892    }
893    // Assert that renaming a key has the effect of moving the value from one
894    // key name to another if the key exists.
895    #[test]
896    fn rename_key_flat_no_overlap() {
897        let value = value!({
898            one: 1,
899            two: 2
900        });
901
902        let mut expected_value = value.clone();
903        let one = expected_value.remove(path!("one"), true).unwrap();
904        expected_value.insert(path!("three"), one);
905
906        let mut base = LogEvent::from_parts(value, EventMetadata::default());
907        base.rename_key(event_path!("one"), event_path!("three"));
908        let (actual_fields, _) = base.into_parts();
909
910        assert_eq!(expected_value, actual_fields);
911    }
912    // Assert that renaming a key has the effect of moving the value from one
913    // key name to another if the key exists and will overwrite another key if
914    // it exists.
915    #[test]
916    fn rename_key_flat_overlap() {
917        let value = value!({
918            one: 1,
919            two: 2
920        });
921
922        let mut expected_value = value.clone();
923        let val = expected_value.remove(path!("one"), true).unwrap();
924        expected_value.insert(path!("two"), val);
925
926        let mut base = LogEvent::from_parts(value, EventMetadata::default());
927        base.rename_key(event_path!("one"), event_path!("two"));
928        let (actual_value, _) = base.into_parts();
929
930        assert_eq!(expected_value, actual_value);
931    }
932
933    #[test]
934    fn insert() {
935        let mut log = LogEvent::default();
936
937        let old = log.insert(event_path!("foo"), "foo");
938
939        assert_eq!(log.get(event_path!("foo")), Some(&"foo".into()));
940        assert_eq!(old, None);
941    }
942
943    #[test]
944    fn insert_existing() {
945        let mut log = LogEvent::default();
946        log.insert(event_path!("foo"), "foo");
947
948        let old = log.insert(event_path!("foo"), "bar");
949
950        assert_eq!(log.get(event_path!("foo")), Some(&"bar".into()));
951        assert_eq!(old, Some("foo".into()));
952    }
953
954    #[test]
955    fn try_insert() {
956        let mut log = LogEvent::default();
957
958        log.try_insert(event_path!("foo"), "foo");
959
960        assert_eq!(log.get(event_path!("foo")), Some(&"foo".into()));
961    }
962
963    #[test]
964    fn try_insert_existing() {
965        let mut log = LogEvent::default();
966        log.insert(event_path!("foo"), "foo");
967
968        log.try_insert(event_path!("foo"), "bar");
969
970        assert_eq!(log.get(event_path!("foo")), Some(&"foo".into()));
971    }
972
973    #[test]
974    fn try_insert_nested() {
975        let mut log = LogEvent::default();
976
977        log.try_insert(event_path!("foo", "bar"), "foo");
978
979        assert_eq!(log.get(event_path!("foo", "bar")), Some(&"foo".into()));
980        assert_eq!(log.get(event_path!("foo.bar")), None);
981    }
982
983    #[test]
984    fn try_insert_existing_nested() {
985        let mut log = LogEvent::default();
986        log.insert(event_path!("foo", "bar"), "foo");
987
988        log.try_insert(event_path!("foo", "bar"), "bar");
989
990        assert_eq!(log.get(event_path!("foo", "bar")), Some(&"foo".into()));
991        assert_eq!(log.get(event_path!("foo.bar")), None);
992    }
993
994    #[test]
995    fn try_insert_flat() {
996        let mut log = LogEvent::default();
997
998        log.try_insert(event_path!("foo"), "foo");
999
1000        assert_eq!(log.get(event_path!("foo")), Some(&"foo".into()));
1001    }
1002
1003    #[test]
1004    fn try_insert_flat_existing() {
1005        let mut log = LogEvent::default();
1006        log.insert(event_path!("foo"), "foo");
1007
1008        log.try_insert(event_path!("foo"), "bar");
1009
1010        assert_eq!(log.get(event_path!("foo")), Some(&"foo".into()));
1011    }
1012
1013    #[test]
1014    fn try_insert_flat_dotted() {
1015        let mut log = LogEvent::default();
1016
1017        log.try_insert(event_path!("foo.bar"), "foo");
1018
1019        assert_eq!(log.get(event_path!("foo.bar")), Some(&"foo".into()));
1020    }
1021
1022    #[test]
1023    fn try_insert_flat_existing_dotted() {
1024        let mut log = LogEvent::default();
1025        log.insert(event_path!("foo.bar"), "foo");
1026
1027        log.try_insert(event_path!("foo.bar"), "bar");
1028
1029        assert_eq!(log.get(event_path!("foo.bar")), Some(&"foo".into()));
1030    }
1031
1032    // This test iterates over the `tests/data/fixtures/log_event` folder and:
1033    //
1034    //   * Ensures the EventLog parsed from bytes and turned into a
1035    //   serde_json::Value are equal to the item being just plain parsed as
1036    //   json.
1037    //
1038    // Basically: This test makes sure we aren't mutilating any content users
1039    // might be sending.
1040    #[test]
1041    fn json_value_to_vector_log_event_to_json_value() {
1042        const FIXTURE_ROOT: &str = "tests/data/fixtures/log_event";
1043
1044        for fixture_file in std::fs::read_dir(FIXTURE_ROOT).unwrap() {
1045            match fixture_file {
1046                Ok(fixture_file) => {
1047                    let path = fixture_file.path();
1048                    tracing::trace!(?path, "Opening.");
1049                    let serde_value = open_fixture(&path).unwrap();
1050
1051                    let vector_value = LogEvent::try_from(serde_value.clone()).unwrap();
1052                    let serde_value_again: serde_json::Value = vector_value.try_into().unwrap();
1053
1054                    assert_eq!(serde_value, serde_value_again);
1055                }
1056                _ => panic!("This test should never read Err'ing test fixtures."),
1057            }
1058        }
1059    }
1060
1061    fn assert_merge_value(
1062        current: impl Into<Value>,
1063        incoming: impl Into<Value>,
1064        expected: impl Into<Value>,
1065    ) {
1066        let mut merged = current.into();
1067        merged.merge(incoming.into());
1068        assert_eq!(merged, expected.into());
1069    }
1070
1071    #[test]
1072    fn merge_value_works_correctly() {
1073        assert_merge_value("hello ", "world", "hello world");
1074
1075        assert_merge_value(true, false, false);
1076        assert_merge_value(false, true, true);
1077
1078        assert_merge_value("my_val", true, true);
1079        assert_merge_value(true, "my_val", "my_val");
1080
1081        assert_merge_value(1, 2, 2);
1082    }
1083
1084    #[test]
1085    fn merge_event_combines_values_accordingly() {
1086        // Specify the fields that will be merged.
1087        // Only the ones listed will be merged from the `incoming` event
1088        // to the `current`.
1089        let fields_to_merge = vec![
1090            "merge".to_string(),
1091            "merge_a".to_string(),
1092            "merge_b".to_string(),
1093            "merge_c".to_string(),
1094        ];
1095
1096        let current = {
1097            let mut log = LogEvent::default();
1098
1099            log.insert(event_path!("merge"), "hello "); // will be concatenated with the `merged` from `incoming`.
1100            log.insert(event_path!("do_not_merge"), "my_first_value"); // will remain as is, since it's not selected for merging.
1101
1102            log.insert(event_path!("merge_a"), true); // will be overwritten with the `merge_a` from `incoming` (since it's a non-bytes kind).
1103            log.insert(event_path!("merge_b"), 123i64); // will be overwritten with the `merge_b` from `incoming` (since it's a non-bytes kind).
1104
1105            log.insert(event_path!("a"), true); // will remain as is since it's not selected for merge.
1106            log.insert(event_path!("b"), 123i64); // will remain as is since it's not selected for merge.
1107
1108            // `c` is not present in the `current`, and not selected for merge,
1109            // so it won't be included in the final event.
1110
1111            log
1112        };
1113
1114        let incoming = {
1115            let mut log = LogEvent::default();
1116
1117            log.insert(event_path!("merge"), "world"); // will be concatenated to the `merge` from `current`.
1118            log.insert(event_path!("do_not_merge"), "my_second_value"); // will be ignored, since it's not selected for merge.
1119
1120            log.insert(event_path!("merge_b"), 456i64); // will be merged in as `456`.
1121            log.insert(event_path!("merge_c"), false); // will be merged in as `false`.
1122
1123            // `a` will remain as-is, since it's not marked for merge and
1124            // neither is it specified in the `incoming` event.
1125            log.insert(event_path!("b"), 456i64); // `b` not marked for merge, will not change.
1126            log.insert(event_path!("c"), true); // `c` not marked for merge, will be ignored.
1127
1128            log
1129        };
1130
1131        let mut merged = current;
1132        merged.merge(incoming, &fields_to_merge);
1133
1134        let expected = {
1135            let mut log = LogEvent::default();
1136            log.insert(event_path!("merge"), "hello world");
1137            log.insert(event_path!("do_not_merge"), "my_first_value");
1138            log.insert(event_path!("a"), true);
1139            log.insert(event_path!("b"), 123i64);
1140            log.insert(event_path!("merge_a"), true);
1141            log.insert(event_path!("merge_b"), 456i64);
1142            log.insert(event_path!("merge_c"), false);
1143            log
1144        };
1145
1146        vector_common::assert_event_data_eq!(merged, expected);
1147    }
1148
1149    #[test]
1150    fn event_fields_iter() {
1151        let mut log = LogEvent::default();
1152        log.insert(event_path!("a"), 0);
1153        log.insert(event_path!("a", "b"), 1);
1154        log.insert(event_path!("c"), 2);
1155        let actual: Vec<(KeyString, Value)> = log
1156            .all_event_fields()
1157            .unwrap()
1158            .map(|(s, v)| (s, v.clone()))
1159            .collect();
1160        assert_eq!(
1161            actual,
1162            vec![("a.b".into(), 1.into()), ("c".into(), 2.into())]
1163        );
1164    }
1165
1166    #[test]
1167    fn metadata_fields_iter() {
1168        let mut log = LogEvent::default();
1169        log.insert(metadata_path!("a"), 0);
1170        log.insert(metadata_path!("a", "b"), 1);
1171        log.insert(metadata_path!("c"), 2);
1172        let actual: Vec<(KeyString, Value)> = log
1173            .all_metadata_fields()
1174            .unwrap()
1175            .map(|(s, v)| (s, v.clone()))
1176            .collect();
1177        assert_eq!(
1178            actual,
1179            vec![("%a.b".into(), 1.into()), ("%c".into(), 2.into())]
1180        );
1181    }
1182
1183    #[test]
1184    fn skip_array_elements() {
1185        let log = LogEvent::from(Value::from(btreemap! {
1186            "arr" => [1],
1187            "obj" => btreemap! {
1188                "arr" => [1,2,3]
1189            },
1190        }));
1191
1192        let actual: Vec<(KeyString, Value)> = log
1193            .all_event_fields_skip_array_elements()
1194            .unwrap()
1195            .map(|(s, v)| (s, v.clone()))
1196            .collect();
1197        assert_eq!(
1198            actual,
1199            vec![
1200                ("arr".into(), [1].into()),
1201                ("obj.arr".into(), [1, 2, 3].into())
1202            ]
1203        );
1204    }
1205
1206    #[test]
1207    fn metadata_set_unique_uuid_v4_source_event_id() {
1208        // Check if event id is UUID v4
1209        let log1 = LogEvent::default();
1210        assert_eq!(
1211            log1.metadata()
1212                .source_event_id()
1213                .expect("source_event_id should be auto-generated for new events")
1214                .get_version(),
1215            Some(Version::Random)
1216        );
1217
1218        // Check if event id is unique on creation
1219        let log2 = LogEvent::default();
1220        assert_ne!(
1221            log1.metadata().source_event_id(),
1222            log2.metadata().source_event_id()
1223        );
1224    }
1225}
vector_core/event/log_event.rs

vector_core/event/
log_event.rs
