Try item sized batching

Author: neuronull

src/sinks/http/batch.rs

@@ -0,0 +1,20 @@
+use codecs::encoding::Framer;
+use vector_core::{
+    event::Event, stream::batcher::limiter::ItemBatchSize, ByteSizeOf, EstimatedJsonEncodedSizeOf,
+};
+
+use crate::codecs::Encoder;
+
+#[derive(Default)]
+pub(super) struct HttpBatchSizer {
+    pub(super) encoder: Encoder<Framer>,
+}
+
+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(),
+            _ => item.size_of(),
+        }
+    }
+}

src/sinks/http/config.rs

@@ -225,7 +225,7 @@ pub(super) fn validate_payload_wrapper(
 #[typetag::serde(name = "http")]
 impl SinkConfig for HttpSinkConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
-        let batch_settings = self.batch.into_batcher_settings()?;
+        let batch_settings = self.batch.validate()?.into_batcher_settings()?;
 
         let encoder = self.build_encoder()?;
         let transformer = self.encoding.transformer();

src/sinks/http/mod.rs

@@ -3,6 +3,7 @@
 //! This module contains the [`vector_core::sink::VectorSink`] instance that is responsible for
 //! taking a stream of [`vector_core::event::Event`]s and forwarding them to an HTTP server.
 
+mod batch;
 mod config;
 mod encoder;
 mod request_builder;

src/sinks/http/sink.rs

@@ -5,7 +5,9 @@ use crate::sinks::{
     util::http_service::{HttpRetryLogic, HttpService},
 };
 
-use super::{request_builder::HttpRequestBuilder, service::HttpSinkRequestBuilder};
+use super::{
+    batch::HttpBatchSizer, request_builder::HttpRequestBuilder, service::HttpSinkRequestBuilder,
+};
 
 pub(super) struct HttpSink {
     service: Svc<HttpService<HttpSinkRequestBuilder>, HttpRetryLogic>,
@@ -29,8 +31,10 @@ impl HttpSink {
 
     async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
         input
-            // Batch the input stream with byte size calculation.
-            .batched(self.batch_settings.into_byte_size_config())
+            // Batch the input stream with size calculation dependent on the configured codec
+            .batched(self.batch_settings.into_item_size_config(HttpBatchSizer {
+                encoder: self.request_builder.encoder.encoder.clone(),
+            }))
             // Build requests with no concurrency limit.
             .request_builder(None, self.request_builder)
             // Filter out any errors that occurred in the request building.

src/sinks/util/http_service/service.rs

@@ -79,6 +79,7 @@ where
     fn call(&mut self, request: HttpRequest) -> Self::Future {
         // for internal metrics reporting
         let raw_byte_size = request.payload.len();
+
         let events_byte_size = request
             .request_metadata
             .into_events_estimated_json_encoded_byte_size();