feat(component validation): add sink validator

src/components/validation/mod.rs

@@ -14,6 +14,14 @@ pub use self::runner::*;
 pub use self::test_case::{TestCase, TestCaseExpectation};
 pub use self::validators::*;
 
+pub mod component_names {
+    pub const TEST_SOURCE_NAME: &str = "test_source";
+    pub const TEST_SINK_NAME: &str = "test_sink";
+    pub const TEST_TRANSFORM_NAME: &str = "test_transform";
+    pub const TEST_INPUT_SOURCE_NAME: &str = "input_source";
+    pub const TEST_OUTPUT_SINK_NAME: &str = "output_sink";
+}
+
 /// Component types that can be validated.
 // TODO: We should centralize this in `vector-common` or something, where both this code and the
 // configuration schema stuff (namely the proc macros that use this) can share it.
@@ -170,6 +178,19 @@ macro_rules! register_validatable_component {
     };
 }
 
+/// Input and Output runners populate this structure as they send and receive events.
+/// The structure is passed into the validator to use as the expected values for the
+/// metrics that the components under test actually output.
+#[derive(Default, Debug)]
+pub struct RunnerMetrics {
+    pub received_events_total: u64,
+    pub received_event_bytes_total: u64,
+    pub received_bytes_total: u64,
+    pub sent_bytes_total: u64,
+    pub sent_event_bytes_total: u64,
+    pub sent_events_total: u64,
+}
+
 #[cfg(all(test, feature = "component-validation-tests"))]
 mod tests {
     use std::{

src/components/validation/resources/event.rs

@@ -1,10 +1,20 @@
+use bytes::BytesMut;
 use serde::Deserialize;
+use snafu::Snafu;
+use tokio_util::codec::Encoder as _;
+
+use crate::codecs::Encoder;
+use codecs::{
+    encoding, JsonSerializer, LengthDelimitedEncoder, LogfmtSerializer, MetricTagValues,
+    NewlineDelimitedEncoder,
+};
 use vector_core::event::{Event, LogEvent};
 
-/// An event used in a test case.
+/// A test case event for deserialization from yaml file.
+/// This is an intermediary step to TestEvent.
 #[derive(Clone, Debug, Deserialize)]
 #[serde(untagged)]
-pub enum TestEvent {
+pub enum RawTestEvent {
     /// The event is used, as-is, without modification.
     Passthrough(EventData),
 
@@ -20,15 +30,6 @@ pub enum TestEvent {
     Modified { modified: bool, event: EventData },
 }
 
-impl TestEvent {
-    pub fn into_event(self) -> Event {
-        match self {
-            Self::Passthrough(event) => event.into_event(),
-            Self::Modified { event, .. } => event.into_event(),
-        }
-    }
-}
-
 #[derive(Clone, Debug, Deserialize)]
 #[serde(untagged)]
 pub enum EventData {
@@ -44,3 +45,93 @@ impl EventData {
         }
     }
 }
+
+/// An event used in a test case.
+/// It is important to have created the event with all fields, immediately after deserializing from the
+/// test case definition yaml file. This ensures that the event data we are using in the expected/actual
+/// metrics collection is based on the same event. Namely, one issue that can arise from creating the event
+/// from the event data twice (once for the expected and once for actual), it can result in a timestamp in
+/// the event which may or may not have the same millisecond precision as it's counterpart.
+#[derive(Clone, Debug, Deserialize)]
+#[serde(try_from = "RawTestEvent")]
+#[serde(untagged)]
+pub enum TestEvent {
+    /// The event is used, as-is, without modification.
+    Passthrough(Event),
+
+    /// The event is potentially modified by the external resource.
+    ///
+    /// The modification made is dependent on the external resource, but this mode is made available
+    /// for when a test case wants to exercise the failure path, but cannot cause a failure simply
+    /// by constructing the event in a certain way i.e. adding an invalid field, or removing a
+    /// required field, or using an invalid field value, and so on.
+    ///
+    /// For transforms and sinks, generally, the only way to cause an error is if the event itself
+    /// is malformed in some way, which can be achieved without this test event variant.
+    Modified { modified: bool, event: Event },
+}
+
+impl TestEvent {
+    #[allow(clippy::missing_const_for_fn)] // const cannot run destructor
+    pub fn into_event(self) -> Event {
+        match self {
+            Self::Passthrough(event) => event,
+            Self::Modified { event, .. } => event,
+        }
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq, Snafu)]
+pub enum RawTestEventParseError {}
+
+impl TryFrom<RawTestEvent> for TestEvent {
+    type Error = RawTestEventParseError;
+
+    fn try_from(other: RawTestEvent) -> Result<Self, Self::Error> {
+        Ok(match other {
+            RawTestEvent::Passthrough(event_data) => {
+                TestEvent::Passthrough(event_data.into_event())
+            }
+            RawTestEvent::Modified { modified, event } => TestEvent::Modified {
+                modified,
+                event: event.into_event(),
+            },
+        })
+    }
+}
+
+pub fn encode_test_event(
+    encoder: &mut Encoder<encoding::Framer>,
+    buf: &mut BytesMut,
+    event: TestEvent,
+) {
+    match event {
+        TestEvent::Passthrough(event) => {
+            // Encode the event normally.
+            encoder
+                .encode(event, buf)
+                .expect("should not fail to encode input event");
+        }
+        TestEvent::Modified { event, .. } => {
+            // This is a little fragile, but we check what serializer this encoder uses, and based
+            // on `Serializer::supports_json`, we choose an opposing codec. For example, if the
+            // encoder supports JSON, we'll use a serializer that doesn't support JSON, and vise
+            // versa.
+            let mut alt_encoder = if encoder.serializer().supports_json() {
+                Encoder::<encoding::Framer>::new(
+                    LengthDelimitedEncoder::new().into(),
+                    LogfmtSerializer::new().into(),
+                )
+            } else {
+                Encoder::<encoding::Framer>::new(
+                    NewlineDelimitedEncoder::new().into(),
+                    JsonSerializer::new(MetricTagValues::default()).into(),
+                )
+            };
+
+            alt_encoder
+                .encode(event, buf)
+                .expect("should not fail to encode input event");
+        }
+    }
+}