vector_core/event/
array.rs

1#![deny(missing_docs)]
2//! This module contains the definitions and wrapper types for handling
3//! arrays of type `Event`, in the various forms they may appear.
4
5use std::{iter, slice, vec};
6
7use futures::{Stream, stream};
8#[cfg(test)]
9use quickcheck::{Arbitrary, Gen};
10use vector_buffers::EventCount;
11use vector_common::{
12    byte_size_of::ByteSizeOf,
13    finalization::{
14        AddBatchNotifier, BatchNotifier, EventFinalizerGroups, EventFinalizers, Finalizable,
15        GroupedFinalizable, MergeFinalizable,
16    },
17    json_size::JsonSize,
18};
19
20use super::{
21    EstimatedJsonEncodedSizeOf, Event, EventDataEq, EventFinalizer, EventMetadata, EventMutRef,
22    EventRef, LogEvent, Metric, TraceEvent,
23};
24
25/// The type alias for an array of `LogEvent` elements.
26pub type LogArray = Vec<LogEvent>;
27
28/// The type alias for an array of `TraceEvent` elements.
29pub type TraceArray = Vec<TraceEvent>;
30
31/// The type alias for an array of `Metric` elements.
32pub type MetricArray = Vec<Metric>;
33
34/// The core trait to abstract over any type that may work as an array
35/// of events. This is effectively the same as the standard
36/// `IntoIterator<Item = Event>` implementations, but that would
37/// conflict with the base implementation for the type aliases below.
38pub trait EventContainer: ByteSizeOf + EstimatedJsonEncodedSizeOf {
39    /// The type of `Iterator` used to turn this container into events.
40    type IntoIter: Iterator<Item = Event>;
41
42    /// The number of events in this container.
43    fn len(&self) -> usize;
44
45    /// Is this container empty?
46    fn is_empty(&self) -> bool {
47        self.len() == 0
48    }
49
50    /// Turn this container into an iterator over `Event`.
51    fn into_events(self) -> Self::IntoIter;
52}
53
54/// Turn a container into a futures stream over the contained `Event`
55/// type.  This would ideally be implemented as a default method on
56/// `trait EventContainer`, but the required feature (associated type
57/// defaults) is still unstable.
58/// See <https://github.com/rust-lang/rust/issues/29661>
59pub fn into_event_stream(container: impl EventContainer) -> impl Stream<Item = Event> + Unpin {
60    stream::iter(container.into_events())
61}
62
63impl EventContainer for Event {
64    type IntoIter = iter::Once<Event>;
65
66    fn len(&self) -> usize {
67        1
68    }
69
70    fn is_empty(&self) -> bool {
71        false
72    }
73
74    fn into_events(self) -> Self::IntoIter {
75        iter::once(self)
76    }
77}
78
79impl EventContainer for LogEvent {
80    type IntoIter = iter::Once<Event>;
81
82    fn len(&self) -> usize {
83        1
84    }
85
86    fn is_empty(&self) -> bool {
87        false
88    }
89
90    fn into_events(self) -> Self::IntoIter {
91        iter::once(self.into())
92    }
93}
94
95impl EventContainer for Metric {
96    type IntoIter = iter::Once<Event>;
97
98    fn len(&self) -> usize {
99        1
100    }
101
102    fn is_empty(&self) -> bool {
103        false
104    }
105
106    fn into_events(self) -> Self::IntoIter {
107        iter::once(self.into())
108    }
109}
110
111impl EventContainer for LogArray {
112    type IntoIter = iter::Map<vec::IntoIter<LogEvent>, fn(LogEvent) -> Event>;
113
114    fn len(&self) -> usize {
115        self.len()
116    }
117
118    fn into_events(self) -> Self::IntoIter {
119        self.into_iter().map(Into::into)
120    }
121}
122
123impl EventContainer for MetricArray {
124    type IntoIter = iter::Map<vec::IntoIter<Metric>, fn(Metric) -> Event>;
125
126    fn len(&self) -> usize {
127        self.len()
128    }
129
130    fn into_events(self) -> Self::IntoIter {
131        self.into_iter().map(Into::into)
132    }
133}
134
135/// An array of one of the `Event` variants exclusively.
136#[derive(Clone, Debug, PartialEq)]
137pub enum EventArray {
138    /// An array of type `LogEvent`
139    Logs(LogArray),
140    /// An array of type `Metric`
141    Metrics(MetricArray),
142    /// An array of type `TraceEvent`
143    Traces(TraceArray),
144}
145
146impl EventArray {
147    /// Iterate over references to this array's events.
148    pub fn iter_events(&self) -> impl Iterator<Item = EventRef<'_>> {
149        match self {
150            Self::Logs(array) => EventArrayIter::Logs(array.iter()),
151            Self::Metrics(array) => EventArrayIter::Metrics(array.iter()),
152            Self::Traces(array) => EventArrayIter::Traces(array.iter()),
153        }
154    }
155
156    /// Iterate over mutable references to this array's events.
157    pub fn iter_events_mut(&mut self) -> impl Iterator<Item = EventMutRef<'_>> {
158        match self {
159            Self::Logs(array) => EventArrayIterMut::Logs(array.iter_mut()),
160            Self::Metrics(array) => EventArrayIterMut::Metrics(array.iter_mut()),
161            Self::Traces(array) => EventArrayIterMut::Traces(array.iter_mut()),
162        }
163    }
164
165    /// Iterate over references to the logs in this array.
166    pub fn iter_logs_mut(&mut self) -> impl Iterator<Item = &mut LogEvent> {
167        match self {
168            Self::Logs(array) => TypedArrayIterMut(Some(array.iter_mut())),
169            _ => TypedArrayIterMut(None),
170        }
171    }
172
173    /// Applies a closure to each event's metadata in this array.
174    pub fn for_each_metadata_mut(&mut self, mut f: impl FnMut(&mut EventMetadata)) {
175        match self {
176            Self::Logs(logs) => {
177                for log in logs {
178                    f(log.metadata_mut());
179                }
180            }
181            Self::Metrics(metrics) => {
182                for metric in metrics {
183                    f(metric.metadata_mut());
184                }
185            }
186            Self::Traces(traces) => {
187                for trace in traces {
188                    f(trace.metadata_mut());
189                }
190            }
191        }
192    }
193}
194
195impl From<Event> for EventArray {
196    fn from(event: Event) -> Self {
197        match event {
198            Event::Log(log) => Self::Logs(vec![log]),
199            Event::Metric(metric) => Self::Metrics(vec![metric]),
200            Event::Trace(trace) => Self::Traces(vec![trace]),
201        }
202    }
203}
204
205impl From<LogEvent> for EventArray {
206    fn from(log: LogEvent) -> Self {
207        Event::from(log).into()
208    }
209}
210
211impl From<Metric> for EventArray {
212    fn from(metric: Metric) -> Self {
213        Event::from(metric).into()
214    }
215}
216
217impl From<TraceEvent> for EventArray {
218    fn from(trace: TraceEvent) -> Self {
219        Event::from(trace).into()
220    }
221}
222
223impl From<LogArray> for EventArray {
224    fn from(array: LogArray) -> Self {
225        Self::Logs(array)
226    }
227}
228
229impl From<MetricArray> for EventArray {
230    fn from(array: MetricArray) -> Self {
231        Self::Metrics(array)
232    }
233}
234
235impl AddBatchNotifier for EventArray {
236    fn add_batch_notifier(&mut self, batch: BatchNotifier) {
237        match self {
238            Self::Logs(array) => array
239                .iter_mut()
240                .for_each(|item| item.add_finalizer(EventFinalizer::new(batch.clone()))),
241            Self::Metrics(array) => array
242                .iter_mut()
243                .for_each(|item| item.add_finalizer(EventFinalizer::new(batch.clone()))),
244            Self::Traces(array) => array
245                .iter_mut()
246                .for_each(|item| item.add_finalizer(EventFinalizer::new(batch.clone()))),
247        }
248    }
249}
250
251impl ByteSizeOf for EventArray {
252    fn allocated_bytes(&self) -> usize {
253        match self {
254            Self::Logs(a) => a.allocated_bytes(),
255            Self::Metrics(a) => a.allocated_bytes(),
256            Self::Traces(a) => a.allocated_bytes(),
257        }
258    }
259}
260
261impl EstimatedJsonEncodedSizeOf for EventArray {
262    fn estimated_json_encoded_size_of(&self) -> JsonSize {
263        match self {
264            Self::Logs(v) => v.estimated_json_encoded_size_of(),
265            Self::Traces(v) => v.estimated_json_encoded_size_of(),
266            Self::Metrics(v) => v.estimated_json_encoded_size_of(),
267        }
268    }
269}
270
271impl EventCount for EventArray {
272    fn event_count(&self) -> usize {
273        match self {
274            Self::Logs(a) => a.len(),
275            Self::Metrics(a) => a.len(),
276            Self::Traces(a) => a.len(),
277        }
278    }
279}
280
281impl EventContainer for EventArray {
282    type IntoIter = EventArrayIntoIter;
283
284    fn len(&self) -> usize {
285        match self {
286            Self::Logs(a) => a.len(),
287            Self::Metrics(a) => a.len(),
288            Self::Traces(a) => a.len(),
289        }
290    }
291
292    fn into_events(self) -> Self::IntoIter {
293        match self {
294            Self::Logs(a) => EventArrayIntoIter::Logs(a.into_iter()),
295            Self::Metrics(a) => EventArrayIntoIter::Metrics(a.into_iter()),
296            Self::Traces(a) => EventArrayIntoIter::Traces(a.into_iter()),
297        }
298    }
299}
300
301impl EventDataEq for EventArray {
302    fn event_data_eq(&self, other: &Self) -> bool {
303        match (self, other) {
304            (Self::Logs(a), Self::Logs(b)) => a.event_data_eq(b),
305            (Self::Metrics(a), Self::Metrics(b)) => a.event_data_eq(b),
306            (Self::Traces(a), Self::Traces(b)) => a.event_data_eq(b),
307            _ => false,
308        }
309    }
310}
311
312impl Finalizable for EventArray {
313    fn take_finalizers(&mut self) -> EventFinalizers {
314        match self {
315            Self::Logs(a) => a.iter_mut().map(Finalizable::take_finalizers).collect(),
316            Self::Metrics(a) => a.iter_mut().map(Finalizable::take_finalizers).collect(),
317            Self::Traces(a) => a.iter_mut().map(Finalizable::take_finalizers).collect(),
318        }
319    }
320
321    fn take_finalizer_groups(&mut self) -> EventFinalizerGroups {
322        match self {
323            Self::Logs(a) => a.iter_mut().map(Finalizable::take_finalizers).collect(),
324            Self::Metrics(a) => a.iter_mut().map(Finalizable::take_finalizers).collect(),
325            Self::Traces(a) => a.iter_mut().map(Finalizable::take_finalizers).collect(),
326        }
327    }
328}
329
330impl GroupedFinalizable for EventArray {
331    fn merge_finalizer_groups(&mut self, finalizers: EventFinalizerGroups) {
332        fn merge_into<T: MergeFinalizable>(items: &mut [T], finalizers: EventFinalizerGroups) {
333            assert_eq!(
334                items.len(),
335                finalizers.len(),
336                "finalizer group count must match EventArray length"
337            );
338
339            for (item, finalizers) in items.iter_mut().zip(finalizers.into_groups()) {
340                item.merge_finalizers(finalizers);
341            }
342        }
343
344        match self {
345            Self::Logs(a) => merge_into(a, finalizers),
346            Self::Metrics(a) => merge_into(a, finalizers),
347            Self::Traces(a) => merge_into(a, finalizers),
348        }
349    }
350}
351
352#[cfg(test)]
353mod tests {
354    use tokio::sync::oneshot::error::TryRecvError;
355    use vector_common::finalization::{BatchStatus, EventStatus};
356
357    use super::*;
358
359    #[test]
360    fn grouped_finalizer_round_trip_preserves_event_ownership() {
361        let (first_batch, mut first_rx) = BatchNotifier::new_with_receiver();
362        let (second_batch, mut second_rx) = BatchNotifier::new_with_receiver();
363
364        let mut first = LogEvent::default();
365        first.add_finalizer(EventFinalizer::new(first_batch));
366
367        let mut second = LogEvent::default();
368        second.add_finalizer(EventFinalizer::new(second_batch));
369
370        let mut array = EventArray::Logs(vec![first, second]);
371        let finalizers = array.take_finalizer_groups();
372        array.merge_finalizer_groups(finalizers);
373
374        let mut events = array.into_events();
375        let mut first = events.next().expect("first event must exist");
376        let mut second = events.next().expect("second event must exist");
377        assert!(events.next().is_none());
378
379        let first_finalizers = first.take_finalizers();
380        first_finalizers.update_status(EventStatus::Delivered);
381        drop(first_finalizers);
382
383        assert_eq!(first_rx.try_recv(), Ok(BatchStatus::Delivered));
384        assert!(matches!(second_rx.try_recv(), Err(TryRecvError::Empty)));
385
386        let second_finalizers = second.take_finalizers();
387        second_finalizers.update_status(EventStatus::Errored);
388        drop(second_finalizers);
389
390        assert_eq!(second_rx.try_recv(), Ok(BatchStatus::Errored));
391    }
392
393    #[test]
394    fn empty_event_array_grouped_round_trip() {
395        let mut array = EventArray::Logs(Vec::new());
396        let finalizers = array.take_finalizer_groups();
397
398        assert!(finalizers.is_empty());
399        array.merge_finalizer_groups(finalizers);
400        assert!(array.is_empty());
401    }
402}
403
404#[cfg(test)]
405impl Arbitrary for EventArray {
406    fn arbitrary(g: &mut Gen) -> Self {
407        let len = u8::arbitrary(g) as usize;
408        let choice: u8 = u8::arbitrary(g);
409        // Quickcheck can't derive Arbitrary for enums, see
410        // https://github.com/BurntSushi/quickcheck/issues/98
411        if choice.is_multiple_of(2) {
412            let mut logs = Vec::new();
413            for _ in 0..len {
414                logs.push(LogEvent::arbitrary(g));
415            }
416            EventArray::Logs(logs)
417        } else {
418            let mut metrics = Vec::new();
419            for _ in 0..len {
420                metrics.push(Metric::arbitrary(g));
421            }
422            EventArray::Metrics(metrics)
423        }
424    }
425
426    fn shrink(&self) -> Box<dyn Iterator<Item = Self>> {
427        match self {
428            EventArray::Logs(logs) => Box::new(logs.shrink().map(EventArray::Logs)),
429            EventArray::Metrics(metrics) => Box::new(metrics.shrink().map(EventArray::Metrics)),
430            EventArray::Traces(traces) => Box::new(traces.shrink().map(EventArray::Traces)),
431        }
432    }
433}
434
435/// The iterator type for `EventArray::iter_events`.
436#[derive(Debug)]
437pub enum EventArrayIter<'a> {
438    /// An iterator over type `LogEvent`.
439    Logs(slice::Iter<'a, LogEvent>),
440    /// An iterator over type `Metric`.
441    Metrics(slice::Iter<'a, Metric>),
442    /// An iterator over type `Trace`.
443    Traces(slice::Iter<'a, TraceEvent>),
444}
445
446impl<'a> Iterator for EventArrayIter<'a> {
447    type Item = EventRef<'a>;
448
449    fn next(&mut self) -> Option<Self::Item> {
450        match self {
451            Self::Logs(i) => i.next().map(EventRef::from),
452            Self::Metrics(i) => i.next().map(EventRef::from),
453            Self::Traces(i) => i.next().map(EventRef::from),
454        }
455    }
456}
457
458/// The iterator type for `EventArray::iter_events_mut`.
459#[derive(Debug)]
460pub enum EventArrayIterMut<'a> {
461    /// An iterator over type `LogEvent`.
462    Logs(slice::IterMut<'a, LogEvent>),
463    /// An iterator over type `Metric`.
464    Metrics(slice::IterMut<'a, Metric>),
465    /// An iterator over type `Trace`.
466    Traces(slice::IterMut<'a, TraceEvent>),
467}
468
469impl<'a> Iterator for EventArrayIterMut<'a> {
470    type Item = EventMutRef<'a>;
471
472    fn next(&mut self) -> Option<Self::Item> {
473        match self {
474            Self::Logs(i) => i.next().map(EventMutRef::from),
475            Self::Metrics(i) => i.next().map(EventMutRef::from),
476            Self::Traces(i) => i.next().map(EventMutRef::from),
477        }
478    }
479}
480
481/// The iterator type for `EventArray::into_events`.
482#[derive(Debug)]
483pub enum EventArrayIntoIter {
484    /// An iterator over type `LogEvent`.
485    Logs(vec::IntoIter<LogEvent>),
486    /// An iterator over type `Metric`.
487    Metrics(vec::IntoIter<Metric>),
488    /// An iterator over type `TraceEvent`.
489    Traces(vec::IntoIter<TraceEvent>),
490}
491
492impl Iterator for EventArrayIntoIter {
493    type Item = Event;
494
495    fn next(&mut self) -> Option<Self::Item> {
496        match self {
497            Self::Logs(i) => i.next().map(Into::into),
498            Self::Metrics(i) => i.next().map(Into::into),
499            Self::Traces(i) => i.next().map(Event::Trace),
500        }
501    }
502}
503
504struct TypedArrayIterMut<'a, T>(Option<slice::IterMut<'a, T>>);
505
506impl<'a, T> Iterator for TypedArrayIterMut<'a, T> {
507    type Item = &'a mut T;
508    fn next(&mut self) -> Option<Self::Item> {
509        self.0.as_mut().and_then(Iterator::next)
510    }
511}
512
513/// Intermediate buffer for conversion of a sequence of individual
514/// `Event`s into a sequence of `EventArray`s by coalescing contiguous
515/// events of the same type into one array. This is used by
516/// `events_into_array`.
517#[derive(Debug, Default)]
518pub struct EventArrayBuffer {
519    buffer: Option<EventArray>,
520    max_size: usize,
521}
522
523impl EventArrayBuffer {
524    fn new(max_size: Option<usize>) -> Self {
525        let max_size = max_size.unwrap_or(usize::MAX);
526        let buffer = None;
527        Self { buffer, max_size }
528    }
529
530    #[must_use]
531    fn push(&mut self, event: Event) -> Option<EventArray> {
532        match (event, &mut self.buffer) {
533            (Event::Log(event), Some(EventArray::Logs(array))) if array.len() < self.max_size => {
534                array.push(event);
535                None
536            }
537            (Event::Metric(event), Some(EventArray::Metrics(array)))
538                if array.len() < self.max_size =>
539            {
540                array.push(event);
541                None
542            }
543            (Event::Trace(event), Some(EventArray::Traces(array)))
544                if array.len() < self.max_size =>
545            {
546                array.push(event);
547                None
548            }
549            (event, current) => current.replace(EventArray::from(event)),
550        }
551    }
552
553    fn take(&mut self) -> Option<EventArray> {
554        self.buffer.take()
555    }
556}
557
558/// Convert the iterator over individual `Event`s into an iterator
559/// over coalesced `EventArray`s.
560pub fn events_into_arrays(
561    events: impl IntoIterator<Item = Event>,
562    max_size: Option<usize>,
563) -> impl Iterator<Item = EventArray> {
564    IntoEventArraysIter {
565        inner: events.into_iter().fuse(),
566        current: EventArrayBuffer::new(max_size),
567    }
568}
569
570/// Iterator type implementing `into_arrays`
571pub struct IntoEventArraysIter<I> {
572    inner: iter::Fuse<I>,
573    current: EventArrayBuffer,
574}
575
576impl<I: Iterator<Item = Event>> Iterator for IntoEventArraysIter<I> {
577    type Item = EventArray;
578    fn next(&mut self) -> Option<Self::Item> {
579        for event in self.inner.by_ref() {
580            if let Some(array) = self.current.push(event) {
581                return Some(array);
582            }
583        }
584        self.current.take()
585    }
586}
vector_core/event/array.rs

vector_core/event/
array.rs
