diff --git a/lib/vector-common/src/event_test_util.rs b/lib/vector-common/src/event_test_util.rs index 39b0f8ebd4db8..97773b528aee3 100644 --- a/lib/vector-common/src/event_test_util.rs +++ b/lib/vector-common/src/event_test_util.rs @@ -13,27 +13,26 @@ thread_local! { /// Will return `Err` if `pattern` is not found in the event record, or is found multiple times. pub fn contains_name_once(pattern: &str) -> Result<(), String> { EVENTS_RECORDED.with(|events| { - let mut n_events = 0; - let mut names = String::new(); - for event in &*events.borrow() { - if event.ends_with(pattern) { - if n_events > 0 { - names.push_str(", "); + let events = events.borrow(); + let matches: Vec<_> = events.iter().filter(|e| e.ends_with(pattern)).collect(); + + match matches.len() { + 0 => Err(format!("Missing event `{pattern}`")), + 1 => Ok(()), + n => { + let mut names = String::new(); + for (i, event) in matches.iter().enumerate() { + if i > 0 { + names.push_str(", "); + } + _ = write!(names, "`{event}`"); } - n_events += 1; - _ = write!(names, "`{event}`"); + Err(format!( + "Multiple ({n}) events matching `{pattern}`: ({names}). Hint! Don't use the `assert_x_` \ + test helpers on round-trip tests (tests that run more than a single component)." + )) } } - if n_events == 0 { - Err(format!("Missing event `{pattern}`")) - } else if n_events > 1 { - Err(format!( - "Multiple ({n_events}) events matching `{pattern}`: ({names}). Hint! Don't use the `assert_x_` \ - test helpers on round-trip tests (tests that run more than a single component)." - )) - } else { - Ok(()) - } }) } @@ -67,3 +66,14 @@ pub fn record_internal_event(event: &str) { EVENTS_RECORDED.with(|er| er.borrow_mut().insert(event.trim().into())); } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rejects_missing_event() { + clear_recorded_events(); + assert!(contains_name_once("BytesSent").is_err()); + } +} diff --git a/lib/vector-common/src/internal_event/bytes_sent.rs b/lib/vector-common/src/internal_event/bytes_sent.rs index cb8a25b2ca032..eb9550e98d7d5 100644 --- a/lib/vector-common/src/internal_event/bytes_sent.rs +++ b/lib/vector-common/src/internal_event/bytes_sent.rs @@ -1,4 +1,4 @@ -use metrics::{counter, Counter}; +use metrics::{Counter, Key, Label, Metadata}; use tracing::trace; use super::{ByteSize, Protocol, SharedString}; @@ -6,8 +6,16 @@ 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![Label::new("protocol", self.protocol.clone())]; + for (k, v) in &self.extra_labels { + labels.push(Label::new(k.clone(), v.clone())); + } + let key = Key::from_parts("component_sent_bytes_total", labels); + metrics::with_recorder(|rec| rec.register_counter(&key, &Metadata::new(module_path!(), metrics::Level::INFO, None))) + }, protocol: SharedString = self.protocol, } @@ -21,6 +29,7 @@ impl From for BytesSent { fn from(protocol: Protocol) -> Self { Self { protocol: protocol.0, + extra_labels: vec![], } } } diff --git a/lib/vector-stream/src/driver.rs b/lib/vector-stream/src/driver.rs index b142deba814f2..696f4c8d639d2 100644 --- a/lib/vector-stream/src/driver.rs +++ b/lib/vector-stream/src/driver.rs @@ -43,6 +43,7 @@ pub struct Driver { input: St, service: Svc, protocol: Option, + extra_labels: Vec<(SharedString, SharedString)>, } impl Driver { @@ -51,6 +52,7 @@ impl Driver { input, service, protocol: None, + extra_labels: vec![], } } @@ -63,6 +65,13 @@ impl Driver { self.protocol = Some(protocol.into()); self } + + /// Add an extra label to the `BytesSent` metric emitted by this driver. + #[must_use] + pub fn label(mut self, key: impl Into, value: impl Into) -> Self { + self.extra_labels.push((key.into(), value.into())); + self + } } impl Driver @@ -91,12 +100,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 { diff --git a/src/aws/mod.rs b/src/aws/mod.rs index 0348f783b0837..979c52f1ada90 100644 --- a/src/aws/mod.rs +++ b/src/aws/mod.rs @@ -179,6 +179,36 @@ where .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( + builder: &T, + auth: &AwsAuthentication, + region: Option, + endpoint: Option, + proxy: &ProxyConfig, + tls_options: Option<&TlsConfig>, + timeout: Option<&AwsTimeout>, +) -> crate::Result +where + T: ClientBuilder, +{ + build_client_inner::( + builder, + auth, + region, + endpoint, + proxy, + tls_options, + timeout, + false, + ) + .await + .map(|(client, _)| client) +} + /// Create the SDK client and resolve the region using the provided settings. pub async fn create_client_and_region( builder: &T, @@ -189,6 +219,33 @@ pub async fn create_client_and_region( tls_options: Option<&TlsConfig>, timeout: Option<&AwsTimeout>, ) -> crate::Result<(T::Client, Region)> +where + T: ClientBuilder, +{ + build_client_inner::( + builder, + auth, + region, + endpoint, + proxy, + tls_options, + timeout, + true, + ) + .await +} + +#[allow(clippy::too_many_arguments)] +async fn build_client_inner( + builder: &T, + auth: &AwsAuthentication, + region: Option, + endpoint: Option, + proxy: &ProxyConfig, + tls_options: Option<&TlsConfig>, + timeout: Option<&AwsTimeout>, + emit_bytes_sent: bool, +) -> crate::Result<(T::Client, Region)> where T: ClientBuilder, { @@ -207,6 +264,7 @@ where let connector = AwsHttpClient { http: connector, region: region.clone(), + emit_bytes_sent, }; // Build the configuration first. @@ -321,6 +379,7 @@ pub async fn sign_request( struct AwsHttpClient { http: T, region: Region, + emit_bytes_sent: bool, } impl HttpClient for AwsHttpClient @@ -337,6 +396,7 @@ where SharedHttpConnector::new(AwsConnector { region: self.region.clone(), http: http_connector, + emit_bytes_sent: self.emit_bytes_sent, }) } } @@ -345,6 +405,7 @@ where struct AwsConnector { http: T, region: Region, + emit_bytes_sent: bool, } impl HttpConnector for AwsConnector @@ -352,6 +413,10 @@ 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); diff --git a/src/sinks/aws_cloudwatch_logs/config.rs b/src/sinks/aws_cloudwatch_logs/config.rs index db7e1d5d4da7a..fbc8f05e420cc 100644 --- a/src/sinks/aws_cloudwatch_logs/config.rs +++ b/src/sinks/aws_cloudwatch_logs/config.rs @@ -9,7 +9,9 @@ use vector_lib::schema; use vrl::value::Kind; use crate::{ - aws::{create_client, AwsAuthentication, ClientBuilder, RegionOrEndpoint}, + aws::{ + create_client_without_transport_metrics, AwsAuthentication, ClientBuilder, RegionOrEndpoint, + }, codecs::{Encoder, EncodingConfig}, config::{ AcknowledgementsConfig, DataType, GenerateConfig, Input, ProxyConfig, SinkConfig, @@ -187,7 +189,7 @@ pub struct CloudwatchLogsSinkConfig { impl CloudwatchLogsSinkConfig { pub async fn create_client(&self, proxy: &ProxyConfig) -> crate::Result { - create_client::( + create_client_without_transport_metrics::( &CloudwatchLogsClientBuilder {}, &self.auth, self.region.region(), @@ -225,8 +227,11 @@ impl SinkConfig for CloudwatchLogsSinkConfig { transformer, encoder, }, - service: svc, + region: self + .region + .region() + .map_or_else(String::new, |r| r.to_string()), }; Ok((VectorSink::from_event_streamsink(sink), healthcheck)) diff --git a/src/sinks/aws_cloudwatch_logs/integration_tests.rs b/src/sinks/aws_cloudwatch_logs/integration_tests.rs index 437408e51f3ea..889d663f0911a 100644 --- a/src/sinks/aws_cloudwatch_logs/integration_tests.rs +++ b/src/sinks/aws_cloudwatch_logs/integration_tests.rs @@ -7,12 +7,14 @@ use aws_sdk_kms::Client as KMSClient; use chrono::Duration; use futures::{stream, StreamExt}; use similar_asserts::assert_eq; +use tracing::Instrument; use vector_lib::codecs::TextSerializerConfig; use vector_lib::lookup; use super::*; use crate::aws::{create_client, ClientBuilder}; use crate::aws::{AwsAuthentication, RegionOrEndpoint}; +use crate::metrics::Controller; use crate::sinks::aws_cloudwatch_logs::config::CloudwatchLogsClientBuilder; use crate::{ config::{log_schema, ProxyConfig, SinkConfig, SinkContext}, @@ -625,3 +627,119 @@ 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 +/// +/// `component_sent_bytes_total` must carry `component_id` / `component_kind` / +/// `component_type` labels so dashboards can attribute byte throughput to +/// individual sinks. The aws_cloudwatch_logs sink emits this metric through +/// `AwsBytesSent` in the shared HTTP connector, which uses a bare `counter!()` +/// that only inherits span labels when a component span is active on the +/// current thread. Because the sink's internal `tower::buffer::Buffer` workers +/// run in separately-spawned tasks, the component span is typically absent and +/// the labels are lost. +#[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); + + // Simulate what the topology does: wrap the sink task in a component span + // (see topology/running.rs spawn_sink). + 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: {:?}", + bytes_total.len(), + bytes_total + .iter() + .map(|m| m.to_string()) + .collect::>() + ); + + let has_region = bytes_total + .iter() + .any(|m| m.tags().and_then(|tags| tags.get("region")).is_some()); + + assert!( + has_region, + "component_sent_bytes_total was emitted but none of the {} instances \ + carry a region label. Found: {:?}", + bytes_total.len(), + bytes_total + .iter() + .map(|m| m.to_string()) + .collect::>() + ); + + let without_component_id: Vec<_> = bytes_total + .iter() + .filter(|m| m.tags().and_then(|tags| tags.get("component_id")).is_none()) + .collect(); + + assert!( + without_component_id.is_empty(), + "Found {} component_sent_bytes_total metric(s) without component_id \ + (transport-layer duplicates that should have been suppressed): {:?}", + without_component_id.len(), + without_component_id + .iter() + .map(|m| m.to_string()) + .collect::>() + ); +} diff --git a/src/sinks/aws_cloudwatch_logs/service.rs b/src/sinks/aws_cloudwatch_logs/service.rs index f914029dbc35f..65e44080b6a0d 100644 --- a/src/sinks/aws_cloudwatch_logs/service.rs +++ b/src/sinks/aws_cloudwatch_logs/service.rs @@ -105,6 +105,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 { @@ -125,6 +126,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)] @@ -175,6 +180,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; @@ -219,7 +225,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 0624f9f204b17..a2375ebf78f30 100644 --- a/src/sinks/aws_cloudwatch_logs/sink.rs +++ b/src/sinks/aws_cloudwatch_logs/sink.rs @@ -23,6 +23,7 @@ pub struct CloudwatchSink { pub batcher_settings: BatcherSettings, pub(super) request_builder: CloudwatchRequestBuilder, pub service: S, + pub region: String, } impl CloudwatchSink @@ -61,6 +62,8 @@ where } }) .into_driver(service) + .protocol("https") + .label("region", self.region) .run() .await }