fix(topology): preserve component failure details (#26549)

Co-authored-by: Thomas <thomas.schneider@datadoghq.com>

Author: gwenaskell

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;