vector_core/event/
mod.rs

1use std::{convert::TryInto, fmt::Debug, sync::Arc};
2
3pub use array::{EventArray, EventContainer, LogArray, MetricArray, TraceArray, into_event_stream};
4pub use estimated_json_encoded_size_of::EstimatedJsonEncodedSizeOf;
5pub use finalization::{
6    BatchNotifier, BatchStatus, BatchStatusReceiver, EventFinalizer, EventFinalizerGroups,
7    EventFinalizers, EventStatus, Finalizable, GroupedFinalizable, MergeFinalizable,
8};
9pub use log_event::LogEvent;
10pub use metadata::{DatadogMetricOriginMetadata, EventMetadata, Secrets, WithMetadata};
11pub use metric::{Metric, MetricKind, MetricTags, MetricValue, StatisticKind};
12pub use r#ref::{EventMutRef, EventRef};
13pub use ser::{MAX_VALUE_NESTING_FRAMES, event_exceeds_max_nesting_cost};
14use serde::{Deserialize, Serialize};
15pub use trace::TraceEvent;
16use vector_buffers::EventCount;
17use vector_common::{
18    EventDataEq, byte_size_of::ByteSizeOf, config::ComponentKey, finalization,
19    internal_event::TaggedEventsSent, json_size::JsonSize, request_metadata::GetEventCountTags,
20};
21pub use vrl::value::{KeyString, ObjectMap, Value};
22pub use vrl_target::MetricTagMode;
23pub use vrl_target::{TargetEvents, VrlTarget};
24
25use crate::config::{LogNamespace, OutputId};
26
27#[cfg(any(test, feature = "generate-fixtures"))]
28pub(crate) mod arbitrary_impl;
29pub mod array;
30pub mod discriminant;
31mod estimated_json_encoded_size_of;
32mod log_event;
33#[cfg(feature = "lua")]
34pub mod lua;
35pub mod merge_state;
36mod metadata;
37pub mod metric;
38pub mod proto;
39mod r#ref;
40mod ser;
41
42#[cfg(test)]
43mod test;
44mod trace;
45pub mod util;
46mod vrl_target;
47
48pub const PARTIAL: &str = "_partial";
49
50#[derive(PartialEq, Debug, Clone, Serialize, Deserialize)]
51#[serde(rename_all = "snake_case")]
52#[allow(clippy::large_enum_variant)]
53pub enum Event {
54    Log(LogEvent),
55    Metric(Metric),
56    Trace(TraceEvent),
57}
58
59impl ByteSizeOf for Event {
60    fn allocated_bytes(&self) -> usize {
61        match self {
62            Event::Log(log_event) => log_event.allocated_bytes(),
63            Event::Metric(metric_event) => metric_event.allocated_bytes(),
64            Event::Trace(trace_event) => trace_event.allocated_bytes(),
65        }
66    }
67}
68
69impl EstimatedJsonEncodedSizeOf for Event {
70    fn estimated_json_encoded_size_of(&self) -> JsonSize {
71        match self {
72            Event::Log(log_event) => log_event.estimated_json_encoded_size_of(),
73            Event::Metric(metric_event) => metric_event.estimated_json_encoded_size_of(),
74            Event::Trace(trace_event) => trace_event.estimated_json_encoded_size_of(),
75        }
76    }
77}
78
79impl EventCount for Event {
80    fn event_count(&self) -> usize {
81        1
82    }
83}
84
85impl Finalizable for Event {
86    fn take_finalizers(&mut self) -> EventFinalizers {
87        match self {
88            Event::Log(log_event) => log_event.take_finalizers(),
89            Event::Metric(metric) => metric.take_finalizers(),
90            Event::Trace(trace_event) => trace_event.take_finalizers(),
91        }
92    }
93}
94
95impl MergeFinalizable for Event {
96    fn merge_finalizers(&mut self, finalizers: EventFinalizers) {
97        match self {
98            Event::Log(log_event) => log_event.merge_finalizers(finalizers),
99            Event::Metric(metric) => metric.merge_finalizers(finalizers),
100            Event::Trace(trace_event) => trace_event.merge_finalizers(finalizers),
101        }
102    }
103}
104
105impl GetEventCountTags for Event {
106    fn get_tags(&self) -> TaggedEventsSent {
107        match self {
108            Event::Log(log) => log.get_tags(),
109            Event::Metric(metric) => metric.get_tags(),
110            Event::Trace(trace) => trace.get_tags(),
111        }
112    }
113}
114
115impl Event {
116    /// Return self as a `LogEvent`
117    ///
118    /// # Panics
119    ///
120    /// This function panics if self is anything other than an `Event::Log`.
121    pub fn as_log(&self) -> &LogEvent {
122        match self {
123            Event::Log(log) => log,
124            _ => panic!("Failed type coercion, {self:?} is not a log event"),
125        }
126    }
127
128    /// Return self as a mutable `LogEvent`
129    ///
130    /// # Panics
131    ///
132    /// This function panics if self is anything other than an `Event::Log`.
133    pub fn as_mut_log(&mut self) -> &mut LogEvent {
134        match self {
135            Event::Log(log) => log,
136            _ => panic!("Failed type coercion, {self:?} is not a log event"),
137        }
138    }
139
140    /// Coerces self into a `LogEvent`
141    ///
142    /// # Panics
143    ///
144    /// This function panics if self is anything other than an `Event::Log`.
145    pub fn into_log(self) -> LogEvent {
146        match self {
147            Event::Log(log) => log,
148            _ => panic!("Failed type coercion, {self:?} is not a log event"),
149        }
150    }
151
152    /// Fallibly coerces self into a `LogEvent`
153    ///
154    /// If the event is a `LogEvent`, then `Some(log_event)` is returned, otherwise `None`.
155    pub fn try_into_log(self) -> Option<LogEvent> {
156        match self {
157            Event::Log(log) => Some(log),
158            _ => None,
159        }
160    }
161
162    /// Return self as a `LogEvent` if possible
163    ///
164    /// If the event is a `LogEvent`, then `Some(&log_event)` is returned, otherwise `None`.
165    pub fn maybe_as_log(&self) -> Option<&LogEvent> {
166        match self {
167            Event::Log(log) => Some(log),
168            _ => None,
169        }
170    }
171
172    /// Return self as a `Metric`
173    ///
174    /// # Panics
175    ///
176    /// This function panics if self is anything other than an `Event::Metric`.
177    pub fn as_metric(&self) -> &Metric {
178        match self {
179            Event::Metric(metric) => metric,
180            _ => panic!("Failed type coercion, {self:?} is not a metric"),
181        }
182    }
183
184    /// Return self as a mutable `Metric`
185    ///
186    /// # Panics
187    ///
188    /// This function panics if self is anything other than an `Event::Metric`.
189    pub fn as_mut_metric(&mut self) -> &mut Metric {
190        match self {
191            Event::Metric(metric) => metric,
192            _ => panic!("Failed type coercion, {self:?} is not a metric"),
193        }
194    }
195
196    /// Coerces self into `Metric`
197    ///
198    /// # Panics
199    ///
200    /// This function panics if self is anything other than an `Event::Metric`.
201    pub fn into_metric(self) -> Metric {
202        match self {
203            Event::Metric(metric) => metric,
204            _ => panic!("Failed type coercion, {self:?} is not a metric"),
205        }
206    }
207
208    /// Fallibly coerces self into a `Metric`
209    ///
210    /// If the event is a `Metric`, then `Some(metric)` is returned, otherwise `None`.
211    pub fn try_into_metric(self) -> Option<Metric> {
212        match self {
213            Event::Metric(metric) => Some(metric),
214            _ => None,
215        }
216    }
217
218    /// Return self as a `TraceEvent`
219    ///
220    /// # Panics
221    ///
222    /// This function panics if self is anything other than an `Event::Trace`.
223    pub fn as_trace(&self) -> &TraceEvent {
224        match self {
225            Event::Trace(trace) => trace,
226            _ => panic!("Failed type coercion, {self:?} is not a trace event"),
227        }
228    }
229
230    /// Return self as a mutable `TraceEvent`
231    ///
232    /// # Panics
233    ///
234    /// This function panics if self is anything other than an `Event::Trace`.
235    pub fn as_mut_trace(&mut self) -> &mut TraceEvent {
236        match self {
237            Event::Trace(trace) => trace,
238            _ => panic!("Failed type coercion, {self:?} is not a trace event"),
239        }
240    }
241
242    /// Coerces self into a `TraceEvent`
243    ///
244    /// # Panics
245    ///
246    /// This function panics if self is anything other than an `Event::Trace`.
247    pub fn into_trace(self) -> TraceEvent {
248        match self {
249            Event::Trace(trace) => trace,
250            _ => panic!("Failed type coercion, {self:?} is not a trace event"),
251        }
252    }
253
254    /// Fallibly coerces self into a `TraceEvent`
255    ///
256    /// If the event is a `TraceEvent`, then `Some(trace)` is returned, otherwise `None`.
257    pub fn try_into_trace(self) -> Option<TraceEvent> {
258        match self {
259            Event::Trace(trace) => Some(trace),
260            _ => None,
261        }
262    }
263
264    pub fn metadata(&self) -> &EventMetadata {
265        match self {
266            Self::Log(log) => log.metadata(),
267            Self::Metric(metric) => metric.metadata(),
268            Self::Trace(trace) => trace.metadata(),
269        }
270    }
271
272    pub fn metadata_mut(&mut self) -> &mut EventMetadata {
273        match self {
274            Self::Log(log) => log.metadata_mut(),
275            Self::Metric(metric) => metric.metadata_mut(),
276            Self::Trace(trace) => trace.metadata_mut(),
277        }
278    }
279
280    /// Destroy the event and return the metadata.
281    pub fn into_metadata(self) -> EventMetadata {
282        match self {
283            Self::Log(log) => log.into_parts().1,
284            Self::Metric(metric) => metric.into_parts().2,
285            Self::Trace(trace) => trace.into_parts().1,
286        }
287    }
288
289    #[must_use]
290    pub fn with_batch_notifier(self, batch: &BatchNotifier) -> Self {
291        match self {
292            Self::Log(log) => log.with_batch_notifier(batch).into(),
293            Self::Metric(metric) => metric.with_batch_notifier(batch).into(),
294            Self::Trace(trace) => trace.with_batch_notifier(batch).into(),
295        }
296    }
297
298    #[must_use]
299    pub fn with_batch_notifier_option(self, batch: &Option<BatchNotifier>) -> Self {
300        match self {
301            Self::Log(log) => log.with_batch_notifier_option(batch).into(),
302            Self::Metric(metric) => metric.with_batch_notifier_option(batch).into(),
303            Self::Trace(trace) => trace.with_batch_notifier_option(batch).into(),
304        }
305    }
306
307    /// Returns a reference to the event metadata source.
308    #[must_use]
309    pub fn source_id(&self) -> Option<&Arc<ComponentKey>> {
310        self.metadata().source_id()
311    }
312
313    /// Sets the `source_id` in the event metadata to the provided value.
314    pub fn set_source_id(&mut self, source_id: Arc<ComponentKey>) {
315        self.metadata_mut().set_source_id(source_id);
316    }
317
318    /// Sets the `upstream_id` in the event metadata to the provided value.
319    pub fn set_upstream_id(&mut self, upstream_id: Arc<OutputId>) {
320        self.metadata_mut().set_upstream_id(upstream_id);
321    }
322
323    /// Sets the `source_type` in the event metadata to the provided value.
324    pub fn set_source_type(&mut self, source_type: &'static str) {
325        self.metadata_mut().set_source_type(source_type);
326    }
327
328    /// Sets the `source_id` in the event metadata to the provided value.
329    #[must_use]
330    pub fn with_source_id(mut self, source_id: Arc<ComponentKey>) -> Self {
331        self.metadata_mut().set_source_id(source_id);
332        self
333    }
334
335    /// Sets the `source_type` in the event metadata to the provided value.
336    #[must_use]
337    pub fn with_source_type(mut self, source_type: &'static str) -> Self {
338        self.metadata_mut().set_source_type(source_type);
339        self
340    }
341
342    /// Sets the `upstream_id` in the event metadata to the provided value.
343    #[must_use]
344    pub fn with_upstream_id(mut self, upstream_id: Arc<OutputId>) -> Self {
345        self.metadata_mut().set_upstream_id(upstream_id);
346        self
347    }
348
349    /// Creates an Event from a JSON value.
350    ///
351    /// # Errors
352    /// If a non-object JSON value is passed in with the `Legacy` namespace, this will return an error.
353    pub fn from_json_value(
354        value: serde_json::Value,
355        log_namespace: LogNamespace,
356    ) -> crate::Result<Self> {
357        match log_namespace {
358            LogNamespace::Vector => Ok(LogEvent::from(Value::from(value)).into()),
359            LogNamespace::Legacy => match value {
360                serde_json::Value::Object(fields) => Ok(LogEvent::from(
361                    fields
362                        .into_iter()
363                        .map(|(k, v)| (k.into(), v.into()))
364                        .collect::<ObjectMap>(),
365                )
366                .into()),
367                _ => Err(crate::Error::from(
368                    "Attempted to convert non-Object JSON into an Event.",
369                )),
370            },
371        }
372    }
373}
374
375impl EventDataEq for Event {
376    fn event_data_eq(&self, other: &Self) -> bool {
377        match (self, other) {
378            (Self::Log(a), Self::Log(b)) => a.event_data_eq(b),
379            (Self::Metric(a), Self::Metric(b)) => a.event_data_eq(b),
380            (Self::Trace(a), Self::Trace(b)) => a.event_data_eq(b),
381            _ => false,
382        }
383    }
384}
385
386impl finalization::AddBatchNotifier for Event {
387    fn add_batch_notifier(&mut self, batch: BatchNotifier) {
388        let finalizer = EventFinalizer::new(batch);
389        match self {
390            Self::Log(log) => log.add_finalizer(finalizer),
391            Self::Metric(metric) => metric.add_finalizer(finalizer),
392            Self::Trace(trace) => trace.add_finalizer(finalizer),
393        }
394    }
395}
396
397impl TryInto<serde_json::Value> for Event {
398    type Error = serde_json::Error;
399
400    fn try_into(self) -> Result<serde_json::Value, Self::Error> {
401        match self {
402            Event::Log(fields) => serde_json::to_value(fields),
403            Event::Metric(metric) => serde_json::to_value(metric),
404            Event::Trace(fields) => serde_json::to_value(fields),
405        }
406    }
407}
408
409impl From<proto::StatisticKind> for StatisticKind {
410    fn from(kind: proto::StatisticKind) -> Self {
411        match kind {
412            proto::StatisticKind::Histogram => StatisticKind::Histogram,
413            proto::StatisticKind::Summary => StatisticKind::Summary,
414        }
415    }
416}
417
418impl From<metric::Sample> for proto::DistributionSample {
419    fn from(sample: metric::Sample) -> Self {
420        Self {
421            value: sample.value,
422            rate: sample.rate,
423        }
424    }
425}
426
427impl From<proto::DistributionSample> for metric::Sample {
428    fn from(sample: proto::DistributionSample) -> Self {
429        Self {
430            value: sample.value,
431            rate: sample.rate,
432        }
433    }
434}
435
436impl From<proto::HistogramBucket> for metric::Bucket {
437    fn from(bucket: proto::HistogramBucket) -> Self {
438        Self {
439            upper_limit: bucket.upper_limit,
440            count: u64::from(bucket.count),
441        }
442    }
443}
444
445impl From<metric::Bucket> for proto::HistogramBucket3 {
446    fn from(bucket: metric::Bucket) -> Self {
447        Self {
448            upper_limit: bucket.upper_limit,
449            count: bucket.count,
450        }
451    }
452}
453
454impl From<proto::HistogramBucket3> for metric::Bucket {
455    fn from(bucket: proto::HistogramBucket3) -> Self {
456        Self {
457            upper_limit: bucket.upper_limit,
458            count: bucket.count,
459        }
460    }
461}
462
463impl From<metric::Quantile> for proto::SummaryQuantile {
464    fn from(quantile: metric::Quantile) -> Self {
465        Self {
466            quantile: quantile.quantile,
467            value: quantile.value,
468        }
469    }
470}
471
472impl From<proto::SummaryQuantile> for metric::Quantile {
473    fn from(quantile: proto::SummaryQuantile) -> Self {
474        Self {
475            quantile: quantile.quantile,
476            value: quantile.value,
477        }
478    }
479}
480
481impl From<LogEvent> for Event {
482    fn from(log: LogEvent) -> Self {
483        Event::Log(log)
484    }
485}
486
487impl From<Metric> for Event {
488    fn from(metric: Metric) -> Self {
489        Event::Metric(metric)
490    }
491}
492
493impl From<TraceEvent> for Event {
494    fn from(trace: TraceEvent) -> Self {
495        Event::Trace(trace)
496    }
497}
498
499pub trait MaybeAsLogMut {
500    fn maybe_as_log_mut(&mut self) -> Option<&mut LogEvent>;
501}
502
503impl MaybeAsLogMut for Event {
504    fn maybe_as_log_mut(&mut self) -> Option<&mut LogEvent> {
505        match self {
506            Event::Log(log) => Some(log),
507            _ => None,
508        }
509    }
510}
vector_core/event/mod.rs

vector_core/event/
mod.rs
