chore: Dropped error field from StreamClosed Error (vectordotdev#17693)

Field has been unused, so I've removed it and cleaned up the usages and
conditionals where it was included previously. `send_batch()` appears to
only return `ClosedError` making some additional matching pointless.

Signed-off-by: Spencer Gilbert <spencer.gilbert@datadoghq.com>

src/internal_events/common.rs

@@ -83,7 +83,6 @@ const STREAM_CLOSED: &str = "stream_closed";
 
 #[derive(Debug)]
 pub struct StreamClosedError {
-    pub error: crate::source_sender::ClosedError,
     pub count: usize,
 }
 

src/sources/amqp.rs

@@ -392,17 +392,17 @@ async fn finalize_event_stream(
             let mut stream = stream.map(|event| event.with_batch_notifier(&batch));
 
             match out.send_event_stream(&mut stream).await {
-                Err(error) => {
-                    emit!(StreamClosedError { error, count: 1 });
+                Err(_) => {
+                    emit!(StreamClosedError { count: 1 });
                 }
                 Ok(_) => {
                     finalizer.add(msg.into(), receiver);
                 }
             }
         }
         None => match out.send_event_stream(&mut stream).await {
-            Err(error) => {
-                emit!(StreamClosedError { error, count: 1 });
+            Err(_) => {
+                emit!(StreamClosedError { count: 1 });
             }
             Ok(_) => {
                 let ack_options = lapin::options::BasicAckOptions::default();

src/sources/apache_metrics/mod.rs

@@ -273,9 +273,9 @@ fn apache_metrics(
                 debug!("Finished sending.");
                 Ok(())
             }
-            Err(error) => {
+            Err(_) => {
                 let (count, _) = stream.size_hint();
-                emit!(StreamClosedError { error, count });
+                emit!(StreamClosedError { count });
                 Err(())
             }
         }

src/sources/aws_ecs_metrics/mod.rs

@@ -207,8 +207,8 @@ async fn aws_ecs_metrics(
                                     endpoint: uri.path(),
                                 });
 
-                                if let Err(error) = out.send_batch(metrics).await {
-                                    emit!(StreamClosedError { error, count });
+                                if (out.send_batch(metrics).await).is_err() {
+                                    emit!(StreamClosedError { count });
                                     return Err(());
                                 }
                             }

src/sources/aws_kinesis_firehose/handlers.rs

@@ -145,10 +145,7 @@ pub(super) async fn firehose(
 
                     let count = events.len();
                     if let Err(error) = context.out.send_batch(events).await {
-                        emit!(StreamClosedError {
-                            error: error.clone(),
-                            count,
-                        });
+                        emit!(StreamClosedError { count });
                         let error = RequestError::ShuttingDown {
                             request_id: request_id.clone(),
                             source: error,

src/sources/aws_s3/sqs.rs

@@ -574,9 +574,9 @@ impl IngestorProcess {
 
         let send_error = match self.out.send_event_stream(&mut stream).await {
             Ok(_) => None,
-            Err(error) => {
+            Err(_) => {
                 let (count, _) = stream.size_hint();
-                emit!(StreamClosedError { error, count });
+                emit!(StreamClosedError { count });
                 Some(crate::source_sender::ClosedError)
             }
         };

src/sources/aws_sqs/source.rs

@@ -180,7 +180,7 @@ impl SqsSource {
                         }
                     }
                 }
-                Err(error) => emit!(StreamClosedError { error, count }),
+                Err(_) => emit!(StreamClosedError { count }),
             }
         }
     }

src/sources/datadog_agent/mod.rs

@@ -443,8 +443,8 @@ pub(crate) async fn handle_request(
             } else {
                 out.send_batch(events).await
             }
-            .map_err(move |error: crate::source_sender::ClosedError| {
-                emit!(StreamClosedError { error, count });
+            .map_err(|_| {
+                emit!(StreamClosedError { count });
                 warp::reject::custom(ApiError::ServerShutdown)
             })?;
             match receiver {

src/sources/demo_logs.rs

@@ -264,8 +264,8 @@ async fn demo_logs_source(
 
                         event
                     });
-                    out.send_batch(events).await.map_err(|error| {
-                        emit!(StreamClosedError { error, count });
+                    out.send_batch(events).await.map_err(|_| {
+                        emit!(StreamClosedError { count });
                     })?;
                 }
                 Err(error) => {

src/sources/docker_logs/mod.rs

@@ -814,13 +814,10 @@ impl EventStreamBuilder {
         let result = {
             let mut stream = events_stream
                 .map(move |event| add_hostname(event, &host_key, &hostname, self.log_namespace));
-            self.out
-                .send_event_stream(&mut stream)
-                .await
-                .map_err(|error| {
-                    let (count, _) = stream.size_hint();
-                    emit!(StreamClosedError { error, count });
-                })
+            self.out.send_event_stream(&mut stream).await.map_err(|_| {
+                let (count, _) = stream.size_hint();
+                emit!(StreamClosedError { count });
+            })
         };
 
         // End of stream

src/sources/eventstoredb_metrics/mod.rs

@@ -137,8 +137,8 @@ fn eventstoredb(
 
                                 events_received.emit(CountByteSize(count, byte_size));
 
-                                if let Err(error) = cx.out.send_batch(metrics).await {
-                                    emit!(StreamClosedError { count, error });
+                                if (cx.out.send_batch(metrics).await).is_err() {
+                                    emit!(StreamClosedError { count });
                                     break;
                                 }
                             }

src/sources/exec/mod.rs

@@ -498,8 +498,8 @@ async fn run_command(
                         for event in &mut events {
                             handle_event(&config, &hostname, &Some(stream.to_string()), pid, event, log_namespace);
                         }
-                        if let Err(error) = out.send_batch(events).await {
-                            emit!(StreamClosedError { count, error });
+                        if (out.send_batch(events).await).is_err() {
+                            emit!(StreamClosedError { count });
                             break;
                         }
                     },

src/sources/file.rs

@@ -659,9 +659,9 @@ pub fn file_source(
                 Ok(()) => {
                     debug!("Finished sending.");
                 }
-                Err(error) => {
+                Err(_) => {
                     let (count, _) = messages.size_hint();
-                    emit!(StreamClosedError { error, count });
+                    emit!(StreamClosedError { count });
                 }
             }
         });

src/sources/file_descriptors/mod.rs

@@ -195,9 +195,9 @@ async fn process_stream(
             debug!("Finished sending.");
             Ok(())
         }
-        Err(error) => {
+        Err(_) => {
             let (count, _) = stream.size_hint();
-            emit!(StreamClosedError { error, count });
+            emit!(StreamClosedError { count });
             Err(())
         }
     }

src/sources/gcp_pubsub.rs

@@ -618,7 +618,7 @@ impl PubsubSource {
 
         let count = events.len();
         match self.out.send_batch(events).await {
-            Err(error) => emit!(StreamClosedError { error, count }),
+            Err(_) => emit!(StreamClosedError { count }),
             Ok(()) => match notifier {
                 None => ack_ids
                     .send(ids)

src/sources/host_metrics/mod.rs

@@ -296,8 +296,8 @@ impl HostMetricsConfig {
             bytes_received.emit(ByteSize(0));
             let metrics = generator.capture_metrics().await;
             let count = metrics.len();
-            if let Err(error) = out.send_batch(metrics).await {
-                emit!(StreamClosedError { count, error });
+            if (out.send_batch(metrics).await).is_err() {
+                emit!(StreamClosedError { count });
                 return Err(());
             }
         }

src/sources/internal_logs.rs

@@ -191,9 +191,9 @@ async fn run(
             Utc::now(),
         );
 
-        if let Err(error) = out.send_event(Event::from(log)).await {
+        if (out.send_event(Event::from(log)).await).is_err() {
             // this wont trigger any infinite loop considering it stops the component
-            emit!(StreamClosedError { error, count: 1 });
+            emit!(StreamClosedError { count: 1 });
             return Err(());
         }
     }

src/sources/internal_metrics.rs

@@ -190,8 +190,8 @@ impl<'a> InternalMetrics<'a> {
                 metric
             });
 
-            if let Err(error) = self.out.send_batch(batch).await {
-                emit!(StreamClosedError { error, count });
+            if (self.out.send_batch(batch).await).is_err() {
+                emit!(StreamClosedError { count });
                 return Err(());
             }
         }

src/sources/journald.rs

@@ -606,8 +606,8 @@ impl<'a> Batch<'a> {
                         finalizer.finalize(cursor, self.receiver).await;
                     }
                 }
-                Err(error) => {
-                    emit!(StreamClosedError { error, count });
+                Err(_) => {
+                    emit!(StreamClosedError { count });
                     // `out` channel is closed, don't restart journalctl.
                     self.exiting = Some(false);
                 }

src/sources/kafka.rs

@@ -449,8 +449,8 @@ async fn parse_message(
                 let (batch, receiver) = BatchNotifier::new_with_receiver();
                 let mut stream = stream.map(|event| event.with_batch_notifier(&batch));
                 match out.send_event_stream(&mut stream).await {
-                    Err(error) => {
-                        emit!(StreamClosedError { error, count });
+                    Err(_) => {
+                        emit!(StreamClosedError { count });
                     }
                     Ok(_) => {
                         // Drop stream to avoid borrowing `msg`: "[...] borrow might be used
@@ -461,8 +461,8 @@ async fn parse_message(
                 }
             }
             None => match out.send_event_stream(&mut stream).await {
-                Err(error) => {
-                    emit!(StreamClosedError { error, count });
+                Err(_) => {
+                    emit!(StreamClosedError { count });
                 }
                 Ok(_) => {
                     if let Err(error) =

src/sources/kubernetes_logs/mod.rs

@@ -871,8 +871,7 @@ impl Source {
             .map(|result| {
                 match result {
                     Ok(Ok(())) => info!(message = "Event processing loop completed gracefully."),
-                    Ok(Err(error)) => emit!(StreamClosedError {
-                        error,
+                    Ok(Err(_)) => emit!(StreamClosedError {
                         count: events_count
                     }),
                     Err(error) => emit!(KubernetesLifecycleError {

src/sources/mongodb_metrics/mod.rs

@@ -147,8 +147,8 @@ impl SourceConfig for MongoDbMetricsConfig {
 
                 let metrics = metrics.into_iter().flatten();
 
-                if let Err(error) = cx.out.send_batch(metrics).await {
-                    emit!(StreamClosedError { error, count });
+                if (cx.out.send_batch(metrics).await).is_err() {
+                    emit!(StreamClosedError { count });
                     return Err(());
                 }
             }

src/sources/nats.rs

@@ -238,8 +238,8 @@ async fn nats_source(
                         event
                     });
 
-                    out.send_batch(events).await.map_err(|error| {
-                        emit!(StreamClosedError { error, count });
+                    out.send_batch(events).await.map_err(|_| {
+                        emit!(StreamClosedError { count });
                     })?;
                 }
                 Err(error) => {

src/sources/nginx_metrics/mod.rs

@@ -135,8 +135,8 @@ impl SourceConfig for NginxMetricsConfig {
 
                 let metrics = metrics.into_iter().flatten();
 
-                if let Err(error) = cx.out.send_batch(metrics).await {
-                    emit!(StreamClosedError { error, count });
+                if (cx.out.send_batch(metrics).await).is_err() {
+                    emit!(StreamClosedError { count });
                     return Err(());
                 }
             }

src/sources/opentelemetry/grpc.rs

@@ -48,7 +48,7 @@ impl LogsService for Service {
             .send_batch_named(LOGS, events)
             .map_err(|error| {
                 let message = error.to_string();
-                emit!(StreamClosedError { error, count });
+                emit!(StreamClosedError { count });
                 Status::unavailable(message)
             })
             .and_then(|_| handle_batch_status(receiver))

src/sources/opentelemetry/http.rs

@@ -122,12 +122,10 @@ async fn handle_request(
             let receiver = BatchNotifier::maybe_apply_to(acknowledgements, &mut events);
             let count = events.len();
 
-            out.send_batch_named(output, events)
-                .await
-                .map_err(move |error| {
-                    emit!(StreamClosedError { error, count });
-                    warp::reject::custom(ApiError::ServerShutdown)
-                })?;
+            out.send_batch_named(output, events).await.map_err(|_| {
+                emit!(StreamClosedError { count });
+                warp::reject::custom(ApiError::ServerShutdown)
+            })?;
 
             match receiver {
                 None => Ok(protobuf(ExportLogsServiceResponse {}).into_response()),

src/sources/postgresql_metrics.rs

@@ -227,8 +227,8 @@ impl SourceConfig for PostgresqlMetricsConfig {
                 });
 
                 let metrics = metrics.into_iter().flatten();
-                if let Err(error) = cx.out.send_batch(metrics).await {
-                    emit!(StreamClosedError { error, count });
+                if (cx.out.send_batch(metrics).await).is_err() {
+                    emit!(StreamClosedError { count });
                     return Err(());
                 }
             }

src/sources/redis/mod.rs

@@ -279,8 +279,8 @@ impl InputHandler {
                         event
                     });
 
-                    if let Err(error) = self.cx.out.send_batch(events).await {
-                        emit!(StreamClosedError { error, count });
+                    if (self.cx.out.send_batch(events).await).is_err() {
+                        emit!(StreamClosedError { count });
                         return Err(());
                     }
                 }

src/sources/socket/udp.rs

@@ -269,8 +269,8 @@ pub(super) fn udp(
 
                                 tokio::select!{
                                     result = out.send_batch(events) => {
-                                        if let Err(error) = result {
-                                            emit!(StreamClosedError { error, count });
+                                        if result.is_err() {
+                                            emit!(StreamClosedError { count });
                                             return Ok(())
                                         }
                                     }

src/sources/statsd/mod.rs

@@ -295,8 +295,8 @@ async fn statsd_udp(
         match frame {
             Ok(((events, _byte_size), _sock)) => {
                 let count = events.len();
-                if let Err(error) = out.send_batch(events).await {
-                    emit!(StreamClosedError { error, count });
+                if (out.send_batch(events).await).is_err() {
+                    emit!(StreamClosedError { count });
                 }
             }
             Err(error) => {

src/sources/syslog.rs

@@ -364,9 +364,9 @@ pub fn udp(
                 debug!("Finished sending.");
                 Ok(())
             }
-            Err(error) => {
+            Err(_) => {
                 let (count, _) = stream.size_hint();
-                emit!(StreamClosedError { error, count });
+                emit!(StreamClosedError { count });
                 Err(())
             }
         }

src/sources/util/http/prelude.rs

@@ -213,10 +213,10 @@ async fn handle_request(
 
             let count = events.len();
             out.send_batch(events)
-                .map_err(move |error: crate::source_sender::ClosedError| {
+                .map_err(|_| {
                     // can only fail if receiving end disconnected, so we are shutting down,
                     // probably not gracefully.
-                    emit!(StreamClosedError { error, count });
+                    emit!(StreamClosedError { count });
                     warp::reject::custom(RejectShuttingDown)
                 })
                 .and_then(|_| handle_batch_status(receiver))

src/sources/util/http_client.rs

@@ -233,9 +233,9 @@ pub(crate) async fn call<
             debug!("Finished sending.");
             Ok(())
         }
-        Err(error) => {
+        Err(_) => {
             let (count, _) = stream.size_hint();
-            emit!(StreamClosedError { error, count });
+            emit!(StreamClosedError { count });
             Err(())
         }
     }