chore(azure_logs_ingestion sink): migrate azure_logs_ingestion sink to the native Hyper 1 client

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(),
         );

src/sinks/azure_logs_ingestion/service.rs

@@ -1,28 +1,28 @@
 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";
 
 /// JSON content type of logs
 const CONTENT_TYPE: &str = "application/json";
 
-static CONTENT_TYPE_VALUE: LazyLock<HeaderValue> =
-    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<http::uri::PathAndQuery> {
+) -> crate::Result<http_1::uri::PathAndQuery> {
     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<dyn TokenCredential>,
         token_scope: String,
     ) -> crate::Result<Self> {
@@ -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<Request<Body>> {
-        let mut request = Request::post(&self.endpoint).body(Body::from(body))?;
+    async fn build_request(&self, body: Bytes) -> crate::Result<Request<Bytes>> {
+        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<AzureLogsIngestionRequest> 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(),

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<Body>| {
+                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;