checkpoint

Signed-off-by: Jean Mertz <[email protected]>

Author: JeanMertz

src/sinks/console/config.rs

@@ -60,7 +60,8 @@ impl GenerateConfig for ConsoleSinkConfig {
 
 #[async_trait::async_trait]
 impl SinkConfig for ConsoleSinkConfig {
-    async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
+    async fn build(&self, ctx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
+        let sources_details = ctx.sources_details.clone();
         let transformer = self.encoding.transformer();
         let (framer, serializer) = self.encoding.build(SinkType::StreamBased)?;
         let encoder = Encoder::<Framer>::new(framer, serializer);
@@ -70,11 +71,13 @@ impl SinkConfig for ConsoleSinkConfig {
                 output: io::stdout(),
                 transformer,
                 encoder,
+                sources_details,
             }),
             Target::Stderr => VectorSink::from_event_streamsink(WriterSink {
                 output: io::stderr(),
                 transformer,
                 encoder,
+                sources_details,
             }),
         };
 

src/sinks/console/sink.rs

@@ -4,6 +4,7 @@ use codecs::encoding::Framer;
 use futures::{stream::BoxStream, StreamExt};
 use tokio::{io, io::AsyncWriteExt};
 use tokio_util::codec::Encoder as _;
+use vector_common::config::SourceDetails;
 use vector_core::{
     internal_event::{ByteSize, BytesSent, EventsSent, InternalEventHandle as _, Protocol},
     EstimatedJsonEncodedSizeOf,
@@ -19,6 +20,7 @@ pub struct WriterSink<T> {
     pub output: T,
     pub transformer: Transformer,
     pub encoder: Encoder<Framer>,
+    pub sources_details: Vec<SourceDetails>,
 }
 
 #[async_trait]
@@ -30,6 +32,11 @@ where
         let bytes_sent = register!(BytesSent::from(Protocol("console".into(),)));
         while let Some(mut event) = input.next().await {
             let event_byte_size = event.estimated_json_encoded_size_of();
+            let source = event
+                .metadata()
+                .source_id()
+                .and_then(|id| self.sources_details.get(id).map(|details| details.key.id()));
+
             self.transformer.transform(&mut event);
 
             let finalizers = event.take_finalizers();
@@ -54,7 +61,7 @@ where
                         byte_size: event_byte_size,
                         count: 1,
                         output: None,
-                        source: None,
+                        source,
                     });
                     bytes_sent.emit(ByteSize(bytes.len()));
                 }