enhancement(internal_events): add source identifier to specific internal events

lib/vector-common/src/internal_event/events_sent.rs

@@ -10,25 +10,28 @@ pub struct EventsSent<'a> {
     pub count: usize,
     pub byte_size: usize,
     pub output: Option<&'a str>,
+    pub source: Option<&'a str>,
 }
 
 impl<'a> InternalEvent for EventsSent<'a> {
     fn emit(self) {
+        let source = self.source.unwrap_or("UNKNOWN");
+
         if let Some(output) = self.output {
-            trace!(message = "Events sent.", count = %self.count, byte_size = %self.byte_size, output = %output);
+            trace!(message = "Events sent.", count = %self.count, byte_size = %self.byte_size, source = %source, output = %output);
         } else {
-            trace!(message = "Events sent.", count = %self.count, byte_size = %self.byte_size);
+            trace!(message = "Events sent.", count = %self.count, byte_size = %self.byte_size, source = %source);
         }
 
         if self.count > 0 {
             if let Some(output) = self.output {
-                counter!("component_sent_events_total", self.count as u64, "output" => output.to_owned());
-                counter!("events_out_total", self.count as u64, "output" => output.to_owned());
-                counter!("component_sent_event_bytes_total", self.byte_size as u64, "output" => output.to_owned());
+                counter!("component_sent_events_total", self.count as u64, "source" => source.to_owned(), "output" => output.to_owned());
+                counter!("events_out_total", self.count as u64, "source" => source.to_owned(), "output" => output.to_owned());
+                counter!("component_sent_event_bytes_total", self.byte_size as u64, "source" => source.to_owned(), "output" => output.to_owned());
             } else {
-                counter!("component_sent_events_total", self.count as u64);
-                counter!("events_out_total", self.count as u64);
-                counter!("component_sent_event_bytes_total", self.byte_size as u64);
+                counter!("component_sent_events_total", self.count as u64, "source" => source.to_owned());
+                counter!("events_out_total", self.count as u64, "source" => source.to_owned());
+                counter!("component_sent_event_bytes_total", self.byte_size as u64, "source" => source.to_owned());
             }
         }
     }

lib/vector-core/src/event/metadata.rs

@@ -35,6 +35,14 @@ pub struct EventMetadata {
     /// TODO(Jean): must not skip serialization to track schemas across restarts.
     #[serde(default = "default_schema_definition", skip)]
     schema_definition: Arc<schema::Definition>,
+
+    /// A unique identifier of the originating source of this event.
+    ///
+    /// Can be used internally to refer back to source details such as its component key.
+    ///
+    /// If `None`, then the event has no originating source (e.g. it was created internally, such
+    /// as in the Lua or Remap transforms).
+    source_id: Option<usize>,
 }
 
 fn default_metadata_value() -> Value {
@@ -98,6 +106,7 @@ impl Default for EventMetadata {
             secrets: Secrets::new(),
             finalizers: Default::default(),
             schema_definition: default_schema_definition(),
+            source_id: None,
         }
     }
 }
@@ -202,6 +211,16 @@ impl EventMetadata {
     pub fn set_schema_definition(&mut self, definition: &Arc<schema::Definition>) {
         self.schema_definition = Arc::clone(definition);
     }
+
+    /// set the source ID.
+    pub fn set_source_id(&mut self, source_id: usize) {
+        self.source_id = Some(source_id);
+    }
+
+    /// Get the source ID.
+    pub fn source_id(&self) -> Option<usize> {
+        self.source_id
+    }
 }
 
 impl EventDataEq for EventMetadata {

lib/vector-core/src/event/ref.rs

@@ -118,6 +118,15 @@ pub enum EventMutRef<'a> {
 }
 
 impl<'a> EventMutRef<'a> {
+    /// Convert an `EventMutRef` to an `EventRef`.
+    pub fn as_event_ref(&'a self) -> EventRef<'a> {
+        match self {
+            EventMutRef::Log(v) => EventRef::Log(v),
+            EventMutRef::Metric(v) => EventRef::Metric(v),
+            EventMutRef::Trace(v) => EventRef::Trace(v),
+        }
+    }
+
     /// Extract the `LogEvent` reference in this.
     ///
     /// # Panics

lib/vector-core/src/stream/driver.rs

@@ -192,6 +192,7 @@ where
                         count: cbs.0,
                         byte_size: cbs.1,
                         output: None,
+                        source: None,
                     });
 
                 // This condition occurs specifically when the `HttpBatchService::call()` is called *within* the `Service::call()`

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

@@ -281,6 +281,7 @@ impl TransformOutputs {
                 count,
                 byte_size,
                 output: Some(DEFAULT_OUTPUT),
+                source: None,
             });
         }
 
@@ -293,6 +294,7 @@ impl TransformOutputs {
                 count,
                 byte_size,
                 output: Some(key.as_ref()),
+                source: None,
             });
         }
 

src/config/sink.rs

