diff --git a/changelog.d/12779_add_aws_connector_observability.enhancement.md b/changelog.d/12779_add_aws_connector_observability.enhancement.md new file mode 100644 index 0000000000000..fb6802ce524ad --- /dev/null +++ b/changelog.d/12779_add_aws_connector_observability.enhancement.md @@ -0,0 +1,3 @@ +Added common internal HTTP metrics to the connector used by AWS sinks. + +authors: gwenaskell diff --git a/src/aws/auth.rs b/src/aws/auth.rs index 7ca3de0285b61..240c579956acf 100644 --- a/src/aws/auth.rs +++ b/src/aws/auth.rs @@ -245,7 +245,11 @@ impl AwsAuthentication { external_id: Option<&str>, session_name: Option<&str>, ) -> crate::Result { - let connector = super::connector(proxy, tls_options)?; + let connector = super::AwsHttpClient { + http: super::connector(proxy, tls_options)?, + region: region.clone(), + emit_bytes_sent: false, + }; let config = SdkConfig::builder() .http_client(connector) .region(region.clone()) @@ -311,8 +315,6 @@ impl AwsAuthentication { profile, region, } => { - let connector = super::connector(proxy, tls_options)?; - // The SDK uses the default profile out of the box, but doesn't provide an optional // type in the builder. We can just hardcode it so that everything works. let profile_files = EnvConfigFiles::builder() @@ -320,6 +322,11 @@ impl AwsAuthentication { .build(); let auth_region = region.clone().map(Region::new).unwrap_or(service_region); + let connector = super::AwsHttpClient { + http: super::connector(proxy, tls_options)?, + region: auth_region.clone(), + emit_bytes_sent: false, + }; let provider_config = ProviderConfig::empty() .with_region(Option::from(auth_region)) .with_http_client(connector); @@ -391,7 +398,11 @@ async fn default_credentials_provider( tls_options: Option<&TlsConfig>, imds: ImdsAuthentication, ) -> crate::Result { - let connector = super::connector(proxy, tls_options)?; + let connector = super::AwsHttpClient { + http: super::connector(proxy, tls_options)?, + region: region.clone(), + emit_bytes_sent: false, + }; let provider_config = ProviderConfig::empty() .with_region(Some(region.clone())) diff --git a/src/aws/mod.rs b/src/aws/mod.rs index 35cb39cfdf9c7..4438e427cc501 100644 --- a/src/aws/mod.rs +++ b/src/aws/mod.rs @@ -11,7 +11,7 @@ use std::{ atomic::{AtomicUsize, Ordering}, }, task::{Context, Poll}, - time::{Duration, SystemTime}, + time::{Duration, Instant, SystemTime}, }; pub use auth::{AwsAuthentication, ImdsAuthentication}; @@ -37,8 +37,7 @@ use aws_smithy_runtime_api::client::{ use aws_smithy_types::body::SdkBody; use aws_types::sdk_config::SharedHttpClient; use bytes::Bytes; -use futures_util::FutureExt; -use http::HeaderMap; +use http::{HeaderMap, HeaderName, header::HeaderValue}; use http_body::{Body, combinators::BoxBody}; use pin_project::pin_project; use regex::RegexSet; @@ -49,7 +48,13 @@ pub use timeout::AwsTimeout; use crate::{ config::ProxyConfig, http::{build_proxy_connector, build_tls_connector, status}, - internal_events::AwsBytesSent, + internal_events::{ + AwsBytesSent, + http_client::{ + AboutToSendHttpRequest, GotHttpResponse, GotHttpWarning, HttpRequestTelemetry, + HttpResponseTelemetry, + }, + }, tls::{MaybeTlsSettings, TlsConfig}, }; @@ -134,6 +139,7 @@ pub fn region_provider( proxy: &ProxyConfig, tls_options: Option<&TlsConfig>, ) -> crate::Result> { + // Region is not yet known here, so we cannot wrap with AwsHttpClient for observability. let config = aws_config::provider_config::ProviderConfig::default() .with_http_client(connector(proxy, tls_options)?); @@ -210,6 +216,7 @@ where let connector = AwsHttpClient { http: connector, region: region.clone(), + emit_bytes_sent: true, }; // Build the configuration first. @@ -324,6 +331,10 @@ pub async fn sign_request( struct AwsHttpClient { http: T, region: Region, + /// When `false`, the connector skips `AwsBytesSent` so that control-plane + /// traffic (STS AssumeRole, IMDS, SSO token exchange) does not inflate + /// `component_sent_bytes_total`. + emit_bytes_sent: bool, } impl HttpClient for AwsHttpClient @@ -340,6 +351,7 @@ where SharedHttpConnector::new(AwsConnector { region: self.region.clone(), http: http_connector, + emit_bytes_sent: self.emit_bytes_sent, }) } } @@ -348,6 +360,122 @@ where struct AwsConnector { http: T, region: Region, + emit_bytes_sent: bool, +} + +// ── Telemetry trait implementations for the AWS SDK HTTP types ──────────────── +// +// `HttpRequest` and `HttpResponse` are the SDK's own structs (not `http::Request` +// / `http::Response`), so they cannot implement the hyper-specific body/version +// accessors. The required method impls (method/uri and status) are enough for +// metric labels; the optional rich-logging fields fall back to `None`. + +impl HttpRequestTelemetry for HttpRequest { + fn method(&self) -> &str { + self.method() + } + + fn uri(&self) -> String { + self.uri().to_string() + } + + fn headers(&self) -> HeaderMap { + smithy_headers_to_map(self.headers()) + } + + fn body_size_hint(&self) -> (u64, Option) { + let hint = Body::size_hint(self.body()); + (hint.lower(), hint.upper()) + } + + fn extra_sensitive_headers(&self) -> &'static [HeaderName] { + &AWS_EXTRA_SENSITIVE_HEADERS + } +} + +impl HttpResponseTelemetry for HttpResponse { + fn status_u16(&self) -> u16 { + self.status().as_u16() + } + + fn headers(&self) -> HeaderMap { + smithy_headers_to_map(self.headers()) + } + + fn body_size_hint(&self) -> (u64, Option) { + let hint = Body::size_hint(self.body()); + (hint.lower(), hint.upper()) + } + + fn extra_sensitive_headers(&self) -> &'static [HeaderName] { + &AWS_EXTRA_SENSITIVE_HEADERS + } +} + +/// AWS-specific headers that carry credential material and must never appear in +/// logs. +/// +/// - `x-amz-security-token` — STS temporary session token +/// - `x-amz-sso_bearer_token` — IAM Identity Center access token (exchangeable for role credentials) +/// - `x-aws-ec2-metadata-token` — IMDSv2 session token (valid for up to 6 h) +static AWS_EXTRA_SENSITIVE_HEADERS: [HeaderName; 3] = [ + HeaderName::from_static("x-amz-security-token"), + HeaderName::from_static("x-amz-sso_bearer_token"), + HeaderName::from_static("x-aws-ec2-metadata-token"), +]; + +/// Converts the AWS SDK's string-pair header iterator into an `http::HeaderMap`. +/// Sanitization (marking sensitive headers) is handled by the trait's default +/// `sanitized_headers` method. +fn smithy_headers_to_map( + headers: &aws_smithy_runtime_api::http::Headers, +) -> HeaderMap { + let mut map = HeaderMap::with_capacity(headers.len()); + for (name, value) in headers { + let Ok(header_name) = http::HeaderName::from_bytes(name.as_bytes()) else { + continue; + }; + let Ok(header_value) = HeaderValue::from_str(value) else { + continue; + }; + map.insert(header_name, header_value); + } + map +} + +// ───────────────────────────────────────────────────────────────────────────── + +impl AwsConnector +where + T: HttpConnector, +{ + /// Calls the inner HTTP connector and emits common telemetry. + /// Does not emit `AwsBytesSent` telemetry. + fn call_inner(&self, req: HttpRequest) -> HttpConnectorFuture { + emit!(AboutToSendHttpRequest { request: &req }); + + let fut: HttpConnectorFuture = self.http.call(req); + + HttpConnectorFuture::new(async move { + let before = Instant::now(); + let result = fut.await; + let roundtrip = before.elapsed(); + + match &result { + Ok(response) => { + emit!(GotHttpResponse { + response, + roundtrip + }); + } + Err(error) => { + emit!(GotHttpWarning { error, roundtrip }); + } + } + + result + }) + } } impl HttpConnector for AwsConnector @@ -355,7 +483,12 @@ where T: HttpConnector, { fn call(&self, req: HttpRequest) -> HttpConnectorFuture { - let bytes_sent = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + if !self.emit_bytes_sent { + return HttpConnectorFuture::new(self.call_inner(req)); + } + + let bytes_sent = Arc::new(AtomicUsize::new(0)); + let req = req.map(|body| { let bytes_sent = Arc::clone(&bytes_sent); body.map_preserve_contents(move |body| { @@ -364,20 +497,25 @@ where }) }); - let fut = self.http.call(req); + let fut = self.call_inner(req); let region = self.region.clone(); - HttpConnectorFuture::new(fut.inspect(move |result| { - let byte_size = bytes_sent.load(Ordering::Relaxed); - if let Ok(result) = result - && result.status().is_success() + HttpConnectorFuture::new(async move { + let result = fut.await; + + if let Ok(response) = &result + && response.status().is_success() { + let byte_size = bytes_sent.load(Ordering::Relaxed); + emit!(AwsBytesSent { byte_size, region: Some(region), }); - } - })) + }; + + result + }) } } diff --git a/src/internal_events/http_client.rs b/src/internal_events/http_client.rs index 38e6fe94f8d7e..34752d4e34611 100644 --- a/src/internal_events/http_client.rs +++ b/src/internal_events/http_client.rs @@ -1,52 +1,138 @@ use std::time::Duration; use http::{ - Request, Response, - header::{self, HeaderMap, HeaderName, HeaderValue}, + HeaderMap, Request, Response, Version, + header::{self, HeaderName, HeaderValue}, }; -use hyper::{Error, body::HttpBody}; +use hyper::body::HttpBody; use vector_lib::{ NamedInternalEvent, counter, histogram, internal_event::{CounterName, HistogramName, InternalEvent, error_stage, error_type}, }; -#[derive(Debug, NamedInternalEvent)] -pub struct AboutToSendHttpRequest<'a, T> { - pub request: &'a Request, +// ── Telemetry traits ────────────────────────────────────────────────────────── + +/// Provides the data required to emit HTTP request telemetry. +/// +/// `method`, `uri`, `sanitized_headers`, and `body_debug` are required; every +/// transport must implement them. +/// +/// `version` is optional (default: `None`) because some transports — notably +/// the AWS SDK connector layer — do not expose HTTP version in their request +/// type. +pub trait HttpRequestTelemetry { + fn method(&self) -> &str; + fn uri(&self) -> String; + fn headers(&self) -> HeaderMap; + /// Returns the body size bounds as `(lower, upper)`. + fn body_size_hint(&self) -> (u64, Option); + /// Returns the HTTP version when the transport exposes it. + fn version(&self) -> Option { + None + } + + /// Transport-specific headers that carry credential material. + /// + /// The names returned here are marked sensitive in addition to the + /// generic list (Authorization, cookies, API keys, …) that + /// [`HttpRequestTelemetry::sanitized_headers`] always applies. Override this — rather than + /// [`HttpRequestTelemetry::sanitized_headers`] — when a transport adds its own credential headers. + fn extra_sensitive_headers(&self) -> &'static [HeaderName] { + &[] + } + + /// Returns the headers with sensitive values redacted. + fn sanitized_headers(&self) -> HeaderMap { + remove_sensitive(self.headers(), self.extra_sensitive_headers()) + } } -fn remove_sensitive(headers: &HeaderMap) -> HeaderMap { - let mut headers = headers.clone(); - let sensitive: &[HeaderName] = &[ - header::AUTHORIZATION, - header::PROXY_AUTHORIZATION, - header::PROXY_AUTHENTICATE, - header::WWW_AUTHENTICATE, - header::COOKIE, - header::SET_COOKIE, - HeaderName::from_static("cookie2"), - HeaderName::from_static("dd-api-key"), - HeaderName::from_static("x-honeycomb-team"), - HeaderName::from_static("x-api-key"), - HeaderName::from_static("api-key"), - ]; - for (name, value) in headers.iter_mut() { - if sensitive.contains(name) { - value.set_sensitive(true); - } +/// Provides the data required to emit HTTP response telemetry. +/// +/// Same rationale as [`HttpRequestTelemetry`]: `version` is optional. +pub trait HttpResponseTelemetry { + fn status_u16(&self) -> u16; + fn headers(&self) -> HeaderMap; + /// Returns the body size bounds as `(lower, upper)`. + fn body_size_hint(&self) -> (u64, Option); + /// Returns the HTTP version when the transport exposes it. + fn version(&self) -> Option { + None } - headers + + /// Transport-specific headers that carry credential material. + /// + /// See [`HttpRequestTelemetry::extra_sensitive_headers`] for details. + fn extra_sensitive_headers(&self) -> &'static [HeaderName] { + &[] + } + + /// Returns the headers with sensitive values redacted. + fn sanitized_headers(&self) -> HeaderMap { + remove_sensitive(self.headers(), self.extra_sensitive_headers()) + } +} + +// ── Implementations for the hyper HTTP types (full data) ────────────────────── + +impl HttpRequestTelemetry for Request { + fn method(&self) -> &str { + self.method().as_str() + } + + fn uri(&self) -> String { + self.uri().to_string() + } + + fn headers(&self) -> HeaderMap { + self.headers().clone() + } + + fn body_size_hint(&self) -> (u64, Option) { + let hint = self.body().size_hint(); + (hint.lower(), hint.upper()) + } + + fn version(&self) -> Option { + Some(self.version()) + } +} + +impl HttpResponseTelemetry for Response { + fn status_u16(&self) -> u16 { + self.status().as_u16() + } + + fn headers(&self) -> HeaderMap { + self.headers().clone() + } + + fn body_size_hint(&self) -> (u64, Option) { + let hint = self.body().size_hint(); + (hint.lower(), hint.upper()) + } + + fn version(&self) -> Option { + Some(self.version()) + } +} + +// ── Events ──────────────────────────────────────────────────────────────────── + +#[derive(Debug, NamedInternalEvent)] +pub struct AboutToSendHttpRequest<'a, T: HttpRequestTelemetry> { + pub request: &'a T, } -impl InternalEvent for AboutToSendHttpRequest<'_, T> { +impl InternalEvent for AboutToSendHttpRequest<'_, T> { fn emit(self) { debug!( message = "Sending HTTP request.", uri = %self.request.uri(), method = %self.request.method(), version = ?self.request.version(), - headers = ?remove_sensitive(self.request.headers()), - body = %FormatBody(self.request.body()), + headers = ?self.request.sanitized_headers(), + body = %FormatBodySizeHint::from(self.request.body_size_hint()), ); counter!(CounterName::HttpClientRequestsSentTotal, "method" => self.request.method().to_string()) .increment(1); @@ -54,37 +140,34 @@ impl InternalEvent for AboutToSendHttpRequest<'_, T> { } #[derive(Debug, NamedInternalEvent)] -pub struct GotHttpResponse<'a, T> { - pub response: &'a Response, +pub struct GotHttpResponse<'a, T: HttpResponseTelemetry> { + pub response: &'a T, pub roundtrip: Duration, } -impl InternalEvent for GotHttpResponse<'_, T> { +impl InternalEvent for GotHttpResponse<'_, T> { fn emit(self) { + let status = self.response.status_u16(); + let status_str = status.to_string(); debug!( message = "HTTP response.", - status = %self.response.status(), + status = %status, version = ?self.response.version(), - headers = ?remove_sensitive(self.response.headers()), - body = %FormatBody(self.response.body()), + headers = ?self.response.sanitized_headers(), + body = %FormatBodySizeHint::from(self.response.body_size_hint()), + ); - counter!( - CounterName::HttpClientResponsesTotal, - "status" => self.response.status().as_u16().to_string(), - ) - .increment(1); + counter!(CounterName::HttpClientResponsesTotal, "status" => status_str.clone()) + .increment(1); histogram!(HistogramName::HttpClientRttSeconds).record(self.roundtrip); - histogram!( - HistogramName::HttpClientResponseRttSeconds, - "status" => self.response.status().as_u16().to_string(), - ) - .record(self.roundtrip); + histogram!(HistogramName::HttpClientResponseRttSeconds, "status" => status_str) + .record(self.roundtrip); } } #[derive(Debug, NamedInternalEvent)] pub struct GotHttpWarning<'a> { - pub error: &'a Error, + pub error: &'a dyn std::error::Error, pub roundtrip: Duration, } @@ -104,13 +187,45 @@ impl InternalEvent for GotHttpWarning<'_> { } } -/// Newtype placeholder to provide a formatter for the request and response body. -struct FormatBody<'a, B>(&'a B); +// ── Helpers ─────────────────────────────────────────────────────────────────── + +fn remove_sensitive( + mut headers: HeaderMap, + extra: &'static [HeaderName], +) -> HeaderMap { + let sensitive: &[HeaderName] = &[ + header::AUTHORIZATION, + header::PROXY_AUTHORIZATION, + header::PROXY_AUTHENTICATE, + header::WWW_AUTHENTICATE, + header::COOKIE, + header::SET_COOKIE, + HeaderName::from_static("cookie2"), + HeaderName::from_static("dd-api-key"), + HeaderName::from_static("x-honeycomb-team"), + HeaderName::from_static("x-api-key"), + HeaderName::from_static("api-key"), + ]; + for (name, value) in headers.iter_mut() { + if sensitive.contains(name) || extra.contains(name) { + value.set_sensitive(true); + } + } + headers +} + +/// Formats a body size hint `(lower, upper)` for debug logging. +struct FormatBodySizeHint(u64, Option); + +impl From<(u64, Option)> for FormatBodySizeHint { + fn from((lower, upper): (u64, Option)) -> Self { + FormatBodySizeHint(lower, upper) + } +} -impl std::fmt::Display for FormatBody<'_, B> { +impl std::fmt::Display for FormatBodySizeHint { fn fmt(&self, fmt: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> { - let size = self.0.size_hint(); - match (size.lower(), size.upper()) { + match (self.0, self.1) { (0, None) => write!(fmt, "[unknown]"), (lower, None) => write!(fmt, "[>={lower} bytes]"), @@ -143,7 +258,7 @@ mod tests { header::AUTHORIZATION, HeaderValue::from_static("Bearer token"), ); - let result = remove_sensitive(&headers); + let result = remove_sensitive(headers, &[]); assert!( is_sensitive(&result, &header::AUTHORIZATION) .iter() @@ -159,7 +274,7 @@ mod tests { headers.append(x_api_key.clone(), HeaderValue::from_static("key-two")); headers.append(x_api_key.clone(), HeaderValue::from_static("key-three")); - let result = remove_sensitive(&headers); + let result = remove_sensitive(headers, &[]); let sensitive_flags = is_sensitive(&result, &x_api_key); assert_eq!(sensitive_flags.len(), 3); assert!( @@ -176,12 +291,22 @@ mod tests { HeaderName::from_static("x-api-key"), HeaderValue::from_static("secret"), ); - let result = remove_sensitive(&headers); + let result = remove_sensitive(headers, &[]); // Lookup with the mixed-case form resolves to the same normalized name. let mixed_case = HeaderName::from_bytes(b"X-Api-Key").unwrap(); assert!(is_sensitive(&result, &mixed_case).iter().all(|&s| s)); } + #[test] + fn extra_headers_are_marked_sensitive() { + static EXTRA: [HeaderName; 1] = [HeaderName::from_static("x-custom-secret")]; + let custom = HeaderName::from_static("x-custom-secret"); + let mut headers = HeaderMap::new(); + headers.insert(custom.clone(), HeaderValue::from_static("s3cr3t")); + let result = remove_sensitive(headers, &EXTRA); + assert!(is_sensitive(&result, &custom).iter().all(|&s| s)); + } + #[test] fn does_not_mark_non_sensitive_headers() { let mut headers = HeaderMap::new(); @@ -189,7 +314,7 @@ mod tests { header::CONTENT_TYPE, HeaderValue::from_static("application/json"), ); - let result = remove_sensitive(&headers); + let result = remove_sensitive(headers, &[]); assert!( is_sensitive(&result, &header::CONTENT_TYPE) .iter()