chore(core): Add a EventCount trait to bufferable types (#11227)

This will be used by future changes to the disk buffer to deal with
event arrays, which may have more than one event per record.

lib/vector-buffers/benches/common.rs

@@ -12,7 +12,7 @@ use vector_buffers::{
         builder::TopologyBuilder,
         channel::{BufferReceiver, BufferSender},
     },
-    BufferType,
+    BufferType, EventCount,
 };
 use vector_common::byte_size_of::ByteSizeOf;
 
@@ -37,6 +37,12 @@ impl<const N: usize> ByteSizeOf for Message<N> {
     }
 }
 
+impl<const N: usize> EventCount for Message<N> {
+    fn event_count(&self) -> usize {
+        1
+    }
+}
+
 #[derive(Debug)]
 pub struct EncodeError;
 

lib/vector-buffers/examples/buffer_perf.rs

@@ -22,7 +22,7 @@ use vector_buffers::{
         builder::TopologyBuilder,
         channel::{BufferReceiver, BufferSender},
     },
-    Acker, BufferType, Bufferable, WhenFull,
+    Acker, BufferType, Bufferable, EventCount, WhenFull,
 };
 use vector_common::byte_size_of::ByteSizeOf;
 
@@ -48,6 +48,12 @@ impl ByteSizeOf for VariableMessage {
     }
 }
 
+impl EventCount for VariableMessage {
+    fn event_count(&self) -> usize {
+        1
+    }
+}
+
 impl FixedEncodable for VariableMessage {
     type EncodeError = EncodeError;
     type DecodeError = DecodeError;

lib/vector-buffers/src/disk_v2/tests/known_errors.rs

@@ -23,6 +23,7 @@ use crate::{
         ReaderError,
     },
     encoding::{AsMetadata, Encodable},
+    EventCount,
 };
 
 #[tokio::test]
@@ -725,6 +726,12 @@ async fn reader_throws_error_when_record_is_undecodable_via_metadata() {
         }
     }
 
+    impl EventCount for ControllableRecord {
+        fn event_count(&self) -> usize {
+            1
+        }
+    }
+
     with_temp_dir(|dir| {
         let data_dir = dir.to_path_buf();
 

lib/vector-buffers/src/disk_v2/tests/mod.rs

@@ -12,7 +12,7 @@ use crate::{
     buffer_usage_data::BufferUsageHandle,
     disk_v2::{Buffer, DiskBufferConfig, Reader, Writer},
     encoding::FixedEncodable,
-    Acker, Bufferable, WhenFull,
+    Acker, Bufferable, EventCount, WhenFull,
 };
 
 mod acknowledgements;
@@ -182,6 +182,12 @@ impl ByteSizeOf for SizedRecord {
     }
 }
 
+impl EventCount for SizedRecord {
+    fn event_count(&self) -> usize {
+        1
+    }
+}
+
 impl FixedEncodable for SizedRecord {
     type EncodeError = io::Error;
     type DecodeError = io::Error;
@@ -221,6 +227,12 @@ impl ByteSizeOf for UndecodableRecord {
     }
 }
 
+impl EventCount for UndecodableRecord {
+    fn event_count(&self) -> usize {
+        1
+    }
+}
+
 impl FixedEncodable for UndecodableRecord {
     type EncodeError = io::Error;
     type DecodeError = io::Error;

lib/vector-buffers/src/lib.rs

@@ -72,12 +72,23 @@ impl Arbitrary for WhenFull {
 ///
 /// This supertrait serves as the base trait for any item that can be pushed into a buffer.
 pub trait Bufferable:
-    ByteSizeOf + Encodable + Debug + Send + Sync + Unpin + Sized + 'static
+    ByteSizeOf + EventCount + Encodable + Debug + Send + Sync + Unpin + Sized + 'static
 {
 }
 
 // Blanket implementation for anything that is already bufferable.
 impl<T> Bufferable for T where
-    T: ByteSizeOf + Encodable + Debug + Send + Sync + Unpin + Sized + 'static
+    T: ByteSizeOf + EventCount + Encodable + Debug + Send + Sync + Unpin + Sized + 'static
 {
 }
+
+pub trait EventCount {
+    fn event_count(&self) -> usize;
+}
+
+#[cfg(test)]
+impl EventCount for u64 {
+    fn event_count(&self) -> usize {
+        1
+    }
+}

lib/vector-buffers/src/test/common/message.rs

@@ -4,7 +4,7 @@ use bytes::{Buf, BufMut};
 use quickcheck::{Arbitrary, Gen};
 use vector_common::byte_size_of::ByteSizeOf;
 
-use crate::encoding::FixedEncodable;
+use crate::{encoding::FixedEncodable, EventCount};
 
 #[derive(Debug)]
 pub struct EncodeError;
@@ -45,6 +45,12 @@ impl ByteSizeOf for Message {
     }
 }
 
+impl EventCount for Message {
+    fn event_count(&self) -> usize {
+        1
+    }
+}
+
 impl Arbitrary for Message {
     fn arbitrary(g: &mut Gen) -> Self {
         Message {

lib/vector-core/src/event/mod.rs

@@ -10,7 +10,7 @@ use chrono::{DateTime, SecondsFormat, Utc};
 use enumflags2::{bitflags, BitFlags, FromBitsError};
 use prost::Message;
 use snafu::Snafu;
-use vector_buffers::encoding::{AsMetadata, Encodable};
+use vector_buffers::{encoding::AsMetadata, encoding::Encodable, EventCount};
 use vector_common::EventDataEq;
 
 use crate::ByteSizeOf;
@@ -64,6 +64,12 @@ impl ByteSizeOf for Event {
     }
 }
 
+impl EventCount for Event {
+    fn event_count(&self) -> usize {
+        1
+    }
+}
+
 impl Finalizable for Event {
     fn take_finalizers(&mut self) -> EventFinalizers {
         match self {