vector_core/event/
proto.rs

1use std::{collections::BTreeMap, sync::Arc};
2
3use chrono::TimeZone;
4use ordered_float::NotNan;
5use uuid::Uuid;
6
7use super::{MetricTags, WithMetadata};
8use crate::{event, metrics::AgentDDSketch};
9
10#[allow(warnings, clippy::all, clippy::pedantic)]
11mod proto_event {
12    include!(concat!(env!("OUT_DIR"), "/event.rs"));
13}
14pub use event_wrapper::Event;
15pub use metric::Value as MetricValue;
16pub use proto_event::*;
17use vrl::value::{ObjectMap, Value as VrlValue};
18
19use super::EventFinalizers;
20use super::metadata::{Inner, default_schema_definition};
21use super::{EventMetadata, array, metric::MetricSketch};
22
23impl event_array::Events {
24    // We can't use the standard `From` traits here because the actual
25    // type of `LogArray` and `TraceArray` are the same.
26    fn from_logs(logs: array::LogArray) -> Self {
27        let logs = logs.into_iter().map(Into::into).collect();
28        Self::Logs(LogArray { logs })
29    }
30
31    fn from_metrics(metrics: array::MetricArray) -> Self {
32        let metrics = metrics.into_iter().map(Into::into).collect();
33        Self::Metrics(MetricArray { metrics })
34    }
35
36    fn from_traces(traces: array::TraceArray) -> Self {
37        let traces = traces.into_iter().map(Into::into).collect();
38        Self::Traces(TraceArray { traces })
39    }
40}
41
42impl From<array::EventArray> for EventArray {
43    fn from(events: array::EventArray) -> Self {
44        let events = Some(match events {
45            array::EventArray::Logs(array) => event_array::Events::from_logs(array),
46            array::EventArray::Metrics(array) => event_array::Events::from_metrics(array),
47            array::EventArray::Traces(array) => event_array::Events::from_traces(array),
48        });
49        Self { events }
50    }
51}
52
53impl From<EventArray> for array::EventArray {
54    fn from(events: EventArray) -> Self {
55        let events = events.events.unwrap();
56
57        match events {
58            event_array::Events::Logs(logs) => {
59                array::EventArray::Logs(logs.logs.into_iter().map(Into::into).collect())
60            }
61            event_array::Events::Metrics(metrics) => {
62                array::EventArray::Metrics(metrics.metrics.into_iter().map(Into::into).collect())
63            }
64            event_array::Events::Traces(traces) => {
65                array::EventArray::Traces(traces.traces.into_iter().map(Into::into).collect())
66            }
67        }
68    }
69}
70
71impl From<Event> for EventWrapper {
72    fn from(event: Event) -> Self {
73        Self { event: Some(event) }
74    }
75}
76
77impl From<Log> for Event {
78    fn from(log: Log) -> Self {
79        Self::Log(log)
80    }
81}
82
83impl From<Metric> for Event {
84    fn from(metric: Metric) -> Self {
85        Self::Metric(metric)
86    }
87}
88
89impl From<Trace> for Event {
90    fn from(trace: Trace) -> Self {
91        Self::Trace(trace)
92    }
93}
94
95impl From<Log> for super::LogEvent {
96    fn from(log: Log) -> Self {
97        #[allow(deprecated)]
98        let metadata = log
99            .metadata_full
100            .map(Into::into)
101            .or_else(|| {
102                log.metadata
103                    .and_then(decode_value)
104                    .map(EventMetadata::default_with_value)
105            })
106            .unwrap_or_default();
107
108        if let Some(value) = log.value {
109            Self::from_parts(decode_value(value).unwrap_or(VrlValue::Null), metadata)
110        } else {
111            // This is for backwards compatibility. Only `value` should be set
112            let fields = log
113                .fields
114                .into_iter()
115                .filter_map(|(k, v)| decode_value(v).map(|value| (k.into(), value)))
116                .collect::<ObjectMap>();
117
118            Self::from_map(fields, metadata)
119        }
120    }
121}
122
123impl From<Trace> for super::TraceEvent {
124    fn from(trace: Trace) -> Self {
125        #[allow(deprecated)]
126        let metadata = trace
127            .metadata_full
128            .map(Into::into)
129            .or_else(|| {
130                trace
131                    .metadata
132                    .and_then(decode_value)
133                    .map(EventMetadata::default_with_value)
134            })
135            .unwrap_or_default();
136
137        let fields = trace
138            .fields
139            .into_iter()
140            .filter_map(|(k, v)| decode_value(v).map(|value| (k.into(), value)))
141            .collect::<ObjectMap>();
142
143        Self::from(super::LogEvent::from_map(fields, metadata))
144    }
145}
146
147impl From<MetricValue> for super::MetricValue {
148    fn from(value: MetricValue) -> Self {
149        match value {
150            MetricValue::Counter(counter) => Self::Counter {
151                value: counter.value,
152            },
153            MetricValue::Gauge(gauge) => Self::Gauge { value: gauge.value },
154            MetricValue::Set(set) => Self::Set {
155                values: set.values.into_iter().collect(),
156            },
157            MetricValue::Distribution1(dist) => Self::Distribution {
158                statistic: dist.statistic().into(),
159                samples: super::metric::zip_samples(dist.values, dist.sample_rates),
160            },
161            MetricValue::Distribution2(dist) => Self::Distribution {
162                statistic: dist.statistic().into(),
163                samples: dist.samples.into_iter().map(Into::into).collect(),
164            },
165            MetricValue::AggregatedHistogram1(hist) => Self::AggregatedHistogram {
166                buckets: super::metric::zip_buckets(
167                    hist.buckets,
168                    hist.counts.iter().map(|h| u64::from(*h)),
169                ),
170                count: u64::from(hist.count),
171                sum: hist.sum,
172            },
173            MetricValue::AggregatedHistogram2(hist) => Self::AggregatedHistogram {
174                buckets: hist.buckets.into_iter().map(Into::into).collect(),
175                count: u64::from(hist.count),
176                sum: hist.sum,
177            },
178            MetricValue::AggregatedHistogram3(hist) => Self::AggregatedHistogram {
179                buckets: hist.buckets.into_iter().map(Into::into).collect(),
180                count: hist.count,
181                sum: hist.sum,
182            },
183            MetricValue::AggregatedSummary1(summary) => Self::AggregatedSummary {
184                quantiles: super::metric::zip_quantiles(summary.quantiles, summary.values),
185                count: u64::from(summary.count),
186                sum: summary.sum,
187            },
188            MetricValue::AggregatedSummary2(summary) => Self::AggregatedSummary {
189                quantiles: summary.quantiles.into_iter().map(Into::into).collect(),
190                count: u64::from(summary.count),
191                sum: summary.sum,
192            },
193            MetricValue::AggregatedSummary3(summary) => Self::AggregatedSummary {
194                quantiles: summary.quantiles.into_iter().map(Into::into).collect(),
195                count: summary.count,
196                sum: summary.sum,
197            },
198            MetricValue::Sketch(sketch) => match sketch.sketch.unwrap() {
199                sketch::Sketch::AgentDdSketch(ddsketch) => Self::Sketch {
200                    sketch: ddsketch.into(),
201                },
202            },
203        }
204    }
205}
206
207impl From<Metric> for super::Metric {
208    fn from(metric: Metric) -> Self {
209        let kind = match metric.kind() {
210            metric::Kind::Incremental => super::MetricKind::Incremental,
211            metric::Kind::Absolute => super::MetricKind::Absolute,
212        };
213
214        let name = metric.name;
215
216        let namespace = (!metric.namespace.is_empty()).then_some(metric.namespace);
217
218        // Sign can never be lost as ts.nanos is always non negative (per proto spec)
219        #[allow(clippy::cast_sign_loss)]
220        let timestamp = metric.timestamp.map(|ts| {
221            chrono::Utc
222                .timestamp_opt(ts.seconds, ts.nanos as u32)
223                .single()
224                .expect("invalid timestamp")
225        });
226
227        let mut tags = MetricTags(
228            metric
229                .tags_v2
230                .into_iter()
231                .map(|(tag, values)| {
232                    (
233                        tag,
234                        values
235                            .values
236                            .into_iter()
237                            .map(|value| super::metric::TagValue::from(value.value))
238                            .collect(),
239                    )
240                })
241                .collect(),
242        );
243        // The current Vector encoding includes copies of the "single" values of tags in `tags_v2`
244        // above. This `extend` will re-add those values, forcing them to become the last added in
245        // the value set.
246        tags.extend(metric.tags_v1);
247        let tags = (!tags.is_empty()).then_some(tags);
248
249        let value = super::MetricValue::from(metric.value.unwrap());
250
251        #[allow(deprecated)]
252        let metadata = metric
253            .metadata_full
254            .map(Into::into)
255            .or_else(|| {
256                metric
257                    .metadata
258                    .and_then(decode_value)
259                    .map(EventMetadata::default_with_value)
260            })
261            .unwrap_or_default();
262
263        Self::new_with_metadata(name, kind, value, metadata)
264            .with_namespace(namespace)
265            .with_tags(tags)
266            .with_timestamp(timestamp)
267            .with_interval_ms(std::num::NonZeroU32::new(metric.interval_ms))
268    }
269}
270
271impl From<EventWrapper> for super::Event {
272    fn from(proto: EventWrapper) -> Self {
273        let event = proto.event.unwrap();
274
275        match event {
276            Event::Log(proto) => Self::Log(proto.into()),
277            Event::Metric(proto) => Self::Metric(proto.into()),
278            Event::Trace(proto) => Self::Trace(proto.into()),
279        }
280    }
281}
282
283impl From<super::LogEvent> for Log {
284    fn from(log_event: super::LogEvent) -> Self {
285        WithMetadata::<Self>::from(log_event).data
286    }
287}
288
289impl From<super::TraceEvent> for Trace {
290    fn from(trace: super::TraceEvent) -> Self {
291        WithMetadata::<Self>::from(trace).data
292    }
293}
294
295impl From<super::LogEvent> for WithMetadata<Log> {
296    fn from(log_event: super::LogEvent) -> Self {
297        let (value, metadata) = log_event.into_parts();
298
299        // Due to the backwards compatibility requirement by the
300        // "event_can_go_from_raw_prost_to_eventarray_encodable" test, "fields" must not
301        // be empty, since that will decode as an empty array. A "dummy" value is placed
302        // in fields instead which is ignored during decoding. To reduce encoding bloat
303        // from a dummy value, it is only used when the root value type is not an object.
304        // Once this backwards compatibility is no longer required, "fields" can
305        // be entirely removed from the Log object
306        let (fields, value) = if let VrlValue::Object(fields) = value {
307            // using only "fields" to prevent having to use the dummy value
308            let fields = fields
309                .into_iter()
310                .map(|(k, v)| (k.into(), encode_value(v)))
311                .collect::<BTreeMap<_, _>>();
312
313            (fields, None)
314        } else {
315            // Must insert at least one field, otherwise the field is omitted entirely on the
316            // Protocol Buffers side. The dummy field value is ultimately ignored in the decoding
317            // step since `value` is provided.
318            let mut dummy_fields = BTreeMap::new();
319            dummy_fields.insert(".".to_owned(), encode_value(VrlValue::Null));
320
321            (dummy_fields, Some(encode_value(value)))
322        };
323
324        #[allow(deprecated)]
325        let data = Log {
326            fields,
327            value,
328            metadata: Some(encode_value(metadata.value().clone())),
329            metadata_full: Some(metadata.clone().into()),
330        };
331
332        Self { data, metadata }
333    }
334}
335
336impl From<super::TraceEvent> for WithMetadata<Trace> {
337    fn from(trace: super::TraceEvent) -> Self {
338        let (fields, metadata) = trace.into_parts();
339        let fields = fields
340            .into_iter()
341            .map(|(k, v)| (k.into(), encode_value(v)))
342            .collect::<BTreeMap<_, _>>();
343
344        #[allow(deprecated)]
345        let data = Trace {
346            fields,
347            metadata: Some(encode_value(metadata.value().clone())),
348            metadata_full: Some(metadata.clone().into()),
349        };
350
351        Self { data, metadata }
352    }
353}
354
355impl From<super::Metric> for Metric {
356    fn from(metric: super::Metric) -> Self {
357        WithMetadata::<Self>::from(metric).data
358    }
359}
360
361impl From<super::MetricValue> for MetricValue {
362    fn from(value: super::MetricValue) -> Self {
363        match value {
364            super::MetricValue::Counter { value } => Self::Counter(Counter { value }),
365            super::MetricValue::Gauge { value } => Self::Gauge(Gauge { value }),
366            super::MetricValue::Set { values } => Self::Set(Set {
367                values: values.into_iter().collect(),
368            }),
369            super::MetricValue::Distribution { samples, statistic } => {
370                Self::Distribution2(Distribution2 {
371                    samples: samples.into_iter().map(Into::into).collect(),
372                    statistic: match statistic {
373                        super::StatisticKind::Histogram => StatisticKind::Histogram,
374                        super::StatisticKind::Summary => StatisticKind::Summary,
375                    }
376                    .into(),
377                })
378            }
379            super::MetricValue::AggregatedHistogram {
380                buckets,
381                count,
382                sum,
383            } => Self::AggregatedHistogram3(AggregatedHistogram3 {
384                buckets: buckets.into_iter().map(Into::into).collect(),
385                count,
386                sum,
387            }),
388            super::MetricValue::AggregatedSummary {
389                quantiles,
390                count,
391                sum,
392            } => Self::AggregatedSummary3(AggregatedSummary3 {
393                quantiles: quantiles.into_iter().map(Into::into).collect(),
394                count,
395                sum,
396            }),
397            super::MetricValue::Sketch { sketch } => match sketch {
398                MetricSketch::AgentDDSketch(ddsketch) => {
399                    let bin_map = ddsketch.bin_map();
400                    let (keys, counts) = bin_map.into_parts();
401                    let keys = keys.into_iter().map(i32::from).collect();
402                    let counts = counts.into_iter().map(u32::from).collect();
403
404                    Self::Sketch(Sketch {
405                        sketch: Some(sketch::Sketch::AgentDdSketch(sketch::AgentDdSketch {
406                            count: ddsketch.count(),
407                            min: ddsketch.min().unwrap_or(f64::MAX),
408                            max: ddsketch.max().unwrap_or(f64::MIN),
409                            sum: ddsketch.sum().unwrap_or(0.0),
410                            avg: ddsketch.avg().unwrap_or(0.0),
411                            k: keys,
412                            n: counts,
413                        })),
414                    })
415                }
416            },
417        }
418    }
419}
420
421impl From<super::Metric> for WithMetadata<Metric> {
422    fn from(metric: super::Metric) -> Self {
423        let (series, data, metadata) = metric.into_parts();
424        let name = series.name.name;
425        let namespace = series.name.namespace.unwrap_or_default();
426
427        // Value never wraps as timestamp_subsec_nanos returns a value <= 1_999_999_999
428        // (as per chrono leap-second specs), which is below i32::MAX
429        #[allow(clippy::cast_possible_wrap)]
430        let timestamp = data.time.timestamp.map(|ts| prost_types::Timestamp {
431            seconds: ts.timestamp(),
432            nanos: ts.timestamp_subsec_nanos() as i32,
433        });
434
435        let interval_ms = data.time.interval_ms.map_or(0, std::num::NonZeroU32::get);
436
437        let tags = series.tags.unwrap_or_default();
438
439        let kind = match data.kind {
440            super::MetricKind::Incremental => metric::Kind::Incremental,
441            super::MetricKind::Absolute => metric::Kind::Absolute,
442        }
443        .into();
444
445        let metric = MetricValue::from(data.value);
446
447        // Include the "single" value of the tags in order to be forward-compatible with older
448        // versions of Vector.
449        let tags_v1 = tags
450            .0
451            .iter()
452            .filter_map(|(tag, values)| {
453                values
454                    .as_single()
455                    .map(|value| (tag.clone(), value.to_string()))
456            })
457            .collect();
458        // These are the full tag values.
459        let tags_v2 = tags
460            .0
461            .into_iter()
462            .map(|(tag, values)| {
463                let values = values
464                    .into_iter()
465                    .map(|value| TagValue {
466                        value: value.into_option(),
467                    })
468                    .collect();
469                (tag, TagValues { values })
470            })
471            .collect();
472
473        #[allow(deprecated)]
474        let data = Metric {
475            name,
476            namespace,
477            timestamp,
478            tags_v1,
479            tags_v2,
480            kind,
481            interval_ms,
482            value: Some(metric),
483            metadata: Some(encode_value(metadata.value().clone())),
484            metadata_full: Some(metadata.clone().into()),
485        };
486
487        Self { data, metadata }
488    }
489}
490
491impl From<super::Event> for Event {
492    fn from(event: super::Event) -> Self {
493        WithMetadata::<Self>::from(event).data
494    }
495}
496
497impl From<super::Event> for WithMetadata<Event> {
498    fn from(event: super::Event) -> Self {
499        match event {
500            super::Event::Log(log_event) => WithMetadata::<Log>::from(log_event).into(),
501            super::Event::Metric(metric) => WithMetadata::<Metric>::from(metric).into(),
502            super::Event::Trace(trace) => WithMetadata::<Trace>::from(trace).into(),
503        }
504    }
505}
506
507impl From<super::Event> for EventWrapper {
508    fn from(event: super::Event) -> Self {
509        WithMetadata::<EventWrapper>::from(event).data
510    }
511}
512
513impl From<super::Event> for WithMetadata<EventWrapper> {
514    fn from(event: super::Event) -> Self {
515        WithMetadata::<Event>::from(event).into()
516    }
517}
518
519impl From<AgentDDSketch> for Sketch {
520    fn from(ddsketch: AgentDDSketch) -> Self {
521        let bin_map = ddsketch.bin_map();
522        let (keys, counts) = bin_map.into_parts();
523        let ddsketch = sketch::AgentDdSketch {
524            count: ddsketch.count(),
525            min: ddsketch.min().unwrap_or(f64::MAX),
526            max: ddsketch.max().unwrap_or(f64::MIN),
527            sum: ddsketch.sum().unwrap_or(0.0),
528            avg: ddsketch.avg().unwrap_or(0.0),
529            k: keys.into_iter().map(i32::from).collect(),
530            n: counts.into_iter().map(u32::from).collect(),
531        };
532        Sketch {
533            sketch: Some(sketch::Sketch::AgentDdSketch(ddsketch)),
534        }
535    }
536}
537
538impl From<sketch::AgentDdSketch> for MetricSketch {
539    fn from(sketch: sketch::AgentDdSketch) -> Self {
540        // These safe conversions are annoying because the Datadog Agent internally uses i16/u16,
541        // but the proto definition uses i32/u32, so we have to jump through these hoops.
542        let keys = sketch
543            .k
544            .into_iter()
545            .map(|k| (k, k > 0))
546            .map(|(k, pos)| {
547                k.try_into()
548                    .unwrap_or(if pos { i16::MAX } else { i16::MIN })
549            })
550            .collect::<Vec<_>>();
551        let counts = sketch
552            .n
553            .into_iter()
554            .map(|n| n.try_into().unwrap_or(u16::MAX))
555            .collect::<Vec<_>>();
556        MetricSketch::AgentDDSketch(
557            AgentDDSketch::from_raw(
558                sketch.count,
559                sketch.min,
560                sketch.max,
561                sketch.sum,
562                sketch.avg,
563                &keys,
564                &counts,
565            )
566            .expect("keys/counts were unexpectedly mismatched"),
567        )
568    }
569}
570
571impl From<super::metadata::Secrets> for Secrets {
572    fn from(value: super::metadata::Secrets) -> Self {
573        Self {
574            entries: value.into_iter().map(|(k, v)| (k, v.to_string())).collect(),
575        }
576    }
577}
578
579impl From<Secrets> for super::metadata::Secrets {
580    fn from(value: Secrets) -> Self {
581        let mut secrets = Self::new();
582        for (k, v) in value.entries {
583            secrets.insert(k, v);
584        }
585
586        secrets
587    }
588}
589
590impl From<super::DatadogMetricOriginMetadata> for DatadogOriginMetadata {
591    fn from(value: super::DatadogMetricOriginMetadata) -> Self {
592        Self {
593            origin_product: value.product(),
594            origin_category: value.category(),
595            origin_service: value.service(),
596        }
597    }
598}
599
600impl From<DatadogOriginMetadata> for super::DatadogMetricOriginMetadata {
601    fn from(value: DatadogOriginMetadata) -> Self {
602        Self::new(
603            value.origin_product,
604            value.origin_category,
605            value.origin_service,
606        )
607    }
608}
609
610impl From<crate::config::OutputId> for OutputId {
611    fn from(value: crate::config::OutputId) -> Self {
612        Self {
613            component: value.component.into_id(),
614            port: value.port,
615        }
616    }
617}
618
619impl From<OutputId> for crate::config::OutputId {
620    fn from(value: OutputId) -> Self {
621        Self::from((value.component, value.port))
622    }
623}
624
625impl From<EventMetadata> for Metadata {
626    fn from(value: EventMetadata) -> Self {
627        let super::metadata::Inner {
628            value,
629            secrets,
630            source_id,
631            source_type,
632            upstream_id,
633            datadog_origin_metadata,
634            source_event_id,
635            ..
636        } = value.into_owned();
637
638        let secrets = (!secrets.is_empty()).then(|| secrets.into());
639
640        Self {
641            value: Some(encode_value(value)),
642            datadog_origin_metadata: datadog_origin_metadata.map(Into::into),
643            source_id: source_id.map(|s| s.to_string()),
644            source_type: source_type.map(|s| s.to_string()),
645            upstream_id: upstream_id.map(|id| id.as_ref().clone()).map(Into::into),
646            secrets,
647            source_event_id: source_event_id.map_or(vec![], std::convert::Into::into),
648        }
649    }
650}
651
652impl From<Metadata> for EventMetadata {
653    fn from(value: Metadata) -> Self {
654        let Metadata {
655            value: metadata_value,
656            source_id,
657            source_type,
658            upstream_id,
659            secrets,
660            datadog_origin_metadata,
661            source_event_id,
662        } = value;
663
664        let metadata_value = metadata_value.and_then(decode_value);
665        let source_id = source_id.map(|s| Arc::new(s.into()));
666        let upstream_id = upstream_id.map(|id| Arc::new(id.into()));
667        let secrets = secrets.map(Into::into);
668        let datadog_origin_metadata = datadog_origin_metadata.map(Into::into);
669        let source_event_id = if source_event_id.is_empty() {
670            None
671        } else {
672            match Uuid::from_slice(&source_event_id) {
673                Ok(id) => Some(id),
674                Err(error) => {
675                    error!(
676                        %error,
677                        source_event_id = %String::from_utf8_lossy(&source_event_id),
678                        "Failed to parse source_event_id.",
679                    );
680                    None
681                }
682            }
683        };
684
685        EventMetadata {
686            inner: Arc::new(Inner {
687                value: metadata_value
688                    .unwrap_or_else(|| vrl::value::Value::Object(ObjectMap::new())),
689                secrets: secrets.unwrap_or_default(),
690                finalizers: EventFinalizers::default(),
691                source_id,
692                source_type: source_type.map(Into::into),
693                upstream_id,
694                schema_definition: default_schema_definition(),
695                dropped_fields: ObjectMap::new(),
696                datadog_origin_metadata,
697                source_event_id,
698            }),
699            last_transform_timestamp: None,
700        }
701    }
702}
703
704fn decode_value(input: Value) -> Option<super::Value> {
705    match input.kind {
706        Some(value::Kind::RawBytes(data)) => Some(super::Value::Bytes(data)),
707        // Sign is never lost as ts.nanos is always non negative (per proto spec)
708        #[allow(clippy::cast_sign_loss)]
709        Some(value::Kind::Timestamp(ts)) => Some(super::Value::Timestamp(
710            chrono::Utc
711                .timestamp_opt(ts.seconds, ts.nanos as u32)
712                .single()
713                .expect("invalid timestamp"),
714        )),
715        Some(value::Kind::Integer(value)) => Some(super::Value::Integer(value)),
716        Some(value::Kind::Float(value)) => Some(super::Value::Float(NotNan::new(value).unwrap())),
717        Some(value::Kind::Boolean(value)) => Some(super::Value::Boolean(value)),
718        Some(value::Kind::Map(map)) => decode_map(map.fields),
719        Some(value::Kind::Array(array)) => decode_array(array.items),
720        Some(value::Kind::Null(_)) => Some(super::Value::Null),
721        None => {
722            error!("Encoded event contains unknown value kind.");
723            None
724        }
725    }
726}
727
728fn decode_map(fields: BTreeMap<String, Value>) -> Option<super::Value> {
729    fields
730        .into_iter()
731        .map(|(key, value)| decode_value(value).map(|value| (key.into(), value)))
732        .collect::<Option<ObjectMap>>()
733        .map(event::Value::Object)
734}
735
736fn decode_array(items: Vec<Value>) -> Option<super::Value> {
737    items
738        .into_iter()
739        .map(decode_value)
740        .collect::<Option<Vec<_>>>()
741        .map(super::Value::Array)
742}
743
744fn encode_value(value: super::Value) -> Value {
745    Value {
746        kind: match value {
747            super::Value::Bytes(b) => Some(value::Kind::RawBytes(b)),
748            super::Value::Regex(regex) => Some(value::Kind::RawBytes(regex.as_bytes())),
749            // Value never wraps as timestamp_subsec_nanos returns a value <= 1_999_999_999
750            // (as per chrono leap-second specs), which is below i32::MAX
751            #[allow(clippy::cast_possible_wrap)]
752            super::Value::Timestamp(ts) => Some(value::Kind::Timestamp(prost_types::Timestamp {
753                seconds: ts.timestamp(),
754                nanos: ts.timestamp_subsec_nanos() as i32,
755            })),
756            super::Value::Integer(value) => Some(value::Kind::Integer(value)),
757            super::Value::Float(value) => Some(value::Kind::Float(value.into_inner())),
758            super::Value::Boolean(value) => Some(value::Kind::Boolean(value)),
759            super::Value::Object(fields) => Some(value::Kind::Map(encode_map(fields))),
760            super::Value::Array(items) => Some(value::Kind::Array(encode_array(items))),
761            super::Value::Null => Some(value::Kind::Null(ValueNull::NullValue as i32)),
762        },
763    }
764}
765
766fn encode_map(fields: ObjectMap) -> ValueMap {
767    ValueMap {
768        fields: fields
769            .into_iter()
770            .map(|(key, value)| (key.into(), encode_value(value)))
771            .collect(),
772    }
773}
774
775fn encode_array(items: Vec<super::Value>) -> ValueArray {
776    ValueArray {
777        items: items.into_iter().map(encode_value).collect(),
778    }
779}
vector_core/event/proto.rs

vector_core/event/
proto.rs
