From ee447a29f4901835afa91ba275d308cb59564163 Mon Sep 17 00:00:00 2001 From: Thomas Date: Tue, 20 Oct 2026 16:00:26 -0400 Subject: [PATCH 1/2] chore(sinks): accept HTTP 1 status codes in shared retry and healthcheck helpers --- src/sinks/clickhouse/config.rs | 10 ++--- src/sinks/clickhouse/service.rs | 7 ++-- src/sinks/gcp/stackdriver/logs/config.rs | 5 +-- src/sinks/gcs_common/config.rs | 7 ++-- .../greptimedb/logs/http_request_builder.rs | 9 ++--- src/sinks/honeycomb/config.rs | 6 +-- src/sinks/keep/config.rs | 6 +-- src/sinks/mod.rs | 11 +++++ src/sinks/util/http.rs | 40 +++++++++++++++---- src/sinks/util/http_v1.rs | 10 +---- 10 files changed, 65 insertions(+), 46 deletions(-) diff --git a/src/sinks/clickhouse/config.rs b/src/sinks/clickhouse/config.rs index 376534ab76e5b..2b358d7338ec3 100644 --- a/src/sinks/clickhouse/config.rs +++ b/src/sinks/clickhouse/config.rs @@ -2,8 +2,8 @@ use std::fmt; -use http::{StatusCode, Uri}; -use http_1::Request; +use http::Uri; +use http_1::{Request, StatusCode}; use vector_lib::codecs::encoding::ArrowStreamSerializerConfig; use vector_lib::codecs::encoding::format::SchemaProvider; use vector_lib::codecs::{ @@ -476,11 +476,9 @@ async fn healthcheck(client: HttpClient, endpoint: Uri, auth: Option) -> c let response = client.send(request).await?; - let status = StatusCode::from_u16(response.status().as_u16()) - .expect("HTTP status codes are valid u16 values"); - match status { + match response.status() { StatusCode::OK => Ok(()), - status => Err(HealthcheckError::UnexpectedStatus { status }.into()), + status => Err(HealthcheckError::unexpected_status(status).into()), } } diff --git a/src/sinks/clickhouse/service.rs b/src/sinks/clickhouse/service.rs index de956f9bc9bd6..1ae8c09963283 100644 --- a/src/sinks/clickhouse/service.rs +++ b/src/sinks/clickhouse/service.rs @@ -3,9 +3,9 @@ use std::fmt::Write as _; use bytes::Bytes; -use http::{StatusCode, Uri}; +use http::Uri; use http_1::{ - Request, + Request, StatusCode, header::{CONTENT_ENCODING, CONTENT_LENGTH, CONTENT_TYPE}, }; use snafu::ResultExt; @@ -39,8 +39,7 @@ impl RetryLogic for ClickhouseRetryLogic { } fn should_retry_response(&self, response: &Self::Response) -> RetryAction { - let status = StatusCode::from_u16(response.response.status().as_u16()) - .expect("HTTP status codes are valid u16 values"); + let status = response.response.status(); if !status.is_success() { debug!( diff --git a/src/sinks/gcp/stackdriver/logs/config.rs b/src/sinks/gcp/stackdriver/logs/config.rs index 88c41b4a1fd0c..0a14b40039067 100644 --- a/src/sinks/gcp/stackdriver/logs/config.rs +++ b/src/sinks/gcp/stackdriver/logs/config.rs @@ -2,7 +2,6 @@ use std::collections::HashMap; -use http::StatusCode; use http_1::Request; use snafu::Snafu; use vector_lib::lookup::lookup_v2::ConfigValuePath; @@ -380,9 +379,7 @@ async fn healthcheck( let response = client.send(request).await?; - let status = StatusCode::from_u16(response.status().as_u16()) - .expect("HTTP status codes are valid u16 values"); - healthcheck_status(status, HealthcheckError::NotFound.into()) + healthcheck_status(response.status(), HealthcheckError::NotFound.into()) } #[cfg(test)] diff --git a/src/sinks/gcs_common/config.rs b/src/sinks/gcs_common/config.rs index 4ac3083fbf67d..31af3d9df9a87 100644 --- a/src/sinks/gcs_common/config.rs +++ b/src/sinks/gcs_common/config.rs @@ -14,6 +14,7 @@ use crate::{ gcs_common::service::GcsResponse, util::{ HttpEndpoint, + http::IntoStatusCode, retries::{RetryAction, RetryLogic}, }, }, @@ -155,14 +156,14 @@ pub fn healthcheck_response( /// # Errors /// Returns an error for forbidden, missing, or unexpected response statuses. pub fn healthcheck_status( - status: http::StatusCode, + status: impl IntoStatusCode, not_found_error: crate::Error, ) -> crate::Result<()> { - match status { + match status.into_status_code() { StatusCode::OK => Ok(()), StatusCode::FORBIDDEN => Err(GcpError::HealthcheckForbidden.into()), StatusCode::NOT_FOUND => Err(not_found_error), - status => Err(HealthcheckError::UnexpectedStatus { status }.into()), + status => Err(HealthcheckError::unexpected_status(status).into()), } } diff --git a/src/sinks/greptimedb/logs/http_request_builder.rs b/src/sinks/greptimedb/logs/http_request_builder.rs index c0eb12be82221..88b83c313a897 100644 --- a/src/sinks/greptimedb/logs/http_request_builder.rs +++ b/src/sinks/greptimedb/logs/http_request_builder.rs @@ -1,9 +1,8 @@ use std::collections::HashMap; use bytes::Bytes; -use http::StatusCode; use http_1::{ - Request, + Request, StatusCode, header::{CONTENT_ENCODING, CONTENT_LENGTH, CONTENT_TYPE}, }; use snafu::ResultExt; @@ -264,11 +263,9 @@ pub(super) async fn http_healthcheck( let response = client.send(request).await?; - let status = StatusCode::from_u16(response.status().as_u16()) - .expect("HTTP status codes are valid u16 values"); - match status { + match response.status() { StatusCode::OK => Ok(()), - status => Err(HealthcheckError::UnexpectedStatus { status }.into()), + status => Err(HealthcheckError::unexpected_status(status).into()), } } diff --git a/src/sinks/honeycomb/config.rs b/src/sinks/honeycomb/config.rs index f8fd60398133b..2fddfcfebb5d1 100644 --- a/src/sinks/honeycomb/config.rs +++ b/src/sinks/honeycomb/config.rs @@ -1,7 +1,6 @@ use bytes::Bytes; use futures::FutureExt; -use http::StatusCode; -use http_1::{HeaderValue, Request}; +use http_1::{HeaderValue, Request, StatusCode}; use http_body_util::BodyExt; use vector_lib::{configurable::configurable_component, sensitive_string::SensitiveString}; use vrl::value::Kind; @@ -207,8 +206,7 @@ async fn healthcheck( let res = client.send(req).await?; - let status = StatusCode::from_u16(res.status().as_u16()) - .expect("HTTP status codes are valid u16 values"); + let status = res.status(); let body = res.into_body().collect().await?.to_bytes(); if status == StatusCode::BAD_REQUEST { diff --git a/src/sinks/keep/config.rs b/src/sinks/keep/config.rs index f7588f14ebd7a..823865648ffcd 100644 --- a/src/sinks/keep/config.rs +++ b/src/sinks/keep/config.rs @@ -2,8 +2,7 @@ use bytes::Bytes; use futures::FutureExt; -use http::StatusCode; -use http_1::Request; +use http_1::{Request, StatusCode}; use http_body_util::BodyExt; use vector_lib::{configurable::configurable_component, sensitive_string::SensitiveString}; use vrl::value::Kind; @@ -184,8 +183,7 @@ async fn healthcheck( let res = client.send(req).await?; - let status = StatusCode::from_u16(res.status().as_u16()) - .expect("HTTP status codes are valid u16 values"); + let status = res.status(); let body = res.into_body().collect().await?.to_bytes(); match status { diff --git a/src/sinks/mod.rs b/src/sinks/mod.rs index 604b30c2f386c..520a94d1b7b7d 100644 --- a/src/sinks/mod.rs +++ b/src/sinks/mod.rs @@ -148,3 +148,14 @@ pub enum HealthcheckError { #[snafu(display("Unexpected status: {}", status))] UnexpectedStatus { status: ::http::StatusCode }, } + +impl HealthcheckError { + /// Builds an [`HealthcheckError::UnexpectedStatus`] from a status code of either `http` + /// crate version. + #[must_use] + pub fn unexpected_status(status: impl util::http::IntoStatusCode) -> Self { + Self::UnexpectedStatus { + status: status.into_status_code(), + } + } +} diff --git a/src/sinks/util/http.rs b/src/sinks/util/http.rs index 00acef3debaa9..5b62fbfe8a552 100644 --- a/src/sinks/util/http.rs +++ b/src/sinks/util/http.rs @@ -635,7 +635,8 @@ impl RetryStrategy { /// /// For the `All` strategy, all non-success status codes will be retried. #[must_use] - pub fn retry_action(&self, status: http::StatusCode) -> RetryAction { + pub fn retry_action(&self, status: impl IntoStatusCode) -> RetryAction { + let status = status.into_status_code(); if status.is_success() { return RetryAction::Successful; } @@ -674,6 +675,28 @@ impl RetryStrategy { } } +/// A response status code from either `http` crate version. +/// +/// Retry logic and healthcheck errors accept both versions through this trait, so callers never +/// convert between them. This is the only place that converts. +// TODO: Use `http_1::StatusCode` natively and remove the `http` 0.2 impl once the HTTP v1 +// migration is complete: https://github.com/vectordotdev/vector/issues/19179 +pub trait IntoStatusCode { + fn into_status_code(self) -> StatusCode; +} + +impl IntoStatusCode for StatusCode { + fn into_status_code(self) -> StatusCode { + self + } +} + +impl IntoStatusCode for http_1::StatusCode { + fn into_status_code(self) -> StatusCode { + StatusCode::from_u16(self.as_u16()).expect("HTTP status codes are valid u16 values") + } +} + #[derive(Debug, Clone)] pub struct HttpRetryLogic { request: PhantomData, @@ -743,9 +766,10 @@ impl HttpErrorClassify for HttpError { } } -impl HttpStatusRetryLogic +impl HttpStatusRetryLogic where - F: Fn(&Res) -> StatusCode + Clone + Send + Sync + 'static, + F: Fn(&Res) -> S + Clone + Send + Sync + 'static, + S: IntoStatusCode, Req: Send + Sync + 'static, Res: Send + Sync + 'static, { @@ -760,9 +784,10 @@ where } } -impl HttpStatusRetryLogic +impl HttpStatusRetryLogic where - F: Fn(&Res) -> StatusCode + Clone + Send + Sync + 'static, + F: Fn(&Res) -> S + Clone + Send + Sync + 'static, + S: IntoStatusCode, Req: Send + Sync + 'static, Res: Send + Sync + 'static, E: HttpErrorClassify + std::error::Error + Send + Sync + 'static, @@ -778,9 +803,10 @@ where } } -impl RetryLogic for HttpStatusRetryLogic +impl RetryLogic for HttpStatusRetryLogic where - F: Fn(&Res) -> StatusCode + Clone + Send + Sync + 'static, + F: Fn(&Res) -> S + Clone + Send + Sync + 'static, + S: IntoStatusCode, Req: Send + Sync + 'static, Res: Send + Sync + 'static, E: HttpErrorClassify + std::error::Error + Send + Sync + 'static, diff --git a/src/sinks/util/http_v1.rs b/src/sinks/util/http_v1.rs index fb47b1642b43f..0e9814db75dff 100644 --- a/src/sinks/util/http_v1.rs +++ b/src/sinks/util/http_v1.rs @@ -146,24 +146,18 @@ where /// 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. -/// -/// # Panics -/// -/// Does not panic in practice: every `http 1` status code is also a valid `http 0.2` status -/// code. #[must_use] pub fn http_response_retry_logic( retry_strategy: super::http::RetryStrategy, ) -> super::http::HttpStatusRetryLogic< - impl Fn(&HttpResponse) -> http::StatusCode + Clone + Send + Sync + 'static, + impl Fn(&HttpResponse) -> http_1::StatusCode + Clone + Send + Sync + 'static, Request, HttpResponse, HttpError, > { super::http::HttpStatusRetryLogic::new_with_error( |response: &HttpResponse| { - let status = http::StatusCode::from_u16(response.response.status().as_u16()) - .expect("HTTP status codes are valid u16 values"); + let status = response.response.status(); if !status.is_success() { debug!( message = "HTTP response.", From ac763bb3cd5eb8aebd92aa4d49f02abbdb86a205 Mon Sep 17 00:00:00 2001 From: Thomas Date: Tue, 20 Oct 2026 16:19:38 -0400 Subject: [PATCH 2/2] chore(azure_logs_ingestion sink): migrate azure_logs_ingestion sink to the native Hyper 1 client --- src/sinks/azure_logs_ingestion/config.rs | 17 ++--- src/sinks/azure_logs_ingestion/service.rs | 48 +++++++++------ src/sinks/azure_logs_ingestion/tests.rs | 75 +++++++++++++++++++++++ 3 files changed, 112 insertions(+), 28 deletions(-) diff --git a/src/sinks/azure_logs_ingestion/config.rs b/src/sinks/azure_logs_ingestion/config.rs index 848794fb7818c..8edfee42e0168 100644 --- a/src/sinks/azure_logs_ingestion/config.rs +++ b/src/sinks/azure_logs_ingestion/config.rs @@ -1,15 +1,17 @@ use std::sync::Arc; use azure_core::credentials::TokenCredential; - -use http::uri::PathAndQuery; +use http_1::uri::PathAndQuery; use vector_lib::{configurable::configurable_component, schema}; use vrl::value::Kind; use crate::{ config::ValidatedSink, - http::{HttpClient, get_http_scheme_from_uri}, + http::{ + client_v1::{HttpClient, HttpError}, + get_http_scheme_from_uri, + }, sinks::{ azure_common::config::AzureAuthentication, prelude::*, @@ -137,22 +139,21 @@ impl AzureLogsIngestionConfig { token_scope: String, timestamp_field: String, ) -> crate::Result<(VectorSink, Healthcheck)> { - let endpoint = endpoint.into_uri(); - let protocol = get_http_scheme_from_uri(&endpoint).to_string(); + let protocol = get_http_scheme_from_uri(&endpoint.clone().into_uri()).to_string(); let tls_settings = TlsSettings::from_options(self.tls.as_ref())?; - let client = HttpClient::new(Some(tls_settings), &cx.proxy)?; + let client = HttpClient::new(tls_settings.into(), &cx.proxy)?; let service = AzureLogsIngestionService::new( client, - endpoint, + endpoint.into_v1(), validated.path_and_query.clone(), credential, token_scope, )?; let healthcheck = service.healthcheck(); - let retry_logic = HttpStatusRetryLogic::new( + let retry_logic = HttpStatusRetryLogic::<_, _, _, HttpError>::new_with_error( |res: &AzureLogsIngestionResponse| res.http_status, self.retry_strategy.clone(), ); diff --git a/src/sinks/azure_logs_ingestion/service.rs b/src/sinks/azure_logs_ingestion/service.rs index 211193217e50e..ad5f7e1082c26 100644 --- a/src/sinks/azure_logs_ingestion/service.rs +++ b/src/sinks/azure_logs_ingestion/service.rs @@ -1,18 +1,21 @@ use std::sync::Arc; -use std::sync::LazyLock; use std::task::{Context, Poll}; use azure_core::credentials::{AccessToken, TokenCredential}; use bytes::Bytes; -use http::{ +use http_1::{ HeaderValue, Request, StatusCode, Uri, header::{self, HeaderMap}, }; -use hyper::Body; +use http_body_util::BodyExt; +use snafu::ResultExt; use tracing::Instrument; -use crate::{http::HttpClient, sinks::prelude::*}; +use crate::{ + http::client_v1::{HttpClient, full_body}, + sinks::{HTTPV1RequestBuilderSnafu, prelude::*}, +}; /// Log Ingestion API version const API_VERSION: &str = "2023-01-01"; @@ -20,9 +23,6 @@ const API_VERSION: &str = "2023-01-01"; /// JSON content type of logs const CONTENT_TYPE: &str = "application/json"; -static CONTENT_TYPE_VALUE: LazyLock = - LazyLock::new(|| HeaderValue::from_static(CONTENT_TYPE)); - #[derive(Debug, Clone)] pub struct AzureLogsIngestionRequest { pub body: Bytes, @@ -85,7 +85,7 @@ pub struct AzureLogsIngestionService { pub(super) fn request_path( dcr_immutable_id: &str, stream_name: &str, -) -> crate::Result { +) -> crate::Result { format!( "/dataCollectionRules/{dcr_immutable_id}/streams/{stream_name}?api-version={API_VERSION}" ) @@ -98,7 +98,7 @@ impl AzureLogsIngestionService { pub fn new( client: HttpClient, endpoint: Uri, - path_and_query: http::uri::PathAndQuery, + path_and_query: http_1::uri::PathAndQuery, credential: Arc, token_scope: String, ) -> crate::Result { @@ -109,7 +109,7 @@ impl AzureLogsIngestionService { let default_headers = { let mut headers = HeaderMap::new(); - headers.insert(header::CONTENT_TYPE, CONTENT_TYPE_VALUE.clone()); + headers.insert(header::CONTENT_TYPE, HeaderValue::from_static(CONTENT_TYPE)); headers }; @@ -122,8 +122,10 @@ impl AzureLogsIngestionService { }) } - async fn build_request(&self, body: Bytes) -> crate::Result> { - let mut request = Request::post(&self.endpoint).body(Body::from(body))?; + async fn build_request(&self, body: Bytes) -> crate::Result> { + let mut request = Request::post(&self.endpoint) + .body(body) + .context(HTTPV1RequestBuilderSnafu)?; // TODO: make this an option, for sovereign clouds let access_token: AccessToken = self @@ -144,19 +146,22 @@ impl AzureLogsIngestionService { pub fn healthcheck(&self) -> Healthcheck { let service = self.clone(); - let mut client = service.client.clone(); Box::pin(async move { let request = service.build_request(Bytes::from("[]")).await?; - let res = client.call(request).in_current_span().await?; - - match res.status() { + let res = service + .client + .send(request.map(full_body)) + .in_current_span() + .await?; + + let status = res.status(); + match status { StatusCode::NO_CONTENT => Ok(()), StatusCode::UNAUTHORIZED => Err("Azure returned 401 Unauthorised. Check that the token_scope matches the sovereign cloud endpoint.".into()), StatusCode::FORBIDDEN => Err("Azure returned 403 Forbidden. Verify that the credential has the Monitoring Metrics Publisher role on the Data Collection Rule.".into()), StatusCode::NOT_FOUND => Err("Azure returned 404 Not Found. Either the URL provided is incorrect, or the request is too large.".into()), _ => { - let status = res.status(); - let body_bytes: Bytes = http_body::Body::collect(res.into_body()).await?.to_bytes(); + let body_bytes: Bytes = res.into_body().collect().await?.to_bytes(); let body_string: String = String::from_utf8(body_bytes.to_vec()).unwrap(); let err_string: String = format!("Azure returned {status}: {body_string}"); Err(err_string.into()) @@ -179,10 +184,13 @@ impl Service for AzureLogsIngestionService { // Emission of Error internal event is handled upstream by the caller. fn call(&mut self, request: AzureLogsIngestionRequest) -> Self::Future { let service = self.clone(); - let mut client = service.client.clone(); Box::pin(async move { let http_request = service.build_request(request.body).await?; - let response = client.call(http_request).in_current_span().await?; + let response = service + .client + .send(http_request.map(full_body)) + .in_current_span() + .await?; Ok(AzureLogsIngestionResponse { http_status: response.status(), raw_byte_size: request.metadata.request_encoded_size(), diff --git a/src/sinks/azure_logs_ingestion/tests.rs b/src/sinks/azure_logs_ingestion/tests.rs index 0ad43228624a5..edb385b198010 100644 --- a/src/sinks/azure_logs_ingestion/tests.rs +++ b/src/sinks/azure_logs_ingestion/tests.rs @@ -298,6 +298,81 @@ async fn correct_request() { ); } +#[tokio::test] +async fn retries_transport_error() { + use hyper::{ + Body, Server, + service::{make_service_fn, service_fn}, + }; + use std::{ + convert::Infallible, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + }; + + let (_guard, addr) = crate::test_util::addr::next_addr(); + let requests = Arc::new(AtomicUsize::new(0)); + let server_requests = Arc::clone(&requests); + let server = Server::bind(&addr).serve(make_service_fn(move |_| { + let requests = Arc::clone(&server_requests); + async move { + Ok::<_, Infallible>(service_fn(move |request: http::Request| { + let requests = Arc::clone(&requests); + async move { + http_body::Body::collect(request.into_body()).await.unwrap(); + if requests.fetch_add(1, Ordering::Relaxed) == 0 { + Err(std::io::Error::other( + "close the first connection without a response", + )) + } else { + Ok(Response::builder().status(204).body(Body::empty()).unwrap()) + } + } + })) + } + })); + let server = tokio::spawn(server); + let config: AzureLogsIngestionConfig = serde_yaml::from_str(indoc::indoc! {r#" + endpoint: "http://localhost:9001" + dcr_immutable_id: dcr-00000000000000000000000000000000 + stream_name: Custom-UnitTest + request: + retry_attempts: 1 + auth: + azure_credential_kind: client_secret_credential + azure_tenant_id: "00000000-0000-0000-0000-000000000000" + azure_client_id: mock-client-id + azure_client_secret: mock-client-secret + "#}) + .unwrap(); + let (sink, _) = config + .build_inner( + &SinkContext::default(), + &config.validate().unwrap(), + HttpEndpoint::parse(&format!("http://{addr}")).unwrap(), + Arc::new(create_mock_credential()), + config.token_scope.clone(), + config.timestamp_field.clone(), + ) + .unwrap(); + let (batch, receiver) = crate::event::BatchNotifier::new_with_receiver(); + let mut log = LogEvent::from("retry after connection closes"); + insert_timestamp_kv(&mut log); + let events = + crate::test_util::map_event_batch_stream(stream::iter([Event::Log(log)]), Some(batch)); + let status = timeout(Duration::from_secs(10), async { + sink.run(events).await.unwrap(); + receiver.await + }) + .await; + server.abort(); + + assert_eq!(status.unwrap(), crate::event::BatchStatus::Delivered); + assert_eq!(requests.load(Ordering::Relaxed), 2); +} + fn create_mock_credential() -> impl TokenCredential { #[derive(Debug)] struct MockCredential;