fix(aws cloudwatch, aws s3): emit component_sent_bytes_total through Driver to ensure component labels

lib/vector-common/src/event_test_util.rs

@@ -19,6 +19,7 @@ pub fn contains_name_once(pattern: &str) -> Result<(), String> {
             .iter()
             .filter(|event| event_name_matches(event, pattern))
             .collect();
+
         match matches.len() {
             0 => Err(format!("Missing event {pattern:?}")),
             1 => Ok(()),
@@ -146,4 +147,10 @@ mod tests {
 
         assert!(contains_name_once("EventsReceived").is_err());
     }
+
+    #[test]
+    fn rejects_missing_event() {
+        clear_recorded_events();
+        assert!(contains_name_once("BytesSent").is_err());
+    }
 }

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

@@ -6,8 +6,15 @@ use super::{ByteSize, Protocol, SharedString};
 crate::registered_event!(
     BytesSent {
         protocol: SharedString,
+        extra_labels: Vec<(SharedString, SharedString)>,
     } => {
-        bytes_sent: Counter = counter!("component_sent_bytes_total", "protocol" => self.protocol.clone()),
+        bytes_sent: Counter = {
+            let mut labels: Vec<(String, String)> = vec![("protocol".to_string(), self.protocol.to_string())];
+            for (k, v) in &self.extra_labels {
+                labels.push((k.to_string(), v.to_string()));
+            }
+            counter!("component_sent_bytes_total", &labels)
+        },
         protocol: SharedString = self.protocol,
     }
 
@@ -21,6 +28,7 @@ impl From<Protocol> for BytesSent {
     fn from(protocol: Protocol) -> Self {
         Self {
             protocol: protocol.0,
+            extra_labels: vec![],
         }
     }
 }

lib/vector-stream/src/driver.rs

@@ -44,6 +44,7 @@ pub struct Driver<St, Svc> {
     input: St,
     service: Svc,
     protocol: Option<SharedString>,
+    extra_labels: Vec<(SharedString, SharedString)>,
 }
 
 impl<St, Svc> Driver<St, Svc> {
@@ -52,6 +53,7 @@ impl<St, Svc> Driver<St, Svc> {
             input,
             service,
             protocol: None,
+            extra_labels: vec![],
         }
     }
 
@@ -64,6 +66,13 @@ impl<St, Svc> Driver<St, Svc> {
         self.protocol = Some(protocol.into());
         self
     }
+
+    /// Add an extra label to `BytesSent` events emitted by this driver.
+    #[must_use]
+    pub fn label(mut self, key: impl Into<SharedString>, value: impl Into<SharedString>) -> Self {
+        self.extra_labels.push((key.into(), value.into()));
+        self
+    }
 }
 
 impl<St, Svc> Driver<St, Svc>
@@ -92,12 +101,18 @@ where
             input,
             mut service,
             protocol,
+            extra_labels,
         } = self;
 
         let batched_input = input.ready_chunks(1024);
         pin!(batched_input);
 
