ehanchement(internal_metrics): reuse BytesReceived event for intern…

…al metrics (#20977) Instead of having a completely separate struct for internal metrics bytes received event, we can reuse the common `BytesReceived` event, defining a new `Protocol` for it. Similar is done for static metrics. Related: #20899
vectordotdev · Jul 31, 2024 · 70e61fc · 70e61fc
1 parent 789848f
commit 70e61fc
Show file tree

Hide file tree

Showing 4 changed files with 7 additions and 29 deletions.
diff --git a/lib/vector-common/src/internal_event/mod.rs b/lib/vector-common/src/internal_event/mod.rs
@@ -147,6 +147,7 @@ impl Protocol {
  pub const TCP: Protocol = Protocol(SharedString::const_str("tcp"));
  pub const UDP: Protocol = Protocol(SharedString::const_str("udp"));
  pub const UNIX: Protocol = Protocol(SharedString::const_str("unix"));
+ pub const INTERNAL: Protocol = Protocol(SharedString::const_str("internal"));
  pub const STATIC: Protocol = Protocol(SharedString::const_str("static"));
 }
 

diff --git a/src/internal_events/internal_metrics.rs b/src/internal_events/internal_metrics.rs
diff --git a/src/internal_events/mod.rs b/src/internal_events/mod.rs
@@ -70,8 +70,6 @@ mod http_client_source;
 mod influxdb;
 #[cfg(feature = "sources-internal_logs")]
 mod internal_logs;
-#[cfg(feature = "sources-internal_metrics")]
-mod internal_metrics;
 #[cfg(all(unix, feature = "sources-journald"))]
 mod journald;
 #[cfg(any(feature = "sources-kafka", feature = "sinks-kafka"))]
@@ -209,8 +207,6 @@ pub(crate) use self::http_client_source::*;
 pub(crate) use self::influxdb::*;
 #[cfg(feature = "sources-internal_logs")]
 pub(crate) use self::internal_logs::*;
-#[cfg(feature = "sources-internal_metrics")]
-pub(crate) use self::internal_metrics::*;
 #[cfg(all(unix, feature = "sources-journald"))]
 pub(crate) use self::journald::*;
 #[cfg(any(feature = "sources-kafka", feature = "sinks-kafka"))]

diff --git a/src/sources/internal_metrics.rs b/src/sources/internal_metrics.rs
@@ -5,13 +5,15 @@ use serde_with::serde_as;
 use tokio::time;
 use tokio_stream::wrappers::IntervalStream;
 use vector_lib::configurable::configurable_component;
-use vector_lib::internal_event::{CountByteSize, InternalEventHandle as _};
+use vector_lib::internal_event::{
+ ByteSize, BytesReceived, CountByteSize, InternalEventHandle as _, Protocol,
+};
 use vector_lib::lookup::lookup_v2::OptionalValuePath;
 use vector_lib::{config::LogNamespace, ByteSizeOf, EstimatedJsonEncodedSizeOf};
 
 use crate::{
  config::{log_schema, SourceConfig, SourceContext, SourceOutput},
- internal_events::{EventsReceived, InternalMetricsBytesReceived, StreamClosedError},
+ internal_events::{EventsReceived, StreamClosedError},
  metrics::Controller,
  shutdown::ShutdownSignal,
  SourceSender,
@@ -156,6 +158,7 @@ struct InternalMetrics<'a> {
 impl<'a> InternalMetrics<'a> {
  async fn run(mut self) -> Result<(), ()> {
  let events_received = register!(EventsReceived);
+ let bytes_received = register!(BytesReceived::from(Protocol::INTERNAL));
  let mut interval =
  IntervalStream::new(time::interval(self.interval)).take_until(self.shutdown);
  while interval.next().await.is_some() {
@@ -167,7 +170,7 @@ impl<'a> InternalMetrics<'a> {
  let byte_size = metrics.size_of();
  let json_size = metrics.estimated_json_encoded_size_of();
 
- emit!(InternalMetricsBytesReceived { byte_size });
+ bytes_received.emit(ByteSize(byte_size));
  events_received.emit(CountByteSize(count, json_size));
 
  let batch = metrics.into_iter().map(|mut metric| {