vector_core/event/
vrl_target.rs

1use std::{
2    borrow::Cow,
3    collections::BTreeMap,
4    convert::TryFrom,
5    marker::PhantomData,
6    num::{NonZero, TryFromIntError},
7};
8
9use lookup::{OwnedTargetPath, OwnedValuePath, PathPrefix, lookup_v2::OwnedSegment};
10use snafu::Snafu;
11use vrl::{
12    compiler::{ProgramInfo, SecretTarget, Target, value::VrlValueConvert},
13    prelude::Collection,
14    value::{Kind, ObjectMap, Value},
15};
16
17use super::{
18    Event, EventMetadata, LogEvent, Metric, MetricKind, TraceEvent,
19    metric::{MetricTags, TagValue, TagValueSet},
20};
21use crate::{
22    config::{LogNamespace, log_schema},
23    schema::Definition,
24};
25
26const VALID_METRIC_PATHS_SET: &str = ".name, .namespace, .interval_ms, .timestamp, .kind, .tags";
27
28/// We can get the `type` of the metric in Remap, but can't set it.
29const VALID_METRIC_PATHS_GET: &str =
30    ".name, .namespace, .interval_ms, .timestamp, .kind, .tags, .type";
31
32/// Metrics aren't interested in paths that have a length longer than 3.
33///
34/// The longest path is 2, and we need to check that a third segment doesn't exist as we don't want
35/// fields such as `.tags.host.thing`.
36const MAX_METRIC_PATH_DEPTH: usize = 3;
37
38/// How metric tags are exposed to and accepted from VRL or Lua.
39#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
40pub enum MetricTagMode {
41    /// Tags are exposed as single strings (last value wins for multi-value
42    /// tags); writes always produce single-value tags.
43    #[default]
44    Single,
45    /// Tags are always exposed as arrays; writes always produce multi-value
46    /// tags regardless of whether the assigned value is scalar or array.
47    Full,
48    /// Tags are exposed using their underlying shape: single-value tags as
49    /// strings, multi-value tags as arrays. Writes: scalar values produce
50    /// single-value tags; arrays of length >= 2 produce multi-value tags.
51    ///
52    /// A length-1 array is normalised to a single-value tag by the metric
53    /// storage layer (`TagValueSet::Set` is never reduced below 2 elements),
54    /// so an assignment like `.tags.region = ["us-east-1"]` round-trips as
55    /// a scalar on the next read. Use `Full` to force array shape regardless
56    /// of length.
57    Auto,
58}
59
60/// An adapter to turn `Event`s into `vrl_lib::Target`s.
61#[allow(clippy::large_enum_variant)]
62#[derive(Debug, Clone)]
63pub enum VrlTarget {
64    // `LogEvent` is essentially just a destructured `event::LogEvent`, but without the semantics
65    // that `fields` must always be a `Map` variant.
66    LogEvent(Value, EventMetadata),
67    Metric {
68        metric: Metric,
69        value: Value,
70        tag_mode: MetricTagMode,
71    },
72    Trace(Value, EventMetadata),
73}
74
75pub enum TargetEvents {
76    One(Event),
77    Logs(TargetIter<LogEvent>),
78    Traces(TargetIter<TraceEvent>),
79}
80
81pub struct TargetIter<T> {
82    iter: std::vec::IntoIter<Value>,
83    metadata: EventMetadata,
84    _marker: PhantomData<T>,
85    log_namespace: LogNamespace,
86}
87
88fn create_log_event(value: Value, metadata: EventMetadata) -> LogEvent {
89    let mut log = LogEvent::new_with_metadata(metadata);
90    log.maybe_insert(log_schema().message_key_target_path(), value);
91    log
92}
93
94impl Iterator for TargetIter<LogEvent> {
95    type Item = Event;
96
97    fn next(&mut self) -> Option<Self::Item> {
98        self.iter.next().map(|v| {
99            match self.log_namespace {
100                LogNamespace::Legacy => match v {
101                    value @ Value::Object(_) => LogEvent::from_parts(value, self.metadata.clone()),
102                    value => create_log_event(value, self.metadata.clone()),
103                },
104                LogNamespace::Vector => LogEvent::from_parts(v, self.metadata.clone()),
105            }
106            .into()
107        })
108    }
109}
110
111impl Iterator for TargetIter<TraceEvent> {
112    type Item = Event;
113
114    fn next(&mut self) -> Option<Self::Item> {
115        self.iter.next().map(|v| {
116            match v {
117                value @ Value::Object(_) => {
118                    TraceEvent::from(LogEvent::from_parts(value, self.metadata.clone()))
119                }
120                value => TraceEvent::from(create_log_event(value, self.metadata.clone())),
121            }
122            .into()
123        })
124    }
125}
126
127impl VrlTarget {
128    pub fn new(event: Event, info: &ProgramInfo, tag_mode: MetricTagMode) -> Self {
129        match event {
130            Event::Log(event) => {
131                let (value, metadata) = event.into_parts();
132                VrlTarget::LogEvent(value, metadata)
133            }
134            Event::Metric(metric) => {
135                // We pre-generate [`Value`] types for the metric fields accessed in
136                // the event. This allows us to then return references to those
137                // values, even if the field is accessed more than once.
138                let value = precompute_metric_value(&metric, info, tag_mode);
139
140                VrlTarget::Metric {
141                    metric,
142                    value,
143                    tag_mode,
144                }
145            }
146            Event::Trace(event) => {
147                let (fields, metadata) = event.into_parts();
148                VrlTarget::Trace(Value::Object(fields), metadata)
149            }
150        }
151    }
152
153    /// Modifies a schema in the same way that the `into_events` function modifies the event
154    pub fn modify_schema_definition_for_into_events(input: Definition) -> Definition {
155        let log_namespaces = input.log_namespaces().clone();
156
157        // both namespaces merge arrays, but only `Legacy` moves field definitions into a "message" field.
158        let merged_arrays = merge_array_definitions(input);
159        Definition::combine_log_namespaces(
160            &log_namespaces,
161            move_field_definitions_into_message(merged_arrays.clone()),
162            merged_arrays,
163        )
164    }
165
166    /// Turn the target back into events.
167    ///
168    /// This returns an iterator of events as one event can be turned into multiple by assigning an
169    /// array to `.` in VRL.
170    pub fn into_events(self, log_namespace: LogNamespace) -> TargetEvents {
171        match self {
172            VrlTarget::LogEvent(value, metadata) => match value {
173                value @ Value::Object(_) => {
174                    TargetEvents::One(LogEvent::from_parts(value, metadata).into())
175                }
176
177                Value::Array(values) => TargetEvents::Logs(TargetIter {
178                    iter: values.into_iter(),
179                    metadata,
180                    _marker: PhantomData,
181                    log_namespace,
182                }),
183
184                v => match log_namespace {
185                    LogNamespace::Vector => {
186                        TargetEvents::One(LogEvent::from_parts(v, metadata).into())
187                    }
188                    LogNamespace::Legacy => TargetEvents::One(create_log_event(v, metadata).into()),
189                },
190            },
191            VrlTarget::Trace(value, metadata) => match value {
192                value @ Value::Object(_) => {
193                    let log = LogEvent::from_parts(value, metadata);
194                    TargetEvents::One(TraceEvent::from(log).into())
195                }
196
197                Value::Array(values) => TargetEvents::Traces(TargetIter {
198                    iter: values.into_iter(),
199                    metadata,
200                    _marker: PhantomData,
201                    log_namespace,
202                }),
203
204                v => TargetEvents::One(create_log_event(v, metadata).into()),
205            },
206            VrlTarget::Metric { metric, .. } => TargetEvents::One(Event::Metric(metric)),
207        }
208    }
209
210    fn metadata(&self) -> &EventMetadata {
211        match self {
212            VrlTarget::LogEvent(_, metadata) | VrlTarget::Trace(_, metadata) => metadata,
213            VrlTarget::Metric { metric, .. } => metric.metadata(),
214        }
215    }
216
217    fn metadata_mut(&mut self) -> &mut EventMetadata {
218        match self {
219            VrlTarget::LogEvent(_, metadata) | VrlTarget::Trace(_, metadata) => metadata,
220            VrlTarget::Metric { metric, .. } => metric.metadata_mut(),
221        }
222    }
223}
224
225/// If the VRL returns a value that is not an array (see [`merge_array_definitions`]),
226/// or an object, that data is moved into the `message` field.
227fn move_field_definitions_into_message(mut definition: Definition) -> Definition {
228    let mut message = definition.event_kind().clone();
229    message.remove_object();
230    message.remove_array();
231
232    if !message.is_never()
233        && let Some(message_key) = log_schema().message_key()
234    {
235        // We need to add the given message type to a field called `message`
236        // in the event.
237        let message = Kind::object(Collection::from(BTreeMap::from([(
238            message_key.to_string().into(),
239            message,
240        )])));
241
242        definition.event_kind_mut().remove_bytes();
243        definition.event_kind_mut().remove_integer();
244        definition.event_kind_mut().remove_float();
245        definition.event_kind_mut().remove_boolean();
246        definition.event_kind_mut().remove_timestamp();
247        definition.event_kind_mut().remove_regex();
248        definition.event_kind_mut().remove_null();
249
250        *definition.event_kind_mut() = definition.event_kind().union(message);
251    }
252
253    definition
254}
255
256/// If the transform returns an array, the elements of this array will be separated
257/// out into it's individual elements and passed downstream.
258///
259/// The potential types that the transform can output are any of the arrays
260/// elements or any non-array elements that are within the definition. All these
261/// definitions need to be merged together.
262fn merge_array_definitions(mut definition: Definition) -> Definition {
263    if let Some(array) = definition.event_kind().as_array() {
264        let array_kinds = array.reduced_kind();
265
266        let kind = definition.event_kind_mut();
267        kind.remove_array();
268        *kind = kind.union(array_kinds);
269    }
270
271    definition
272}
273
274fn set_metric_tag_values(
275    name: String,
276    value: &Value,
277    metric: &mut Metric,
278    tag_mode: MetricTagMode,
279) {
280    // `Auto` dispatches by the *shape* of the assigned value: arrays go through
281    // the multi-value path, scalars go through the single-value path. `Full`
282    // and `Single` are unchanged.
283    let multi_value = match tag_mode {
284        MetricTagMode::Single => false,
285        MetricTagMode::Full => true,
286        MetricTagMode::Auto => matches!(value, Value::Array(_)),
287    };
288
289    if multi_value {
290        let values = if let Value::Array(values) = value {
291            values.as_slice()
292        } else {
293            std::slice::from_ref(value)
294        };
295
296        let tag_values = values
297            .iter()
298            .filter_map(|value| match value {
299                Value::Bytes(bytes) => {
300                    Some(TagValue::Value(String::from_utf8_lossy(bytes).to_string()))
301                }
302                Value::Null => Some(TagValue::Bare),
303                _ => None,
304            })
305            .collect::<Vec<_>>();
306
307        metric.set_multi_value_tag(name, tag_values);
308    } else {
309        // set a single tag value
310        if let Ok(tag_value) = value.try_bytes_utf8_lossy().map(Cow::into_owned) {
311            metric.replace_tag(name, tag_value);
312        } else if value.is_null() {
313            metric.set_multi_value_tag(name, vec![TagValue::Bare]);
314        }
315    }
316}
317
318impl Target for VrlTarget {
319    fn target_insert(&mut self, target_path: &OwnedTargetPath, value: Value) -> Result<(), String> {
320        let path = &target_path.path;
321        match target_path.prefix {
322            PathPrefix::Event => match self {
323                VrlTarget::LogEvent(log, _) | VrlTarget::Trace(log, _) => {
324                    log.insert(path, value);
325                    Ok(())
326                }
327                VrlTarget::Metric {
328                    metric,
329                    value: metric_value,
330                    tag_mode,
331                } => {
332                    if path.is_root() {
333                        return Err(MetricPathError::SetPathError.to_string());
334                    }
335
336                    if let Some(paths) = path.to_alternative_components(MAX_METRIC_PATH_DEPTH) {
337                        match paths.as_slice() {
338                            ["tags"] => {
339                                let value =
340                                    value.clone().try_object().map_err(|e| e.to_string())?;
341
342                                metric.remove_tags();
343                                for (field, value) in &value {
344                                    set_metric_tag_values(
345                                        field.as_str().into(),
346                                        value,
347                                        metric,
348                                        *tag_mode,
349                                    );
350                                }
351                            }
352                            ["tags", field] => {
353                                set_metric_tag_values(
354                                    (*field).to_owned(),
355                                    &value,
356                                    metric,
357                                    *tag_mode,
358                                );
359                            }
360                            ["name"] => {
361                                let value = value.clone().try_bytes().map_err(|e| e.to_string())?;
362                                metric.series.name.name =
363                                    String::from_utf8_lossy(&value).into_owned();
364                            }
365                            ["namespace"] => {
366                                let value = value.clone().try_bytes().map_err(|e| e.to_string())?;
367                                metric.series.name.namespace =
368                                    Some(String::from_utf8_lossy(&value).into_owned());
369                            }
370                            ["interval_ms"] => {
371                                let value: i64 =
372                                    value.clone().try_into_i64().map_err(|e| e.to_string())?;
373                                let value: u32 = value
374                                    .try_into()
375                                    .map_err(|e: TryFromIntError| e.to_string())?;
376                                let value = NonZero::try_from(value).map_err(|e| e.to_string())?;
377                                metric.data.time.interval_ms = Some(value);
378                            }
379                            ["timestamp"] => {
380                                let value =
381                                    value.clone().try_timestamp().map_err(|e| e.to_string())?;
382                                metric.data.time.timestamp = Some(value);
383                            }
384                            ["kind"] => {
385                                metric.data.kind = MetricKind::try_from(value.clone())?;
386                            }
387                            _ => {
388                                return Err(MetricPathError::InvalidPath {
389                                    path: &path.to_string(),
390                                    expected: VALID_METRIC_PATHS_SET,
391                                }
392                                .to_string());
393                            }
394                        }
395
396                        metric_value.insert(path, value);
397
398                        return Ok(());
399                    }
400
401                    Err(MetricPathError::InvalidPath {
402                        path: &path.to_string(),
403                        expected: VALID_METRIC_PATHS_SET,
404                    }
405                    .to_string())
406                }
407            },
408            PathPrefix::Metadata => {
409                self.metadata_mut()
410                    .value_mut()
411                    .insert(&target_path.path, value);
412                Ok(())
413            }
414        }
415    }
416
417    #[allow(clippy::redundant_closure_for_method_calls)] // false positive
418    fn target_get(&self, target_path: &OwnedTargetPath) -> Result<Option<&Value>, String> {
419        match target_path.prefix {
420            PathPrefix::Event => match self {
421                VrlTarget::LogEvent(log, _) | VrlTarget::Trace(log, _) => {
422                    Ok(log.get(&target_path.path))
423                }
424                VrlTarget::Metric { value, .. } => target_get_metric(&target_path.path, value),
425            },
426            PathPrefix::Metadata => Ok(self.metadata().value().get(&target_path.path)),
427        }
428    }
429
430    fn target_get_mut(
431        &mut self,
432        target_path: &OwnedTargetPath,
433    ) -> Result<Option<&mut Value>, String> {
434        match target_path.prefix {
435            PathPrefix::Event => match self {
436                VrlTarget::LogEvent(log, _) | VrlTarget::Trace(log, _) => {
437                    Ok(log.get_mut(&target_path.path))
438                }
439                VrlTarget::Metric { value, .. } => target_get_mut_metric(&target_path.path, value),
440            },
441            PathPrefix::Metadata => Ok(self.metadata_mut().value_mut().get_mut(&target_path.path)),
442        }
443    }
444
445    fn target_remove(
446        &mut self,
447        target_path: &OwnedTargetPath,
448        compact: bool,
449    ) -> Result<Option<vrl::value::Value>, String> {
450        match target_path.prefix {
451            PathPrefix::Event => match self {
452                VrlTarget::LogEvent(log, _) | VrlTarget::Trace(log, _) => {
453                    Ok(log.remove(&target_path.path, compact))
454                }
455                VrlTarget::Metric {
456                    metric,
457                    value,
458                    tag_mode,
459                } => {
460                    if target_path.path.is_root() {
461                        return Err(MetricPathError::SetPathError.to_string());
462                    }
463
464                    if let Some(paths) = target_path
465                        .path
466                        .to_alternative_components(MAX_METRIC_PATH_DEPTH)
467                    {
468                        let removed_value = match paths.as_slice() {
469                            ["namespace"] => metric.series.name.namespace.take().map(Into::into),
470                            ["timestamp"] => metric.data.time.timestamp.take().map(Into::into),
471                            ["interval_ms"] => metric
472                                .data
473                                .time
474                                .interval_ms
475                                .take()
476                                .map(u32::from)
477                                .map(Into::into),
478                            ["tags"] => metric
479                                .series
480                                .tags
481                                .take()
482                                .map(|tags| metric_tags_to_vrl_value(tags, *tag_mode)),
483                            ["tags", field] => metric
484                                .remove_tag_set(field)
485                                .map(|tag_set| tag_value_set_to_vrl_value(&tag_set, *tag_mode)),
486                            _ => {
487                                return Err(MetricPathError::InvalidPath {
488                                    path: &target_path.path.to_string(),
489                                    expected: VALID_METRIC_PATHS_SET,
490                                }
491                                .to_string());
492                            }
493                        };
494
495                        value.remove(&target_path.path, false);
496
497                        Ok(removed_value)
498                    } else {
499                        Ok(None)
500                    }
501                }
502            },
503            PathPrefix::Metadata => Ok(self
504                .metadata_mut()
505                .value_mut()
506                .remove(&target_path.path, compact)),
507        }
508    }
509}
510
511impl SecretTarget for VrlTarget {
512    fn get_secret(&self, key: &str) -> Option<&str> {
513        self.metadata().secrets().get_secret(key)
514    }
515
516    fn insert_secret(&mut self, key: &str, value: &str) {
517        self.metadata_mut().secrets_mut().insert_secret(key, value);
518    }
519
520    fn remove_secret(&mut self, key: &str) {
521        self.metadata_mut().secrets_mut().remove_secret(key);
522    }
523}
524
525/// Retrieves a value from a the provided metric using the path.
526/// Currently the root path and the following paths are supported:
527/// - `name`
528/// - `namespace`
529/// - `interval_ms`
530/// - `timestamp`
531/// - `kind`
532/// - `tags`
533/// - `tags.<tagname>`
534/// - `type`
535///
536/// Any other paths result in a `MetricPathError::InvalidPath` being returned.
537fn target_get_metric<'a>(
538    path: &OwnedValuePath,
539    value: &'a Value,
540) -> Result<Option<&'a Value>, String> {
541    if path.is_root() {
542        return Ok(Some(value));
543    }
544
545    let value = value.get(path);
546
547    let Some(paths) = path.to_alternative_components(MAX_METRIC_PATH_DEPTH) else {
548        return Ok(None);
549    };
550
551    match paths.as_slice() {
552        ["name"]
553        | ["kind"]
554        | ["type"]
555        | ["tags", _]
556        | ["namespace"]
557        | ["timestamp"]
558        | ["interval_ms"]
559        | ["tags"] => Ok(value),
560        _ => Err(MetricPathError::InvalidPath {
561            path: &path.to_string(),
562            expected: VALID_METRIC_PATHS_GET,
563        }
564        .to_string()),
565    }
566}
567
568fn target_get_mut_metric<'a>(
569    path: &OwnedValuePath,
570    value: &'a mut Value,
571) -> Result<Option<&'a mut Value>, String> {
572    if path.is_root() {
573        return Ok(Some(value));
574    }
575
576    let value = value.get_mut(path);
577
578    let Some(paths) = path.to_alternative_components(MAX_METRIC_PATH_DEPTH) else {
579        return Ok(None);
580    };
581
582    match paths.as_slice() {
583        ["name"]
584        | ["kind"]
585        | ["tags", _]
586        | ["namespace"]
587        | ["timestamp"]
588        | ["interval_ms"]
589        | ["tags"] => Ok(value),
590        _ => Err(MetricPathError::InvalidPath {
591            path: &path.to_string(),
592            expected: VALID_METRIC_PATHS_SET,
593        }
594        .to_string()),
595    }
596}
597
598struct MetricProperty {
599    property: &'static str,
600    getter: fn(&Metric) -> Option<Value>,
601    set: bool,
602}
603
604impl MetricProperty {
605    fn new(property: &'static str, getter: fn(&Metric) -> Option<Value>) -> Self {
606        Self {
607            property,
608            getter,
609            set: false,
610        }
611    }
612
613    fn insert(&mut self, metric: &Metric, map: &mut ObjectMap) {
614        if self.set {
615            return;
616        }
617        if let Some(value) = (self.getter)(metric) {
618            map.insert(self.property.into(), value);
619            self.set = true;
620        }
621    }
622}
623
624fn get_single_value_tags(metric: &Metric) -> Option<Value> {
625    metric
626        .tags()
627        .cloned()
628        .map(|tags| metric_tags_to_vrl_value(tags, MetricTagMode::Single))
629}
630
631fn get_multi_value_tags(metric: &Metric) -> Option<Value> {
632    metric
633        .tags()
634        .cloned()
635        .map(|tags| metric_tags_to_vrl_value(tags, MetricTagMode::Full))
636}
637
638/// `Auto` keeps the underlying shape: a tag with exactly one value renders
639/// as a scalar (string for `Value(_)`, null for `Bare`), and a tag with two
640/// or more values renders as an array. The discriminator must be `len()` --
641/// `as_single()` deliberately collapses multi-value sets to their last
642/// value to support `Single`-mode semantics, which is the opposite of what
643/// we want here.
644fn get_auto_value_tags(metric: &Metric) -> Option<Value> {
645    metric
646        .tags()
647        .cloned()
648        .map(|tags| metric_tags_to_vrl_value(tags, MetricTagMode::Auto))
649}
650
651fn tag_ref_to_vrl_value(value: Option<&str>) -> Value {
652    value.map_or(Value::Null, Value::from)
653}
654
655fn tag_value_set_to_vrl_value(tag_set: &TagValueSet, tag_mode: MetricTagMode) -> Value {
656    match tag_mode {
657        MetricTagMode::Single => tag_set.as_single().map_or(Value::Null, Value::from),
658        MetricTagMode::Full => Value::Array(tag_set.iter().map(tag_ref_to_vrl_value).collect()),
659        MetricTagMode::Auto => match tag_set.len() {
660            1 => match tag_set.iter().next() {
661                Some(Some(s)) => Value::from(s),
662                _ => Value::Null,
663            },
664            _ => Value::Array(tag_set.iter().map(tag_ref_to_vrl_value).collect()),
665        },
666    }
667}
668
669fn metric_tags_to_vrl_value(tags: MetricTags, tag_mode: MetricTagMode) -> Value {
670    match tag_mode {
671        MetricTagMode::Single => tags
672            .into_iter_single()
673            .map(|(tag, value)| (tag.into(), value.into()))
674            .collect::<ObjectMap>()
675            .into(),
676        MetricTagMode::Full | MetricTagMode::Auto => tags
677            .iter_sets()
678            .map(|(tag, tag_set)| (tag.into(), tag_value_set_to_vrl_value(tag_set, tag_mode)))
679            .collect::<ObjectMap>()
680            .into(),
681    }
682}
683
684/// pre-compute the `Value` structure of the metric.
685///
686/// This structure is partially populated based on the fields accessed by
687/// the VRL program as informed by `ProgramInfo`.
688fn precompute_metric_value(metric: &Metric, info: &ProgramInfo, tag_mode: MetricTagMode) -> Value {
689    let mut name = MetricProperty::new("name", |metric| Some(metric.name().to_owned().into()));
690    let mut kind = MetricProperty::new("kind", |metric| Some(metric.kind().into()));
691    let mut type_ = MetricProperty::new("type", |metric| Some(metric.value().clone().into()));
692    let mut namespace = MetricProperty::new("namespace", |metric| {
693        metric.namespace().map(String::from).map(Into::into)
694    });
695    let mut interval_ms =
696        MetricProperty::new("interval_ms", |metric| metric.interval_ms().map(Into::into));
697    let mut timestamp =
698        MetricProperty::new("timestamp", |metric| metric.timestamp().map(Into::into));
699    let mut tags = MetricProperty::new(
700        "tags",
701        match tag_mode {
702            MetricTagMode::Single => get_single_value_tags,
703            MetricTagMode::Full => get_multi_value_tags,
704            MetricTagMode::Auto => get_auto_value_tags,
705        },
706    );
707
708    let mut map = ObjectMap::default();
709
710    for target_path in &info.target_queries {
711        // Accessing a root path requires us to pre-populate all fields.
712        if target_path == &OwnedTargetPath::event_root() {
713            let mut properties = [
714                &mut name,
715                &mut kind,
716                &mut type_,
717                &mut namespace,
718                &mut interval_ms,
719                &mut timestamp,
720                &mut tags,
721            ];
722            for property in &mut properties {
723                property.insert(metric, &mut map);
724            }
725            break;
726        }
727
728        // For non-root paths, we continuously populate the value with the
729        // relevant data.
730        if let Some(OwnedSegment::Field(field)) = target_path.path.segments.first() {
731            let property = match field.as_ref() {
732                "name" => Some(&mut name),
733                "kind" => Some(&mut kind),
734                "type" => Some(&mut type_),
735                "namespace" => Some(&mut namespace),
736                "timestamp" => Some(&mut timestamp),
737                "interval_ms" => Some(&mut interval_ms),
738                "tags" => Some(&mut tags),
739                _ => None,
740            };
741            if let Some(property) = property {
742                property.insert(metric, &mut map);
743            }
744        }
745    }
746
747    map.into()
748}
749
750#[derive(Debug, Snafu)]
751enum MetricPathError<'a> {
752    #[snafu(display("cannot set root path"))]
753    SetPathError,
754
755    #[snafu(display("invalid path {}: expected one of {}", path, expected))]
756    InvalidPath { path: &'a str, expected: &'a str },
757}
758
759#[cfg(test)]
760mod test {
761    use chrono::{Utc, offset::TimeZone};
762    use lookup::owned_value_path;
763    use similar_asserts::assert_eq;
764    use vrl::{btreemap, value::kind::Index};
765
766    use super::{super::MetricValue, *};
767    use crate::metric_tags;
768
769    #[test]
770    fn test_field_definitions_in_message() {
771        let definition =
772            Definition::new_with_default_metadata(Kind::bytes(), [LogNamespace::Legacy]);
773        assert_eq!(
774            Definition::new_with_default_metadata(
775                Kind::object(BTreeMap::from([("message".into(), Kind::bytes())])),
776                [LogNamespace::Legacy]
777            ),
778            move_field_definitions_into_message(definition)
779        );
780
781        // Test when a message field already exists.
782        let definition = Definition::new_with_default_metadata(
783            Kind::object(BTreeMap::from([("message".into(), Kind::integer())])).or_bytes(),
784            [LogNamespace::Legacy],
785        );
786        assert_eq!(
787            Definition::new_with_default_metadata(
788                Kind::object(BTreeMap::from([(
789                    "message".into(),
790                    Kind::bytes().or_integer()
791                )])),
792                [LogNamespace::Legacy]
793            ),
794            move_field_definitions_into_message(definition)
795        );
796    }
797
798    #[test]
799    fn test_merged_array_definitions_simple() {
800        // Test merging the array definitions where the schema definition
801        // is simple, containing only one possible type in the array.
802        let object: BTreeMap<vrl::value::kind::Field, Kind> = [
803            ("carrot".into(), Kind::bytes()),
804            ("potato".into(), Kind::integer()),
805        ]
806        .into();
807
808        let kind = Kind::array(Collection::from_unknown(Kind::object(object)));
809
810        let definition = Definition::new_with_default_metadata(kind, [LogNamespace::Legacy]);
811
812        let kind = Kind::object(BTreeMap::from([
813            ("carrot".into(), Kind::bytes()),
814            ("potato".into(), Kind::integer()),
815        ]));
816
817        let wanted = Definition::new_with_default_metadata(kind, [LogNamespace::Legacy]);
818        let merged = merge_array_definitions(definition);
819
820        assert_eq!(wanted, merged);
821    }
822
823    #[test]
824    fn test_merged_array_definitions_complex() {
825        // Test merging the array definitions where the schema definition
826        // is fairly complex containing multiple different possible types.
827        let object: BTreeMap<vrl::value::kind::Field, Kind> = [
828            ("carrot".into(), Kind::bytes()),
829            ("potato".into(), Kind::integer()),
830        ]
831        .into();
832
833        let array: BTreeMap<Index, Kind> = [
834            (Index::from(0), Kind::integer()),
835            (Index::from(1), Kind::boolean()),
836            (
837                Index::from(2),
838                Kind::object(BTreeMap::from([("peas".into(), Kind::bytes())])),
839            ),
840        ]
841        .into();
842
843        let mut kind = Kind::bytes();
844        kind.add_object(object);
845        kind.add_array(array);
846
847        let definition = Definition::new_with_default_metadata(kind, [LogNamespace::Legacy]);
848
849        let mut kind = Kind::bytes();
850        kind.add_integer();
851        kind.add_boolean();
852        kind.add_object(BTreeMap::from([
853            ("carrot".into(), Kind::bytes().or_undefined()),
854            ("potato".into(), Kind::integer().or_undefined()),
855            ("peas".into(), Kind::bytes().or_undefined()),
856        ]));
857
858        let wanted = Definition::new_with_default_metadata(kind, [LogNamespace::Legacy]);
859        let merged = merge_array_definitions(definition);
860
861        assert_eq!(wanted, merged);
862    }
863
864    #[test]
865    fn log_get() {
866        let cases = vec![
867            (
868                BTreeMap::new(),
869                owned_value_path!(),
870                Ok(Some(BTreeMap::new().into())),
871            ),
872            (
873                BTreeMap::from([("foo".into(), "bar".into())]),
874                owned_value_path!(),
875                Ok(Some(BTreeMap::from([("foo".into(), "bar".into())]).into())),
876            ),
877            (
878                BTreeMap::from([("foo".into(), "bar".into())]),
879                owned_value_path!("foo"),
880                Ok(Some("bar".into())),
881            ),
882            (
883                BTreeMap::from([("foo".into(), "bar".into())]),
884                owned_value_path!("bar"),
885                Ok(None),
886            ),
887            (
888                btreemap! { "foo" => vec![btreemap! { "bar" => true }] },
889                owned_value_path!("foo", 0, "bar"),
890                Ok(Some(true.into())),
891            ),
892            (
893                btreemap! { "foo" => btreemap! { "bar baz" => btreemap! { "baz" => 2 } } },
894                owned_value_path!("foo", r"bar baz", "baz"),
895                Ok(Some(2.into())),
896            ),
897        ];
898
899        for (value, path, expect) in cases {
900            let value: ObjectMap = value;
901            let info = ProgramInfo {
902                fallible: false,
903                abortable: false,
904                target_queries: vec![],
905                target_assignments: vec![],
906            };
907            let target = VrlTarget::new(
908                Event::Log(LogEvent::from(value)),
909                &info,
910                MetricTagMode::Single,
911            );
912            let path = OwnedTargetPath::event(path);
913
914            assert_eq!(
915                Target::target_get(&target, &path).map(Option::<&Value>::cloned),
916                expect
917            );
918        }
919    }
920
921    #[allow(clippy::too_many_lines)]
922    #[test]
923    fn log_insert() {
924        let cases = vec![
925            (
926                BTreeMap::from([("foo".into(), "bar".into())]),
927                owned_value_path!(0),
928                btreemap! { "baz" => "qux" }.into(),
929                btreemap! { "baz" => "qux" },
930                Ok(()),
931            ),
932            (
933                BTreeMap::from([("foo".into(), "bar".into())]),
934                owned_value_path!("foo"),
935                "baz".into(),
936                btreemap! { "foo" => "baz" },
937                Ok(()),
938            ),
939            (
940                BTreeMap::from([("foo".into(), "bar".into())]),
941                owned_value_path!("foo", 2, "bar baz", "a", "b"),
942                true.into(),
943                btreemap! {
944                    "foo" => vec![
945                        Value::Null,
946                        Value::Null,
947                        btreemap! {
948                            "bar baz" => btreemap! { "a" => btreemap! { "b" => true } },
949                        }.into()
950                    ]
951                },
952                Ok(()),
953            ),
954            (
955                btreemap! { "foo" => vec![0, 1, 2] },
956                owned_value_path!("foo", 5),
957                "baz".into(),
958                btreemap! {
959                    "foo" => vec![
960                        0.into(),
961                        1.into(),
962                        2.into(),
963                        Value::Null,
964                        Value::Null,
965                        Value::from("baz"),
966                    ],
967                },
968                Ok(()),
969            ),
970            (
971                BTreeMap::from([("foo".into(), "bar".into())]),
972                owned_value_path!("foo", 0),
973                "baz".into(),
974                btreemap! { "foo" => vec!["baz"] },
975                Ok(()),
976            ),
977            (
978                btreemap! { "foo" => Value::Array(vec![]) },
979                owned_value_path!("foo", 0),
980                "baz".into(),
981                btreemap! { "foo" => vec!["baz"] },
982                Ok(()),
983            ),
984            (
985                btreemap! { "foo" => Value::Array(vec![0.into()]) },
986                owned_value_path!("foo", 0),
987                "baz".into(),
988                btreemap! { "foo" => vec!["baz"] },
989                Ok(()),
990            ),
991            (
992                btreemap! { "foo" => Value::Array(vec![0.into(), 1.into()]) },
993                owned_value_path!("foo", 0),
994                "baz".into(),
995                btreemap! { "foo" => Value::Array(vec!["baz".into(), 1.into()]) },
996                Ok(()),
997            ),
998            (
999                btreemap! { "foo" => Value::Array(vec![0.into(), 1.into()]) },
1000                owned_value_path!("foo", 1),
1001                "baz".into(),
1002                btreemap! { "foo" => Value::Array(vec![0.into(), "baz".into()]) },
1003                Ok(()),
1004            ),
1005        ];
1006
1007        for (object, path, value, expect, result) in cases {
1008            let object: ObjectMap = object;
1009            let info = ProgramInfo {
1010                fallible: false,
1011                abortable: false,
1012                target_queries: vec![],
1013                target_assignments: vec![],
1014            };
1015            let mut target = VrlTarget::new(
1016                Event::Log(LogEvent::from(object)),
1017                &info,
1018                MetricTagMode::Single,
1019            );
1020            let expect = LogEvent::from(expect);
1021            let value: Value = value;
1022            let path = OwnedTargetPath::event(path);
1023
1024            assert_eq!(
1025                Target::target_insert(&mut target, &path, value.clone()),
1026                result
1027            );
1028            assert_eq!(
1029                Target::target_get(&target, &path).map(Option::<&Value>::cloned),
1030                Ok(Some(value))
1031            );
1032            assert_eq!(
1033                match target.into_events(LogNamespace::Legacy) {
1034                    TargetEvents::One(event) => vec![event],
1035                    TargetEvents::Logs(events) => events.collect::<Vec<_>>(),
1036                    TargetEvents::Traces(events) => events.collect::<Vec<_>>(),
1037                }
1038                .first()
1039                .cloned()
1040                .unwrap(),
1041                Event::Log(expect)
1042            );
1043        }
1044    }
1045
1046    #[test]
1047    fn log_remove() {
1048        let cases = vec![
1049            (
1050                BTreeMap::from([("foo".into(), "bar".into())]),
1051                owned_value_path!("foo"),
1052                false,
1053                Some(BTreeMap::new().into()),
1054            ),
1055            (
1056                BTreeMap::from([("foo".into(), "bar".into())]),
1057                owned_value_path!(r"foo bar", "foo"),
1058                false,
1059                Some(btreemap! { "foo" => "bar"}.into()),
1060            ),
1061            (
1062                btreemap! { "foo" => "bar", "baz" => "qux" },
1063                owned_value_path!(),
1064                false,
1065                Some(BTreeMap::new().into()),
1066            ),
1067            (
1068                btreemap! { "foo" => "bar", "baz" => "qux" },
1069                owned_value_path!(),
1070                true,
1071                Some(BTreeMap::new().into()),
1072            ),
1073            (
1074                btreemap! { "foo" => vec![0] },
1075                owned_value_path!("foo", 0),
1076                false,
1077                Some(btreemap! { "foo" => Value::Array(vec![]) }.into()),
1078            ),
1079            (
1080                btreemap! { "foo" => vec![0] },
1081                owned_value_path!("foo", 0),
1082                true,
1083                Some(BTreeMap::new().into()),
1084            ),
1085            (
1086                btreemap! {
1087                    "foo" => btreemap! { "bar baz" => vec![0] },
1088                    "bar" => "baz",
1089                },
1090                owned_value_path!("foo", r"bar baz", 0),
1091                false,
1092                Some(
1093                    btreemap! {
1094                        "foo" => btreemap! { "bar baz" => Value::Array(vec![]) },
1095                        "bar" => "baz",
1096                    }
1097                    .into(),
1098                ),
1099            ),
1100            (
1101                btreemap! {
1102                    "foo" => btreemap! { "bar baz" => vec![0] },
1103                    "bar" => "baz",
1104                },
1105                owned_value_path!("foo", r"bar baz", 0),
1106                true,
1107                Some(btreemap! { "bar" => "baz" }.into()),
1108            ),
1109        ];
1110
1111        for (object, path, compact, expect) in cases {
1112            let info = ProgramInfo {
1113                fallible: false,
1114                abortable: false,
1115                target_queries: vec![],
1116                target_assignments: vec![],
1117            };
1118            let mut target = VrlTarget::new(
1119                Event::Log(LogEvent::from(object)),
1120                &info,
1121                MetricTagMode::Single,
1122            );
1123            let path = OwnedTargetPath::event(path);
1124            let removed = Target::target_get(&target, &path).unwrap().cloned();
1125
1126            assert_eq!(
1127                Target::target_remove(&mut target, &path, compact),
1128                Ok(removed)
1129            );
1130            assert_eq!(
1131                Target::target_get(&target, &OwnedTargetPath::event_root())
1132                    .map(Option::<&Value>::cloned),
1133                Ok(expect)
1134            );
1135        }
1136    }
1137
1138    #[test]
1139    fn log_into_events() {
1140        use vrl::btreemap;
1141
1142        let cases = vec![
1143            (
1144                Value::from(btreemap! {"foo" => "bar"}),
1145                vec![btreemap! {"foo" => "bar"}],
1146            ),
1147            (Value::from(1), vec![btreemap! {"message" => 1}]),
1148            (Value::from("2"), vec![btreemap! {"message" => "2"}]),
1149            (Value::from(true), vec![btreemap! {"message" => true}]),
1150            (
1151                Value::from(vec![
1152                    Value::from(1),
1153                    Value::from("2"),
1154                    Value::from(true),
1155                    Value::from(btreemap! {"foo" => "bar"}),
1156                ]),
1157                vec![
1158                    btreemap! {"message" => 1},
1159                    btreemap! {"message" => "2"},
1160                    btreemap! {"message" => true},
1161                    btreemap! {"foo" => "bar"},
1162                ],
1163            ),
1164        ];
1165
1166        for (value, expect) in cases {
1167            let metadata = EventMetadata::default();
1168            let info = ProgramInfo {
1169                fallible: false,
1170                abortable: false,
1171                target_queries: vec![],
1172                target_assignments: vec![],
1173            };
1174            let mut target = VrlTarget::new(
1175                Event::Log(LogEvent::new_with_metadata(metadata.clone())),
1176                &info,
1177                MetricTagMode::Single,
1178            );
1179
1180            Target::target_insert(&mut target, &OwnedTargetPath::event_root(), value).unwrap();
1181
1182            assert_eq!(
1183                match target.into_events(LogNamespace::Legacy) {
1184                    TargetEvents::One(event) => vec![event],
1185                    TargetEvents::Logs(events) => events.collect::<Vec<_>>(),
1186                    TargetEvents::Traces(events) => events.collect::<Vec<_>>(),
1187                },
1188                expect
1189                    .into_iter()
1190                    .map(|v| Event::Log(LogEvent::from_map(v, metadata.clone())))
1191                    .collect::<Vec<_>>()
1192            );
1193        }
1194    }
1195
1196    #[test]
1197    fn metric_all_fields() {
1198        let metric = Metric::new(
1199            "zub",
1200            MetricKind::Absolute,
1201            MetricValue::Counter { value: 1.23 },
1202        )
1203        .with_namespace(Some("zoob"))
1204        .with_tags(Some(metric_tags!("tig" => "tog")))
1205        .with_timestamp(Some(
1206            Utc.with_ymd_and_hms(2020, 12, 10, 12, 0, 0)
1207                .single()
1208                .expect("invalid timestamp"),
1209        ))
1210        .with_interval_ms(Some(NonZero::<u32>::new(507).unwrap()));
1211
1212        let info = ProgramInfo {
1213            fallible: false,
1214            abortable: false,
1215            target_queries: vec![
1216                OwnedTargetPath::event(owned_value_path!("name")),
1217                OwnedTargetPath::event(owned_value_path!("namespace")),
1218                OwnedTargetPath::event(owned_value_path!("interval_ms")),
1219                OwnedTargetPath::event(owned_value_path!("timestamp")),
1220                OwnedTargetPath::event(owned_value_path!("kind")),
1221                OwnedTargetPath::event(owned_value_path!("type")),
1222                OwnedTargetPath::event(owned_value_path!("tags")),
1223            ],
1224            target_assignments: vec![],
1225        };
1226        let target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Single);
1227
1228        assert_eq!(
1229            Ok(Some(
1230                btreemap! {
1231                    "name" => "zub",
1232                    "namespace" => "zoob",
1233                    "interval_ms" => 507,
1234                    "timestamp" => Utc.with_ymd_and_hms(2020, 12, 10, 12, 0, 0).single().expect("invalid timestamp"),
1235                    "tags" => btreemap! { "tig" => "tog" },
1236                    "kind" => "absolute",
1237                    "type" => "counter",
1238                }
1239                .into()
1240            )),
1241            target
1242                .target_get(&OwnedTargetPath::event_root())
1243                .map(Option::<&Value>::cloned)
1244        );
1245    }
1246
1247    #[test]
1248    fn metric_fields() {
1249        struct Case {
1250            path: OwnedValuePath,
1251            current: Option<Value>,
1252            new: Value,
1253            delete: bool,
1254        }
1255
1256        let metric = Metric::new(
1257            "name",
1258            MetricKind::Absolute,
1259            MetricValue::Counter { value: 1.23 },
1260        )
1261        .with_tags(Some(metric_tags!("tig" => "tog")));
1262
1263        let cases = vec![
1264            Case {
1265                path: owned_value_path!("name"),
1266                current: Some(Value::from("name")),
1267                new: Value::from("namefoo"),
1268                delete: false,
1269            },
1270            Case {
1271                path: owned_value_path!("namespace"),
1272                current: None,
1273                new: "namespacefoo".into(),
1274                delete: true,
1275            },
1276            Case {
1277                path: owned_value_path!("timestamp"),
1278                current: None,
1279                new: Utc
1280                    .with_ymd_and_hms(2020, 12, 8, 12, 0, 0)
1281                    .single()
1282                    .expect("invalid timestamp")
1283                    .into(),
1284                delete: true,
1285            },
1286            Case {
1287                path: owned_value_path!("interval_ms"),
1288                current: None,
1289                new: 123_456.into(),
1290                delete: true,
1291            },
1292            Case {
1293                path: owned_value_path!("kind"),
1294                current: Some(Value::from("absolute")),
1295                new: "incremental".into(),
1296                delete: false,
1297            },
1298            Case {
1299                path: owned_value_path!("tags", "thing"),
1300                current: None,
1301                new: "footag".into(),
1302                delete: true,
1303            },
1304        ];
1305
1306        let info = ProgramInfo {
1307            fallible: false,
1308            abortable: false,
1309            target_queries: vec![
1310                OwnedTargetPath::event(owned_value_path!("name")),
1311                OwnedTargetPath::event(owned_value_path!("namespace")),
1312                OwnedTargetPath::event(owned_value_path!("timestamp")),
1313                OwnedTargetPath::event(owned_value_path!("interval_ms")),
1314                OwnedTargetPath::event(owned_value_path!("kind")),
1315            ],
1316            target_assignments: vec![],
1317        };
1318        let mut target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Single);
1319
1320        for Case {
1321            path,
1322            current,
1323            new,
1324            delete,
1325        } in cases
1326        {
1327            let path = OwnedTargetPath::event(path);
1328
1329            assert_eq!(
1330                Ok(current),
1331                target.target_get(&path).map(Option::<&Value>::cloned)
1332            );
1333            assert_eq!(Ok(()), target.target_insert(&path, new.clone()));
1334            assert_eq!(
1335                Ok(Some(new.clone())),
1336                target.target_get(&path).map(Option::<&Value>::cloned)
1337            );
1338
1339            if delete {
1340                assert_eq!(Ok(Some(new)), target.target_remove(&path, true));
1341                assert_eq!(
1342                    Ok(None),
1343                    target.target_get(&path).map(Option::<&Value>::cloned)
1344                );
1345            }
1346        }
1347    }
1348
1349    #[test]
1350    fn metric_set_tags() {
1351        let metric = Metric::new(
1352            "name",
1353            MetricKind::Absolute,
1354            MetricValue::Counter { value: 1.23 },
1355        )
1356        .with_tags(Some(metric_tags!("tig" => "tog")));
1357
1358        let info = ProgramInfo {
1359            fallible: false,
1360            abortable: false,
1361            target_queries: vec![],
1362            target_assignments: vec![],
1363        };
1364        let mut target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Single);
1365        let _result = target.target_insert(
1366            &OwnedTargetPath::event(owned_value_path!("tags")),
1367            Value::Object(BTreeMap::from([("a".into(), "b".into())])),
1368        );
1369
1370        match target {
1371            VrlTarget::Metric {
1372                metric,
1373                value: _,
1374                tag_mode: _,
1375            } => {
1376                assert!(metric.tags().is_some());
1377                assert_eq!(metric.tags().unwrap(), &crate::metric_tags!("a" => "b"));
1378            }
1379            _ => panic!("must be a metric"),
1380        }
1381    }
1382
1383    #[test]
1384    fn metric_invalid_paths() {
1385        let metric = Metric::new(
1386            "name",
1387            MetricKind::Absolute,
1388            MetricValue::Counter { value: 1.23 },
1389        );
1390
1391        let validpaths_get = [
1392            ".name",
1393            ".namespace",
1394            ".interval_ms",
1395            ".timestamp",
1396            ".kind",
1397            ".tags",
1398            ".type",
1399        ];
1400
1401        let validpaths_set = [
1402            ".name",
1403            ".namespace",
1404            ".interval_ms",
1405            ".timestamp",
1406            ".kind",
1407            ".tags",
1408        ];
1409
1410        let info = ProgramInfo {
1411            fallible: false,
1412            abortable: false,
1413            target_queries: vec![],
1414            target_assignments: vec![],
1415        };
1416        let mut target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Single);
1417
1418        assert_eq!(
1419            Err(format!(
1420                "invalid path zork: expected one of {}",
1421                validpaths_get.join(", ")
1422            )),
1423            target.target_get(&OwnedTargetPath::event(owned_value_path!("zork")))
1424        );
1425
1426        assert_eq!(
1427            Err(format!(
1428                "invalid path zork: expected one of {}",
1429                validpaths_set.join(", ")
1430            )),
1431            target.target_insert(
1432                &OwnedTargetPath::event(owned_value_path!("zork")),
1433                "thing".into()
1434            )
1435        );
1436
1437        assert_eq!(
1438            Err(format!(
1439                "invalid path zork: expected one of {}",
1440                validpaths_set.join(", ")
1441            )),
1442            target.target_remove(&OwnedTargetPath::event(owned_value_path!("zork")), true)
1443        );
1444
1445        assert_eq!(
1446            Err(format!(
1447                "invalid path tags.foo.flork: expected one of {}",
1448                validpaths_get.join(", ")
1449            )),
1450            target.target_get(&OwnedTargetPath::event(owned_value_path!(
1451                "tags", "foo", "flork"
1452            )))
1453        );
1454    }
1455
1456    #[test]
1457    fn test_metric_insert_get_multi_value_tag() {
1458        let metric = Metric::new(
1459            "name",
1460            MetricKind::Absolute,
1461            MetricValue::Counter { value: 1.23 },
1462        );
1463        let info = ProgramInfo {
1464            fallible: false,
1465            abortable: false,
1466            target_queries: vec![],
1467            target_assignments: vec![],
1468        };
1469
1470        let mut target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Full);
1471
1472        let value = Value::Array(vec!["a".into(), "".into(), Value::Null, "b".into()]);
1473        target
1474            .target_insert(
1475                &OwnedTargetPath::event(owned_value_path!("tags", "foo")),
1476                value,
1477            )
1478            .unwrap();
1479
1480        let vrl_tags_value = target
1481            .target_get(&OwnedTargetPath::event(owned_value_path!("tags")))
1482            .unwrap()
1483            .unwrap();
1484
1485        assert_eq!(
1486            vrl_tags_value,
1487            &Value::Object(BTreeMap::from([(
1488                "foo".into(),
1489                Value::Array(vec!["a".into(), "".into(), Value::Null, "b".into()])
1490            )]))
1491        );
1492
1493        let VrlTarget::Metric { metric, .. } = target else {
1494            unreachable!()
1495        };
1496
1497        // get single value (should be the last one)
1498        assert_eq!(metric.tag_value("foo"), Some("b".into()));
1499    }
1500
1501    fn auto_mode_program_info() -> ProgramInfo {
1502        ProgramInfo {
1503            fallible: false,
1504            abortable: false,
1505            target_queries: vec![OwnedTargetPath::event(owned_value_path!("tags"))],
1506            target_assignments: vec![],
1507        }
1508    }
1509
1510    /// Auto exposes a single-value tag as a plain string -- the same shape
1511    /// `Single` would produce -- so existing VRL programs that read scalar
1512    /// tags work unchanged.
1513    #[test]
1514    fn metric_auto_read_single_value_tag_returns_string() {
1515        let metric = Metric::new(
1516            "m",
1517            MetricKind::Absolute,
1518            MetricValue::Counter { value: 1.0 },
1519        )
1520        .with_tags(Some(metric_tags!("env" => "prod")));
1521
1522        let info = auto_mode_program_info();
1523        let target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Auto);
1524
1525        let tags = target
1526            .target_get(&OwnedTargetPath::event(owned_value_path!("tags")))
1527            .unwrap()
1528            .unwrap()
1529            .clone();
1530
1531        assert_eq!(
1532            tags,
1533            Value::Object(BTreeMap::from([("env".into(), Value::from("prod"))])),
1534        );
1535    }
1536
1537    /// Auto exposes a multi-value tag as an array -- the same shape `Full`
1538    /// would produce -- so multi-value information survives the round trip
1539    /// instead of being collapsed to the last value.
1540    #[test]
1541    fn metric_auto_read_multi_value_tag_returns_array() {
1542        use super::super::metric::{MetricTags, TagValueSet};
1543        let metric = Metric::new(
1544            "m",
1545            MetricKind::Absolute,
1546            MetricValue::Counter { value: 1.0 },
1547        )
1548        .with_tags(Some(MetricTags(BTreeMap::from([(
1549            "shard".to_string(),
1550            TagValueSet::from(vec![
1551                TagValue::from("a".to_string()),
1552                TagValue::from("b".to_string()),
1553            ]),
1554        )]))));
1555
1556        let info = auto_mode_program_info();
1557        let target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Auto);
1558
1559        let tags = target
1560            .target_get(&OwnedTargetPath::event(owned_value_path!("tags")))
1561            .unwrap()
1562            .unwrap()
1563            .clone();
1564
1565        assert_eq!(
1566            tags,
1567            Value::Object(BTreeMap::from([(
1568                "shard".into(),
1569                Value::Array(vec!["a".into(), "b".into()])
1570            )])),
1571        );
1572    }
1573
1574    /// Auto reads a mixed metric (one single, one multi) preserving both
1575    /// shapes -- the property the new mode exists for.
1576    #[test]
1577    fn metric_auto_read_mixed_tags_preserves_both_shapes() {
1578        use super::super::metric::{MetricTags, TagValueSet};
1579        let mut tags = BTreeMap::new();
1580        tags.insert(
1581            "env".to_string(),
1582            TagValueSet::from(vec!["prod".to_string()]),
1583        );
1584        tags.insert(
1585            "shard".to_string(),
1586            TagValueSet::from(vec!["a".to_string(), "b".to_string()]),
1587        );
1588        let metric = Metric::new(
1589            "m",
1590            MetricKind::Absolute,
1591            MetricValue::Counter { value: 1.0 },
1592        )
1593        .with_tags(Some(MetricTags(tags)));
1594
1595        let info = auto_mode_program_info();
1596        let target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Auto);
1597
1598        let tags = target
1599            .target_get(&OwnedTargetPath::event(owned_value_path!("tags")))
1600            .unwrap()
1601            .unwrap()
1602            .clone();
1603
1604        assert_eq!(
1605            tags,
1606            Value::Object(BTreeMap::from([
1607                ("env".into(), Value::from("prod")),
1608                ("shard".into(), Value::Array(vec!["a".into(), "b".into()])),
1609            ])),
1610        );
1611    }
1612
1613    /// Auto write: assigning a scalar produces a single-value tag (so a VRL
1614    /// program that did `.tags.env = "staging"` round-trips as `Single`-style).
1615    #[test]
1616    fn metric_auto_write_string_creates_single_tag() {
1617        let metric = Metric::new(
1618            "m",
1619            MetricKind::Absolute,
1620            MetricValue::Counter { value: 1.0 },
1621        );
1622        let info = auto_mode_program_info();
1623        let mut target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Auto);
1624
1625        target
1626            .target_insert(
1627                &OwnedTargetPath::event(owned_value_path!("tags", "env")),
1628                Value::from("staging"),
1629            )
1630            .unwrap();
1631
1632        let VrlTarget::Metric { metric, .. } = target else {
1633            unreachable!()
1634        };
1635        let metric_tags = metric.tags().expect("tag should be set");
1636        let tag_set = metric_tags
1637            .iter_sets()
1638            .find(|(k, _)| *k == "env")
1639            .expect("env tag missing")
1640            .1;
1641        assert_eq!(
1642            tag_set.len(),
1643            1,
1644            "scalar assignment must produce a 1-element set"
1645        );
1646        assert_eq!(tag_set.as_single(), Some("staging"));
1647    }
1648
1649    /// Auto write: assigning an array produces a multi-value tag (the
1650    /// behaviour `Single` would silently drop and `Full` would force on
1651    /// every other tag).
1652    #[test]
1653    fn metric_auto_write_array_creates_multi_value_tag() {
1654        let metric = Metric::new(
1655            "m",
1656            MetricKind::Absolute,
1657            MetricValue::Counter { value: 1.0 },
1658        );
1659        let info = auto_mode_program_info();
1660        let mut target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Auto);
1661
1662        target
1663            .target_insert(
1664                &OwnedTargetPath::event(owned_value_path!("tags", "shard")),
1665                Value::Array(vec!["a".into(), "b".into(), "c".into()]),
1666            )
1667            .unwrap();
1668
1669        let VrlTarget::Metric { metric, .. } = target else {
1670            unreachable!()
1671        };
1672        let metric_tags = metric.tags().expect("tag should be set");
1673        let tag_set = metric_tags
1674            .iter_sets()
1675            .find(|(k, _)| *k == "shard")
1676            .expect("shard tag missing")
1677            .1;
1678        assert_eq!(tag_set.len(), 3);
1679        let collected: Vec<Option<&str>> = tag_set.iter().collect();
1680        assert_eq!(
1681            collected,
1682            vec![Some("a"), Some("b"), Some("c")],
1683            "array assignment must preserve every element"
1684        );
1685    }
1686
1687    /// Auto round-trip across two passes: a length-1 array write reaches the
1688    /// metric storage as `TagValueSet::Single` (the storage layer never holds
1689    /// a `Set` with fewer than 2 elements), so when a *subsequent* VRL pass
1690    /// constructs a new `VrlTarget` around that metric, the tag surfaces as a
1691    /// scalar string rather than a 1-element array. This pins that behaviour
1692    /// as intentional -- a true array-shape preserving round-trip needs
1693    /// `MetricTagMode::Full`.
1694    ///
1695    /// Within a single pass, the `VrlTarget` caches writes verbatim in its
1696    /// precomputed `Value`, so an intra-pass `.tags.region` read sees the
1697    /// original array; the collapse is observable only across passes.
1698    #[test]
1699    fn metric_auto_one_element_array_normalises_to_scalar() {
1700        let metric = Metric::new(
1701            "m",
1702            MetricKind::Absolute,
1703            MetricValue::Counter { value: 1.0 },
1704        );
1705        let info = auto_mode_program_info();
1706        let mut first_pass = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Auto);
1707
1708        first_pass
1709            .target_insert(
1710                &OwnedTargetPath::event(owned_value_path!("tags", "region")),
1711                Value::Array(vec!["us-east-1".into()]),
1712            )
1713            .unwrap();
1714
1715        let VrlTarget::Metric { metric, .. } = first_pass else {
1716            unreachable!()
1717        };
1718        let tag_set = metric
1719            .tags()
1720            .expect("tag should be set")
1721            .iter_sets()
1722            .find(|(k, _)| *k == "region")
1723            .expect("region tag missing")
1724            .1;
1725        assert_eq!(
1726            tag_set.len(),
1727            1,
1728            "storage layer must collapse a 1-element multi-value tag to a Single",
1729        );
1730
1731        let second_pass = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Auto);
1732        let tags = second_pass
1733            .target_get(&OwnedTargetPath::event(owned_value_path!("tags")))
1734            .unwrap()
1735            .unwrap()
1736            .clone();
1737
1738        assert_eq!(
1739            tags,
1740            Value::Object(BTreeMap::from([(
1741                "region".into(),
1742                Value::from("us-east-1"),
1743            )])),
1744            "a length-1 array written in one Auto pass must read back as a \
1745             scalar on the next pass",
1746        );
1747    }
1748
1749    /// Auto read: an empty tag set (`len() == 0`) keeps its array shape so a
1750    /// noop `.tags = .tags` round-trip preserves it. Without this guarantee
1751    /// an empty multi-value tag would silently morph into a bare single-value
1752    /// tag through one Auto round-trip.
1753    #[test]
1754    fn metric_auto_read_empty_tag_set_returns_array() {
1755        use super::super::metric::{MetricTags, TagValueSet};
1756        let mut tags = BTreeMap::new();
1757        tags.insert("empty".to_string(), TagValueSet::default());
1758        let metric = Metric::new(
1759            "m",
1760            MetricKind::Absolute,
1761            MetricValue::Counter { value: 1.0 },
1762        )
1763        .with_tags(Some(MetricTags(tags)));
1764
1765        let info = auto_mode_program_info();
1766        let target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Auto);
1767
1768        let tags = target
1769            .target_get(&OwnedTargetPath::event(owned_value_path!("tags")))
1770            .unwrap()
1771            .unwrap()
1772            .clone();
1773
1774        assert_eq!(
1775            tags,
1776            Value::Object(BTreeMap::from([("empty".into(), Value::Array(vec![]))])),
1777            "empty tag set must surface as an empty array, not as null",
1778        );
1779    }
1780
1781    /// Auto write: assigning `null` falls through the scalar path, producing
1782    /// a single bare tag (matches `Single` semantics so the two modes agree
1783    /// on the unambiguous case).
1784    #[test]
1785    fn metric_auto_write_null_creates_single_bare_tag() {
1786        let metric = Metric::new(
1787            "m",
1788            MetricKind::Absolute,
1789            MetricValue::Counter { value: 1.0 },
1790        );
1791        let info = auto_mode_program_info();
1792        let mut target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Auto);
1793
1794        target
1795            .target_insert(
1796                &OwnedTargetPath::event(owned_value_path!("tags", "bare")),
1797                Value::Null,
1798            )
1799            .unwrap();
1800
1801        let VrlTarget::Metric { metric, .. } = target else {
1802            unreachable!()
1803        };
1804        let metric_tags = metric.tags().expect("tag should be set");
1805        let tag_set = metric_tags
1806            .iter_sets()
1807            .find(|(k, _)| *k == "bare")
1808            .expect("bare tag missing")
1809            .1;
1810        assert_eq!(tag_set.len(), 1);
1811        let collected: Vec<Option<&str>> = tag_set.iter().collect();
1812        assert_eq!(
1813            collected,
1814            vec![None],
1815            "null must produce a single bare value"
1816        );
1817    }
1818
1819    /// `del(.tags.<field>)` must return the same shape that a read would have
1820    /// exposed, so moving multi-value tags in `Auto` mode does not silently
1821    /// drop values.
1822    #[test]
1823    fn metric_auto_remove_multi_value_tag_returns_array() {
1824        use super::super::metric::{MetricTags, TagValue, TagValueSet};
1825        let metric = Metric::new(
1826            "m",
1827            MetricKind::Absolute,
1828            MetricValue::Counter { value: 1.0 },
1829        )
1830        .with_tags(Some(MetricTags(BTreeMap::from([(
1831            "shard".to_string(),
1832            TagValueSet::from(vec![
1833                TagValue::from("a".to_string()),
1834                TagValue::from("b".to_string()),
1835            ]),
1836        )]))));
1837
1838        let info = auto_mode_program_info();
1839        let mut target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Auto);
1840
1841        let removed = target
1842            .target_remove(
1843                &OwnedTargetPath::event(owned_value_path!("tags", "shard")),
1844                true,
1845            )
1846            .unwrap()
1847            .expect("shard tag should be removed");
1848
1849        assert_eq!(removed, Value::Array(vec!["a".into(), "b".into()]));
1850
1851        let VrlTarget::Metric { metric, .. } = target else {
1852            unreachable!()
1853        };
1854        assert!(metric.tags().is_none());
1855    }
1856
1857    /// `del(.tags)` in `Auto` mode must preserve mixed single/multi shapes.
1858    #[test]
1859    fn metric_auto_remove_all_tags_preserves_shapes() {
1860        use super::super::metric::{MetricTags, TagValueSet};
1861        let mut tags = BTreeMap::new();
1862        tags.insert(
1863            "env".to_string(),
1864            TagValueSet::from(vec!["prod".to_string()]),
1865        );
1866        tags.insert(
1867            "shard".to_string(),
1868            TagValueSet::from(vec!["a".to_string(), "b".to_string()]),
1869        );
1870        let metric = Metric::new(
1871            "m",
1872            MetricKind::Absolute,
1873            MetricValue::Counter { value: 1.0 },
1874        )
1875        .with_tags(Some(MetricTags(tags)));
1876
1877        let info = auto_mode_program_info();
1878        let mut target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Auto);
1879
1880        let removed = target
1881            .target_remove(&OwnedTargetPath::event(owned_value_path!("tags")), true)
1882            .unwrap()
1883            .expect("tags should be removed");
1884
1885        assert_eq!(
1886            removed,
1887            Value::Object(BTreeMap::from([
1888                ("env".into(), Value::from("prod")),
1889                ("shard".into(), Value::Array(vec!["a".into(), "b".into()]))
1890            ]))
1891        );
1892    }
1893
1894    /// `del(.tags.<field>)` in `Full` mode must return the full value array.
1895    #[test]
1896    fn metric_full_remove_multi_value_tag_returns_array() {
1897        use super::super::metric::{MetricTags, TagValue, TagValueSet};
1898        let metric = Metric::new(
1899            "m",
1900            MetricKind::Absolute,
1901            MetricValue::Counter { value: 1.0 },
1902        )
1903        .with_tags(Some(MetricTags(BTreeMap::from([(
1904            "shard".to_string(),
1905            TagValueSet::from(vec![
1906                TagValue::from("a".to_string()),
1907                TagValue::from("b".to_string()),
1908            ]),
1909        )]))));
1910
1911        let info = ProgramInfo {
1912            fallible: false,
1913            abortable: false,
1914            target_queries: vec![],
1915            target_assignments: vec![],
1916        };
1917        let mut target = VrlTarget::new(Event::Metric(metric), &info, MetricTagMode::Full);
1918
1919        let removed = target
1920            .target_remove(
1921                &OwnedTargetPath::event(owned_value_path!("tags", "shard")),
1922                true,
1923            )
1924            .unwrap()
1925            .expect("shard tag should be removed");
1926
1927        assert_eq!(removed, Value::Array(vec!["a".into(), "b".into()]));
1928    }
1929}
vector_core/event/vrl_target.rs

vector_core/event/
vrl_target.rs
