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

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 }

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<BoxFuture<'static, Result<Request<Bytes>>>>,
+    inner: HttpService<SematextMetricsRequestBuilder, ()>,
 }
 
 #[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,15 +187,16 @@ impl ValidatedSink for SematextMetricsConfig {
 }
 
 fn write_uri(endpoint: &HttpEndpoint) -> Result<Uri> {
-    encode_uri(
+    let uri = encode_uri(
         endpoint,
         "write",
         &[
             ("db", Some("metrics".into())),
             ("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<Vec<Metric>> for SematextMetricsService {
-    type Response = http::Response<bytes::Bytes>;
-    type Error = crate::Error;
+    type Response = HttpResponse;
+    type Error = HttpError;
     type Future = BoxFuture<'static, std::result::Result<Self::Response, Self::Error>>;
 
     fn poll_ready(
@@ -241,14 +252,22 @@ impl Service<Vec<Metric>> for SematextMetricsService {
     }
 
     fn call(&mut self, items: Vec<Metric>) -> 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<Request<Bytes>>> + 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<Bytes>> {
+        Request::post(self.uri.clone())
+            .header("Content-Type", "text/plain")
+            .body(request.take_payload())
+            .context(HTTPV1RequestBuilderSnafu)
+            .map_err(Into::into)
     }
 }