fix(healthchecks): instrument healthchecks with component span

changelog.d/20356_component_tags_healthcheck.fix.md

@@ -0,0 +1,2 @@
+Any events emitted during the sink healthchecks now get tagged with the correct
+sink tags.

src/aws/mod.rs

@@ -164,7 +164,22 @@ pub async fn create_client<T: ClientBuilder>(
  proxy: &ProxyConfig,
  tls_options: &Option<TlsConfig>,
 ) -> crate::Result<T::Client> {
- create_client_and_region::<T>(auth, region, endpoint, proxy, tls_options)
+ create_client_and_region::<T>(auth, region, endpoint, proxy, tls_options, true)
+ .await
+ .map(|(client, _)| client)
+}
+
+/// Create the SDK client using the provided settings, with the additional option to select
+/// whether we should instrument the http calls with a bytes_sent metric.
+pub async fn create_client_instrument<T: ClientBuilder>(
+ auth: &AwsAuthentication,
+ region: Option<Region>,
+ endpoint: Option<String>,
+ proxy: &ProxyConfig,
+ tls_options: &Option<TlsConfig>,
+ instrument: bool,
+) -> crate::Result<T::Client> {
+ create_client_and_region::<T>(auth, region, endpoint, proxy, tls_options, instrument)
  .await
  .map(|(client, _)| client)
 }
@@ -176,6 +191,7 @@ pub async fn create_client_and_region<T: ClientBuilder>(
  endpoint: Option<String>,
  proxy: &ProxyConfig,
  tls_options: &Option<TlsConfig>,
+ instrument: bool,
 ) -> crate::Result<(T::Client, Region)> {
  let retry_config = RetryConfig::disabled();
 
@@ -190,6 +206,7 @@ pub async fn create_client_and_region<T: ClientBuilder>(
 
  // Create a custom http connector that will emit the required metrics for us.
  let connector = AwsHttpClient {
+ instrument,
  http: connector,
  region: region.clone(),
  };
@@ -277,6 +294,7 @@ pub async fn sign_request(
 struct AwsHttpClient<T> {
  http: T,
  region: Region,
+ instrument: bool,
 }
 
 impl<T> HttpClient for AwsHttpClient<T>
@@ -291,6 +309,7 @@ where
  let http_connector = self.http.http_connector(settings, components);
 
  SharedHttpConnector::new(AwsConnector {
+ instrument: self.instrument,
  region: self.region.clone(),
  http: http_connector,
  })
@@ -299,6 +318,7 @@ where
 
 #[derive(Clone, Debug)]
 struct AwsConnector<T> {
+ instrument: bool,
  http: T,
  region: Region,
 }
@@ -319,11 +339,12 @@ where
 
  let fut = self.http.call(req);
  let region = self.region.clone();
+ let instrument = self.instrument;
 
  HttpConnectorFuture::new(fut.inspect(move |result| {
  let byte_size = bytes_sent.load(Ordering::Relaxed);
  if let Ok(result) = result {
- if result.status().is_success() {
+ if result.status().is_success() && instrument {
  emit!(AwsBytesSent {
  byte_size,
  region: Some(region),

src/sinks/aws_s3/config.rs

@@ -1,4 +1,3 @@
-use aws_sdk_s3::Client as S3Client;
 use tower::ServiceBuilder;
 use vector_lib::codecs::{
  encoding::{Framer, FramingConfig},
@@ -10,8 +9,9 @@ use vector_lib::TimeZone;
 
 use super::sink::S3RequestOptions;
 use crate::{
- aws::{AwsAuthentication, RegionOrEndpoint},
+ aws::{create_client_instrument, AwsAuthentication, RegionOrEndpoint},
  codecs::{Encoder, EncodingConfigWithFraming, SinkType},
+ common::s3::S3ClientBuilder,
  config::{AcknowledgementsConfig, GenerateConfig, Input, ProxyConfig, SinkConfig, SinkContext},
  sinks::{
  s3_common::{
@@ -177,7 +177,7 @@ impl GenerateConfig for S3SinkConfig {
 impl SinkConfig for S3SinkConfig {
  async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
  let service = self.create_service(&cx.proxy).await?;
- let healthcheck = self.build_healthcheck(service.client())?;
+ let healthcheck = self.build_healthcheck(&cx.proxy).await?;
  let sink = self.build_processor(service, cx)?;
  Ok((sink, healthcheck))
  }
@@ -246,7 +246,17 @@ impl S3SinkConfig {
  Ok(VectorSink::from_event_streamsink(sink))
  }
 
- pub fn build_healthcheck(&self, client: S3Client) -> crate::Result<Healthcheck> {
+ pub async fn build_healthcheck(&self, proxy: &ProxyConfig) -> crate::Result<Healthcheck> {
+ let client = create_client_instrument::<S3ClientBuilder>(
+ &self.auth,
+ self.region.region(),
+ self.region.endpoint(),
+ proxy,
+ &self.tls,
+ // Make use the healthcheck doesn't emit a `BytesSent` event.
+ false,
+ )
+ .await?;
  s3_common::config::build_healthcheck(self.bucket.clone(), client)
  }
 

src/sinks/aws_s3/integration_tests.rs

@@ -384,12 +384,9 @@ async fn s3_healthchecks() {
  create_bucket(&bucket, false).await;
 
  let config = config(&bucket, 1);
- let service = config
- .create_service(&ProxyConfig::from_env())
- .await
- .unwrap();
  config
- .build_healthcheck(service.client())
+ .build_healthcheck(&ProxyConfig::from_env())
+ .await
  .unwrap()
  .await
  .unwrap();
@@ -398,12 +395,9 @@ async fn s3_healthchecks() {
 #[tokio::test]
 async fn s3_healthchecks_invalid_bucket() {
  let config = config("s3_healthchecks_invalid_bucket", 1);
- let service = config
- .create_service(&ProxyConfig::from_env())
- .await
- .unwrap();
  assert!(config
- .build_healthcheck(service.client())
+ .build_healthcheck(&ProxyConfig::from_env())
+ .await
  .unwrap()
  .await
  .is_err());

src/sources/aws_s3/mod.rs

@@ -252,6 +252,7 @@ impl AwsS3Config {
  endpoint,
  proxy,
  &sqs.tls_options,
+ true,
  )
  .await?;