chore(http sink): refactor to new style

docs/tutorials/sinks/2_http_sink.md

@@ -290,6 +290,12 @@ where the data is actually sent.
 
 # Service
 
+**⚠ NOTE! This section implements an HTTP tower `Service` from scratch, for the
+purpose of demonstration only. Many sinks will require implementing `Service`
+in this way. Any new HTTP based sink should ideally utilize the structure
+`HttpService`, which handles most of what this section discusses in a shared
+structure that HTTP based sinks can utilize.**
+
 We need to create a [`Tower`][tower] service that is responsible for actually
 sending our final encoded data.
 

src/sinks/appsignal/service.rs

@@ -24,6 +24,9 @@ use super::request_builder::AppsignalRequest;
 
 #[derive(Clone)]
 pub(super) struct AppsignalService {
+    // TODO: `HttpBatchService` has been deprecated for direct use in sinks.
+    //       This sink should undergo a refactor to utilize the `HttpService`
+    //       instead, which extracts much of the boilerplate code for `Service`.
     pub(super) batch_service:
         HttpBatchService<Ready<Result<http::Request<Bytes>, crate::Error>>, AppsignalRequest>,
 }

src/sinks/datadog/events/service.rs

@@ -43,6 +43,9 @@ impl DriverResponse for DatadogEventsResponse {
 
 #[derive(Clone)]
 pub struct DatadogEventsService {
+    // TODO: `HttpBatchService` has been deprecated for direct use in sinks.
+    //       This sink should undergo a refactor to utilize the `HttpService`
+    //       instead, which extracts much of the boilerplate code for `Service`.
     batch_http_service:
         HttpBatchService<Ready<Result<http::Request<Bytes>, crate::Error>>, DatadogEventsRequest>,
 }

src/sinks/elasticsearch/service.rs

@@ -68,6 +68,9 @@ impl MetaDescriptive for ElasticsearchRequest {
 
 #[derive(Clone)]
 pub struct ElasticsearchService {
+    // TODO: `HttpBatchService` has been deprecated for direct use in sinks.
+    //       This sink should undergo a refactor to utilize the `HttpService`
+    //       instead, which extracts much of the boilerplate code for `Service`.
     batch_service: HttpBatchService<
         BoxFuture<'static, Result<http::Request<Bytes>, crate::Error>>,
         ElasticsearchRequest,

src/sinks/http/batch.rs

@@ -16,7 +16,9 @@ pub(super) struct HttpBatchSizer {
 impl ItemBatchSize<Event> for HttpBatchSizer {
     fn size(&self, item: &Event) -> usize {
         match self.encoder.serializer() {
-            codecs::encoding::Serializer::Json(_) => item.estimated_json_encoded_size_of().get(),
+            codecs::encoding::Serializer::Json(_) | codecs::encoding::Serializer::NativeJson(_) => {
+                item.estimated_json_encoded_size_of().get()
+            }
             _ => item.size_of(),
         }
     }

src/sinks/http/config.rs

@@ -275,7 +275,7 @@ impl SinkConfig for HttpSinkConfig {
                 .to_string()
         });
 
-        let http_service_request_builder = HttpSinkRequestBuilder::new(
+        let http_sink_request_builder = HttpSinkRequestBuilder::new(
             self.uri.with_default_parts(),
             self.method,
             self.auth.choose_one(&self.uri.auth)?,
@@ -284,7 +284,7 @@ impl SinkConfig for HttpSinkConfig {
             content_encoding,
         );
 
-        let service = HttpService::new(client, http_service_request_builder);
+        let service = HttpService::new(client, http_sink_request_builder);
 
         let request_limits = self.request.tower.unwrap_with(&Default::default());
 

src/sinks/http/encoder.rs

@@ -53,6 +53,14 @@ impl SinkEncoder<Vec<Event>> for HttpEncoder {
         let mut byte_size = telemetry().create_request_count_byte_size();
         let mut body = BytesMut::new();
 
+        match (self.encoder.serializer(), self.encoder.framer()) {
+            (Json(_), CharacterDelimited(CharacterDelimitedEncoder { delimiter: b',' })) => {
+                body.put(self.payload_prefix.as_bytes());
+                body.put_u8(b'[');
+            }
+            _ => {}
+        }
+
         for mut event in events {
             self.transformer.transform(&mut event);
 
@@ -72,14 +80,7 @@ impl SinkEncoder<Vec<Event>> for HttpEncoder {
                 }
             }
             (Json(_), CharacterDelimited(CharacterDelimitedEncoder { delimiter: b',' })) => {
-                // TODO(https://github.com/vectordotdev/vector/issues/11253):
-                // Prepend before building a request body to eliminate the
-                // additional copy here.
-                let message = body.split();
-                body.put(self.payload_prefix.as_bytes());
-                body.put_u8(b'[');
-                if !message.is_empty() {
-                    body.unsplit(message);
+                if !body.is_empty() {
                     // remove trailing comma from last record
                     body.truncate(body.len() - 1);
                 }

src/sinks/http/sink.rs

@@ -31,7 +31,6 @@ where
     }
 
     async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
-        let service = ServiceBuilder::new().service(self.service);
         input
             // Batch the input stream with size calculation based on the configured codec
             .batched(self.batch_settings.into_item_size_config(HttpBatchSizer {
@@ -51,7 +50,7 @@ where
             })
             // Generate the driver that will send requests and handle retries,
             // event finalization, and logging/internal metric reporting.
-            .into_driver(service)
+            .into_driver(self.service)
             .run()
             .await
     }

src/sinks/splunk_hec/common/util.rs

@@ -50,6 +50,9 @@ pub fn create_client(
     Ok(HttpClient::new(tls_settings, proxy_config)?)
 }
 
+// TODO: `HttpBatchService` has been deprecated for direct use in sinks.
+//       This sink should undergo a refactor to utilize the `HttpService`
+//       instead, which extracts much of the boilerplate code for `Service`.
 pub fn build_http_batch_service(
     client: HttpClient,
     http_request_builder: Arc<HttpRequestBuilder>,

src/sinks/util/http.rs

@@ -24,8 +24,9 @@ use vector_core::{ByteSizeOf, EstimatedJsonEncodedSizeOf};
 
 use super::{
     retries::{RetryAction, RetryLogic},
-    sink, uri, Batch, EncodedEvent, Partition, TowerBatchedSink, TowerPartitionSink,
-    TowerRequestConfig, TowerRequestSettings,
+    sink::{self, Response as _},
+    uri, Batch, EncodedEvent, Partition, TowerBatchedSink, TowerPartitionSink, TowerRequestConfig,
+    TowerRequestSettings,
 };
 use crate::{
     event::Event,
@@ -356,6 +357,12 @@ where
     }
 }
 
+/// @struct HttpBatchService
+///
+/// NOTE: This has been deprecated, please do not use <directly> when creating new sinks.
+///       The `HttpService` currently wraps this structure. Eventually all sinks currently using the
+///       HttpBatchService directly should be updated to use `HttpService`. At which time we can
+///       remove this struct and inline the functionality into the `HttpService` directly.
 pub struct HttpBatchService<F, B = Bytes> {
     inner: HttpClient<Body>,
     request_builder: Arc<dyn Fn(B) -> F + Send + Sync>,
@@ -653,8 +660,10 @@ pub struct HttpResponse {
 
 impl DriverResponse for HttpResponse {
     fn event_status(&self) -> EventStatus {
-        if self.http_response.status().is_success() {
+        if self.http_response.is_successful() {
             EventStatus::Delivered
+        } else if self.http_response.is_transient() {
+            EventStatus::Errored
         } else {
             EventStatus::Rejected
         }
@@ -669,12 +678,12 @@ impl DriverResponse for HttpResponse {
     }
 }
 
-/// HTTP request builder for batched HTTP stream sinks using the generic `HttpService`
+/// HTTP request builder for HTTP stream sinks using the generic `HttpService`
 pub trait HttpServiceRequestBuilder {
     fn build(&self, body: Bytes) -> Request<Bytes>;
 }
 
-/// Generic 'Service' implementation for batched HTTP stream sinks.
+/// Generic 'Service' implementation for HTTP stream sinks.
 #[derive(Clone)]
 pub struct HttpService<B> {
     batch_service: