diff --git a/changelog.d/socket_bind_error_context.fix.md b/changelog.d/socket_bind_error_context.fix.md new file mode 100644 index 0000000000000..bb2127116cd6f --- /dev/null +++ b/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 diff --git a/lib/vector-core/src/sink.rs b/lib/vector-core/src/sink.rs index ab866c1ce5bd9..e3e76d0d684e2 100644 --- a/lib/vector-core/src/sink.rs +++ b/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,9 +7,56 @@ 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) -> 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 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 + Send + Unpin>), - Stream(Box + Send>), + Stream(Box + Send>), } impl VectorSink { @@ -17,10 +64,14 @@ impl VectorSink { /// /// # Errors /// - /// It is unclear under what conditions this function will error. - pub async fn run(self, input: impl Stream + Send) -> Result<(), ()> { + /// Returns an error if the sink reports a failure. + pub async fn run(self, input: impl Stream + 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(self, input: I) -> Result<(), ()> + pub async fn run_events(self, input: I) -> Result<(), SinkError> where I: IntoIterator + 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 + Send> { + pub fn into_stream(self) -> Box + 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 + 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 + 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 + 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 { - async fn run(self: Box, input: stream::BoxStream<'_, T>) -> Result<(), ()>; +pub trait StreamSink { + async fn run(self: Box, input: stream::BoxStream<'_, T>) -> Result<(), E>; +} + +struct OpaqueStream(Box); + +#[async_trait::async_trait] +impl StreamSink for OpaqueStream +where + T: StreamSink + Send, +{ + async fn run( + self: Box, + input: stream::BoxStream<'_, EventArray>, + ) -> Result<(), SinkError> { + self.0.run(input).await.map_err(|()| SinkError::default()) + } } /// Wrapper for sinks implementing `Sink` to implement @@ -175,8 +255,12 @@ struct EventStream { } #[async_trait::async_trait] -impl + Send> StreamSink for EventStream { - async fn run(self: Box, input: stream::BoxStream<'_, EventArray>) -> Result<(), ()> { +impl StreamSink for EventStream +where + T: StreamSink + Send, + E: Send, +{ + async fn run(self: Box, input: stream::BoxStream<'_, EventArray>) -> Result<(), E> { let input = Box::pin(input.flat_map(into_event_stream)); self.sink.run(input).await } diff --git a/lib/vector-core/src/source.rs b/lib/vector-core/src/source.rs index fb042bc7b3123..f3aa4ce90c8de 100644 --- a/lib/vector-core/src/source.rs +++ b/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(future: F) -> Source +where + F: Future> + 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(future: F) -> Source +where + F: Future> + Send + 'static, + E: Into, +{ + Box::pin(future.err_into()) +} diff --git a/lib/vector-core/src/source_sender/mod.rs b/lib/vector-core/src/source_sender/mod.rs index fb06b237cb4ea..9ab0e73c8918b 100644 --- a/lib/vector-core/src/source_sender/mod.rs +++ b/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}; diff --git a/src/config/unit_test/unit_test_components.rs b/src/config/unit_test/unit_test_components.rs index 047d874938654..6690cfbfa9e06 100644 --- a/src/config/unit_test/unit_test_components.rs +++ b/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(()) })) } diff --git a/src/enrichment_tables/memory/config.rs b/src/enrichment_tables/memory/config.rs index 488edb6d092d2..31e001d5632e7 100644 --- a/src/enrichment_tables/memory/config.rs +++ b/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(), )) } diff --git a/src/internal_events/amqp.rs b/src/internal_events/amqp.rs index e7864acc44e98..18b770ad72851 100644 --- a/src/internal_events/amqp.rs +++ b/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, diff --git a/src/internal_events/journald.rs b/src/internal_events/journald.rs index 02f6300cb6905..af7b34569806f 100644 --- a/src/internal_events/journald.rs +++ b/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.", diff --git a/src/sinks/blackhole/config.rs b/src/sinks/blackhole/config.rs index 81c4374293c59..b9243ca100b5a 100644 --- a/src/sinks/blackhole/config.rs +++ b/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)) } } diff --git a/src/sinks/blackhole/mod.rs b/src/sinks/blackhole/mod.rs index 2fddea593fe28..f909f7b02a2ce 100644 --- a/src/sinks/blackhole/mod.rs +++ b/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; diff --git a/src/sinks/console/config.rs b/src/sinks/console/config.rs index 7d939b0390ec9..d442f1ee2589b 100644 --- a/src/sinks/console/config.rs +++ b/src/sinks/console/config.rs @@ -102,12 +102,12 @@ impl ValidatedSink for ConsoleSinkConfig { let encoder = Encoder::::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, diff --git a/src/sinks/console/sink.rs b/src/sinks/console/sink.rs index b6c5fc39ee95c..5d5d990d065c5 100644 --- a/src/sinks/console/sink.rs +++ b/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 { } #[async_trait] -impl StreamSink for WriterSink +impl StreamSink for WriterSink where T: io::AsyncWrite + Send + Sync + Unpin, { - async fn run(mut self: Box, mut input: BoxStream<'_, Event>) -> Result<(), ()> { + async fn run(mut self: Box, 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,9 +38,9 @@ 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 { @@ -47,7 +48,7 @@ where // 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::::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::() + .expect("the sink should retain the I/O error") + .kind(), + io::ErrorKind::BrokenPipe + ); + } } diff --git a/src/sinks/humio/metrics.rs b/src/sinks/humio/metrics.rs index e1823430fb5b1..834063255dc87 100644 --- a/src/sinks/humio/metrics.rs +++ b/src/sinks/humio/metrics.rs @@ -8,7 +8,7 @@ use vector_lib::{ lookup, lookup::lookup_v2::{ConfigValuePath, OptionalTargetPath, OptionalValuePath}, sensitive_string::SensitiveString, - sink::StreamSink, + sink::{SinkError, StreamSink}, }; use super::{ @@ -247,8 +247,8 @@ pub struct HumioMetricsSink { } #[async_trait] -impl StreamSink for HumioMetricsSink { - async fn run(self: Box, input: BoxStream<'_, EventArray>) -> Result<(), ()> { +impl StreamSink for HumioMetricsSink { + async fn run(self: Box, input: BoxStream<'_, EventArray>) -> Result<(), SinkError> { let mut transform = self.transform; self.inner .run(input.map(move |events| { diff --git a/src/sinks/prometheus/exporter.rs b/src/sinks/prometheus/exporter.rs index 8baa6eec397f3..ce00441b4d766 100644 --- a/src/sinks/prometheus/exporter.rs +++ b/src/sinks/prometheus/exporter.rs @@ -29,6 +29,7 @@ use vector_lib::{ ByteSize, BytesSent, CountByteSize, EventsSent, InternalEventHandle as _, Output, Protocol, Registered, }, + sink::SinkError, }; use super::collector::{MetricCollector, StringCollector, metric_identifiers_have_no_line_breaks}; @@ -219,7 +220,10 @@ impl ValidatedSink for PrometheusExporterConfig { let sink = PrometheusExporter::new(self.clone()); let healthcheck = future::ok(()).boxed(); - Ok((VectorSink::from_event_streamsink(sink), healthcheck)) + Ok(( + VectorSink::from_detailed_event_streamsink(sink), + healthcheck, + )) } } @@ -544,11 +548,12 @@ impl PrometheusExporter { } #[async_trait] -impl StreamSink for PrometheusExporter { - async fn run(mut self: Box, mut input: BoxStream<'_, Event>) -> Result<(), ()> { - self.start_server_if_needed() - .await - .map_err(|error| error!("Failed to start Prometheus exporter: {error}."))?; +impl StreamSink for PrometheusExporter { + async fn run(mut self: Box, mut input: BoxStream<'_, Event>) -> Result<(), SinkError> { + self.start_server_if_needed().await.map_err(|error| { + error!("Failed to start Prometheus exporter: {error}."); + SinkError::new(error) + })?; let mut last_flush = Instant::now(); let flush_period = self.config.flush_period_secs; @@ -1186,7 +1191,7 @@ mod tests { let sink = PrometheusExporter::new(config); let metrics_handle = Arc::clone(&sink.metrics); - let sink = VectorSink::from_event_streamsink(sink); + let sink = VectorSink::from_detailed_event_streamsink(sink); sink.run(stream::iter(events).map(Into::into)) .await .unwrap(); @@ -1226,7 +1231,7 @@ mod tests { let metrics_handle = Arc::clone(&sink.metrics); - let sink = VectorSink::from_event_streamsink(sink); + let sink = VectorSink::from_detailed_event_streamsink(sink); let input_events = stream::iter(events).map(Into::into); sink.run(input_events).await.unwrap(); @@ -1339,7 +1344,7 @@ mod tests { .map(Event::Metric) .collect::>(); - let sink = VectorSink::from_event_streamsink(sink); + let sink = VectorSink::from_detailed_event_streamsink(sink); let input_events = stream::iter(events).map(Into::into); sink.run(input_events).await.unwrap(); @@ -1456,7 +1461,7 @@ mod tests { .map(Event::Metric) .collect::>(); - let sink = VectorSink::from_event_streamsink(sink); + let sink = VectorSink::from_detailed_event_streamsink(sink); let input_events = stream::iter(events).map(Into::into); sink.run(input_events).await.unwrap(); @@ -1525,7 +1530,7 @@ mod tests { .map(Event::Metric) .collect::>(); - let sink = VectorSink::from_event_streamsink(sink); + let sink = VectorSink::from_detailed_event_streamsink(sink); let input_events = stream::iter(events).map(Into::into); sink.run(input_events).await.unwrap(); diff --git a/src/sinks/sematext/logs.rs b/src/sinks/sematext/logs.rs index ac8a84d2ddeec..2ee880bf93a2e 100644 --- a/src/sinks/sematext/logs.rs +++ b/src/sinks/sematext/logs.rs @@ -1,7 +1,9 @@ use async_trait::async_trait; use futures::stream::{BoxStream, StreamExt}; use indoc::indoc; -use vector_lib::{configurable::configurable_component, sensitive_string::SensitiveString}; +use vector_lib::{ + configurable::configurable_component, sensitive_string::SensitiveString, sink::SinkError, +}; use vrl::event_path; use super::Region; @@ -161,12 +163,12 @@ impl SematextLogsConfig { } struct MapTimestampStream { - inner: Box + Send>, + inner: Box + Send>, } #[async_trait] -impl StreamSink for MapTimestampStream { - async fn run(self: Box, input: BoxStream<'_, EventArray>) -> Result<(), ()> { +impl StreamSink for MapTimestampStream { + async fn run(self: Box, input: BoxStream<'_, EventArray>) -> Result<(), SinkError> { let mapped_input = input.map(map_timestamp).boxed(); self.inner.run(mapped_input).await } diff --git a/src/sinks/websocket_server/config.rs b/src/sinks/websocket_server/config.rs index 1d6fcd9580cf7..939a1528ab2cd 100644 --- a/src/sinks/websocket_server/config.rs +++ b/src/sinks/websocket_server/config.rs @@ -178,7 +178,7 @@ impl ValidatedSink for WebSocketListenerSinkConfig { let ws_sink = WebSocketListenerSink::from_validated(self.clone(), validated, tls, &cx)?; Ok(( - VectorSink::from_event_streamsink(ws_sink), + VectorSink::from_detailed_event_streamsink(ws_sink), Box::pin(async move { Ok(()) }), )) } diff --git a/src/sinks/websocket_server/sink.rs b/src/sinks/websocket_server/sink.rs index 201e6153b2cb7..66096bd193cd4 100644 --- a/src/sinks/websocket_server/sink.rs +++ b/src/sinks/websocket_server/sink.rs @@ -49,7 +49,7 @@ use vector_lib::{ internal_event::{ ByteSize, BytesSent, CountByteSize, EventsSent, InternalEventHandle, Output, Protocol, }, - sink::StreamSink, + sink::{SinkError, StreamSink}, tls::{MaybeTlsIncomingStream, MaybeTlsListener, MaybeTlsSettings}, }; @@ -360,8 +360,8 @@ impl WebSocketListenerSink { } #[async_trait] -impl StreamSink for WebSocketListenerSink { - async fn run(mut self: Box, input: BoxStream<'_, Event>) -> Result<(), ()> { +impl StreamSink for WebSocketListenerSink { + async fn run(mut self: Box, input: BoxStream<'_, Event>) -> Result<(), SinkError> { let input = input.fuse().peekable(); pin_mut!(input); @@ -369,7 +369,7 @@ impl StreamSink for WebSocketListenerSink { let events_sent = register!(EventsSent::from(Output(None))); let encode_as_binary = self.encoder.serializer().is_binary(); - let listener = self.tls.bind(&self.address).await.map_err(|_| ())?; + let listener = self.tls.bind(&self.address).await.map_err(SinkError::new)?; let peers = Arc::new(Mutex::new(HashMap::default())); let message_buffer = Arc::new(Mutex::new(VecDeque::with_capacity( @@ -685,7 +685,7 @@ mod tests { .unwrap(); run_and_assert_sink_compliance( - VectorSink::from_event_streamsink(sink), + VectorSink::from_detailed_event_streamsink(sink), stream::once(ready(event)), &SINK_TAGS, ) @@ -839,7 +839,7 @@ mod tests { let sink = WebSocketListenerSink::new(config, &SinkContext::default()).unwrap(); let compliance_assertion = tokio::spawn(run_and_assert_sink_compliance( - VectorSink::from_event_streamsink(sink), + VectorSink::from_detailed_event_streamsink(sink), events, &SINK_TAGS, )); diff --git a/src/sources/amqp.rs b/src/sources/amqp.rs index e31c7263c75a1..7e42bd4906e42 100644 --- a/src/sources/amqp.rs +++ b/src/sources/amqp.rs @@ -318,9 +318,9 @@ async fn receive_event( log_namespace: LogNamespace, finalizer: Option<&UnorderedFinalizer>, msg: Delivery, -) -> Result<(), ()> { +) -> Result<(), crate::sources::SourceError> { let payload = Cursor::new(Bytes::copy_from_slice(&msg.data)); - let decoder = config.decoder(log_namespace).map_err(|_e| ())?; + let decoder = config.decoder(log_namespace)?; let mut stream = DecoderFramedRead::new(payload, decoder); // Extract timestamp from AMQP message @@ -428,7 +428,7 @@ async fn run_amqp_source( channel: Channel, log_namespace: LogNamespace, acknowledgements: bool, -) -> Result<(), ()> { +) -> Result<(), crate::sources::SourceError> { let (finalizer, mut ack_stream) = UnorderedFinalizer::::maybe_new(acknowledgements, Some(shutdown.clone())); @@ -440,6 +440,7 @@ async fn run_amqp_source( .await .map_err(|error| { error!(message = "Failed to apply basic_qos.", ?error); + crate::sources::SourceError::from(error) })?; } @@ -454,6 +455,7 @@ async fn run_amqp_source( .await .map_err(|error| { error!(message = "Failed to consume.", ?error); + crate::sources::SourceError::from(error) })? .fuse(); let mut shutdown = shutdown.fuse(); @@ -469,8 +471,8 @@ async fn run_amqp_source( if let Some(try_m) = opt_m { match try_m { Err(error) => { - emit!(AmqpEventError { error }); - return Err(()); + emit!(AmqpEventError { error: &error }); + return Err(crate::sources::SourceError::from(error)); } Ok(msg) => { receive_event(&config, &mut out, log_namespace, finalizer.as_ref(), msg).await? diff --git a/src/sources/apache_metrics/mod.rs b/src/sources/apache_metrics/mod.rs index c6a5d37f8495a..2de3d013f96d2 100644 --- a/src/sources/apache_metrics/mod.rs +++ b/src/sources/apache_metrics/mod.rs @@ -149,7 +149,7 @@ fn apache_metrics( mut out: SourceSender, proxy: ProxyConfig, ) -> super::Source { - Box::pin(async move { + crate::sources::source_with_error(async move { let mut stream = IntervalStream::new(tokio::time::interval(interval)) .take_until(shutdown) .map(move |_| stream::iter(urls.clone())) @@ -267,10 +267,10 @@ fn apache_metrics( debug!("Finished sending."); Ok(()) } - Err(_) => { + Err(error) => { let (count, _) = stream.size_hint(); emit!(StreamClosedError { count }); - Err(()) + Err(error) } } }) diff --git a/src/sources/aws_ecs_metrics/mod.rs b/src/sources/aws_ecs_metrics/mod.rs index 79cba4b6f2386..4cc0d3a18bdd2 100644 --- a/src/sources/aws_ecs_metrics/mod.rs +++ b/src/sources/aws_ecs_metrics/mod.rs @@ -157,7 +157,7 @@ impl SourceConfig for AwsEcsMetricsSourceConfig { let namespace = Some(self.namespace.clone()).filter(|namespace| !namespace.is_empty()); let http_client = HttpClient::new(None, &cx.proxy)?; - Ok(Box::pin(aws_ecs_metrics( + Ok(crate::sources::source_with_error(aws_ecs_metrics( http_client, self.stats_endpoint(), self.scrape_interval_secs, @@ -183,7 +183,7 @@ async fn aws_ecs_metrics( namespace: Option, mut out: SourceSender, shutdown: ShutdownSignal, -) -> Result<(), ()> { +) -> vector_lib::source_sender::SendResult { let mut interval = IntervalStream::new(time::interval(interval)).take_until(shutdown); let bytes_received = register!(BytesReceived::from(Protocol::HTTP)); while interval.next().await.is_some() { @@ -210,9 +210,9 @@ async fn aws_ecs_metrics( endpoint: uri.path(), }); - if (out.send_batch(metrics).await).is_err() { + if let Err(error) = out.send_batch(metrics).await { emit!(StreamClosedError { count }); - return Err(()); + return Err(error); } } Err(error) => { diff --git a/src/sources/aws_kinesis_firehose/mod.rs b/src/sources/aws_kinesis_firehose/mod.rs index b3e5a8cc8e3f1..4f4db3f5a85c8 100644 --- a/src/sources/aws_kinesis_firehose/mod.rs +++ b/src/sources/aws_kinesis_firehose/mod.rs @@ -219,9 +219,7 @@ impl SourceConfig for AwsKinesisFirehoseConfig { .serve(make_svc) .with_graceful_shutdown(shutdown.map(|_| ())) .await - .map_err(|err| { - error!("An error occurred: {err:?}."); - })?; + .inspect_err(|err| error!("An error occurred: {err:?}."))?; Ok(()) })) diff --git a/src/sources/aws_s3/mod.rs b/src/sources/aws_s3/mod.rs index a1090e3b6fa50..b2f702e3c7b5b 100644 --- a/src/sources/aws_s3/mod.rs +++ b/src/sources/aws_s3/mod.rs @@ -193,7 +193,7 @@ impl SourceConfig for AwsS3Config { .transpose()?; match self.strategy { - Strategy::Sqs => Ok(Box::pin( + Strategy::Sqs => Ok(crate::sources::opaque_source( self.create_sqs_ingestor(multiline_config, &cx.proxy, log_namespace) .await? .run(cx, self.acknowledgements, log_namespace), diff --git a/src/sources/aws_sqs/config.rs b/src/sources/aws_sqs/config.rs index b1a936a4e69e8..eb4e080e8443f 100644 --- a/src/sources/aws_sqs/config.rs +++ b/src/sources/aws_sqs/config.rs @@ -111,7 +111,7 @@ impl SourceConfig for AwsSqsConfig { .build()?; let acknowledgements = cx.do_acknowledgements(self.acknowledgements); - Ok(Box::pin( + Ok(crate::sources::opaque_source( SqsSource { client, queue_url: self.queue_url.clone(), diff --git a/src/sources/datadog_agent/mod.rs b/src/sources/datadog_agent/mod.rs index fd7b966830ab0..2e9e8f5671d8a 100644 --- a/src/sources/datadog_agent/mod.rs +++ b/src/sources/datadog_agent/mod.rs @@ -254,9 +254,7 @@ impl SourceConfig for DatadogAgentConfig { .serve(make_svc) .with_graceful_shutdown(shutdown.map(|_| ())) .await - .map_err(|err| { - error!("An error occurred: {err:?}."); - })?; + .inspect_err(|err| error!("An error occurred: {err:?}."))?; Ok(()) })) diff --git a/src/sources/demo_logs.rs b/src/sources/demo_logs.rs index a83dae13bf364..34f9ef4783323 100644 --- a/src/sources/demo_logs.rs +++ b/src/sources/demo_logs.rs @@ -211,7 +211,7 @@ async fn demo_logs_source( mut shutdown: ShutdownSignal, mut out: SourceSender, log_namespace: LogNamespace, -) -> Result<(), ()> { +) -> vector_lib::source_sender::SendResult { let interval: Option = (interval != Duration::ZERO).then_some(interval); let mut interval = interval.map(time::interval); @@ -263,7 +263,7 @@ async fn demo_logs_source( event }); - out.send_batch(events).await.map_err(|_| { + out.send_batch(events).await.inspect_err(|_| { emit!(StreamClosedError { count }); })?; } @@ -293,7 +293,7 @@ impl SourceConfig for DemoLogsConfig { let decoder = DecodingConfig::new(self.framing.clone(), self.decoding.clone(), log_namespace) .build()?; - Ok(Box::pin(demo_logs_source( + Ok(crate::sources::source_with_error(demo_logs_source( self.interval, self.count, self.format.clone(), diff --git a/src/sources/exec/mod.rs b/src/sources/exec/mod.rs index 07f7dddcd2780..c44c64a45fd7f 100644 --- a/src/sources/exec/mod.rs +++ b/src/sources/exec/mod.rs @@ -255,7 +255,7 @@ impl SourceConfig for ExecConfig { Mode::Scheduled => { let exec_interval_secs = self.exec_interval_secs_or_default(); - Ok(Box::pin(run_scheduled( + Ok(crate::sources::opaque_source(run_scheduled( self.clone(), hostname, exec_interval_secs, @@ -269,7 +269,7 @@ impl SourceConfig for ExecConfig { let respawn_on_exit = self.respawn_on_exit_or_default(); let respawn_interval_secs = self.respawn_interval_secs_or_default(); - Ok(Box::pin(run_streaming( + Ok(crate::sources::opaque_source(run_streaming( self.clone(), hostname, respawn_on_exit, diff --git a/src/sources/file.rs b/src/sources/file.rs index 4db8beea97cde..b29c0a7a9cd7e 100644 --- a/src/sources/file.rs +++ b/src/sources/file.rs @@ -479,11 +479,9 @@ pub fn file_source( ) -> super::Source { // the include option must be specified but also must contain at least one entry. if config.include.is_empty() { - error!( - message = "`include` configuration option must contain at least one file pattern.", - internal_log_rate_limit = false - ); - return Box::pin(future::ready(Err(()))); + let error = "`include` configuration option must contain at least one file pattern."; + error!(message = error, internal_log_rate_limit = false); + return Box::pin(future::ready(Err(error.into()))); } let exclude_patterns = config @@ -586,7 +584,7 @@ pub fn file_source( let checkpoints = checkpointer.view(); let include_file_metric_tag = config.internal_metrics.include_file_tag; - Box::pin(async move { + crate::sources::opaque_source(async move { info!(message = "Starting file server.", include = ?include, exclude = ?exclude); let mut encoding_decoder = encoding_charset.map(Decoder::new); diff --git a/src/sources/file_descriptors/mod.rs b/src/sources/file_descriptors/mod.rs index b3c60d50a54c0..f818729d0e6e5 100644 --- a/src/sources/file_descriptors/mod.rs +++ b/src/sources/file_descriptors/mod.rs @@ -74,7 +74,7 @@ pub trait FileDescriptorConfig: NamedComponent { read_from_fd(reader, sender); }); - Ok(Box::pin(process_stream( + Ok(crate::sources::source_with_error(process_stream( receiver, decoder, out, @@ -122,7 +122,7 @@ async fn process_stream( source_type: &'static str, hostname: Option, log_namespace: LogNamespace, -) -> Result<(), ()> { +) -> vector_lib::source_sender::SendResult { let bytes_received = register!(BytesReceived::from(Protocol::NONE)); let events_received = register!(EventsReceived); let stream = receiver.inspect(|result| { @@ -190,10 +190,10 @@ async fn process_stream( debug!("Finished sending."); Ok(()) } - Err(_) => { + Err(error) => { let (count, _) = stream.size_hint(); emit!(StreamClosedError { count }); - Err(()) + Err(error) } } } diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs index 1bf501c098f59..9cfba0879cc03 100644 --- a/src/sources/gcp_pubsub.rs +++ b/src/sources/gcp_pubsub.rs @@ -327,7 +327,10 @@ impl SourceConfig for PubsubConfig { events_received: register!(EventsReceived), } .run_all(self.max_concurrency, self.poll_time_seconds) - .map_err(|error| error!(message = "Source failed.", %error)); + .map_err(|error| { + error!(message = "Source failed.", %error); + crate::sources::SourceError::from(error) + }); Ok(Box::pin(source)) } diff --git a/src/sources/host_metrics/mod.rs b/src/sources/host_metrics/mod.rs index 191acb3f4e846..c60b3a32d7c52 100644 --- a/src/sources/host_metrics/mod.rs +++ b/src/sources/host_metrics/mod.rs @@ -304,7 +304,9 @@ impl SourceConfig for HostMetricsConfig { let mut config = self.clone(); config.namespace = config.namespace.filter(|namespace| !namespace.is_empty()); - Ok(Box::pin(config.run(cx.out, cx.shutdown))) + Ok(crate::sources::source_with_error( + config.run(cx.out, cx.shutdown), + )) } fn outputs(&self, _global_log_namespace: LogNamespace) -> Vec { @@ -322,7 +324,11 @@ impl HostMetricsConfig { self.scrape_interval_secs = Duration::from_secs_f64(value); } - async fn run(self, mut out: SourceSender, shutdown: ShutdownSignal) -> Result<(), ()> { + async fn run( + self, + mut out: SourceSender, + shutdown: ShutdownSignal, + ) -> vector_lib::source_sender::SendResult { let duration = self.scrape_interval_secs; let mut interval = IntervalStream::new(time::interval(duration)).take_until(shutdown); @@ -334,9 +340,9 @@ impl HostMetricsConfig { bytes_received.emit(ByteSize(0)); let metrics = generator.capture_metrics().await; let count = metrics.len(); - if (out.send_batch(metrics).await).is_err() { + if let Err(error) = out.send_batch(metrics).await { emit!(StreamClosedError { count }); - return Err(()); + return Err(error); } } diff --git a/src/sources/http_client/client.rs b/src/sources/http_client/client.rs index 9363a58f54c0f..d8eaf3290b169 100644 --- a/src/sources/http_client/client.rs +++ b/src/sources/http_client/client.rs @@ -5,7 +5,6 @@ use std::{collections::HashMap, time::Duration}; use bytes::{Bytes, BytesMut}; use chrono::Utc; -use futures_util::FutureExt; use http::{Uri, response::Parts}; use serde_with::serde_as; use snafu::ResultExt; @@ -378,7 +377,12 @@ impl SourceConfig for HttpClientConfig { shutdown: cx.shutdown, }; - Ok(call(inputs, context, cx.out, self.method).boxed()) + Ok(crate::sources::source_with_error(call( + inputs, + context, + cx.out, + self.method, + ))) } fn outputs(&self, global_log_namespace: LogNamespace) -> Vec { diff --git a/src/sources/internal_logs.rs b/src/sources/internal_logs.rs index 74143c6c5b3d7..ae85aa2bbb12a 100644 --- a/src/sources/internal_logs.rs +++ b/src/sources/internal_logs.rs @@ -113,7 +113,7 @@ impl SourceConfig for InternalLogsConfig { let log_namespace = cx.log_namespace(self.log_namespace); - Ok(Box::pin(run( + Ok(crate::sources::source_with_error(run( host_key, pid_key, subscription, @@ -145,7 +145,7 @@ async fn run( mut out: SourceSender, shutdown: ShutdownSignal, log_namespace: LogNamespace, -) -> Result<(), ()> { +) -> vector_lib::source_sender::SendResult { let hostname = crate::get_hostname(); let pid = std::process::id(); @@ -198,10 +198,10 @@ async fn run( Utc::now(), ); - if (out.send_event(Event::from(log)).await).is_err() { + if let Err(error) = out.send_event(Event::from(log)).await { // this wont trigger any infinite loop considering it stops the component emit!(StreamClosedError { count: 1 }); - return Err(()); + return Err(error); } } diff --git a/src/sources/internal_metrics.rs b/src/sources/internal_metrics.rs index 60846eda4d8af..c18ffc2b26b33 100644 --- a/src/sources/internal_metrics.rs +++ b/src/sources/internal_metrics.rs @@ -112,7 +112,7 @@ impl SourceConfig for InternalMetricsConfig { .as_deref() .and_then(|tag| (!tag.is_empty()).then(|| tag.to_owned())); - Ok(Box::pin( + Ok(crate::sources::source_with_error( InternalMetrics { namespace, host_key, @@ -146,7 +146,7 @@ struct InternalMetrics<'a> { } impl InternalMetrics<'_> { - async fn run(mut self) -> Result<(), ()> { + async fn run(mut self) -> vector_lib::source_sender::SendResult { let events_received = register!(EventsReceived); let bytes_received = register!(BytesReceived::from(Protocol::INTERNAL)); let mut interval = @@ -181,9 +181,9 @@ impl InternalMetrics<'_> { metric }); - if (self.out.send_batch(batch).await).is_err() { + if let Err(error) = self.out.send_batch(batch).await { emit!(StreamClosedError { count }); - return Err(()); + return Err(error); } } diff --git a/src/sources/journald.rs b/src/sources/journald.rs index b2663d5adc4c9..9fa636878caa4 100644 --- a/src/sources/journald.rs +++ b/src/sources/journald.rs @@ -440,18 +440,22 @@ struct JournaldSource { } impl JournaldSource { - async fn run_shutdown(self, shutdown: ShutdownSignal) -> Result<(), ()> { + async fn run_shutdown( + self, + shutdown: ShutdownSignal, + ) -> Result<(), crate::sources::SourceError> { let checkpointer = StatefulCheckpointer::new(self.checkpoint_path.clone()) .await .map_err(|error| { emit!(JournaldCheckpointFileOpenError { - error, + error: &error, path: self .checkpoint_path .to_str() .unwrap_or("unknown") .to_string(), }); + crate::sources::SourceError::from(error) })?; let checkpointer = SharedCheckpointer::new(checkpointer); diff --git a/src/sources/kafka.rs b/src/sources/kafka.rs index ca743071b3a3e..684ee49438cba 100644 --- a/src/sources/kafka.rs +++ b/src/sources/kafka.rs @@ -436,7 +436,7 @@ async fn kafka_source( shutdown: ShutdownSignal, eof: bool, log_namespace: LogNamespace, -) -> Result<(), ()> { +) -> Result<(), crate::sources::SourceError> { let span = info_span!("kafka_source"); let consumer = Arc::new(consumer); @@ -452,7 +452,7 @@ async fn kafka_source( let topics: Vec<&str> = config.topics.iter().map(|s| s.as_str()).collect(); if let Err(e) = consumer.subscribe(&topics).context(SubscribeSnafu) { error!("{e}"); - return Err(()); + return Err(crate::sources::SourceError::from(e)); } let coordination_task = { diff --git a/src/sources/kubernetes_logs/mod.rs b/src/sources/kubernetes_logs/mod.rs index 4b1c2432f6afe..dfb6871f53af9 100644 --- a/src/sources/kubernetes_logs/mod.rs +++ b/src/sources/kubernetes_logs/mod.rs @@ -364,6 +364,7 @@ impl SourceConfig for Config { .map(|result| { result.map_err(|error| { error!(message = "Source future failed.", %error); + crate::sources::SourceError::from(error) }) }), )) diff --git a/src/sources/mod.rs b/src/sources/mod.rs index bac6ad727e76f..3f4fd97bed6bb 100644 --- a/src/sources/mod.rs +++ b/src/sources/mod.rs @@ -99,7 +99,7 @@ pub mod windows_event_log; pub mod util; -pub use vector_lib::source::Source; +pub use vector_lib::source::{Source, SourceError, opaque_source, source_with_error}; #[allow(dead_code)] // Easier than listing out all the features that use this /// Common build errors diff --git a/src/sources/mongodb_metrics/mod.rs b/src/sources/mongodb_metrics/mod.rs index 2043d642adf60..5568d1224bd3d 100644 --- a/src/sources/mongodb_metrics/mod.rs +++ b/src/sources/mongodb_metrics/mod.rs @@ -17,6 +17,7 @@ use tokio::time; use tokio_stream::wrappers::IntervalStream; use vector_lib::{ ByteSizeOf, EstimatedJsonEncodedSizeOf, configurable::configurable_component, metric_tags, + source_sender::SendError, }; use crate::{ @@ -135,7 +136,7 @@ impl SourceConfig for MongoDbMetricsConfig { let duration = self.scrape_interval_secs; let shutdown = cx.shutdown; - Ok(Box::pin(async move { + Ok(crate::sources::source_with_error(async move { let mut interval = IntervalStream::new(time::interval(duration)).take_until(shutdown); while interval.next().await.is_some() { let start = Instant::now(); @@ -148,13 +149,13 @@ impl SourceConfig for MongoDbMetricsConfig { let metrics: Vec = metrics.into_iter().flatten().collect(); let count = metrics.len(); - if (cx.out.send_batch(metrics).await).is_err() { - emit!(StreamClosedError { count }); - return Err(()); - } + cx.out + .send_batch(metrics) + .await + .inspect_err(|_| emit!(StreamClosedError { count }))?; } - Ok(()) + Ok::<(), SendError>(()) })) } diff --git a/src/sources/mqtt/source.rs b/src/sources/mqtt/source.rs index d0f3efea5b0f7..03a8bc1cac23d 100644 --- a/src/sources/mqtt/source.rs +++ b/src/sources/mqtt/source.rs @@ -14,7 +14,7 @@ use crate::{ internal_events::{EndpointBytesReceived, StreamClosedError}, serde::OneOrMany, shutdown::ShutdownSignal, - sources::{mqtt::MqttSourceConfig, util}, + sources::{SourceError, mqtt::MqttSourceConfig, util}, }; pub struct MqttSource { @@ -39,15 +39,16 @@ impl MqttSource { }) } - pub async fn run(self, mut out: SourceSender, shutdown: ShutdownSignal) -> Result<(), ()> { + pub async fn run( + self, + mut out: SourceSender, + shutdown: ShutdownSignal, + ) -> Result<(), SourceError> { let (client, mut connection) = self.connector.connect(); match &self.config.topic { OneOrMany::One(topic) => { - client - .subscribe(topic, QoS::AtLeastOnce) - .await - .map_err(|_| ())?; + client.subscribe(topic, QoS::AtLeastOnce).await?; } OneOrMany::Many(topics) => { client @@ -57,8 +58,7 @@ impl MqttSource { .cloned() .map(|topic| SubscribeFilter::new(topic, QoS::AtLeastOnce)), ) - .await - .map_err(|_| ())?; + .await?; } } diff --git a/src/sources/nats/config.rs b/src/sources/nats/config.rs index 6a69b9d6de5cf..df3addf0a7ad8 100644 --- a/src/sources/nats/config.rs +++ b/src/sources/nats/config.rs @@ -202,7 +202,7 @@ impl SourceConfig for NatsSourceConfig { let connection = self.connect().await?; let messages = create_consumer_stream(&connection, js_config).await?; - Ok(Box::pin(run_nats_jetstream( + Ok(crate::sources::source_with_error(run_nats_jetstream( self.clone(), connection, messages, @@ -215,7 +215,7 @@ impl SourceConfig for NatsSourceConfig { NatsMode::Core => { let (connection, subscription) = create_subscription(self).await?; - Ok(Box::pin(run_nats_core( + Ok(crate::sources::source_with_error(run_nats_core( self.clone(), connection, subscription, diff --git a/src/sources/nats/integration_tests.rs b/src/sources/nats/integration_tests.rs index ffab3402cbf1a..23898e394c129 100644 --- a/src/sources/nats/integration_tests.rs +++ b/src/sources/nats/integration_tests.rs @@ -612,7 +612,7 @@ struct JetStreamTestHarness { subject: String, stream_name: String, consumer_name: String, - _source_handle: tokio::task::JoinHandle>, + _source_handle: tokio::task::JoinHandle>, } impl JetStreamTestHarness { diff --git a/src/sources/nats/source.rs b/src/sources/nats/source.rs index d22a4ca37cba7..84687065d3b59 100644 --- a/src/sources/nats/source.rs +++ b/src/sources/nats/source.rs @@ -33,7 +33,7 @@ pub enum ProcessingStatus { /// A non-recoverable error occurred while decoding the payload. Failed, /// The downstream channel is closed, and the source should shut down. - ChannelClosed, + ChannelClosed(vector_lib::source_sender::SendError), } /// Processes a single NATS message, sending decoded events downstream. @@ -84,9 +84,9 @@ pub async fn process_message( event }); - if out.send_batch(events).await.is_err() { + if let Err(error) = out.send_batch(events).await { emit!(StreamClosedError { count }); - return ProcessingStatus::ChannelClosed; + return ProcessingStatus::ChannelClosed(error); } } Err(error) => { @@ -137,7 +137,7 @@ pub async fn run_nats_jetstream( log_namespace: LogNamespace, mut shutdown: ShutdownSignal, mut out: SourceSender, -) -> Result<(), ()> { +) -> vector_lib::source_sender::SendResult { let events_received = register!(EventsReceived); let bytes_received = register!(BytesReceived::from(Protocol::TCP)); let mut backoff = ExponentialBackoff::default().max_delay(std::time::Duration::from_secs(30)); @@ -179,7 +179,7 @@ pub async fn run_nats_jetstream( error!(message = "Failed to acknowledge JetStream message.", %err); } } - ProcessingStatus::ChannelClosed => return Err(()), + ProcessingStatus::ChannelClosed(error) => return Err(error), // Do not acknowledge on failure; the message will be redelivered. ProcessingStatus::Failed => {} } @@ -232,7 +232,7 @@ pub async fn run_nats_core( log_namespace: LogNamespace, mut shutdown: ShutdownSignal, mut out: SourceSender, -) -> Result<(), ()> { +) -> vector_lib::source_sender::SendResult { let events_received = register!(EventsReceived); let bytes_received = register!(BytesReceived::from(Protocol::TCP)); @@ -261,8 +261,8 @@ pub async fn run_nats_core( ) .await; - if let ProcessingStatus::ChannelClosed = status { - return Err(()); + if let ProcessingStatus::ChannelClosed(error) = status { + return Err(error); } }, None => { diff --git a/src/sources/nginx_metrics/mod.rs b/src/sources/nginx_metrics/mod.rs index cdde937dd05ef..7f1d4c2c2d59f 100644 --- a/src/sources/nginx_metrics/mod.rs +++ b/src/sources/nginx_metrics/mod.rs @@ -120,7 +120,7 @@ impl SourceConfig for NginxMetricsConfig { let duration = self.scrape_interval_secs; let shutdown = cx.shutdown; - Ok(Box::pin(async move { + Ok(crate::sources::source_with_error(async move { let mut interval = IntervalStream::new(time::interval(duration)).take_until(shutdown); while interval.next().await.is_some() { let start = Instant::now(); @@ -133,9 +133,9 @@ impl SourceConfig for NginxMetricsConfig { let metrics: Vec = metrics.into_iter().flatten().collect(); let count = metrics.len(); - if (cx.out.send_batch(metrics).await).is_err() { + if let Err(error) = cx.out.send_batch(metrics).await { emit!(StreamClosedError { count }); - return Err(()); + return Err(error); } } diff --git a/src/sources/okta/client.rs b/src/sources/okta/client.rs index 0caf7a8109b12..8073a3515605e 100644 --- a/src/sources/okta/client.rs +++ b/src/sources/okta/client.rs @@ -153,7 +153,7 @@ impl SourceConfig for OktaConfig { warn_if_interval_too_low(self.timeout, self.interval); - Ok(run( + Ok(crate::sources::source_with_error(run( url, tls, cx.proxy, @@ -163,8 +163,7 @@ impl SourceConfig for OktaConfig { log_namespace, cx.shutdown, cx.out, - ) - .boxed()) + ))) } fn outputs(&self, global_log_namespace: LogNamespace) -> Vec { @@ -294,7 +293,7 @@ async fn run( log_namespace: LogNamespace, shutdown: ShutdownSignal, mut out: SourceSender, -) -> Result<(), ()> { +) -> Result<(), crate::Error> { let url_mutex = Arc::new(Mutex::new(url.clone())); let decoder = DecodingConfig::new( FramingConfig::Bytes, @@ -302,15 +301,17 @@ async fn run( log_namespace, ) .build() - .map_err(|ref e| { + .inspect_err(|e| { emit!(DecoderDeserializeError { error: e }); })?; let client = HttpClient::new(tls, &proxy).map_err(|e| { + let error = Arc::new(e); emit!(HttpClientHttpError { - error: Box::new(e), + error: Box::new(Arc::clone(&error)), url: url.to_string() }); + crate::Error::from(error) })?; let mut stream = IntervalStream::new(tokio::time::interval(interval)) @@ -390,10 +391,10 @@ async fn run( debug!("Finished sending."); Ok(()) } - Err(_) => { + Err(error) => { let (count, _) = stream.size_hint(); emit!(StreamClosedError { count }); - Err(()) + Err(error.into()) } } } diff --git a/src/sources/postgresql_metrics.rs b/src/sources/postgresql_metrics.rs index 19dbcc59adbab..6da7742286f12 100644 --- a/src/sources/postgresql_metrics.rs +++ b/src/sources/postgresql_metrics.rs @@ -218,7 +218,7 @@ impl SourceConfig for PostgresqlMetricsConfig { let duration = self.scrape_interval_secs; let shutdown = cx.shutdown; - Ok(Box::pin(async move { + Ok(crate::sources::source_with_error(async move { let mut interval = IntervalStream::new(time::interval(duration)).take_until(shutdown); while interval.next().await.is_some() { let start = Instant::now(); @@ -231,9 +231,9 @@ impl SourceConfig for PostgresqlMetricsConfig { let metrics: Vec = metrics.into_iter().flatten().collect(); let count = metrics.len(); - if (cx.out.send_batch(metrics).await).is_err() { + if let Err(error) = cx.out.send_batch(metrics).await { emit!(StreamClosedError { count }); - return Err(()); + return Err(error); } } diff --git a/src/sources/prometheus/scrape.rs b/src/sources/prometheus/scrape.rs index b142cd9b816eb..c50afc6c4462b 100644 --- a/src/sources/prometheus/scrape.rs +++ b/src/sources/prometheus/scrape.rs @@ -1,7 +1,6 @@ use std::{collections::HashMap, time::Duration}; use bytes::Bytes; -use futures_util::FutureExt; use http::{Uri, response::Parts}; use serde_with::serde_as; use snafu::ResultExt; @@ -172,7 +171,12 @@ impl SourceConfig for PrometheusScrapeConfig { shutdown: cx.shutdown, }; - Ok(call(inputs, builder, cx.out, HttpMethod::Get).boxed()) + Ok(crate::sources::source_with_error(call( + inputs, + builder, + cx.out, + HttpMethod::Get, + ))) } fn outputs(&self, _global_log_namespace: LogNamespace) -> Vec { diff --git a/src/sources/pulsar.rs b/src/sources/pulsar.rs index 99de419961ea5..62e7955f7726f 100644 --- a/src/sources/pulsar.rs +++ b/src/sources/pulsar.rs @@ -205,7 +205,7 @@ impl SourceConfig for PulsarSourceConfig { .build()?; let acknowledgements = cx.do_acknowledgements(self.acknowledgements); - Ok(Box::pin(pulsar_source( + Ok(crate::sources::opaque_source(pulsar_source( consumer, decoder, cx.shutdown, diff --git a/src/sources/socket/mod.rs b/src/sources/socket/mod.rs index 946fad688ba88..2e864cc646ec7 100644 --- a/src/sources/socket/mod.rs +++ b/src/sources/socket/mod.rs @@ -924,7 +924,7 @@ mod test { assert_eq!(shutdown_result, Ok(true)); let source_result = source_handle.await.expect("source task should not panic"); - assert_eq!(source_result, Ok(())); + assert!(source_result.is_ok()); } #[tokio::test] @@ -1048,7 +1048,10 @@ mod test { sender: SourceSender, source_id: &ComponentKey, shutdown: &mut SourceShutdownCoordinator, - ) -> (SocketAddr, JoinHandle>) { + ) -> ( + SocketAddr, + JoinHandle>, + ) { let (shutdown_signal, _) = shutdown.register_source(source_id, false); init_udp_inner(sender, source_id, shutdown_signal, None, false).await } @@ -1083,7 +1086,10 @@ mod test { shutdown_signal: ShutdownSignal, config: Option, use_vector_namespace: bool, - ) -> (SocketAddr, JoinHandle>) { + ) -> ( + SocketAddr, + JoinHandle>, + ) { let (guard, address, mut config) = match config { Some(config) => match config.address() { SocketListenAddr::SocketAddr(addr) => (None, addr, config), diff --git a/src/sources/socket/udp.rs b/src/sources/socket/udp.rs index 1cebb7f4d630d..bd6dbced6a67e 100644 --- a/src/sources/socket/udp.rs +++ b/src/sources/socket/udp.rs @@ -28,6 +28,7 @@ use crate::{ net, serde::default_decoding, shutdown::ShutdownSignal, + sources::SourceError, sources::{ Source, socket::SocketConfig, @@ -180,8 +181,9 @@ pub(super) fn udp( .map_err(|error| { emit!(SocketBindError { mode: SocketMode::Udp, - error, - }) + error: &error, + }); + SourceError::from(error) })?; if !config.multicast_groups.is_empty() { @@ -204,10 +206,11 @@ pub(super) fn udp( .join_multicast_v4(group_addr, interface) .map_err(|error| { emit!(SocketMulticastGroupJoinError { - error, + error: &error, group_addr, interface, - }) + }); + SourceError::from(error) })?; info!(message = "Joined multicast group.", group = %group_addr); } @@ -249,10 +252,11 @@ pub(super) fn udp( } } - return Err(emit!(SocketReceiveError { + emit!(SocketReceiveError { mode: SocketMode::Udp, - error - })); + error: &error + }); + return Err(SourceError::from(error)); } }; diff --git a/src/sources/splunk_hec/mod.rs b/src/sources/splunk_hec/mod.rs index c27b9825c2fab..07221186b8628 100644 --- a/src/sources/splunk_hec/mod.rs +++ b/src/sources/splunk_hec/mod.rs @@ -292,9 +292,7 @@ impl SplunkConfig { .serve(make_svc) .with_graceful_shutdown(shutdown.map(|_| ())) .await - .map_err(|err| { - error!("An error occurred: {err:?}."); - })?; + .inspect_err(|err| error!("An error occurred: {err:?}."))?; Ok(()) })) diff --git a/src/sources/static_metrics.rs b/src/sources/static_metrics.rs index 1398562ec82d6..da55c2d19a554 100644 --- a/src/sources/static_metrics.rs +++ b/src/sources/static_metrics.rs @@ -103,7 +103,7 @@ impl SourceConfig for StaticMetricsConfig { let metrics = self.metrics.clone(); - Ok(Box::pin( + Ok(crate::sources::source_with_error( StaticMetrics { namespace, metrics, @@ -133,7 +133,7 @@ struct StaticMetrics { } impl StaticMetrics { - async fn run(mut self) -> Result<(), ()> { + async fn run(mut self) -> vector_lib::source_sender::SendResult { let events_received = register!(EventsReceived); let bytes_received = register!(BytesReceived::from(Protocol::STATIC)); let mut interval = @@ -185,9 +185,9 @@ impl StaticMetrics { .into_iter() .map(|metric| metric.with_timestamp(Some(Utc::now()))); - if (self.out.send_batch(batch).await).is_err() { + if let Err(error) = self.out.send_batch(batch).await { emit!(StreamClosedError { count }); - return Err(()); + return Err(error); } } @@ -225,6 +225,28 @@ mod tests { } } + #[tokio::test] + async fn send_error_retains_cause() { + let (out, receiver) = SourceSender::new_test(); + drop(receiver); + + let source = StaticMetrics { + namespace: "static".to_owned(), + metrics: vec![default_metric()], + interval: time::Duration::from_secs(1), + out, + shutdown: ShutdownSignal::noop(), + }; + + let error = crate::sources::source_with_error(source.run()) + .await + .expect_err("sending to a closed output should fail"); + assert_eq!( + error.downcast_ref::(), + Some(&vector_lib::source_sender::SendError::Closed) + ); + } + #[tokio::test] async fn default_empty() { let events = events_from_config(StaticMetricsConfig::default()).await; diff --git a/src/sources/statsd/mod.rs b/src/sources/statsd/mod.rs index 2131aead55903..1b85a2bf2042b 100644 --- a/src/sources/statsd/mod.rs +++ b/src/sources/statsd/mod.rs @@ -6,7 +6,7 @@ use std::{ }; use bytes::Bytes; -use futures::{StreamExt, TryFutureExt}; +use futures::StreamExt; use listenfd::ListenFd; use serde_with::serde_as; use smallvec::{SmallVec, smallvec}; @@ -347,16 +347,17 @@ async fn statsd_udp( config: UdpConfig, shutdown: ShutdownSignal, mut out: SourceSender, -) -> Result<(), ()> { +) -> Result<(), super::SourceError> { let listenfd = ListenFd::from_env(); let socket = try_bind_udp_socket(config.address, listenfd) + .await .map_err(|error| { emit!(SocketBindError { mode: SocketMode::Udp, - error - }) - }) - .await?; + error: &error, + }); + super::SourceError::from(error) + })?; if let Some(receive_buffer_bytes) = config.receive_buffer_bytes && let Err(error) = net::set_receive_buffer_size(&socket, receive_buffer_bytes) diff --git a/src/sources/syslog.rs b/src/sources/syslog.rs index 83c70addd3ae2..ce23716274a83 100644 --- a/src/sources/syslog.rs +++ b/src/sources/syslog.rs @@ -354,7 +354,8 @@ pub fn udp( emit!(SocketBindError { mode: SocketMode::Udp, error: &error, - }) + }); + super::SourceError::from(error) })?; if let Some(receive_buffer_bytes) = receive_buffer_bytes @@ -415,10 +416,10 @@ pub fn udp( debug!("Finished sending."); Ok(()) } - Err(_) => { + Err(error) => { let (count, _) = stream.size_hint(); emit!(StreamClosedError { count }); - Err(()) + Err(error.into()) } } }) @@ -1234,6 +1235,34 @@ mod test { .await; } + #[tokio::test] + async fn test_udp_syslog_send_error_retains_cause() { + let (_guard, in_addr) = next_addr(); + let config = SyslogConfig::from_mode(Mode::Udp { + address: in_addr.into(), + receive_buffer_bytes: None, + }); + + let key = ComponentKey::from("in"); + let (tx, rx) = SourceSender::new_test(); + drop(rx); + let (context, _shutdown) = SourceContext::new_shutdown(&key, tx); + let source = config.build(context).await.unwrap(); + let source = tokio::spawn(source); + + sleep(Duration::from_millis(150)).await; + let socket = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let message = SyslogMessageRfc5424::random(0, 30, 4, 3, 3).to_string(); + socket.send_to(message.as_bytes(), in_addr).await.unwrap(); + + let error = tokio::time::timeout(Duration::from_secs(5), source) + .await + .expect("source error timed out") + .unwrap() + .expect_err("source should fail when its output is closed"); + assert_eq!(error.to_string(), "Sender is closed."); + } + #[tokio::test] async fn test_udp_syslog() { assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async { diff --git a/src/sources/util/framestream.rs b/src/sources/util/framestream.rs index 743f6408df400..f52a4c8c5eb2d 100644 --- a/src/sources/util/framestream.rs +++ b/src/sources/util/framestream.rs @@ -427,7 +427,8 @@ pub fn build_framestream_tcp_source( emit!(SocketBindError { mode: SocketMode::Tcp, error: &error, - }) + }); + crate::sources::SourceError::from(error) })?; info!( @@ -1239,7 +1240,10 @@ mod test { addr: &SocketAddr, frame_handler: impl TcpFrameHandler + Send + Sync + Clone + 'static, pipeline: SourceSender, - ) -> (JoinHandle>, SourceShutdownCoordinator) { + ) -> ( + JoinHandle>, + SourceShutdownCoordinator, + ) { let source_id = ComponentKey::from(source_id); let mut shutdown = SourceShutdownCoordinator::default(); let (shutdown_signal, _) = shutdown.register_source(&source_id, false); @@ -1261,7 +1265,7 @@ mod test { pipeline: SourceSender, ) -> ( PathBuf, - JoinHandle>, + JoinHandle>, SourceShutdownCoordinator, ) { let source_id = ComponentKey::from(source_id); @@ -1386,7 +1390,7 @@ mod test { mut sock_stream: U, rx: V, mut shutdown: SourceShutdownCoordinator, - source_handle: JoinHandle>, + source_handle: JoinHandle>, ) { //1 - send READY frame (with content_type) let content_type = Bytes::from(&b"test_content"[..]); @@ -1441,7 +1445,7 @@ mod test { mut sock_sink: T, mut sock_stream: U, mut shutdown: SourceShutdownCoordinator, - source_handle: JoinHandle>, + source_handle: JoinHandle>, ) { //1 - send READY frame (with content_type) let content_type = Bytes::from(&b"test_content"[..]); diff --git a/src/sources/util/http/prelude.rs b/src/sources/util/http/prelude.rs index 7afddbd9463d7..a4c4e3bad1384 100644 --- a/src/sources/util/http/prelude.rs +++ b/src/sources/util/http/prelude.rs @@ -272,18 +272,14 @@ pub trait HttpSource: Clone + Send + Sync + 'static { let listener = tls .bind(&address) .await - .map_err(|err| { - error!("An error occurred: {err:?}."); - })? + .inspect_err(|err| error!("An error occurred: {err:?}."))? .with_keepalive(keepalive_settings.tcp_keepalive); Server::builder(hyper::server::accept::from_stream(listener.accept_stream())) .serve(make_svc) .with_graceful_shutdown(cx.shutdown.map(|_| ())) .await - .map_err(|err| { - error!("An error occurred: {err:?}."); - })?; + .inspect_err(|err| error!("An error occurred: {err:?}."))?; Ok(()) })) diff --git a/src/sources/util/http_client.rs b/src/sources/util/http_client.rs index b0730ebe04861..a412ea85dcb78 100644 --- a/src/sources/util/http_client.rs +++ b/src/sources/util/http_client.rs @@ -198,7 +198,7 @@ pub(crate) async fn call< context_builder: B, mut out: SourceSender, http_method: HttpMethod, -) -> Result<(), ()> { +) -> vector_lib::source_sender::SendResult { // Building the HttpClient should not fail as it is just setting up the client with the // proxy and tls settings. let client = @@ -341,10 +341,10 @@ pub(crate) async fn call< debug!("Finished sending."); Ok(()) } - Err(_) => { + Err(error) => { let (count, _) = stream.size_hint(); emit!(StreamClosedError { count }); - Err(()) + Err(error) } } } diff --git a/src/sources/util/net/tcp/mod.rs b/src/sources/util/net/tcp/mod.rs index 89edd0675c316..10aff19893d12 100644 --- a/src/sources/util/net/tcp/mod.rs +++ b/src/sources/util/net/tcp/mod.rs @@ -43,6 +43,7 @@ use crate::{ TcpSocketTlsConnectionError, TcpSocketTlsHandshakeTimeout, TcpSourceConnectionClosed, }, net::is_graceful_tls_shutdown, + sources::SourceError, sources::util::{AfterReadExt, LenientFramedRead}, }; @@ -147,7 +148,8 @@ where emit!(SocketBindError { mode: SocketMode::Tcp, error: &error, - }) + }); + SourceError::from(error) })?; info!( diff --git a/src/sources/util/unix_datagram.rs b/src/sources/util/unix_datagram.rs index 49c9970524665..cb657d9da7ae3 100644 --- a/src/sources/util/unix_datagram.rs +++ b/src/sources/util/unix_datagram.rs @@ -66,7 +66,7 @@ async fn listen( mut shutdown: ShutdownSignal, handle_events: impl Fn(&mut [Event], Option) + Clone + Send + Sync + 'static, mut out: SourceSender, -) -> Result<(), ()> { +) -> Result<(), crate::sources::SourceError> { let mut buf = BytesMut::with_capacity(max_length); let bytes_received = register!(BytesReceived::from(Protocol::UNIX)); loop { @@ -78,7 +78,8 @@ async fn listen( emit!(SocketReceiveError { mode: SocketMode::Unix, error: &error - }) + }); + crate::sources::SourceError::from(error) })?; let span = info_span!("datagram"); diff --git a/src/sources/vector/mod.rs b/src/sources/vector/mod.rs index c85e06ba046fa..11498bc49eb2f 100644 --- a/src/sources/vector/mod.rs +++ b/src/sources/vector/mod.rs @@ -233,6 +233,7 @@ impl SourceConfig for VectorConfig { ) .map_err(|error| { error!(message = "Source future failed.", %error); + crate::sources::SourceError::from(error) }); Ok(Box::pin(source)) diff --git a/src/sources/websocket/config.rs b/src/sources/websocket/config.rs index 8214f302eb4ca..a4dd68f7c3b70 100644 --- a/src/sources/websocket/config.rs +++ b/src/sources/websocket/config.rs @@ -172,7 +172,7 @@ impl SourceConfig for WebSocketConfig { let source = WebSocketSource::new(self.clone(), params); - Ok(Box::pin(source.run(cx).map_err(|_err| ()))) + Ok(Box::pin(source.run(cx).map_err(Into::into))) } fn outputs(&self, global_log_namespace: LogNamespace) -> Vec { diff --git a/src/test_util/mock/sinks/oneshot.rs b/src/test_util/mock/sinks/oneshot.rs index 378a9168e6900..f7c804a5b480e 100644 --- a/src/test_util/mock/sinks/oneshot.rs +++ b/src/test_util/mock/sinks/oneshot.rs @@ -63,11 +63,11 @@ impl ValidatedSink for OneshotSinkConfig { let mut guard = self.tx.lock().expect("who cares if the lock is poisoned"); guard.take() }; - let sink = Box::new(OneshotSink { tx }); + let sink = OneshotSink { tx }; let healthcheck = Box::pin(async { Ok(()) }); - Ok((VectorSink::Stream(sink), healthcheck)) + Ok((VectorSink::from_streamsink(sink), healthcheck)) } } diff --git a/src/test_util/mock/sources/basic.rs b/src/test_util/mock/sources/basic.rs index 24a64f30aa04c..cb025d0fc1eea 100644 --- a/src/test_util/mock/sources/basic.rs +++ b/src/test_util/mock/sources/basic.rs @@ -129,7 +129,7 @@ impl SourceConfig for BasicSourceConfig { if let Err(e) = out.send_event(array).await { error!(message = "Error sending in sink..", %e); - return Err(()) + return Err(e.into()) } }, diff --git a/src/test_util/mock/sources/error.rs b/src/test_util/mock/sources/error.rs index e6fef2845e641..05f4029fb060a 100644 --- a/src/test_util/mock/sources/error.rs +++ b/src/test_util/mock/sources/error.rs @@ -1,5 +1,5 @@ use async_trait::async_trait; -use futures_util::{FutureExt, future::err}; +use futures_util::future::err; use vector_lib::{ config::{DataType, LogNamespace, SourceOutput}, configurable::configurable_component, @@ -23,7 +23,7 @@ impl_generate_config_from_default!(ErrorSourceConfig); #[typetag::serde(name = "test_error")] impl SourceConfig for ErrorSourceConfig { async fn build(&self, _cx: SourceContext) -> crate::Result { - Ok(err(()).boxed()) + Ok(vector_lib::source::opaque_source(err(()))) } fn outputs(&self, _global_log_namespace: LogNamespace) -> Vec { diff --git a/src/test_util/mock/sources/tripwire.rs b/src/test_util/mock/sources/tripwire.rs index f5be67c4ef13b..d6676ba4a228d 100644 --- a/src/test_util/mock/sources/tripwire.rs +++ b/src/test_util/mock/sources/tripwire.rs @@ -61,8 +61,10 @@ impl SourceConfig for TripwireSourceConfig { .then(crate::shutdown::tripwire_handler) .boxed(), ) - .map(|_| drop(out)) - .unit_error(), + .map(|_| { + drop(out); + Ok(()) + }), )) } diff --git a/src/topology/builder.rs b/src/topology/builder.rs index f63012fb1c723..7906b044f45bf 100644 --- a/src/topology/builder.rs +++ b/src/topology/builder.rs @@ -459,17 +459,12 @@ impl<'a> Builder<'a> { Ok(e) = &mut pump_error_rx => Err(e), // The source finished normally. - result = server => result.map_err(|()| TaskError::Opaque), + result = server => result.map_err(Into::into), }; - // Even though we already tried to receive any pump task error above, we may have exited - // on the source itself returning an error due to task scheduling, where the pump task - // encountered an error, sent it over the oneshot, but we were polling the source - // already and hit an error trying to send to the now-shutdown pump task. - // - // Since the error from the source is opaque at the moment (i.e. `()`), we try a final - // time to see if the pump task encountered an error, using _that_ instead if so, to - // propagate the true error that caused the source to have to stop. + // The pump may fail while the source is being polled, before its error is received + // above. Prefer the pump error if both tasks fail: a source send error can be a + // consequence of the pump stopping, even when the source reports its cause. if let Ok(e) = pump_error_rx.try_recv() { result = Err(e); } @@ -480,7 +475,7 @@ impl<'a> Builder<'a> { Ok(TaskOutput::Source) } Err(e) => { - debug!("Source finished with an error."); + debug!(error = %e, "Source finished with an error."); Err(e) } } @@ -763,10 +758,8 @@ impl<'a> Builder<'a> { debug!("Sink finished normally."); TaskOutput::Sink(rx) }) - .map_err(|()| { - debug!("Sink finished with an error."); - TaskError::Opaque - }) + .inspect_err(|error| debug!(%error, "Sink finished with an error.")) + .map_err(Into::into) }; let task = Task::new(key.clone(), typetag, sink); @@ -1378,7 +1371,8 @@ impl Runner { self.send_outputs(&mut outputs_buf).await .map_err(TaskError::wrapped)?; } - _ => unreachable!("join error or bad poll"), + Some(Err(error)) => return Err(TaskError::from(error)), + None => unreachable!("bad poll"), } } diff --git a/src/topology/mod.rs b/src/topology/mod.rs index 966371f0a85d7..98f457925e644 100644 --- a/src/topology/mod.rs +++ b/src/topology/mod.rs @@ -66,10 +66,14 @@ async fn handle_errors( AssertUnwindSafe(task) .catch_unwind() .await - .map_err(|_| TaskError::Panicked) + .map_err(|payload| TaskError::panicked(payload.as_ref())) .and_then(|res| res) .map_err(|e| { - error!("An error occurred that Vector couldn't handle: {e}."); + let panic = match &e { + TaskError::Panicked { panic } => Some(panic.as_str()), + _ => None, + }; + error!(panic, "An error occurred that Vector couldn't handle: {e}."); _ = abort_tx.send(error(e.to_string())); e }) diff --git a/src/topology/task.rs b/src/topology/task.rs index bbd6b4e47d93f..f7c15b2d35f02 100644 --- a/src/topology/task.rs +++ b/src/topology/task.rs @@ -1,4 +1,5 @@ use std::{ + any::Any, fmt, future::Future, pin::Pin, @@ -9,7 +10,9 @@ use futures::{FutureExt, future::BoxFuture}; use pin_project::pin_project; use snafu::Snafu; use tokio::task::JoinError; -use vector_lib::{buffers::topology::channel::BufferReceiverStream, event::EventArray}; +use vector_lib::{ + buffers::topology::channel::BufferReceiverStream, event::EventArray, sink::SinkError, +}; use crate::{config::ComponentKey, utilization::Utilization}; @@ -27,9 +30,7 @@ pub(crate) enum TaskError { #[snafu(display("the task was cancelled before it completed"))] Cancelled, #[snafu(display("the task panicked and was aborted"))] - Panicked, - #[snafu(display("the task completed with an error"))] - Opaque, + Panicked { panic: String }, #[snafu(display("{}", source))] Wrapped { source: crate::Error }, } @@ -38,6 +39,28 @@ impl TaskError { pub fn wrapped(e: crate::Error) -> Self { Self::Wrapped { source: e } } + + pub fn panicked(payload: &(dyn Any + Send)) -> Self { + let panic = payload + .downcast_ref::() + .map(String::as_str) + .or_else(|| payload.downcast_ref::<&str>().copied()) + .unwrap_or("non-string panic payload") + .to_owned(); + Self::Panicked { panic } + } +} + +impl From for TaskError { + fn from(error: crate::Error) -> Self { + Self::wrapped(error) + } +} + +impl From for TaskError { + fn from(error: SinkError) -> Self { + Self::wrapped(error.into_inner()) + } } impl From for TaskError { @@ -45,7 +68,7 @@ impl From for TaskError { if e.is_cancelled() { Self::Cancelled } else { - Self::Panicked + Self::panicked(e.into_panic().as_ref()) } } } diff --git a/src/topology/test/crash.rs b/src/topology/test/crash.rs index 2416ae9052e5c..87c5ee6a4bd77 100644 --- a/src/topology/test/crash.rs +++ b/src/topology/test/crash.rs @@ -1,20 +1,185 @@ +use std::net::{TcpListener, UdpSocket}; + use futures_util::StreamExt; -use tokio::time::{Duration, sleep}; +use tokio::time::{Duration, sleep, timeout}; use tokio_stream::wrappers::UnboundedReceiverStream; use crate::{ config::AcknowledgementsConfig, config::Config, + signal::ShutdownError, sinks::socket::SocketSinkConfig, - sources::socket::SocketConfig, + sources::socket::{SocketConfig, udp::UdpConfig}, test_util::{ CountReceiver, addr::next_addr, - mock::{error_sink, error_source, panic_sink, panic_source}, + mock::{basic_sink, error_sink, error_source, panic_sink, panic_source}, random_lines, send_lines, start_topology, trace_init, wait_for_tcp, }, }; +#[cfg(feature = "sinks-prometheus")] +use crate::sinks::prometheus::exporter::PrometheusExporterConfig; +#[cfg(feature = "sinks-websocket_server")] +use crate::sinks::websocket_server::WebSocketListenerSinkConfig; +#[cfg(feature = "sources-statsd")] +use crate::sources::statsd::{StatsdConfig, UdpConfig as StatsdUdpConfig}; +#[cfg(feature = "sources-syslog")] +use crate::sources::syslog::{Mode as SyslogMode, SyslogConfig}; +#[cfg(any(feature = "sinks-prometheus", feature = "sinks-websocket_server"))] +use crate::test_util::mock::basic_source; + +async fn first_shutdown_error(config: Config) -> ShutdownError { + let (topology, mut errors) = start_topology(config, false).await; + let error = timeout(Duration::from_secs(5), errors.recv()) + .await + .expect("component error timed out") + .expect("component error missing"); + topology.stop().await; + error +} + +#[cfg(feature = "sinks-websocket_server")] +#[tokio::test] +async fn test_websocket_sink_bind_error_is_reported() { + trace_init(); + + let (_guard, address) = next_addr(); + let _listener = TcpListener::bind(address).unwrap(); + let expected_error = TcpListener::bind(address).unwrap_err().to_string(); + let (_input, source) = basic_source(); + + let sink = WebSocketListenerSinkConfig { + address, + ..Default::default() + }; + let mut config = Config::builder(); + config.add_source("in", source); + config.add_sink("out", &["in"], sink); + + let error = first_shutdown_error(config.build().unwrap()).await; + assert!(matches!( + error, + ShutdownError::SinkAborted { error, .. } if error.contains(&expected_error) + )); +} + +#[cfg(feature = "sinks-prometheus")] +#[tokio::test] +async fn test_prometheus_exporter_bind_error_is_reported() { + trace_init(); + + let (_guard, address) = next_addr(); + let _listener = TcpListener::bind(address).unwrap(); + let expected_error = TcpListener::bind(address).unwrap_err().to_string(); + let (_input, source) = basic_source(); + + let sink = PrometheusExporterConfig { + address, + ..Default::default() + }; + let mut config = Config::builder(); + config.add_source("in", source); + config.add_sink("out", &["in"], sink); + + let error = first_shutdown_error(config.build().unwrap()).await; + assert!(matches!( + error, + ShutdownError::SinkAborted { error, .. } if error.contains(&expected_error) + )); +} + +#[tokio::test] +async fn test_tcp_socket_bind_error_is_reported() { + trace_init(); + + let (_guard, address) = next_addr(); + let _listener = TcpListener::bind(address).unwrap(); + let expected_error = TcpListener::bind(address).unwrap_err().to_string(); + + let mut config = Config::builder(); + config.add_source("in", SocketConfig::make_basic_tcp_config(address)); + config.add_sink("out", &["in"], basic_sink(1).1); + + let error = first_shutdown_error(config.build().unwrap()).await; + assert!(matches!( + error, + ShutdownError::SourceAborted { error, .. } + if error.contains("TCP bind failed") && error.contains(&expected_error) + )); +} + +#[tokio::test] +async fn test_udp_socket_bind_error_is_reported() { + trace_init(); + + let (_guard, address) = next_addr(); + let _socket = UdpSocket::bind(address).unwrap(); + let expected_error = UdpSocket::bind(address).unwrap_err().to_string(); + + let mut config = Config::builder(); + config.add_source( + "in", + SocketConfig::from(UdpConfig::from_address(address.into())), + ); + config.add_sink("out", &["in"], basic_sink(1).1); + + let error = first_shutdown_error(config.build().unwrap()).await; + assert!(matches!( + error, + ShutdownError::SourceAborted { error, .. } if error == expected_error + )); +} + +#[cfg(feature = "sources-syslog")] +#[tokio::test] +async fn test_udp_syslog_bind_error_is_reported() { + trace_init(); + + let (_guard, address) = next_addr(); + let _socket = UdpSocket::bind(address).unwrap(); + let expected_error = UdpSocket::bind(address).unwrap_err().to_string(); + + let mut config = Config::builder(); + config.add_source( + "in", + SyslogConfig::from_mode(SyslogMode::Udp { + address: address.into(), + receive_buffer_bytes: None, + }), + ); + config.add_sink("out", &["in"], basic_sink(1).1); + + let error = first_shutdown_error(config.build().unwrap()).await; + assert!(matches!( + error, + ShutdownError::SourceAborted { error, .. } if error == expected_error + )); +} + +#[cfg(feature = "sources-statsd")] +#[tokio::test] +async fn test_udp_statsd_bind_error_is_reported() { + trace_init(); + + let (_guard, address) = next_addr(); + let _socket = UdpSocket::bind(address).unwrap(); + let expected_error = UdpSocket::bind(address).unwrap_err().to_string(); + + let mut config = Config::builder(); + config.add_source( + "in", + StatsdConfig::Udp(StatsdUdpConfig::from_address(address.into())), + ); + config.add_sink("out", &["in"], basic_sink(1).1); + + let error = first_shutdown_error(config.build().unwrap()).await; + assert!(matches!( + error, + ShutdownError::SourceAborted { error, .. } if error == expected_error + )); +} + /// Ensures that an unrelated source completing immediately with an error does not prematurely terminate the topology. #[tokio::test] async fn test_source_error() { diff --git a/src/topology/test/mod.rs b/src/topology/test/mod.rs index 853deb43c409b..c3d8ce34f25ec 100644 --- a/src/topology/test/mod.rs +++ b/src/topology/test/mod.rs @@ -44,6 +44,7 @@ mod doesnt_reload; #[cfg(all(feature = "sources-http_server", feature = "sinks-http"))] mod end_to_end; mod latency_metrics; +mod panic; #[cfg(all( feature = "sources-prometheus", feature = "sinks-prometheus", diff --git a/src/topology/test/panic.rs b/src/topology/test/panic.rs new file mode 100644 index 0000000000000..b70d942fa00a1 --- /dev/null +++ b/src/topology/test/panic.rs @@ -0,0 +1,93 @@ +use std::{ + fmt, + sync::{Arc, Mutex}, +}; + +use tokio::sync::mpsc; +use tracing::{ + Event, Subscriber, + field::{Field, Visit}, +}; +use tracing_subscriber::{ + Layer, + layer::{Context, SubscriberExt}, +}; + +use crate::{ + config::ComponentKey, + signal::ShutdownError, + topology::{ + handle_errors, + task::{TaskError, TaskOutput}, + }, +}; + +#[derive(Default)] +struct LogFields { + message: Option, + panic: Option, +} + +impl Visit for LogFields { + fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) { + if field.name() == "message" { + self.message = Some(format!("{value:?}")); + } + } + + fn record_str(&mut self, field: &Field, value: &str) { + if field.name() == "panic" { + self.panic = Some(value.to_owned()); + } + } +} + +struct Capture(Arc>>); + +impl Layer for Capture { + fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { + let mut fields = LogFields::default(); + event.record(&mut fields); + self.0.lock().unwrap().push(fields); + } +} + +#[tokio::test(flavor = "current_thread")] +async fn panic_payload_is_logged_as_a_field() { + let logs = Arc::new(Mutex::new(Vec::new())); + let subscriber = tracing_subscriber::registry().with(Capture(Arc::clone(&logs))); + let _guard = tracing::subscriber::set_default(subscriber); + let (abort_tx, mut abort_rx) = mpsc::unbounded_channel(); + + let task = async { + std::panic::panic_any("source panic value"); + #[allow(unreachable_code)] + Ok(TaskOutput::Source) + }; + let result = handle_errors(task, abort_tx, |error| ShutdownError::SourceAborted { + key: ComponentKey::from("source"), + error, + }) + .await; + + assert!(matches!(result, Err(TaskError::Panicked { .. }))); + let shutdown_error = abort_rx.recv().await.unwrap(); + assert!(!shutdown_error.to_string().contains("source panic value")); + + let logs = logs.lock().unwrap(); + let log = logs.iter().find(|log| log.panic.is_some()).unwrap(); + assert_eq!(log.panic.as_deref(), Some("source panic value")); + assert!(!log.message.as_ref().unwrap().contains("source panic value")); +} + +#[tokio::test(flavor = "current_thread")] +async fn joined_task_panic_retains_payload() { + let error = tokio::spawn(async { std::panic::panic_any(String::from("child panic value")) }) + .await + .unwrap_err(); + let error = TaskError::from(error); + assert!(matches!( + error, + TaskError::Panicked { panic } if panic == "child panic value" + )); +}