chore(observability): emit component_sent events by source and service

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

@@ -1,7 +1,7 @@
 use metrics::{register_counter, Counter};
 use tracing::trace;
 
-use super::{CountByteSize, Output, SharedString};
+use super::{CountByteSize, Output, RegisterEvent, SharedString};
 
 pub const DEFAULT_OUTPUT: &str = "_default";
 
@@ -96,6 +96,19 @@ crate::registered_event!(
     }
 );
 
+/// TODO: This can probably become a part of the previous macro.
+impl RegisterEvent<(Option<String>, Option<String>), TaggedEventsSent> for TaggedEventsSent {
+    fn register(
+        tags: &(Option<String>, Option<String>),
+    ) -> <TaggedEventsSent as super::RegisterInternalEvent>::Handle {
+        super::register(TaggedEventsSent::new(
+            tags.0.clone(),
+            tags.1.clone(),
+            Output(None),
+        ))
+    }
+}
+
 impl TaggedEventsSent {
     #[must_use]
     pub fn new(source: Option<String>, service: Option<String>, output: Output) -> Self {

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

@@ -248,23 +248,38 @@ macro_rules! registered_event {
     };
 }
 
-#[derive(Clone)]
-pub struct Cached<Tags, Event, Register> {
-    cache: Arc<RwLock<BTreeMap<Tags, Event>>>,
-    register: Register,
+pub trait RegisterEvent<Tags, Event>
+where
+    Event: RegisterInternalEvent,
+{
+    fn register(tags: &Tags) -> <Event as RegisterInternalEvent>::Handle;
+}
+
+pub struct Cached<Tags, Event: RegisterInternalEvent> {
+    cache: Arc<RwLock<BTreeMap<Tags, <Event as RegisterInternalEvent>::Handle>>>,
+}
+
+/// Deriving `Clone` for `Cached` doesn't work since the `Event` type is not clone,
+/// we can happily implement our own `clone` however since we are just cloning
+/// the `Arc`.
+impl<Tags, Event: RegisterInternalEvent> Clone for Cached<Tags, Event> {
+    fn clone(&self) -> Self {
+        Self {
+            cache: Arc::clone(&self.cache),
+        }
+    }
 }
 
-impl<Tags, Event, Register, Data> Cached<Tags, Event, Register>
+impl<Tags, Event, EventHandle, Data> Cached<Tags, Event>
 where
     Data: Sized,
-    Register: Fn(&Tags) -> Event,
-    Event: InternalEventHandle<Data = Data>,
+    EventHandle: InternalEventHandle<Data = Data>,
     Tags: Ord + Clone,
+    Event: RegisterInternalEvent<Handle = EventHandle> + RegisterEvent<Tags, Event>,
 {
-    pub fn new(register: Register) -> Self {
+    pub fn new() -> Self {
         Self {
             cache: Arc::new(RwLock::new(BTreeMap::new())),
-            register,
         }
     }
 
@@ -273,7 +288,7 @@ where
         if let Some(event) = read.get(tags) {
             event.emit(value);
         } else {
-            let event = (self.register)(tags);
+            let event = <Event as RegisterEvent<Tags, Event>>::register(tags);
             event.emit(value);
 
             // Ensure the read lock is dropped so we can write.

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

@@ -5,8 +5,8 @@ use tokio::{pin, select};
 use tower::Service;
 use tracing::Instrument;
 use vector_common::internal_event::{
-    register, ByteSize, BytesSent, Cached, CallError, InternalEventHandle as _, Output,
-    PollReadyError, Registered, SharedString, TaggedEventsSent,
+    register, ByteSize, BytesSent, Cached, CallError, InternalEventHandle as _, PollReadyError,
+    Registered, SharedString, TaggedEventsSent,
 };
 use vector_common::request_metadata::{MetaDescriptive, RequestCountByteSize, RequestMetadata};
 
@@ -99,13 +99,7 @@ where
         pin!(batched_input);
 
         let bytes_sent = protocol.map(|protocol| register(BytesSent { protocol }));
-        let events_sent = Cached::new(|tags: &(Option<String>, Option<String>)| {
-            register(TaggedEventsSent::new(
-                tags.0.clone(),
-                tags.1.clone(),
-                Output(None),
-            ))
-        });
+        let events_sent = Cached::new();
 
         loop {
             // Core behavior of the loop:
@@ -205,16 +199,14 @@ where
         Ok(())
     }
 
-    fn handle_response<T>(
+    fn handle_response(
         result: Result<Svc::Response, Svc::Error>,
         request_id: usize,
         finalizers: EventFinalizers,
         metadata: &RequestMetadata,
         bytes_sent: &Option<Registered<BytesSent>>,
-        events_sent: &Cached<(Option<String>, Option<String>), Registered<TaggedEventsSent>, T>,
-    ) where
-        T: Fn(&(Option<String>, Option<String>)) -> Registered<TaggedEventsSent>,
-    {
+        events_sent: &Cached<(Option<String>, Option<String>), TaggedEventsSent>,
+    ) {
         match result {
             Err(error) => {
                 Self::emit_call_error(Some(error), request_id, metadata.event_count());