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

docs/tutorials/sinks/2_http_sink.md

@@ -387,9 +387,9 @@ impl DriverResponse for BasicResponse {
         EventStatus::Delivered
     }
 
-    fn events_sent(&self) -> CountByteSize {
+    fn events_sent(&self) -> RequestCountByteSize {
         // (events count, byte size)
-        CountByteSize(1, self.byte_size)
+        CountByteSize(1, self.byte_size).into()
     }
 }
 ```

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

@@ -44,3 +44,65 @@ impl From<Output> for EventsSent {
         Self { output: output.0 }
     }
 }
+
+crate::registered_event!(
+    TaggedEventsSent {
+        output: Option<SharedString>,
+        source: Option<String>,
+        service: Option<String>,
+    } => {
+        events: Counter = if let Some(output) = &self.output {
+            register_counter!("component_sent_events_total", "output" => output.clone(),
+                "source" => self.source.clone().unwrap_or("-".to_string()),
+                "service" => self.service.clone().unwrap_or("-".to_string()))
+        } else {
+            register_counter!("component_sent_events_total",
+                "source" => self.source.clone().unwrap_or("-".to_string()),
+                "service" => self.service.clone().unwrap_or("-".to_string()))
+        },
+        events_out: Counter = if let Some(output) = &self.output {
+            register_counter!("events_out_total", "output" => output.clone())
+        } else {
+            register_counter!("events_out_total")
+        },
+        event_bytes: Counter = if let Some(output) = &self.output {
+            register_counter!("component_sent_event_bytes_total",
+                "output" => output.clone(),
+                "source" => self.source.clone().unwrap_or("-".to_string()),
+                "service" => self.service.clone().unwrap_or("-".to_string()))
+        } else {
+            register_counter!("component_sent_event_bytes_total",
+                "source" => self.source.clone().unwrap_or("-".to_string()),
+                "service" => self.service.clone().unwrap_or("-".to_string()))
+        },
+        output: Option<SharedString> = self.output,
+    }
+
+    fn emit(&self, data: CountByteSize) {
+        let CountByteSize(count, byte_size) = data;
+
+        match &self.output {
+            Some(output) => {
+                trace!(message = "Events sent.", count = %count, byte_size = %byte_size, output = %output);
+            }
+            None => {
+                trace!(message = "Events sent.", count = %count, byte_size = %byte_size);
+            }
+        }
+
+        self.events.increment(count as u64);
+        self.events_out.increment(count as u64);
+        self.event_bytes.increment(byte_size as u64);
+    }
+);
+
+impl TaggedEventsSent {
+    #[must_use]
+    pub fn new(source: Option<String>, service: Option<String>, output: Output) -> Self {
+        Self {
+            output: output.0,
+            source: source.map(|source| source.to_string()),
+            service: service.map(|service| service.to_string()),
+        }
+    }
+}

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

@@ -6,13 +6,19 @@ mod events_sent;
 mod prelude;
 pub mod service;
 
+use std::{
+    collections::BTreeMap,
+    ops::{Add, AddAssign},
+    sync::{Arc, RwLock},
+};
+
 pub use metrics::SharedString;
 
 pub use bytes_received::BytesReceived;
 pub use bytes_sent::BytesSent;
 pub use component_events_dropped::{ComponentEventsDropped, INTENTIONAL, UNINTENTIONAL};
 pub use events_received::EventsReceived;
-pub use events_sent::{EventsSent, DEFAULT_OUTPUT};
+pub use events_sent::{EventsSent, TaggedEventsSent, DEFAULT_OUTPUT};
 pub use prelude::{error_stage, error_type};
 pub use service::{CallError, PollReadyError};
 
@@ -107,9 +113,24 @@ pub struct ByteSize(pub usize);
 pub struct Count(pub usize);
 
 /// Holds the tuple `(count_of_events, size_of_events_in_bytes)`.
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, Debug)]
 pub struct CountByteSize(pub usize, pub usize);
 
+impl AddAssign for CountByteSize {
+    fn add_assign(&mut self, rhs: Self) {
+        self.0 += rhs.0;
+        self.1 += rhs.1;
+    }
+}
+
+impl Add<CountByteSize> for CountByteSize {
+    type Output = CountByteSize;
+
+    fn add(self, rhs: CountByteSize) -> Self::Output {
+        CountByteSize(self.0 + rhs.0, self.1 + rhs.1)
+    }
+}
+
 // Wrapper types used to hold parameters for registering events
 
 pub struct Output(pub Option<SharedString>);
@@ -224,3 +245,38 @@ macro_rules! registered_event {
         }
     };
 }
+
+#[derive(Clone)]
+pub struct Cached<Tags, Event, Register> {
+    cache: Arc<RwLock<BTreeMap<Tags, Event>>>,
+    register: Register,
+}
+
+impl<Tags, Event, Register, Data> Cached<Tags, Event, Register>
+where
+    Data: Sized,
+    Register: Fn(&Tags) -> Event,
+    Event: InternalEventHandle<Data = Data>,
+    Tags: Ord + Clone,
+{
+    pub fn new(register: Register) -> Self {
+        Self {
+            cache: Arc::new(RwLock::new(BTreeMap::new())),
+            register,
+        }
+    }
+
+    pub fn emit(&self, tags: &Tags, value: Data) {
+        let read = self.cache.read().unwrap();
+        if let Some(event) = read.get(tags) {
+            event.emit(value);
+        } else {
+            let event = (self.register)(tags);
+            event.emit(value);
+
+            // Ensure the read lock is dropped so we can write.
+            drop(read);
+            self.cache.write().unwrap().insert(tags.clone(), event);
+        }
+    }
+}

lib/vector-common/src/request_metadata.rs

@@ -1,14 +1,79 @@
+use std::collections::HashMap;
 use std::ops::Add;
 
+use crate::internal_event::CountByteSize;
+
+pub type EventCountTags = (Option<String>, Option<String>);
+
+pub trait GetEventCountTags {
+    fn get_tags(&self) -> EventCountTags;
+}
+
+/// Struct that keeps track of the estimated json size of a given
+/// batch of events by source and service.
+#[derive(Clone, Debug, Default)]
+pub struct RequestCountByteSize {
+    sizes: HashMap<EventCountTags, CountByteSize>,
+}
+
+impl RequestCountByteSize {
+    pub fn sizes(&self) -> &HashMap<EventCountTags, CountByteSize> {
+        &self.sizes
+    }
+
+    pub fn add_event<E>(&mut self, event: &E, json_size: usize)
+    where
+        E: GetEventCountTags,
+    {
+        let size = CountByteSize(1, json_size);
+        let tags = event.get_tags();
+
+        match self.sizes.get_mut(&tags) {
+            Some(current) => {
+                *current += size;
+            }
+            None => {
+                self.sizes.insert(tags, size);
+            }
+        }
+    }
+}
+
+impl From<CountByteSize> for RequestCountByteSize {
+    fn from(value: CountByteSize) -> Self {
+        let mut sizes = HashMap::new();
+        sizes.insert((None, None), value);
+
+        Self { sizes }
+    }
+}
+
+impl<'a> Add<&'a RequestCountByteSize> for RequestCountByteSize {
+    type Output = RequestCountByteSize;
+
+    fn add(mut self, other: &'a Self::Output) -> Self::Output {
+        for (key, value) in &other.sizes {
+            match self.sizes.get_mut(&key) {
+                Some(size) => *size += *value,
+                None => {
+                    self.sizes.insert(key.clone(), *value);
+                }
+            }
+        }
+
+        Self { sizes: self.sizes }
+    }
+}
+
 /// Metadata for batch requests.
-#[derive(Clone, Copy, Debug, Default)]
+#[derive(Clone, Debug, Default)]
 pub struct RequestMetadata {
     /// Number of events represented by this batch request.
     event_count: usize,
     /// Size, in bytes, of the in-memory representation of all events in this batch request.
     events_byte_size: usize,
     /// Size, in bytes, of the estimated JSON-encoded representation of all events in this batch request.
-    events_estimated_json_encoded_byte_size: usize,
+    events_estimated_json_encoded_byte_size: RequestCountByteSize,
     /// Uncompressed size, in bytes, of the encoded events in this batch request.
     request_encoded_size: usize,
     /// On-the-wire size, in bytes, of the batch request itself after compression, etc.
@@ -25,7 +90,7 @@ impl RequestMetadata {
         events_byte_size: usize,
         request_encoded_size: usize,
         request_wire_size: usize,
-        events_estimated_json_encoded_byte_size: usize,
+        events_estimated_json_encoded_byte_size: RequestCountByteSize,
     ) -> Self {
         Self {
             event_count,
@@ -47,8 +112,8 @@ impl RequestMetadata {
     }
 
     #[must_use]
-    pub const fn events_estimated_json_encoded_byte_size(&self) -> usize {
-        self.events_estimated_json_encoded_byte_size
+    pub fn events_estimated_json_encoded_byte_size(&self) -> &RequestCountByteSize {
+        &self.events_estimated_json_encoded_byte_size
     }
 
     #[must_use]
@@ -64,7 +129,7 @@ impl RequestMetadata {
     /// Constructs a `RequestMetadata` by summation of the "batch" of `RequestMetadata` provided.
     #[must_use]
     pub fn from_batch<T: IntoIterator<Item = RequestMetadata>>(metadata_iter: T) -> Self {
-        let mut metadata_sum = RequestMetadata::new(0, 0, 0, 0, 0);
+        let mut metadata_sum = RequestMetadata::new(0, 0, 0, 0, Default::default());
 
         for metadata in metadata_iter {
             metadata_sum = metadata_sum + &metadata;
@@ -82,7 +147,7 @@ impl<'a> Add<&'a RequestMetadata> for RequestMetadata {
             event_count: self.event_count + other.event_count,
             events_byte_size: self.events_byte_size + other.events_byte_size,
             events_estimated_json_encoded_byte_size: self.events_estimated_json_encoded_byte_size
-                + other.events_estimated_json_encoded_byte_size,
+                + &other.events_estimated_json_encoded_byte_size,
             request_encoded_size: self.request_encoded_size + other.request_encoded_size,
             request_wire_size: self.request_wire_size + other.request_wire_size,
         }
@@ -92,5 +157,5 @@ impl<'a> Add<&'a RequestMetadata> for RequestMetadata {
 /// Objects implementing this trait have metadata that describes the request.
 pub trait MetaDescriptive {
     /// Returns the `RequestMetadata` associated with this object.
-    fn get_metadata(&self) -> RequestMetadata;
+    fn get_metadata(&self) -> &RequestMetadata;
 }

lib/vector-core/src/event/metric/mod.rs

@@ -11,7 +11,10 @@ use std::{
 };
 
 use chrono::{DateTime, Utc};
-use vector_common::EventDataEq;
+use vector_common::{
+    request_metadata::{EventCountTags, GetEventCountTags},
+    EventDataEq,
+};
 use vector_config::configurable_component;
 
 use crate::{
@@ -476,6 +479,15 @@ impl Finalizable for Metric {
     }
 }
 
+impl GetEventCountTags for Metric {
+    fn get_tags(&self) -> EventCountTags {
+        let source = self.metadata().source_id().map(|output| output.to_string());
+        // Metrics do not contain a service field.
+
+        (source, None)
+    }
+}
+
 /// Metric kind.
 ///
 /// Metrics can be either absolute of incremental. Absolute metrics represent a sort of "last write wins" scenario,

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

@@ -19,7 +19,11 @@ pub use r#ref::{EventMutRef, EventRef};
 use serde::{Deserialize, Serialize};
 pub use trace::TraceEvent;
 use vector_buffers::EventCount;
-use vector_common::{finalization, EventDataEq};
+use vector_common::{
+    finalization,
+    request_metadata::{EventCountTags, GetEventCountTags},
+    EventDataEq,
+};
 pub use vrl::value::Value;
 #[cfg(feature = "vrl")]
 pub use vrl_target::{TargetEvents, VrlTarget};
@@ -90,6 +94,20 @@ impl Finalizable for Event {
     }
 }
 
+impl GetEventCountTags for Event {
+    fn get_tags(&self) -> EventCountTags {
+        let source = self.metadata().source_id().map(|id| id.to_string());
+        let service = if let Event::Log(log) = self {
+            log.get_by_meaning("service")
+                .map(|service| service.to_string())
+        } else {
+            None
+        };
+
+        (source, service)
+    }
+}
+
 impl Event {
     /// Return self as a `LogEvent`
     ///