diff --git a/lib/vector-common/src/event_test_util.rs b/lib/vector-common/src/event_test_util.rs index 4e2e9e728b440..1ef7a5df9da6e 100644 --- a/lib/vector-common/src/event_test_util.rs +++ b/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()); + } } diff --git a/src/sinks/aws_cloudwatch_logs/integration_tests.rs b/src/sinks/aws_cloudwatch_logs/integration_tests.rs index f25a849552348..0d1d9da3b9b29 100644 --- a/src/sinks/aws_cloudwatch_logs/integration_tests.rs +++ b/src/sinks/aws_cloudwatch_logs/integration_tests.rs @@ -6,6 +6,7 @@ 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::*; @@ -13,6 +14,7 @@ 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::>() + .join(", ")) + .unwrap_or_default() + )) + .collect::>() + ); +} diff --git a/src/sinks/aws_cloudwatch_logs/service.rs b/src/sinks/aws_cloudwatch_logs/service.rs index cdb34702aa413..2f79649ed17df 100644 --- a/src/sinks/aws_cloudwatch_logs/service.rs +++ b/src/sinks/aws_cloudwatch_logs/service.rs @@ -111,6 +111,7 @@ impl From> 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 { + Some(self.byte_size) + } } #[derive(Snafu, Debug)] @@ -181,6 +186,7 @@ impl Service 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 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() } diff --git a/src/sinks/aws_cloudwatch_logs/sink.rs b/src/sinks/aws_cloudwatch_logs/sink.rs index da272b2fafd28..2f56b9ef01715 100644 --- a/src/sinks/aws_cloudwatch_logs/sink.rs +++ b/src/sinks/aws_cloudwatch_logs/sink.rs @@ -64,6 +64,7 @@ where } }) .into_driver(service) + .protocol("https") .run() .await } diff --git a/src/sinks/aws_s3/integration_tests.rs b/src/sinks/aws_s3/integration_tests.rs index 31907d70b5c0f..701ea214fc197 100644 --- a/src/sinks/aws_s3/integration_tests.rs +++ b/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 { 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::>() + .join(", ")) + .unwrap_or_default() + )) + .collect::>() + ); +} diff --git a/src/sinks/s3_common/service.rs b/src/sinks/s3_common/service.rs index 08b5cad0e0016..9fab27cbe7af5 100644 --- a/src/sinks/s3_common/service.rs +++ b/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 { + Some(self.byte_size) + } } /// Wrapper for the AWS SDK S3 client. @@ -118,6 +123,7 @@ impl Service 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 for S3Service { key = request.metadata.s3_key ); - S3Response { events_byte_size } + S3Response { + events_byte_size, + byte_size, + } }) }) } diff --git a/src/sinks/s3_common/sink.rs b/src/sinks/s3_common/sink.rs index 729a67153f641..46ee41f9342fc 100644 --- a/src/sinks/s3_common/sink.rs +++ b/src/sinks/s3_common/sink.rs @@ -60,6 +60,7 @@ where } }) .into_driver(self.service) + .protocol("https") .run() .await }