chore(observability): robustly synchronize component validation framework tasks

Cargo.toml

@@ -896,7 +896,9 @@ enterprise-tests = [
 ]
 
 component-validation-runner = ["dep:tonic", "sources-internal_logs", "sources-internal_metrics", "sources-vector", "sinks-vector"]
-component-validation-tests = ["component-validation-runner", "sources", "transforms", "sinks"]
+# For now, only include components that implement ValidatableComponent.
+# In the future, this can change to simply reference the targets `sources`, `transforms`, `sinks`
+component-validation-tests = ["component-validation-runner", "sources-http_client", "sources-http_server", "sinks-http"]
 
 # Grouping together features for benchmarks. We exclude the API client due to it causing the build process to run out
 # of memory when those additional dependencies are built in CI.

src/components/validation/resources/event.rs

@@ -1,6 +1,5 @@
 use bytes::BytesMut;
 use serde::Deserialize;
-use serde_json::Value;
 use snafu::Snafu;
 use tokio_util::codec::Encoder as _;
 
@@ -30,16 +29,9 @@ pub enum RawTestEvent {
     /// is malformed in some way, which can be achieved without this test event variant.
     AlternateEncoder { fail_encoding_of: EventData },
 
-    /// The event is created, and the specified field is added to it.
-    ///
-    /// This allows the ability to hit code paths where some codecs require specific fields to be of specific
-    /// types, thus allowing us to encode into the input runner without error, but encoding in the component
-    /// under test can be set up to fail.
-    WithField {
-        event: EventData,
-        name: String,
-        value: Value,
-        fail: Option<bool>,
+    /// The event will be rejected by the external resource.
+    ResourceReject {
+        external_resource_rejects: EventData,
     },
 }
 
@@ -79,12 +71,9 @@ pub enum TestEvent {
     /// configured encoding, which should cause an error when the event is decoded.
     FailWithAlternateEncoder(Event),
 
-    /// The event has an additional field injected prior to encoding, which should cause
-    /// an error when the event is decoded.
-    ///
-    /// This is useful for testing encodings that have strict schemas and cannot
-    /// handle arbitrary fields or differing data types for certain fields.
-    FailWithInjectedField(Event),
+    /// The event encodes successfully but when the external resource receives that event, it should
+    /// throw a failure.
+    FailWithExternalResource(Event),
 }
 
 impl TestEvent {
@@ -93,15 +82,15 @@ impl TestEvent {
         match self {
             Self::Passthrough(event) => event,
             Self::FailWithAlternateEncoder(event) => event,
-            Self::FailWithInjectedField(event) => event,
+            Self::FailWithExternalResource(event) => event,
         }
     }
 
     pub fn get_event(&mut self) -> &mut Event {
         match self {
             Self::Passthrough(event) => event,
             Self::FailWithAlternateEncoder(event) => event,
-            Self::FailWithInjectedField(event) => event,
+            Self::FailWithExternalResource(event) => event,
         }
     }
 
@@ -110,7 +99,24 @@ impl TestEvent {
         match self {
             Self::Passthrough(event) => (false, event),
             Self::FailWithAlternateEncoder(event) => (true, event),
-            Self::FailWithInjectedField(event) => (true, event),
+            Self::FailWithExternalResource(event) => (true, event),
+        }
+    }
+
+    /// True if the event should fail, false otherwise.
+    pub const fn should_fail(&self) -> bool {
+        match self {
+            Self::Passthrough(_) => false,
+            Self::FailWithAlternateEncoder(_) | Self::FailWithExternalResource(_) => true,
+        }
+    }
+
+    /// True if the event should be rejected by the external resource in order to
+    /// trigger a failure path.
+    pub const fn should_reject(&self) -> bool {
+        match self {
+            Self::Passthrough(_) | Self::FailWithAlternateEncoder(_) => false,
+            Self::FailWithExternalResource(_) => true,
         }
     }
 }
@@ -127,22 +133,9 @@ impl From<RawTestEvent> for TestEvent {
             RawTestEvent::AlternateEncoder {
                 fail_encoding_of: event_data,
             } => TestEvent::FailWithAlternateEncoder(event_data.into_event()),
-            RawTestEvent::WithField {
-                event,
-                name,
-                value,
-                fail,
-            } => {
-                let mut event = event.into_event();
-                let log_event = event.as_mut_log();
-                log_event.insert(name.as_str(), value);
-
-                if fail.unwrap_or_default() {
-                    TestEvent::FailWithInjectedField(event)
-                } else {
-                    TestEvent::Passthrough(event)
-                }
-            }
+            RawTestEvent::ResourceReject {
+                external_resource_rejects: event_data,
+            } => TestEvent::FailWithExternalResource(event_data.into_event()),
         }
     }
 }
