diff --git a/Cargo.lock b/Cargo.lock index 621ae63e7c633..61392f3c4cbfa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -12840,15 +12840,20 @@ version = "0.6.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d4e6559d53cc268e5031cd8429d05415bc4cb4aefc4aa5d6cc35fbf5b924a1f8" dependencies = [ + "async-compression", "base64 0.22.1", "bitflags 2.13.2", "bytes", + "futures-core", "futures-util", "http 1.5.0", "http-body 1.0.1", + "http-body-util", "iri-string", "mime", "pin-project-lite", + "tokio", + "tokio-util", "tower 0.5.3", "tower-layer", "tower-service", @@ -13653,6 +13658,7 @@ dependencies = [ "tonic-reflection", "tower 0.5.3", "tower-http 0.4.4", + "tower-http 0.6.8", "tower-test", "tracing 0.1.44", "tracing-core 0.1.36", diff --git a/Cargo.toml b/Cargo.toml index fea2a188f9b89..e8f1b03c82ee6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -357,6 +357,7 @@ opendal-http-transport-reqwest = { version = "0.58.2", default-features = false, # Tower tower = { version = "0.5.2", default-features = false, features = ["buffer", "limit", "retry", "timeout", "util", "balance", "discover"] } tower-http = { version = "0.4.4", default-features = false, features = ["compression-full", "decompression-full", "trace"] } +tower-http-1 = { package = "tower-http", version = "0.6.8", default-features = false, features = ["decompression-full"] } # Serde serde.workspace = true serde_bytes = { version = "0.11.17", default-features = false, features = ["std"], optional = true } diff --git a/src/sinks/sematext/metrics.rs b/src/sinks/sematext/metrics.rs index 601065e893b12..701f2d26fcdfc 100644 --- a/src/sinks/sematext/metrics.rs +++ b/src/sinks/sematext/metrics.rs @@ -1,13 +1,15 @@ -use std::{collections::HashMap, future::ready, task::Poll}; +use std::{collections::HashMap, task::Poll}; use bytes::{Bytes, BytesMut}; use futures::{FutureExt, SinkExt, future::BoxFuture, stream}; -use http::{StatusCode, Uri}; -use hyper::{Body, Request}; +use http::StatusCode; +use http_1::{Request, Uri}; use indoc::indoc; +use snafu::ResultExt; use tower::Service; use vector_lib::{ ByteSizeOf, EstimatedJsonEncodedSizeOf, configurable::configurable_component, + internal_event::CountByteSize, request_metadata::RequestMetadata, sensitive_string::SensitiveString, }; @@ -21,25 +23,29 @@ use crate::{ Event, KeyString, metric::{Metric, MetricValue}, }, - http::HttpClient, + http::client_v1::{HttpClient, HttpError, empty_body}, internal_events::{SematextMetricsEncodeEventError, SematextMetricsInvalidMetricError}, sinks::{ - Healthcheck, HealthcheckError, VectorSink, + HTTPV1RequestBuilderSnafu, Healthcheck, HealthcheckError, VectorSink, influxdb::{Field, ProtocolVersion, encode_timestamp, encode_uri, influx_line_protocol}, util::{ BatchConfig, BatchSettings, EncodedEvent, HttpEndpoint, SinkBatchSettings, TowerRequestConfig, buffer::metrics::{MetricNormalize, MetricNormalizer, MetricSet, MetricsBuffer}, - http::{HttpBatchService, HttpRetryLogic}, + http::{HttpRequest, RetryStrategy}, + http_v1::{ + HttpResponse, HttpService, HttpServiceRequestBuilder, http_response_retry_logic, + }, }, }, + tls::MaybeTlsSettings, vector_version, }; #[derive(Clone)] struct SematextMetricsService { config: SematextMetricsConfig, - inner: HttpBatchService>>>, + inner: HttpService, } #[derive(Clone, Copy, Debug, Default)] @@ -104,15 +110,17 @@ impl GenerateConfig for SematextMetricsConfig { async fn healthcheck(endpoint: String, client: HttpClient) -> Result<()> { let uri = HttpEndpoint::parse(&endpoint)? .append_path("health")? - .into_uri(); + .into_v1(); let request = Request::get(uri) - .body(Body::empty()) + .body(empty_body()) .map_err(|e| e.to_string())?; let response = client.send(request).await?; - match response.status() { + let status = StatusCode::from_u16(response.status().as_u16()) + .expect("HTTP status codes are valid u16 values"); + match status { StatusCode::OK | StatusCode::NO_CONTENT => Ok(()), other => Err(HealthcheckError::UnexpectedStatus { status: other }.into()), } @@ -169,7 +177,7 @@ impl ValidatedSink for SematextMetricsConfig { ) -> Result<(VectorSink, Healthcheck)> { let ValidatedSematextMetrics { endpoint, .. } = validated; - let client = HttpClient::new(None, cx.proxy())?; + let client = HttpClient::new(MaybeTlsSettings::from_config(None, false)?, cx.proxy())?; let healthcheck = healthcheck(endpoint.clone(), client.clone()).boxed(); let sink = SematextMetricsService::from_validated(self.clone(), validated, client); @@ -179,7 +187,7 @@ impl ValidatedSink for SematextMetricsConfig { } fn write_uri(endpoint: &HttpEndpoint) -> Result { - encode_uri( + let uri = encode_uri( endpoint, "write", &[ @@ -187,7 +195,8 @@ fn write_uri(endpoint: &HttpEndpoint) -> Result { ("v", Some(format!("vector-{}", vector_version()))), ("precision", Some("ns".into())), ], - ) + )?; + Ok(HttpEndpoint::new(uri)?.into_v1()) } impl SematextMetricsService { @@ -198,7 +207,9 @@ impl SematextMetricsService { ) -> VectorSink { let ValidatedSematextMetrics { uri, batch, .. } = validated; let request = config.request.into_settings(); - let http_service = HttpBatchService::new(client, create_build_request(uri.clone())); + let http_service = + HttpService::new(client, SematextMetricsRequestBuilder { uri: uri.clone() }) + .with_response_decompression(); let sematext_service = SematextMetricsService { config, inner: http_service, @@ -207,7 +218,7 @@ impl SematextMetricsService { let sink = request .batch_sink( - HttpRetryLogic::default(), + http_response_retry_logic(RetryStrategy::Default), sematext_service, MetricsBuffer::new(batch.size), batch.timeout, @@ -229,8 +240,8 @@ impl SematextMetricsService { } impl Service> for SematextMetricsService { - type Response = http::Response; - type Error = crate::Error; + type Response = HttpResponse; + type Error = HttpError; type Future = BoxFuture<'static, std::result::Result>; fn poll_ready( @@ -241,14 +252,22 @@ impl Service> for SematextMetricsService { } fn call(&mut self, items: Vec) -> Self::Future { + let event_count = items.len(); let input = encode_events( self.config.token.inner(), &self.config.default_namespace, items, ); - let body = input.item; + let metadata = RequestMetadata::new( + event_count, + input.byte_size, + input.item.len(), + input.item.len(), + CountByteSize(event_count, input.json_byte_size).into(), + ); - self.inner.call(body) + self.inner + .call(HttpRequest::new(input.item, input.finalizers, metadata, ())) } } @@ -268,16 +287,18 @@ impl MetricNormalize for SematextMetricNormalize { } } -fn create_build_request( - uri: http::Uri, -) -> impl Fn(Bytes) -> BoxFuture<'static, Result>> + Sync + Send + 'static { - move |body| { - Box::pin(ready( - Request::post(uri.clone()) - .header("Content-Type", "text/plain") - .body(body) - .map_err(Into::into), - )) +#[derive(Clone)] +struct SematextMetricsRequestBuilder { + uri: Uri, +} + +impl HttpServiceRequestBuilder<()> for SematextMetricsRequestBuilder { + fn build(&self, mut request: HttpRequest<()>) -> Result> { + Request::post(self.uri.clone()) + .header("Content-Type", "text/plain") + .body(request.take_payload()) + .context(HTTPV1RequestBuilderSnafu) + .map_err(Into::into) } } diff --git a/src/sinks/util/http_v1.rs b/src/sinks/util/http_v1.rs index fb47b1642b43f..881a805488a03 100644 --- a/src/sinks/util/http_v1.rs +++ b/src/sinks/util/http_v1.rs @@ -8,7 +8,8 @@ use bytes::Bytes; use futures::future::BoxFuture; use http_1::{Request, Response, Uri}; use http_body_util::BodyExt; -use tower::Service; +use tower::{Service, service_fn}; +use tower_http_1::decompression::Decompression; use tracing::debug; use vector_lib::{ event::EventStatus, @@ -76,6 +77,7 @@ impl DriverResponse for HttpResponse { pub struct HttpService { client: HttpClient, request_builder: Arc, + response_decompression: bool, _phantom: PhantomData T>, } @@ -88,9 +90,19 @@ where Self { client, request_builder: Arc::new(request_builder), + response_decompression: false, _phantom: PhantomData, } } + + /// Requests and decodes compressed responses, matching legacy HTTP batch services. + /// + /// Other services continue to request identity encoding. + #[must_use] + pub const fn with_response_decompression(mut self) -> Self { + self.response_decompression = true; + self + } } impl Service> for HttpService @@ -109,6 +121,7 @@ where fn call(&mut self, mut request: HttpRequest) -> Self::Future { let request_builder = Arc::clone(&self.request_builder); let client = self.client.clone(); + let response_decompression = self.response_decompression; let metadata = std::mem::take(request.metadata_mut()); let raw_byte_size = metadata.request_encoded_size(); let events_byte_size = metadata.into_events_estimated_json_encoded_byte_size(); @@ -122,20 +135,21 @@ where .map_err(HttpError::from)?; let byte_size = request.body().len(); let (protocol, endpoint) = protocol_endpoint(request.uri()); - let response = client.send(request.map(full_body)).await?; - - if response.status().is_success() { - emit!(EndpointBytesSent { - byte_size, - protocol: &protocol, - endpoint: &endpoint, - }); - } - - let (parts, body) = response.into_parts(); - let body = body.collect().await.map_err(HttpError::new)?.to_bytes(); + let request = request.map(full_body); + let response = if response_decompression { + let mut service = Decompression::new(service_fn(|request| client.send(request))); + let response = service.call(request).await?; + collect_response(response, byte_size, &protocol, &endpoint) + .await + .map_err(HttpError::from)? + } else { + let response = client.send(request).await?; + collect_response(response, byte_size, &protocol, &endpoint) + .await + .map_err(HttpError::new)? + }; Ok(HttpResponse { - response: Response::from_parts(parts, body), + response, events_byte_size, raw_byte_size, }) @@ -143,6 +157,28 @@ where } } +async fn collect_response( + response: Response, + byte_size: usize, + protocol: &str, + endpoint: &str, +) -> Result, B::Error> +where + B: http_body_1::Body, +{ + if response.status().is_success() { + emit!(EndpointBytesSent { + byte_size, + protocol, + endpoint, + }); + } + + let (parts, body) = response.into_parts(); + let body = body.collect().await?.to_bytes(); + Ok(Response::from_parts(parts, body)) +} + /// Retry logic for sinks that use [`HttpService`], based on the response status code. /// /// ClickHouse implements its own retry logic; other native http 1 sinks use this. @@ -190,3 +226,106 @@ fn protocol_endpoint(uri: &Uri) -> (String, String) { .expect("a valid HTTP/1 URI is valid as an HTTP URI"); super::uri::protocol_endpoint(uri) } + +#[cfg(test)] +mod tests { + use std::{convert::Infallible, io::Write}; + + use flate2::write::GzEncoder; + use vector_lib::{finalization::EventFinalizers, request_metadata::RequestMetadata}; + + use super::*; + use crate::{ + config::ProxyConfig, + sinks::util::{ + http::RetryStrategy, + retries::{RetryAction, RetryLogic}, + }, + test_util::http::spawn_blackhole_http_server, + tls::MaybeTlsSettings, + }; + + #[derive(Clone)] + struct RequestBuilder { + uri: Uri, + } + + impl HttpServiceRequestBuilder<()> for RequestBuilder { + fn build(&self, mut request: HttpRequest<()>) -> Result, crate::Error> { + Ok(Request::post(self.uri.clone()).body(request.take_payload())?) + } + } + + async fn compressed_response_service(body: Vec) -> HttpService { + let uri = spawn_blackhole_http_server(move |request| { + let body = body.clone(); + async move { + assert_eq!(request.headers()["accept-encoding"], "zstd,gzip,deflate,br"); + let request_body = http_body::Body::collect(request.into_body()) + .await + .unwrap() + .to_bytes(); + assert_eq!(request_body, "events"); + Ok::<_, Infallible>( + http::Response::builder() + .status(http::StatusCode::SERVICE_UNAVAILABLE) + .header("content-encoding", "gzip") + .header("content-length", body.len()) + .body(hyper::Body::from(body)) + .unwrap(), + ) + } + }) + .await; + let client = HttpClient::new( + MaybeTlsSettings::from_config(None, false).unwrap(), + &ProxyConfig::default(), + ) + .unwrap(); + let builder = RequestBuilder { + uri: crate::sinks::util::HttpEndpoint::new(uri) + .unwrap() + .into_v1(), + }; + HttpService::new(client, builder).with_response_decompression() + } + + fn request() -> HttpRequest<()> { + HttpRequest::new( + Bytes::from_static(b"events"), + EventFinalizers::default(), + RequestMetadata::default(), + (), + ) + } + + #[tokio::test] + async fn decompresses_retry_response() { + let message = b"retry after the service has recovered"; + let mut encoder = GzEncoder::new(Vec::new(), flate2::Compression::default()); + encoder.write_all(message).unwrap(); + let mut service = compressed_response_service(encoder.finish().unwrap()).await; + let response = service.call(request()).await.unwrap(); + + assert_eq!( + response.response.status(), + http_1::StatusCode::SERVICE_UNAVAILABLE + ); + assert_eq!(response.response.body().as_ref(), message); + assert!(!response.response.headers().contains_key("content-encoding")); + assert!(!response.response.headers().contains_key("content-length")); + let retry_logic = http_response_retry_logic::>(RetryStrategy::Default); + assert!(matches!( + retry_logic.should_retry_response(&response), + RetryAction::Retry(_) + )); + } + + #[tokio::test] + async fn rejects_invalid_compressed_response() { + let mut service = compressed_response_service(b"not a gzip response".to_vec()).await; + let error = service.call(request()).await.unwrap_err(); + let retry_logic = http_response_retry_logic::>(RetryStrategy::Default); + assert!(!retry_logic.is_retriable_error(&error)); + } +}