fix handling of encoding request splits + add more accurate batch size impl

Author: tobz

src/sinks/statsd/mod.rs

@@ -1,3 +1,4 @@
+mod batch;
 mod config;
 mod encoder;
 mod normalizer;

src/sinks/statsd/request_builder.rs

@@ -115,6 +115,9 @@ impl IncrementalRequestBuilder<Vec<Metric>> for StatsdRequestBuilder {
                 if request_buf_len != 0
                     && (request_buf_len + self.encode_buf.len() > self.request_max_size)
                 {
+                    // The metric, as encoded, would cause us to exceed our maximum request size, so
+                    // store it off to the side and finalize the current request.
+                    pending = Some(metric);
                     break;
                 }
 

src/sinks/statsd/sink.rs

@@ -16,7 +16,8 @@ use vector_core::{
 use crate::sinks::util::SinkBuilderExt;
 
 use super::{
-    normalizer::StatsdNormalizer, request_builder::StatsdRequestBuilder, service::StatsdRequest,
+    batch::StatsdBatchSizer, normalizer::StatsdNormalizer, request_builder::StatsdRequestBuilder,
+    service::StatsdRequest,
 };
 
 pub(crate) struct StatsdSink<S> {
@@ -57,7 +58,10 @@ where
             // other metric types in type-specific ways i.e. incremental gauge updates use a
             // different syntax, etc.
             .normalized_with_default::<StatsdNormalizer>()
-            .batched(self.batch_settings.into_byte_size_config())
+            .batched(
+                self.batch_settings
+                    .into_item_size_config(StatsdBatchSizer::default()),
+            )
             // We build our requests "incrementally", which means that for a single batch of
             // metrics, we might generate N requests to represent all of the metrics in the batch.
             //