@@ -153,7 +146,7 @@ pub fn encode_test_event(
     event: TestEvent,
 ) {
     match event {
-        TestEvent::Passthrough(event) | TestEvent::FailWithInjectedField(event) => {
+        TestEvent::Passthrough(event) | TestEvent::FailWithExternalResource(event) => {
             // Encode the event normally.
             encoder
                 .encode(event, buf)

src/components/validation/resources/http.rs

@@ -66,6 +66,7 @@ impl HttpResourceConfig {
         codec: ResourceCodec,
         output_tx: mpsc::Sender<Vec<Event>>,
         task_coordinator: &TaskCoordinator<Configuring>,
+        input_events: Vec<TestEvent>,
         runner_metrics: &Arc<Mutex<RunnerMetrics>>,
     ) -> vector_lib::Result<()> {
         match direction {
@@ -77,9 +78,14 @@ impl HttpResourceConfig {
                 task_coordinator,
             )),
             // The sink will push data to us.
-            ResourceDirection::Push => {
-                spawn_output_http_server(self, codec, output_tx, task_coordinator, runner_metrics)
-            }
+            ResourceDirection::Push => spawn_output_http_server(
+                self,
+                codec,
+                output_tx,
+                task_coordinator,
+                input_events,
+                runner_metrics,
+            ),
         }
     }
 }
@@ -132,10 +138,11 @@ fn spawn_input_http_server(
     // events and working with the HTTP server as they're consumed.
     let resource_started = task_coordinator.track_started();
     let resource_completed = task_coordinator.track_completed();
+    let mut resource_shutdown_rx = task_coordinator.register_for_shutdown();
 
     tokio::spawn(async move {
         resource_started.mark_as_done();
-        debug!("HTTP server external input resource started.");
+        info!("HTTP server external input resource started.");
 
         let mut input_finished = false;
 
@@ -152,7 +159,7 @@ fn spawn_input_http_server(
                         outstanding_events.push_back(event);
                     },
                     None => {
-                        trace!("HTTP server external input resource input is finished.");
+                        info!("HTTP server external input resource input is finished.");
                         input_finished = true;
                     },
                 },
@@ -175,16 +182,18 @@ fn spawn_input_http_server(
         }
         // Mark ourselves as completed now that we've sent all inputs to the source, and
         // additionally signal the HTTP server to also gracefully shutdown.
-        _ = http_server_shutdown_tx.send(());
+        info!("HTTP server external input resource signalling ready for shutdown.");
 
-        // TODO - currently we are getting lucky in the testing of `http_client` source... if the source tries to query
-        // this server but we have already shut down the thread, then it will generate an error which can throw off our error
-        // validation.
-        // I think the solution involves adding synchronization to wait here for the runner to tell us to shutdown.
+        // Wait for the runner to signal us to shutdown
+        resource_shutdown_rx.wait().await;
+
+        // Shutdown the server
+        _ = http_server_shutdown_tx.send(());
 
+        info!("HTTP server external input resource marking as done.");
         resource_completed.mark_as_done();
 
-        debug!("HTTP server external input resource completed.");
+        info!("HTTP server external input resource completed.");
     });
 }
 
@@ -205,7 +214,7 @@ fn spawn_input_http_client(
         // Mark ourselves as started. We don't actually do anything until we get our first input
         // message, though.
         started.mark_as_done();
-        debug!("HTTP client external input resource started.");
+        info!("HTTP client external input resource started.");
 
         let client = Client::builder().build_http::<Body>();
         let request_uri = config.uri;
@@ -238,7 +247,7 @@ fn spawn_input_http_client(
         // Mark ourselves as completed now that we've sent all inputs to the source.
         completed.mark_as_done();
 
-        debug!("HTTP client external input resource completed.");
+        info!("HTTP client external input resource completed.");
     });
 }
 
