fix(topology): preserve component failure details

changelog.d/socket_bind_error_context.fix.md

@@ -0,0 +1,3 @@
+Vector now reports the underlying error when a source or sink task fails at runtime; previously, it only said `the task completed with an error`. That includes failures to bind a listening socket (for example `Address already in use (os error 48)`), broker subscribe and consume errors, checkpoint file errors, sink runtime errors, and sources that stop because their downstream channel closed. Vector now includes the panic payload in its logs.
+
+authors: gwenaskell

lib/vector-core/src/sink.rs

@@ -1,4 +1,4 @@
-use std::{fmt, iter::IntoIterator, pin::Pin};
+use std::{fmt, iter::IntoIterator, ops::Deref, pin::Pin};
 
 use futures::{
     Sink, SinkExt, Stream, StreamExt, stream,
@@ -7,20 +7,71 @@ use futures::{
 
 use crate::event::{Event, EventArray, EventContainer, into_event_stream};
 
+/// Error returned by a running sink.
+#[repr(transparent)]
+#[derive(Debug)]
+pub struct SinkError(vector_common::Error);
+
+impl SinkError {
+    pub fn new(error: impl Into<vector_common::Error>) -> Self {
+        Self(error.into())
+    }
+
+    pub fn into_inner(self) -> vector_common::Error {
+        self.0
+    }
+}
+
+impl Default for SinkError {
+    fn default() -> Self {
+        Self("the sink task completed with an error".into())
+    }
+}
+
+impl From<vector_common::Error> for SinkError {
+    fn from(error: vector_common::Error) -> Self {
+        Self(error)
+    }
+}
+
+impl fmt::Display for SinkError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
+impl std::error::Error for SinkError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        Some(self.0.as_ref())
+    }
+}
+
+impl Deref for SinkError {
+    type Target = vector_common::Error;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
 pub enum VectorSink {
     Sink(Box<dyn Sink<EventArray, Error = ()> + Send + Unpin>),
-    Stream(Box<dyn StreamSink<EventArray> + Send>),
+    Stream(Box<dyn StreamSink<EventArray, SinkError> + Send>),
 }
 
 impl VectorSink {
     /// Run the `VectorSink`
     ///
     /// # Errors
     ///
-    /// It is unclear under what conditions this function will error.
-    pub async fn run(self, input: impl Stream<Item = EventArray> + Send) -> Result<(), ()> {
+    /// Returns an error if the sink reports a failure.
+    pub async fn run(self, input: impl Stream<Item = EventArray> + Send) -> Result<(), SinkError> {
         match self {
-            Self::Sink(sink) => input.map(Ok).forward(sink).await,
+            Self::Sink(sink) => input
+                .map(Ok)
+                .forward(sink)
+                .await
+                .map_err(|()| SinkError::default()),
             Self::Stream(s) => s.run(Box::pin(input)).await,
         }
     }
@@ -30,7 +81,7 @@ impl VectorSink {
     /// # Errors
     ///
     /// See `VectorSink::run` for errors.
-    pub async fn run_events<I>(self, input: I) -> Result<(), ()>
+    pub async fn run_events<I>(self, input: I) -> Result<(), SinkError>
     where
         I: IntoIterator<Item = Event> + Send,
         I::IntoIter: Send,
@@ -55,7 +106,7 @@ impl VectorSink {
     /// # Panics
     ///
     /// This function will panic if the self instance is not `VectorSink::Stream`.
-    pub fn into_stream(self) -> Box<dyn StreamSink<EventArray> + Send> {
+    pub fn into_stream(self) -> Box<dyn StreamSink<EventArray, SinkError> + Send> {
         match self {
             Self::Stream(stream) => stream,
             _ => panic!("Failed type coercion, {self:?} is not a Stream"),
@@ -75,6 +126,20 @@ impl VectorSink {
 
     /// Converts an event stream into a `VectorSink`
     pub fn from_event_streamsink(sink: impl StreamSink<Event> + Send + 'static) -> Self {
+        Self::from_streamsink(EventStream {
+            sink: Box::new(sink),
+        })
+    }
+
+    /// Converts a stream sink with opaque runtime errors into a `VectorSink`.
+    pub fn from_streamsink(sink: impl StreamSink<EventArray> + Send + 'static) -> Self {
+        Self::Stream(Box::new(OpaqueStream(Box::new(sink))))
+    }
+
+    /// Converts a sink that retains its runtime errors into a `VectorSink`.
+    pub fn from_detailed_event_streamsink(
+        sink: impl StreamSink<Event, SinkError> + Send + 'static,
+    ) -> Self {
         let sink = Box::new(sink);
         VectorSink::Stream(Box::new(EventStream { sink }))
     }
@@ -89,8 +154,23 @@ impl fmt::Debug for VectorSink {
 // === StreamSink ===
 
 #[async_trait::async_trait]
-pub trait StreamSink<T> {
-    async fn run(self: Box<Self>, input: stream::BoxStream<'_, T>) -> Result<(), ()>;
+pub trait StreamSink<T, E = ()> {
+    async fn run(self: Box<Self>, input: stream::BoxStream<'_, T>) -> Result<(), E>;
+}
+
+struct OpaqueStream<T>(Box<T>);
+
+#[async_trait::async_trait]
+impl<T> StreamSink<EventArray, SinkError> for OpaqueStream<T>
+where
+    T: StreamSink<EventArray> + Send,
+{
+    async fn run(
+        self: Box<Self>,
+        input: stream::BoxStream<'_, EventArray>,
+    ) -> Result<(), SinkError> {
+        self.0.run(input).await.map_err(|()| SinkError::default())
+    }
 }
 
 /// Wrapper for sinks implementing `Sink<Event>` to implement
@@ -175,8 +255,12 @@ struct EventStream<T> {
 }
 
 #[async_trait::async_trait]
-impl<T: StreamSink<Event> + Send> StreamSink<EventArray> for EventStream<T> {
-    async fn run(self: Box<Self>, input: stream::BoxStream<'_, EventArray>) -> Result<(), ()> {
+impl<T, E> StreamSink<EventArray, E> for EventStream<T>
+where
+    T: StreamSink<Event, E> + Send,
+    E: Send,
+{
+    async fn run(self: Box<Self>, input: stream::BoxStream<'_, EventArray>) -> Result<(), E> {
         let input = Box::pin(input.flat_map(into_event_stream));
         self.sink.run(input).await
     }

lib/vector-core/src/source.rs

@@ -1,3 +1,30 @@
-use futures::future::BoxFuture;
+use std::future::Future;
 
-pub type Source = BoxFuture<'static, Result<(), ()>>;
+use futures::{TryFutureExt, future::BoxFuture};
+
+/// Error returned by a running source.
+pub type SourceError = vector_common::Error;
+
+/// Future that runs a source until shutdown or failure.
+pub type Source = BoxFuture<'static, Result<(), SourceError>>;
+
+fn default_source_error() -> SourceError {
+    "the source task completed with an error".into()
+}
+
+/// Adapt a source that only reports whether it failed.
+pub fn opaque_source<F>(future: F) -> Source
+where
+    F: Future<Output = Result<(), ()>> + Send + 'static,
+{
+    Box::pin(future.map_err(|()| default_source_error()))
+}
+
+/// Adapt a source that returns an error with a cause.
+pub fn source_with_error<F, E>(future: F) -> Source
+where
+    F: Future<Output = Result<(), E>> + Send + 'static,
+    E: Into<SourceError>,
+{
+    Box::pin(future.err_into())
+}

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

@@ -14,6 +14,8 @@ mod tests;
 
 pub use builder::Builder;
 pub use errors::SendError;
+/// Result of sending a source's events downstream.
+pub type SendResult = Result<(), SendError>;
 use output::{Output, OutputMetrics};
 pub use sender::{SourceSender, SourceSenderItem};
 

src/config/unit_test/unit_test_components.rs

@@ -41,7 +41,7 @@ impl SourceConfig for UnitTestSourceConfig {
         Ok(Box::pin(async move {
             let mut out = cx.out;
             let _shutdown = cx.shutdown;
-            out.send_batch(events).await.map_err(|_| ())?;
+            out.send_batch(events).await?;
             Ok(())
         }))
     }
@@ -98,7 +98,7 @@ impl SourceConfig for UnitTestStreamSourceConfig {
         Ok(Box::pin(async move {
             let mut out = cx.out;
             let _shutdown = cx.shutdown;
-            out.send_event_stream(stream).await.map_err(|_| ())?;
+            out.send_event_stream(stream).await?;
             Ok(())
         }))
     }

src/enrichment_tables/memory/config.rs

@@ -375,7 +375,7 @@ impl SourceConfig for MemoryConfig {
 
         let log_namespace = cx.log_namespace(self.log_namespace);
 
-        Ok(Box::pin(
+        Ok(crate::sources::opaque_source(
             memory.as_source(cx.shutdown, cx.out, log_namespace).run(),
         ))
     }

src/internal_events/amqp.rs

@@ -27,11 +27,11 @@ pub mod source {
     }
 
     #[derive(Debug, NamedInternalEvent)]
-    pub struct AmqpEventError {
-        pub error: lapin::Error,
+    pub struct AmqpEventError<'a> {
+        pub error: &'a lapin::Error,
     }
 
-    impl InternalEvent for AmqpEventError {
+    impl InternalEvent for AmqpEventError<'_> {
         fn emit(self) {
             error!(message = "Failed to read message.",
                    error = ?self.error,

src/internal_events/journald.rs

@@ -98,12 +98,12 @@ impl InternalEvent for JournaldCheckpointSetError {
 }
 
 #[derive(Debug, NamedInternalEvent)]
-pub struct JournaldCheckpointFileOpenError {
-    pub error: std::io::Error,
+pub struct JournaldCheckpointFileOpenError<'a> {
+    pub error: &'a std::io::Error,
     pub path: String,
 }
 
-impl InternalEvent for JournaldCheckpointFileOpenError {
+impl InternalEvent for JournaldCheckpointFileOpenError<'_> {
     fn emit(self) {
         error!(
             message = "Unable to open checkpoint file.",

src/sinks/blackhole/config.rs

@@ -81,7 +81,7 @@ impl ValidatedSink for BlackholeConfig {
         let sink = BlackholeSink::new(self.clone());
         let healthcheck = future::ok(()).boxed();
 
-        Ok((VectorSink::Stream(Box::new(sink)), healthcheck))
+        Ok((VectorSink::from_streamsink(sink), healthcheck))
     }
 }
 

src/sinks/blackhole/mod.rs

@@ -30,7 +30,7 @@ mod tests {
             acknowledgements: AcknowledgementsConfig::default(),
         };
         let sink = BlackholeSink::new(config);
-        let sink = VectorSink::Stream(Box::new(sink));
+        let sink = VectorSink::from_streamsink(sink);
 
         let (_input_lines, events) = random_events_with_stream(100, 10, None);
         run_and_assert_sink_compliance(sink, events, &SINK_TAGS).await;

src/sinks/console/config.rs

@@ -102,12 +102,12 @@ impl ValidatedSink for ConsoleSinkConfig {
         let encoder = Encoder::<Framer>::new(framer, serializer);
 
         let sink: VectorSink = match self.target {
-            Target::Stdout => VectorSink::from_event_streamsink(WriterSink {
+            Target::Stdout => VectorSink::from_detailed_event_streamsink(WriterSink {
                 output: io::stdout(),
                 transformer,
                 encoder,
             }),
-            Target::Stderr => VectorSink::from_event_streamsink(WriterSink {
+            Target::Stderr => VectorSink::from_detailed_event_streamsink(WriterSink {
                 output: io::stderr(),
                 transformer,
                 encoder,

src/sinks/console/sink.rs

@@ -9,6 +9,7 @@ use vector_lib::{
     internal_event::{
         ByteSize, BytesSent, CountByteSize, EventsSent, InternalEventHandle as _, Output, Protocol,
     },
+    sink::SinkError,
 };
 
 use crate::{
@@ -24,11 +25,11 @@ pub struct WriterSink<T> {
 }
 
 #[async_trait]
-impl<T> StreamSink<Event> for WriterSink<T>
+impl<T> StreamSink<Event, SinkError> for WriterSink<T>
 where
     T: io::AsyncWrite + Send + Sync + Unpin,
 {
-    async fn run(mut self: Box<Self>, mut input: BoxStream<'_, Event>) -> Result<(), ()> {
+    async fn run(mut self: Box<Self>, mut input: BoxStream<'_, Event>) -> Result<(), SinkError> {
         let bytes_sent = register!(BytesSent::from(Protocol("console".into(),)));
         let events_sent = register!(EventsSent::from(Output(None)));
         while let Some(mut event) = input.next().await {
@@ -37,17 +38,17 @@ where
 
             let finalizers = event.take_finalizers();
             let mut bytes = BytesMut::new();
-            self.encoder.encode(event, &mut bytes).map_err(|_| {
-                // Error is handled by `Encoder`.
+            self.encoder.encode(event, &mut bytes).map_err(|error| {
                 finalizers.update_status(EventStatus::Errored);
+                SinkError::new(error)
             })?;
 
             if let Err(error) = self.output.write_all(&bytes).await {
                 // Error when writing to stdout/stderr is likely irrecoverable,
                 // so stop the sink.
                 error!(message = "Error writing to output. Stopping sink.", %error, internal_log_rate_limit = false);
                 finalizers.update_status(EventStatus::Errored);
-                return Err(());
+                return Err(SinkError::new(error));
             }
 
             finalizers.update_status(EventStatus::Delivered);
@@ -90,10 +91,39 @@ mod test {
         };
 
         run_and_assert_sink_compliance(
-            VectorSink::from_event_streamsink(sink),
+            VectorSink::from_detailed_event_streamsink(sink),
             stream::once(ready(event)),
             &SINK_TAGS,
         )
         .await;
     }
+
+    #[tokio::test]
+    async fn write_error_retains_cause() {
+        let (output, input) = io::duplex(64);
+        drop(input);
+
+        let encoder = Encoder::<Framer>::new(
+            NewlineDelimitedEncoder::default().into(),
+            JsonSerializerConfig::default().build().into(),
+        );
+        let sink = WriterSink {
+            output,
+            transformer: Transformer::default(),
+            encoder,
+        };
+
+        let error = VectorSink::from_detailed_event_streamsink(sink)
+            .run_events([Event::Log(LogEvent::from("foo"))])
+            .await
+            .expect_err("writing to a closed pipe should fail");
+
+        assert_eq!(
+            error
+                .downcast_ref::<io::Error>()
+                .expect("the sink should retain the I/O error")
+                .kind(),
+            io::ErrorKind::BrokenPipe
+        );
+    }
 }