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

lib/vector-common/src/event_test_util.rs

@@ -15,10 +15,28 @@ thread_local! {
 pub fn contains_name_once(pattern: &str) -> Result<(), String> {
     EVENTS_RECORDED.with(|events| {
         let events = events.borrow();
-        let matches: Vec<_> = events
+        let mut matches: Vec<_> = events
             .iter()
             .filter(|event| event_name_matches(event, pattern))
             .collect();
+
+        // When both a component-scoped event (e.g. `BytesSent`) and a transport-level
+        // prefixed variant (e.g. `AwsBytesSent`) are present, drop the prefixed form.
+        // This lets sinks that emit through the Driver coexist with the AWS connector's
+        // `AwsBytesSent`, while sinks not yet migrated still pass via `AwsBytesSent` alone.
+        if matches.len() > 1 {
+            let has_exact = matches.iter().any(|event| {
+                let segment = event.rsplit_once("::").map_or(event.as_str(), |(_, s)| s);
+                segment == pattern
+            });
+            if has_exact {
+                matches.retain(|event| {
+                    let segment = event.rsplit_once("::").map_or(event.as_str(), |(_, s)| s);
+                    !demotable_prefix_match(segment, pattern)
+                });
+            }
+        }
+
         match matches.len() {
             0 => Err(format!("Missing event {pattern:?}")),
             1 => Ok(()),
@@ -62,6 +80,17 @@ fn ignore_prefixed_match(segment: &str, pattern: &str) -> bool {
     matches!(pattern, "EventsReceived" | "EventsSent") && segment.starts_with("Buffer")
 }
 
+/// Returns true for prefixed event variants that should be dropped when the exact
+/// (component-scoped) form is also present. Unlike `ignore_prefixed_match`, these
+/// prefixed forms are still accepted as the *sole* match — they are only demoted
+/// when they would cause a duplicate.
+fn demotable_prefix_match(segment: &str, pattern: &str) -> bool {
+    // The AWS HTTP connector emits `AwsBytesSent` at the transport layer. Sinks that
+    // also route byte reporting through the Driver emit a component-scoped `BytesSent`.
+    // When both are present, prefer the Driver's `BytesSent`.
+    pattern == "BytesSent" && segment.starts_with("Aws")
+}
+
 /// Record an emitted internal event. This is somewhat dumb at this
 /// point, just recording the pure string value of the `emit!` call
 /// parameter. At some point, making all internal events implement
@@ -146,4 +175,21 @@ mod tests {
 
         assert!(contains_name_once("EventsReceived").is_err());
     }
+
+    #[test]
+    fn contains_name_once_demotes_aws_bytes_sent_when_exact_present() {
+        reset_events();
+        record_internal_event("BytesSent");
+        record_internal_event("AwsBytesSent");
+
+        assert!(contains_name_once("BytesSent").is_ok());
+    }
+
+    #[test]
+    fn contains_name_once_accepts_aws_bytes_sent_alone() {
+        reset_events();
+        record_internal_event("AwsBytesSent");
+
+        assert!(contains_name_once("BytesSent").is_ok());
+    }
 }

src/sinks/aws_cloudwatch_logs/integration_tests.rs

@@ -6,13 +6,15 @@ use aws_sdk_kms::Client as KMSClient;
 use chrono::Duration;
 use futures::{StreamExt, stream};
 use similar_asserts::assert_eq;
+use tracing::Instrument;
 use vector_lib::{codecs::TextSerializerConfig, lookup};
 
 use super::*;
 use crate::{
     aws::{AwsAuthentication, ClientBuilder, RegionOrEndpoint, create_client},
     config::{ProxyConfig, SinkConfig, SinkContext, log_schema},
     event::{Event, LogEvent, Value},
+    metrics::Controller,
     sinks::{aws_cloudwatch_logs::config::CloudwatchLogsClientBuilder, util::BatchConfig},
     template::Template,
     test_util::{
@@ -621,3 +623,86 @@ async fn ensure_group() {
 fn gen_name() -> String {
     format!("test-{}", random_string(10).to_lowercase())
 }
+
+/// Regression test for https://github.com/vectordotdev/vector/issues/20356
+#[tokio::test]
+async fn cloudwatch_component_sent_bytes_total_has_component_labels() {
+    trace_init();
+
+    ensure_group().await;
+
+    let controller = Controller::get().unwrap();
+    controller.reset();
+
+    let stream_name = gen_name();
+    let config = CloudwatchLogsSinkConfig {
+        stream_name: Template::try_from(stream_name.as_str()).unwrap(),
+        group_name: Template::try_from(GROUP_NAME).unwrap(),
+        region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()),
+        encoding: TextSerializerConfig::default().into(),
+        create_missing_group: true,
+        create_missing_stream: true,
+        retention: Default::default(),
+        compression: Default::default(),
+        batch: Default::default(),
+        request: Default::default(),
+        tls: Default::default(),
+        assume_role: None,
+        auth: Default::default(),
+        acknowledgements: Default::default(),
+        kms_key: None,
+        tags: None,
+    };
+
+    let (sink, _) = config.build(SinkContext::default()).await.unwrap();
+
+    let (_input_lines, events) = random_lines_with_stream(100, 11, None);
+
+    let component_span = tracing::error_span!(
+        "sink",
+        component_kind = "sink",
+        component_id = "test_cloudwatch",
+        component_type = "aws_cloudwatch_logs",
+    );
+
+    sink.run(events.map(Into::into))
+        .instrument(component_span)
+        .await
+        .expect("Running sink failed");
+
+    let metrics = controller.capture_metrics();
+
+    let bytes_total: Vec<_> = metrics
+        .iter()
+        .filter(|m| m.name() == "component_sent_bytes_total")
+        .collect();
+
+    assert!(
+        !bytes_total.is_empty(),
+        "Expected at least one component_sent_bytes_total metric to be emitted"
+    );
+
+    let has_component_id = bytes_total
+        .iter()
+        .any(|m| m.tags().and_then(|tags| tags.get("component_id")).is_some());
+
+    assert!(
+        has_component_id,
+        "component_sent_bytes_total was emitted but none of the {} instances \
+         carry a component_id label.  Found tags: {:?}",
+        bytes_total.len(),
+        bytes_total
+            .iter()
+            .map(|m| format!(
+                "{{{}}}",
+                m.tags()
+                    .map(|t| t
+                        .iter_single()
+                        .map(|(k, v)| format!("{}={}", k, v))
+                        .collect::<Vec<_>>()
+                        .join(", "))
+                    .unwrap_or_default()
+            ))
+            .collect::<Vec<_>>()
+    );
+}

src/sinks/aws_cloudwatch_logs/service.rs

@@ -111,6 +111,7 @@ impl From<SdkError<DescribeLogStreamsError, HttpResponse>> for CloudwatchError {
 #[derive(Debug)]
 pub struct CloudwatchResponse {
     events_byte_size: GroupedCountByteSize,
+    byte_size: usize,
 }
 
 impl crate::sinks::util::sink::Response for CloudwatchResponse {
@@ -131,6 +132,10 @@ impl DriverResponse for CloudwatchResponse {
     fn events_sent(&self) -> &GroupedCountByteSize {
         &self.events_byte_size
     }
+
+    fn bytes_sent(&self) -> Option<usize> {
+        Some(self.byte_size)
+    }
 }
 
 #[derive(Snafu, Debug)]
@@ -181,6 +186,7 @@ impl Service<BatchCloudwatchRequest> for CloudwatchLogsPartitionSvc {
 
     fn call(&mut self, mut req: BatchCloudwatchRequest) -> Self::Future {
         let metadata = std::mem::take(req.metadata_mut());
+        let byte_size = metadata.request_encoded_size();
         let events_byte_size = metadata.into_events_estimated_json_encoded_byte_size();
 
         let key = req.key;
@@ -225,7 +231,10 @@ impl Service<BatchCloudwatchRequest> for CloudwatchLogsPartitionSvc {
         };
 
         svc.oneshot(events)
-            .map_ok(move |_x| CloudwatchResponse { events_byte_size })
+            .map_ok(move |_x| CloudwatchResponse {
+                events_byte_size,
+                byte_size,
+            })
             .map_err(Into::into)
             .boxed()
     }

src/sinks/aws_cloudwatch_logs/sink.rs

@@ -64,6 +64,7 @@ where
                 }
             })
             .into_driver(service)
+            .protocol("https")
             .run()
             .await
     }

src/sinks/aws_s3/integration_tests.rs

@@ -19,6 +19,7 @@ use flate2::read::MultiGzDecoder;
 use futures::{Stream, stream};
 use similar_asserts::assert_eq;
 use tokio_stream::StreamExt;
+use tracing::Instrument;
 use vector_lib::{
     codecs::{TextSerializerConfig, encoding::FramingConfig},
     config::proxy::ProxyConfig,
@@ -30,6 +31,7 @@ use crate::{
     aws::{AwsAuthentication, RegionOrEndpoint, create_client},
     common::s3::S3ClientBuilder,
     config::SinkContext,
+    metrics::Controller,
     sinks::{
         aws_s3::config::default_filename_time_format,
         s3_common::config::{S3Options, S3ServerSideEncryption},
@@ -40,7 +42,7 @@ use crate::{
             AWS_SINK_TAGS, COMPONENT_ERROR_TAGS, run_and_assert_sink_compliance,
             run_and_assert_sink_error,
         },
-        random_lines_with_stream, random_string,
+        random_lines_with_stream, random_string, trace_init,
     },
 };
 
@@ -628,3 +630,71 @@ async fn get_zstd_lines(obj: GetObjectOutput) -> Vec<String> {
 async fn get_object_output_body(obj: GetObjectOutput) -> impl std::io::Read {
     obj.body.collect().await.unwrap().reader()
 }
+
+/// Regression test for https://github.com/vectordotdev/vector/issues/20356
+#[tokio::test]
+async fn s3_component_sent_bytes_total_has_component_labels() {
+    trace_init();
+
+    let controller = Controller::get().unwrap();
+    controller.reset();
+
+    let cx = SinkContext::default();
+    let bucket = uuid::Uuid::new_v4().to_string();
+    create_bucket(&bucket, false).await;
+
+    let config = config(&bucket, 1000000);
+    let service = config.create_service(&cx.globals.proxy).await.unwrap();
+    let sink = config.build_processor(service, cx).unwrap();
+
+    let (_lines, events, receiver) = make_events_batch(100, 10);
+
+    let component_span = tracing::error_span!(
+        "sink",
+        component_kind = "sink",
+        component_id = "test_s3",
+        component_type = "aws_s3",
+    );
+
+    sink.run(events)
+        .instrument(component_span)
+        .await
+        .expect("Running sink failed");
+    assert_eq!(receiver.await, BatchStatus::Delivered);
+
+    let metrics = controller.capture_metrics();
+
+    let bytes_total: Vec<_> = metrics
+        .iter()
+        .filter(|m| m.name() == "component_sent_bytes_total")
+        .collect();
+
+    assert!(
+        !bytes_total.is_empty(),
+        "Expected at least one component_sent_bytes_total metric to be emitted"
+    );
+
+    let has_component_id = bytes_total
+        .iter()
+        .any(|m| m.tags().and_then(|tags| tags.get("component_id")).is_some());
+
+    assert!(
+        has_component_id,
+        "component_sent_bytes_total was emitted but none of the {} instances \
+         carry a component_id label.  Found tags: {:?}",
+        bytes_total.len(),
+        bytes_total
+            .iter()
+            .map(|m| format!(
+                "{{{}}}",
+                m.tags()
+                    .map(|t| t
+                        .iter_single()
+                        .map(|(k, v)| format!("{}={}", k, v))
+                        .collect::<Vec<_>>()
+                        .join(", "))
+                    .unwrap_or_default()
+            ))
+            .collect::<Vec<_>>()
+    );
+}

src/sinks/s3_common/service.rs

@@ -53,6 +53,7 @@ pub struct S3Metadata {
 #[derive(Debug)]
 pub struct S3Response {
     events_byte_size: GroupedCountByteSize,
+    byte_size: usize,
 }
 
 impl DriverResponse for S3Response {
@@ -63,6 +64,10 @@ impl DriverResponse for S3Response {
     fn events_sent(&self) -> &GroupedCountByteSize {
         &self.events_byte_size
     }
+
+    fn bytes_sent(&self) -> Option<usize> {
+        Some(self.byte_size)
+    }
 }
 
 /// Wrapper for the AWS SDK S3 client.
@@ -118,6 +123,7 @@ impl Service<S3Request> for S3Service {
             tagging.finish()
         });
 
+        let byte_size = request.body.len();
         let events_byte_size = request
             .request_metadata
             .into_events_estimated_json_encoded_byte_size();
@@ -153,7 +159,10 @@ impl Service<S3Request> for S3Service {
                     key = request.metadata.s3_key
                 );
 
-                S3Response { events_byte_size }
+                S3Response {
+                    events_byte_size,
+                    byte_size,
+                }
             })
         })
     }

src/sinks/s3_common/sink.rs

@@ -60,6 +60,7 @@ where
                 }
             })
             .into_driver(self.service)
+            .protocol("https")
             .run()
             .await
     }