@@ -249,6 +258,7 @@ fn spawn_output_http_server(
     codec: ResourceCodec,
     output_tx: mpsc::Sender<Vec<Event>>,
     task_coordinator: &TaskCoordinator<Configuring>,
+    input_events: Vec<TestEvent>,
     runner_metrics: &Arc<Mutex<RunnerMetrics>>,
 ) -> vector_lib::Result<()> {
     // This HTTP server will wait for events to be sent by a sink, and collect them and send them on
@@ -257,13 +267,26 @@ fn spawn_output_http_server(
     // First, we'll build and spawn our HTTP server.
     let decoder = codec.into_decoder()?;
 
+    // the number of events we expect to receive from the sink.
+    // any happy path events + any events that are flagged as external resource should reject.
+    let expected_output_events = input_events
+        .iter()
+        .filter(|te| !te.should_fail() || te.should_reject())
+        .count();
+
+    let should_reject = input_events.iter().filter(|te| te.should_reject()).count() > 0;
+
+    let (server_sent_all_tx, mut server_sent_all_rx) = mpsc::channel(1);
+
     let (_, http_server_shutdown_tx) = spawn_http_server(
         task_coordinator,
         &config,
         runner_metrics,
         move |request, output_runner_metrics| {
             let output_tx = output_tx.clone();
             let mut decoder = decoder.clone();
+            let mut decoded_events = 0;
+            let server_sent_all_tx = server_sent_all_tx.clone();
 
             async move {
                 match hyper::body::to_bytes(request.into_body()).await {
@@ -273,28 +296,46 @@ fn spawn_output_http_server(
                         loop {
                             match decoder.decode_eof(&mut body) {
                                 Ok(Some((events, byte_size))) => {
-                                    let mut output_runner_metrics =
-                                        output_runner_metrics.lock().await;
-                                    debug!("HTTP server external output resource decoded {byte_size} bytes.");
-
-                                    // Update the runner metrics for the received events. This will later
-                                    // be used in the Validators, as the "expected" case.
-                                    output_runner_metrics.received_bytes_total += byte_size as u64;
-
-                                    output_runner_metrics.received_events_total +=
-                                        events.len() as u64;
-
-                                    events.iter().for_each(|event| {
-                                        output_runner_metrics.received_event_bytes_total +=
-                                            event.estimated_json_encoded_size_of().get() as u64;
-                                    });
-
-                                    output_tx
-                                        .send(events.to_vec())
-                                        .await
-                                        .expect("should not fail to send output event");
+                                    if should_reject {
+                                        info!("HTTP server external output resource decoded {byte_size} bytes but test case configured to reject.");
+                                    } else {
+                                        let mut output_runner_metrics =
+                                            output_runner_metrics.lock().await;
+                                        info!("HTTP server external output resource decoded {byte_size} bytes.");
+
+                                        // Update the runner metrics for the received events. This will later
+                                        // be used in the Validators, as the "expected" case.
+                                        output_runner_metrics.received_bytes_total +=
+                                            byte_size as u64;
+
+                                        output_runner_metrics.received_events_total +=
+                                            events.len() as u64;
+
+                                        events.iter().for_each(|event| {
+                                            output_runner_metrics.received_event_bytes_total +=
+                                                event.estimated_json_encoded_size_of().get() as u64;
+                                        });
+
+                                        output_tx
+                                            .send(events.to_vec())
+                                            .await
+                                            .expect("should not fail to send output event");
+                                    }
+                                    decoded_events += events.len();
+                                }
+                                Ok(None) => {
+                                    if decoded_events == expected_output_events {
+                                        // Signal the outer thread that all expected events have been processed
+                                        let _ = server_sent_all_tx.send(1).await;
+                                    }
+                                    if should_reject {
+                                        // This status code is not retried and should result in the component under test
+                                        // emitting error events
+                                        return StatusCode::BAD_REQUEST.into_response();
+                                    } else {
+                                        return StatusCode::OK.into_response();
+                                    }
                                 }
-                                Ok(None) => return StatusCode::OK.into_response(),
                                 Err(_) => return StatusCode::INTERNAL_SERVER_ERROR.into_response(),
                             }
                         }
@@ -312,14 +353,25 @@ fn spawn_output_http_server(
 
     tokio::spawn(async move {
         resource_started.mark_as_done();
-        debug!("HTTP server external output resource started.");
+        info!("HTTP server external output resource started.");
 
+        // Wait for the runner to tell us to shutdown
         resource_shutdown_rx.wait().await;
-        _ = http_server_shutdown_tx.send(());
+
+        // ensure we've processed all events we expect to receive from the sink, if any.
+        if expected_output_events > 0 {
+            server_sent_all_rx.recv().await;
+        }
+
+        // signal the server to shutdown
+        let _ = http_server_shutdown_tx.send(());
+
+        // mark ourselves as done
         resource_completed.mark_as_done();
 
-        debug!("HTTP server external output resource completed.");
+        info!("HTTP server external output resource completed.");
     });
+
     Ok(())
 }
 

src/components/validation/resources/mod.rs

@@ -340,6 +340,7 @@ impl ExternalResource {
         self,
         output_tx: mpsc::Sender<Vec<Event>>,
         task_coordinator: &TaskCoordinator<Configuring>,
+        input_events: Vec<TestEvent>,
         runner_metrics: &Arc<Mutex<RunnerMetrics>>,
     ) -> vector_lib::Result<()> {
         match self.definition {
@@ -348,6 +349,7 @@ impl ExternalResource {
                 self.codec,
                 output_tx,
                 task_coordinator,
+                input_events,
                 runner_metrics,
             ),
         }