@@ -9,7 +9,10 @@ use vector_core::{
 };
 
 use super::{id::Inputs, schema, ComponentKey, ProxyConfig, Resource};
-use crate::sinks::{util::UriSerde, Healthcheck, Sinks};
+use crate::{
+    sinks::{util::UriSerde, Healthcheck, Sinks},
+    topology::builder::SourceDetails,
+};
 
 /// Fully resolved sink component.
 #[configurable_component]
@@ -211,6 +214,7 @@ pub struct SinkContext {
     pub globals: GlobalOptions,
     pub proxy: ProxyConfig,
     pub schema: schema::Options,
+    pub sources_details: Vec<SourceDetails>,
 }
 
 impl SinkContext {
@@ -221,6 +225,7 @@ impl SinkContext {
             globals: GlobalOptions::default(),
             proxy: ProxyConfig::default(),
             schema: schema::Options::default(),
+            sources_details: vec![],
         }
     }
 
@@ -231,4 +236,8 @@ impl SinkContext {
     pub const fn proxy(&self) -> &ProxyConfig {
         &self.proxy
     }
+
+    pub fn source_details(&self, id: usize) -> Option<&SourceDetails> {
+        self.sources_details.get(id)
+    }
 }

src/sinks/blackhole/sink.rs

@@ -96,6 +96,7 @@ impl StreamSink<EventArray> for BlackholeSink {
                 count: events.len(),
                 byte_size: message_len,
                 output: None,
+                source: None,
             });
 
             emit!(BytesSent {

src/sinks/console/sink.rs

@@ -54,6 +54,7 @@ where
                         byte_size: event_byte_size,
                         count: 1,
                         output: None,
+                        source: None,
                     });
                     bytes_sent.emit(ByteSize(bytes.len()));
                 }

src/sinks/file/mod.rs

@@ -28,6 +28,7 @@ use crate::{
     internal_events::{FileBytesSent, FileIoError, FileOpen, TemplateRenderingError},
     sinks::util::StreamSink,
     template::Template,
+    topology::builder::SourceDetails,
 };
 mod bytes_path;
 use std::convert::TryFrom;
@@ -152,9 +153,9 @@ impl OutFile {
 impl SinkConfig for FileSinkConfig {
     async fn build(
         &self,
-        _cx: SinkContext,
+        ctx: SinkContext,
     ) -> crate::Result<(super::VectorSink, super::Healthcheck)> {
-        let sink = FileSink::new(self)?;
+        let sink = FileSink::new(self, ctx.sources_details.clone())?;
         Ok((
             super::VectorSink::from_event_streamsink(sink),
             future::ok(()).boxed(),
@@ -178,10 +179,14 @@ pub struct FileSink {
     idle_timeout: Duration,
     files: ExpiringHashMap<Bytes, OutFile>,
     compression: Compression,
+    sources_details: Vec<SourceDetails>,
 }
 
 impl FileSink {
-    pub fn new(config: &FileSinkConfig) -> crate::Result<Self> {
+    pub fn new(
+        config: &FileSinkConfig,
+        sources_details: Vec<SourceDetails>,
+    ) -> crate::Result<Self> {
         let transformer = config.encoding.transformer();
         let (framer, serializer) = config.encoding.build(SinkType::StreamBased)?;
         let encoder = Encoder::<Framer>::new(framer, serializer);
@@ -193,6 +198,7 @@ impl FileSink {
             idle_timeout: Duration::from_secs(config.idle_timeout_secs.unwrap_or(30)),
             files: ExpiringHashMap::default(),
             compression: config.compression,
+            sources_details,
         })
     }
 
@@ -335,13 +341,19 @@ impl FileSink {
         trace!(message = "Writing an event to file.", path = ?path);
         let event_size = event.estimated_json_encoded_size_of();
         let finalizers = event.take_finalizers();
+        let source = event
+            .metadata()
+            .source_id()
+            .and_then(|id| self.sources_details.get(id).map(|details| details.key.id()));
+
         match write_event_to_file(file, event, &self.transformer, &mut self.encoder).await {
             Ok(byte_size) => {
                 finalizers.update_status(EventStatus::Delivered);
                 emit!(EventsSent {
                     count: 1,
                     byte_size: event_size,
                     output: None,
+                    source,
                 });
                 emit!(FileBytesSent {
                     byte_size,
@@ -439,7 +451,7 @@ mod tests {
             acknowledgements: Default::default(),
         };
 
-        let sink = FileSink::new(&config).unwrap();
+        let sink = FileSink::new(&config, vec![]).unwrap();
         let (input, _events) = random_lines_with_stream(100, 64, None);
 
         let events = Box::pin(stream::iter(
@@ -475,7 +487,7 @@ mod tests {
             acknowledgements: Default::default(),
         };
 
-        let sink = FileSink::new(&config).unwrap();
+        let sink = FileSink::new(&config, vec![]).unwrap();
         let (input, _) = random_lines_with_stream(100, 64, None);
 
         let events = Box::pin(stream::iter(
@@ -511,7 +523,7 @@ mod tests {
             acknowledgements: Default::default(),
         };
 
-        let sink = FileSink::new(&config).unwrap();
+        let sink = FileSink::new(&config, vec![]).unwrap();
         let (input, _) = random_lines_with_stream(100, 64, None);
 
         let events = Box::pin(stream::iter(
@@ -552,7 +564,7 @@ mod tests {
             acknowledgements: Default::default(),
         };
 
-        let sink = FileSink::new(&config).unwrap();
+        let sink = FileSink::new(&config, vec![]).unwrap();
 
         let (mut input, _events) = random_events_with_stream(32, 8, None);
         input[0].as_mut_log().insert("date", "2019-26-07");
@@ -637,7 +649,7 @@ mod tests {
             acknowledgements: Default::default(),
         };
 
-        let sink = FileSink::new(&config).unwrap();
+        let sink = FileSink::new(&config, vec![]).unwrap();
         let (mut input, _events) = random_lines_with_stream(10, 64, None);
 
         let (mut tx, rx) = futures::channel::mpsc::channel(0);

src/sinks/nats.rs

@@ -20,6 +20,7 @@ use crate::{
     sinks::util::StreamSink,
     template::{Template, TemplateParseError},
     tls::TlsEnableableConfig,
+    topology::builder::SourceDetails,
 };
 
 #[derive(Debug, Snafu)]
@@ -99,9 +100,9 @@ impl GenerateConfig for NatsSinkConfig {
 impl SinkConfig for NatsSinkConfig {
     async fn build(
         &self,
-        _cx: SinkContext,
+        ctx: SinkContext,
     ) -> crate::Result<(super::VectorSink, super::Healthcheck)> {
-        let sink = NatsSink::new(self.clone()).await?;
+        let sink = NatsSink::new(self.clone(), ctx).await?;
         let healthcheck = healthcheck(self.clone()).boxed();
         Ok((super::VectorSink::from_event_streamsink(sink), healthcheck))
     }
@@ -140,20 +141,23 @@ pub struct NatsSink {
     encoder: Encoder<()>,
     connection: nats::asynk::Connection,
     subject: Template,
+    sources_details: Vec<SourceDetails>,
 }
 
 impl NatsSink {
-    async fn new(config: NatsSinkConfig) -> Result<Self, BuildError> {
+    async fn new(config: NatsSinkConfig, ctx: SinkContext) -> Result<Self, BuildError> {
         let connection = config.connect().await?;
         let transformer = config.encoding.transformer();
         let serializer = config.encoding.build().context(EncodingSnafu)?;
         let encoder = Encoder::<()>::new(serializer);
+        let sources_details = ctx.sources_details.clone();
 
         Ok(NatsSink {
             connection,
             transformer,
             encoder,
             subject: Template::try_from(config.subject).context(SubjectTemplateSnafu)?,
+            sources_details,
         })
     }
 }
@@ -182,6 +186,10 @@ impl StreamSink<Event> for NatsSink {
             self.transformer.transform(&mut event);
 
             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()));
 
             let mut bytes = BytesMut::new();
             if self.encoder.encode(event, &mut bytes).is_err() {
@@ -202,7 +210,8 @@ impl StreamSink<Event> for NatsSink {
                     emit!(EventsSent {
                         byte_size: event_byte_size,
                         count: 1,
-                        output: None
+                        output: None,
+                        source,
                     });
                     bytes_sent.emit(ByteSize(bytes.len()));
                 }

src/sinks/prometheus/exporter.rs

@@ -422,7 +422,8 @@ fn handle(
             emit!(EventsSent {
                 count,
                 byte_size,
-                output: None
+                output: None,
+                source: None,
             });
         }
 

src/sinks/pulsar.rs

@@ -373,6 +373,7 @@ impl Sink<Event> for PulsarSink {
                         count: metadata.event_count(),
                         byte_size: metadata.events_estimated_json_encoded_byte_size(),
                         output: None,
+                        source: None,
                     });
 
                     this.bytes_sent

src/sinks/util/sink.rs

@@ -451,7 +451,8 @@ where
                         emit!(EventsSent {
                             count,
                             byte_size,
-                            output: None
+                            output: None,
+                            source: None,
                         });
                         // TODO: Emit a BytesSent event here too
                     }

src/sinks/websocket/config.rs

@@ -65,9 +65,9 @@ impl GenerateConfig for WebSocketSinkConfig {
 
 #[async_trait::async_trait]
 impl SinkConfig for WebSocketSinkConfig {
-    async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
+    async fn build(&self, ctx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
         let connector = self.build_connector()?;
-        let ws_sink = WebSocketSink::new(self, connector.clone())?;
+        let ws_sink = WebSocketSink::new(self, connector.clone(), ctx)?;
 
         Ok((
             VectorSink::from_event_streamsink(ws_sink),