feat(component validation): validate component_errors_total for sou…

…rces (vectordotdev#17965)

closes vectordotdev#16841

This one already had some infrastructure in place with the
`TestEvent::Modified` variant. Requires
vectordotdev#17956 .

src/components/validation/mod.rs

@@ -181,6 +181,7 @@ pub struct RunnerMetrics {
     pub sent_bytes_total: u64, // a reciprocal for received_bytes_total
     pub sent_event_bytes_total: u64,
     pub sent_events_total: u64,
+    pub errors_total: u64,
 }
 
 #[cfg(all(test, feature = "component-validation-tests"))]

src/components/validation/resources/http.rs

@@ -107,8 +107,10 @@ fn spawn_input_http_server(
 
                     buffer.into_response()
                 } else {
-                    // No outstanding events to send, so just provide an empty response.
-                    StatusCode::NO_CONTENT.into_response()
+                    // We'll send an empty 200 in the response since some
+                    // sources throw errors for anything other than a valid
+                    // response.
+                    StatusCode::OK.into_response()
                 }
             }
         });

src/components/validation/runner/mod.rs

@@ -282,7 +282,7 @@ impl Runner {
             // send all inputs until we have no more, when we can't send more because we need to
             // drive output collection to allow forward progress to be made, etc.)
 
-            // We sleep for one second here because while we do wait for the component topology to
+            // We sleep for two seconds here because while we do wait for the component topology to
             // mark itself as started, starting the topology does not necessarily mean that all
             // component tasks are actually ready for input, etc.
             //
@@ -542,7 +542,9 @@ fn spawn_input_driver(
             };
 
             // account for failure case
-            if !modified {
+            if modified {
+                input_runner_metrics.errors_total += 1;
+            } else {
                 input_runner_metrics.sent_events_total += 1;
 
                 input_runner_metrics.sent_event_bytes_total +=

src/components/validation/validators/component_spec/mod.rs

@@ -50,7 +50,7 @@ impl Validator for ComponentSpecValidator {
             TestCaseExpectation::Success => {
                 if inputs.len() != outputs.len() {
                     return Err(vec![format!(
-                        "Sent {} inputs but only received {} outputs.",
+                        "Sent {} inputs but received {} outputs.",
                         inputs.len(),
                         outputs.len()
                     )]);

src/components/validation/validators/component_spec/sources.rs

@@ -14,6 +14,7 @@ pub enum SourceMetricType {
     ReceivedBytesTotal,
     SentEventsTotal,
     SentEventBytesTotal,
+    ErrorsTotal,
 }
 
 impl SourceMetricType {
@@ -24,6 +25,7 @@ impl SourceMetricType {
             SourceMetricType::ReceivedBytesTotal => "component_received_bytes_total",
             SourceMetricType::SentEventsTotal => "component_sent_events_total",
             SourceMetricType::SentEventBytesTotal => "component_sent_event_bytes_total",
+            SourceMetricType::ErrorsTotal => "component_errors_total",
         }
     }
 }
@@ -47,6 +49,7 @@ pub fn validate_sources(
         validate_component_received_bytes_total,
         validate_component_sent_events_total,
         validate_component_sent_event_bytes_total,
+        validate_component_errors_total,
     ];
 
     for v in validations.iter() {
@@ -226,3 +229,14 @@ fn validate_component_sent_event_bytes_total(
         expected_bytes,
     )
 }
+
+fn validate_component_errors_total(
+    telemetry_events: &[Event],
+    runner_metrics: &RunnerMetrics,
+) -> Result<Vec<String>, Vec<String>> {
+    validate_events_total(
+        telemetry_events,
+        &SourceMetricType::ErrorsTotal,
+        runner_metrics.errors_total,
+    )
+}