-        let bytes_sent = protocol.map(|protocol| register(BytesSent { protocol }));
+        let bytes_sent = protocol.map(|protocol| {
+            register(BytesSent {
+                protocol,
+                extra_labels,
+            })
+        });
         let events_sent = RegisteredEventCache::new(());
 
         loop {

src/aws/mod.rs

@@ -177,9 +177,47 @@ pub async fn create_client<T>(
 where
     T: ClientBuilder,
 {
-    create_client_and_region::<T>(builder, auth, region, endpoint, proxy, tls_options, timeout)
-        .await
-        .map(|(client, _)| client)
+    build_client_inner::<T>(
+        builder,
+        auth,
+        region,
+        endpoint,
+        proxy,
+        tls_options,
+        timeout,
+        true,
+    )
+    .await
+    .map(|(client, _)| client)
+}
+
+/// Like [`create_client`], but suppresses transport-level `AwsBytesSent` emission.
+///
+/// Use this for sinks that report bytes through the [`Driver`] to avoid double-counting
+/// `component_sent_bytes_total`.
+pub async fn create_client_without_transport_metrics<T>(
+    builder: &T,
+    auth: &AwsAuthentication,
+    region: Option<Region>,
+    endpoint: Option<String>,
+    proxy: &ProxyConfig,
+    tls_options: Option<&TlsConfig>,
+    timeout: Option<&AwsTimeout>,
+) -> crate::Result<(T::Client, Region)>
+where
+    T: ClientBuilder,
+{
+    build_client_inner::<T>(
+        builder,
+        auth,
+        region,
+        endpoint,
+        proxy,
+        tls_options,
+        timeout,
+        false,
+    )
+    .await
 }
 
 /// Create the SDK client and resolve the region using the provided settings.
@@ -192,6 +230,33 @@ pub async fn create_client_and_region<T>(
     tls_options: Option<&TlsConfig>,
     timeout: Option<&AwsTimeout>,
 ) -> crate::Result<(T::Client, Region)>
+where
+    T: ClientBuilder,
+{
+    build_client_inner::<T>(
+        builder,
+        auth,
+        region,
+        endpoint,
+        proxy,
+        tls_options,
+        timeout,
+        true,
+    )
+    .await
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn build_client_inner<T>(
+    builder: &T,
+    auth: &AwsAuthentication,
+    region: Option<Region>,
+    endpoint: Option<String>,
+    proxy: &ProxyConfig,
+    tls_options: Option<&TlsConfig>,
+    timeout: Option<&AwsTimeout>,
+    emit_bytes_sent: bool,
+) -> crate::Result<(T::Client, Region)>
 where
     T: ClientBuilder,
 {
@@ -210,6 +275,7 @@ where
     let connector = AwsHttpClient {
         http: connector,
         region: region.clone(),
+        emit_bytes_sent,
     };
 
     // Build the configuration first.
@@ -324,6 +390,7 @@ pub async fn sign_request(
 struct AwsHttpClient<T> {
     http: T,
     region: Region,
+    emit_bytes_sent: bool,
 }
 
 impl<T> HttpClient for AwsHttpClient<T>
@@ -340,6 +407,7 @@ where
         SharedHttpConnector::new(AwsConnector {
             region: self.region.clone(),
             http: http_connector,
+            emit_bytes_sent: self.emit_bytes_sent,
         })
     }
 }
@@ -348,13 +416,18 @@ where
 struct AwsConnector<T> {
     http: T,
     region: Region,
+    emit_bytes_sent: bool,
 }
 
 impl<T> HttpConnector for AwsConnector<T>
 where
     T: HttpConnector,
 {
     fn call(&self, req: HttpRequest) -> HttpConnectorFuture {
+        if !self.emit_bytes_sent {
+            return self.http.call(req);
+        }
+
         let bytes_sent = Arc::new(std::sync::atomic::AtomicUsize::new(0));
         let req = req.map(|body| {
             let bytes_sent = Arc::clone(&bytes_sent);

src/sinks/aws_cloudwatch_logs/config.rs

@@ -7,8 +7,12 @@ use tower::ServiceBuilder;
 use vector_lib::{codecs::JsonSerializerConfig, configurable::configurable_component, schema};
 use vrl::value::Kind;
 
+use aws_config::Region;
+
 use crate::{
-    aws::{AwsAuthentication, ClientBuilder, RegionOrEndpoint, create_client},
+    aws::{
+        AwsAuthentication, ClientBuilder, RegionOrEndpoint, create_client_without_transport_metrics,
+    },
     codecs::{Encoder, EncodingConfig},
     config::{
         AcknowledgementsConfig, DataType, GenerateConfig, Input, ProxyConfig, SinkConfig,
@@ -185,8 +189,11 @@ pub struct CloudwatchLogsSinkConfig {
 }
 
 impl CloudwatchLogsSinkConfig {
-    pub async fn create_client(&self, proxy: &ProxyConfig) -> crate::Result<CloudwatchLogsClient> {
-        create_client::<CloudwatchLogsClientBuilder>(
+    pub async fn create_client(
+        &self,
+        proxy: &ProxyConfig,
+    ) -> crate::Result<(CloudwatchLogsClient, Region)> {
+        create_client_without_transport_metrics::<CloudwatchLogsClientBuilder>(
             &CloudwatchLogsClientBuilder {},
             &self.auth,
             self.region.region(),
@@ -205,12 +212,13 @@ impl SinkConfig for CloudwatchLogsSinkConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
         let batcher_settings = self.batch.into_batcher_settings()?;
         let request_settings = self.request.tower.into_settings();
-        let client = self.create_client(cx.proxy()).await?;
+        let (client, resolved_region) = self.create_client(cx.proxy()).await?;
         let svc = ServiceBuilder::new()
             .settings(request_settings, CloudwatchRetryLogic::new())
             .service(CloudwatchLogsPartitionSvc::new(
                 self.clone(),
                 client.clone(),
+                resolved_region.to_string(),
             )?);
         let transformer = self.encoding.transformer();
         let serializer = self.encoding.build()?;
@@ -224,8 +232,8 @@ impl SinkConfig for CloudwatchLogsSinkConfig {
                 transformer,
                 encoder,
             },
-
             service: svc,
+            region: resolved_region.to_string(),
         };
 
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))