From 8a6ccf2c0f872e212cc452bab0e301dad460fc0d Mon Sep 17 00:00:00 2001
From: Matt Klein <mklein@bitdrift.io>
Date: Sat, 13 Sep 2025 08:33:42 -0600
Subject: [PATCH] prom outflow: add max request size config

This will split outbound requests if the compressed size is too large.
This is off by default and must be enabled in config.

Signed-off-by: Matt Klein <mklein@bitdrift.io>
---
 Dockerfile                                    |   2 +-
 pulse-common/src/k8s/pods_info.rs             |   4 +-
 pulse-metrics/src/pipeline/inflow/otlp/mod.rs |   2 +-
 .../pipeline/inflow/prom_scrape/scraper.rs    |   2 +-
 .../src/pipeline/outflow/http/remote_write.rs | 299 +++++++++++-------
 .../src/pipeline/outflow/otlp/mod.rs          |   1 +
 .../src/pipeline/outflow/prom/mod.rs          |   4 +
 .../config/outflow/v1/prom_remote_write.proto |   8 +-
 .../src/protos/opentelemetry/common.rs        |   2 +-
 .../src/protos/opentelemetry/metrics.rs       |   2 +-
 .../protos/opentelemetry/metrics_service.rs   |   2 +-
 .../src/protos/opentelemetry/resource.rs      |   2 +-
 .../pulse/config/bootstrap/v1/bootstrap.rs    |   2 +-
 .../protos/pulse/config/common/v1/common.rs   |   2 +-
 .../pulse/config/common/v1/file_watcher.rs    |   2 +-
 .../protos/pulse/config/common/v1/retry.rs    |   2 +-
 .../protos/pulse/config/inflow/v1/inflow.rs   |   2 +-
 .../pulse/config/inflow/v1/inflow_common.rs   |   2 +-
 .../protos/pulse/config/inflow/v1/k8s_prom.rs |   2 +-
 .../config/inflow/v1/metric_generator.rs      |   2 +-
 .../src/protos/pulse/config/inflow/v1/otlp.rs |   2 +-
 .../config/inflow/v1/prom_remote_write.rs     |   2 +-
 .../src/protos/pulse/config/inflow/v1/wire.rs |   2 +-
 .../protos/pulse/config/outflow/v1/otlp.rs    |   2 +-
 .../protos/pulse/config/outflow/v1/outflow.rs |   2 +-
 .../pulse/config/outflow/v1/outflow_common.rs |   2 +-
 .../config/outflow/v1/prom_remote_write.rs    |  62 ++--
 .../pulse/config/outflow/v1/queue_policy.rs   |   2 +-
 .../protos/pulse/config/outflow/v1/wire.rs    |   2 +-
 .../pulse/config/processor/v1/aggregation.rs  |   2 +-
 .../pulse/config/processor/v1/buffer.rs       |   2 +-
 .../processor/v1/cardinality_limiter.rs       |   2 +-
 .../processor/v1/cardinality_tracker.rs       |   2 +-
 .../protos/pulse/config/processor/v1/drop.rs  |   2 +-
 .../pulse/config/processor/v1/elision.rs      |   2 +-
 .../pulse/config/processor/v1/internode.rs    |   2 +-
 .../pulse/config/processor/v1/mutate.rs       |   2 +-
 .../config/processor/v1/populate_cache.rs     |   2 +-
 .../pulse/config/processor/v1/processor.rs    |   2 +-
 .../protos/pulse/config/processor/v1/regex.rs |   2 +-
 .../pulse/config/processor/v1/sampler.rs      |   2 +-
 .../pulse/drop_tester/v1/drop_tester.rs       |   2 +-
 .../protos/pulse/internode/v1/internode.rs    |   2 +-
 .../src/protos/pulse/internode/v1/metric.rs   |   2 +-
 .../protos/pulse/vrl_tester/v1/vrl_tester.rs  |   2 +-
 pulse-proxy/src/test/integration/prom.rs      |  46 +++
 46 files changed, 316 insertions(+), 186 deletions(-)

diff --git a/Dockerfile b/Dockerfile
index 6bc00f9..6ee0200 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,4 +1,4 @@
-FROM ubuntu:jammy as base
+FROM ubuntu:noble as base
 
 RUN apt-get update \
     && apt-get upgrade -y \
diff --git a/pulse-common/src/k8s/pods_info.rs b/pulse-common/src/k8s/pods_info.rs
index 487789e..6241c92 100644
--- a/pulse-common/src/k8s/pods_info.rs
+++ b/pulse-common/src/k8s/pods_info.rs
@@ -129,7 +129,7 @@ impl PodsInfoSingleton {
             k8s_config
               .services_cache_interval
               .as_ref()
-              .map_or(15.minutes(), bd_time::ProtoDurationExt::to_time_duration),
+              .map_or_else(|| 15.minutes(), bd_time::ProtoDurationExt::to_time_duration),
             Box::new(RealServiceFetcher),
           ))
         } else {
@@ -514,7 +514,7 @@ impl ResourceWatchCallbacks<Pod> for PodsInfoCache {
   async fn init_apply(&mut self, pod: Pod) {
     self
       .initializing_state
-      .get_or_insert(PodsInfo::default())
+      .get_or_insert_with(PodsInfo::default)
       .apply_pod(
         &self.node_info,
         &pod,
diff --git a/pulse-metrics/src/pipeline/inflow/otlp/mod.rs b/pulse-metrics/src/pipeline/inflow/otlp/mod.rs
index 8aca890..6060bae 100644
--- a/pulse-metrics/src/pipeline/inflow/otlp/mod.rs
+++ b/pulse-metrics/src/pipeline/inflow/otlp/mod.rs
@@ -76,7 +76,7 @@ impl OtlpInflow {
           config.bind.to_string(),
           config
             .path
-            .map_or("/v1/metrics".to_string(), |p| p.to_string()),
+            .map_or_else(|| "/v1/metrics".to_string(), |p| p.to_string()),
           config.downstream_id_source.unwrap_or_default(),
           context,
           Box::new(move |_inflow, headers, body, downstream_id_provider| {
diff --git a/pulse-metrics/src/pipeline/inflow/prom_scrape/scraper.rs b/pulse-metrics/src/pipeline/inflow/prom_scrape/scraper.rs
index 1793c9e..d96b719 100644
--- a/pulse-metrics/src/pipeline/inflow/prom_scrape/scraper.rs
+++ b/pulse-metrics/src/pipeline/inflow/prom_scrape/scraper.rs
@@ -871,7 +871,7 @@ pub async fn make(
       let interval = http_service_discovery
         .fetch_interval
         .as_ref()
-        .map_or(1.minutes(), bd_time::ProtoDurationExt::to_time_duration)
+        .map_or_else(|| 1.minutes(), bd_time::ProtoDurationExt::to_time_duration)
         .interval(MissedTickBehavior::Delay);
       let target = HttpServiceDiscoveryEndpointTarget::new(
         http_service_discovery,
diff --git a/pulse-metrics/src/pipeline/outflow/http/remote_write.rs b/pulse-metrics/src/pipeline/outflow/http/remote_write.rs
index 09ab1d9..978f9ab 100644
--- a/pulse-metrics/src/pipeline/outflow/http/remote_write.rs
+++ b/pulse-metrics/src/pipeline/outflow/http/remote_write.rs
@@ -58,6 +58,7 @@ struct HttpRemoteWriteOutflowStats {
   requests_total: IntCounter,
   requests_time: Histogram,
   requests_size: Histogram,
+  requests_split: IntCounter,
   offload_queue_tx: IntCounter,
   offload_queue_rx: IntCounter,
   tx_bytes: IntCounter,
@@ -87,6 +88,7 @@ impl HttpRemoteWriteOutflowStats {
           1_572_864.0, // 1.5 MiB
         ],
       ),
+      requests_split: stats.counter("requests_split"),
       offload_queue_tx: stats.counter("offload_queue_tx"),
       offload_queue_rx: stats.counter("offload_queue_rx"),
       tx_bytes: stats.counter("tx_bytes"),
@@ -101,7 +103,7 @@ impl HttpRemoteWriteOutflowStats {
 enum SendRequest {
   // Request coming from the standard inflow/batch system.
   Normal {
-    compressed_write_request: Bytes,
+    compressed_write_requests: Vec<Bytes>,
     received_at: Vec<Instant>,
     extra_headers: Option<Arc<HeaderMap>>,
   },
@@ -232,136 +234,140 @@ impl HttpRemoteWriteOutflow {
     let auto_requests_in_flight = AutoGauge::new(self.stats.requests_in_flight.clone());
 
     tokio::spawn(async move {
-      let (compressed_write_request, num_metrics, received_at, extra_headers, serialized) =
-        match send_request {
-          SendRequest::Normal {
-            compressed_write_request,
-            received_at,
-            extra_headers,
-          } => (
-            compressed_write_request,
-            received_at.len().try_into().unwrap(),
-            Some(received_at),
-            extra_headers,
-            None,
-          ),
-          SendRequest::OffloadQueue { serialized } => (
-            serialized.compressed_write_request(),
-            serialized.num_metrics(),
-            None,
-            serialized.extra_headers(),
-            Some(serialized),
-          ),
-        };
+      let (compressed_write_requests, num_metrics, received_at, extra_headers) = match send_request
+      {
+        SendRequest::Normal {
+          compressed_write_requests,
+          received_at,
+          extra_headers,
+        } => (
+          compressed_write_requests,
+          received_at.len().try_into().unwrap(),
+          Some(received_at),
+          extra_headers,
+        ),
+        SendRequest::OffloadQueue { serialized } => (
+          vec![serialized.compressed_write_request()],
+          serialized.num_metrics(),
+          None,
+          serialized.extra_headers(),
+        ),
+      };
 
       log::debug!("sending batch of {num_metrics} metric(s)");
-      self
-        .stats
-        .requests_size
-        .observe(compressed_write_request.len().lossy_to_f64());
-      let time = self.stats.requests_time.start_timer();
-      let res = self
-        .retry
-        .retry_notify(
-          (self.backoff)(),
-          || async {
-            self
-              .stats
-              .tx_bytes
-              .inc_by(compressed_write_request.len() as u64);
-
-            match self
-              .client
-              .send_write_request(
-                compressed_write_request.clone(),
-                extra_headers.as_ref().map(std::convert::AsRef::as_ref),
-              )
-              .await
-            {
-              Ok(()) => Ok(()),
-              Err(e) => {
-                // Skip retries if shutdown is pending.
-                if should_retry(&e)
-                  && shutdown.component_status() != ComponentStatus::PendingShutdown
-                {
-                  Err(backoff::Error::transient(e))
-                } else {
-                  Err(backoff::Error::permanent(e))
-                }
-              },
-            }
-          },
-          || {
-            self.stats.requests_retry.inc();
-          },
-        )
-        .await;
-
-      drop(time);
-      self.stats.requests_total.inc();
-
-      // TODO(mattklein123): We don't attempt to serialize received_at for offload given it's an
-      // edge case. Perhaps we should do this?
-      if let Some(received_at) = received_at {
+      if compressed_write_requests.len() > 1 {
         self
           .stats
-          .outflow_stats
-          .messages_e2e_timer_observe(&received_at);
+          .requests_split
+          .inc_by(compressed_write_requests.len() as u64);
       }
+      for compressed_write_request in compressed_write_requests {
+        self
+          .stats
+          .requests_size
+          .observe(compressed_write_request.len().lossy_to_f64());
+        let time = self.stats.requests_time.start_timer();
+        let res = self
+          .retry
+          .retry_notify(
+            (self.backoff)(),
+            || async {
+              self
+                .stats
+                .tx_bytes
+                .inc_by(compressed_write_request.len() as u64);
+
+              match self
+                .client
+                .send_write_request(
+                  compressed_write_request.clone(),
+                  extra_headers.as_ref().map(std::convert::AsRef::as_ref),
+                )
+                .await
+              {
+                Ok(()) => Ok(()),
+                Err(e) => {
+                  // Skip retries if shutdown is pending.
+                  if should_retry(&e)
+                    && shutdown.component_status() != ComponentStatus::PendingShutdown
+                  {
+                    Err(backoff::Error::transient(e))
+                  } else {
+                    Err(backoff::Error::permanent(e))
+                  }
+                },
+              }
+            },
+            || {
+              self.stats.requests_retry.inc();
+            },
+          )
+          .await;
+
+        drop(time);
+        self.stats.requests_total.inc();
 
-      match res {
-        Ok(()) => {
-          self
-            .stats
-            .outflow_stats
-            .messages_outgoing_success
-            .inc_by(num_metrics);
-        },
-        Err(e) => {
-          // This is incremented whether we offload or not, so that we get accurate SR for
-          // upstream. Alarming should happen on drops.
-          self.stats.requests_fail.inc();
-
-          if maybe_queue_for_retry(
-            self.offload_queue.as_ref(),
-            &self.retry_policy.offload_queue,
-            &e,
-            serialized.unwrap_or_else(|| {
+        match res {
+          Ok(()) => {
+            self
+              .stats
+              .outflow_stats
+              .messages_outgoing_success
+              .inc_by(num_metrics);
+          },
+          Err(e) => {
+            // This is incremented whether we offload or not, so that we get accurate SR for
+            // upstream. Alarming should happen on drops.
+            self.stats.requests_fail.inc();
+
+            if maybe_queue_for_retry(
+              self.offload_queue.as_ref(),
+              &self.retry_policy.offload_queue,
+              &e,
               SerializedOffloadRequest::new(
                 &compressed_write_request,
-                extra_headers,
+                extra_headers.clone(),
                 num_metrics,
                 &RealTimeProvider {},
-              )
-            }),
-            &RealTimeProvider {},
-          )
-          .await
-          {
-            self.stats.offload_queue_tx.inc();
+              ),
+              &RealTimeProvider {},
+            )
+            .await
+            {
+              self.stats.offload_queue_tx.inc();
+              warn_every!(
+                15.seconds(),
+                "remote write request failed, but sent to retry queue: size={}, outflow=\"{}\": {}",
+                compressed_write_request.len(),
+                self.name,
+                e
+              );
+              return;
+            }
+
+            self
+              .stats
+              .outflow_stats
+              .messages_outgoing_failed
+              .inc_by(num_metrics);
             warn_every!(
               15.seconds(),
-              "remote write request failed, but sent to retry queue: size={}, outflow=\"{}\": {}",
+              "remote write request failed: size={}, outflow=\"{}\": {}",
               compressed_write_request.len(),
               self.name,
               e
             );
-            return;
-          }
-
-          self
-            .stats
-            .outflow_stats
-            .messages_outgoing_failed
-            .inc_by(num_metrics);
-          warn_every!(
-            15.seconds(),
-            "remote write request failed: size={}, outflow=\"{}\": {}",
-            compressed_write_request.len(),
-            self.name,
-            e
-          );
-        },
+          },
+        }
+      }
+
+      // TODO(mattklein123): We don't attempt to serialize received_at for offload given it's an
+      // edge case. Perhaps we should do this?
+      if let Some(received_at) = received_at {
+        self
+          .stats
+          .outflow_stats
+          .messages_e2e_timer_observe(&received_at);
       }
 
       drop(shutdown);
@@ -431,7 +437,7 @@ impl HttpRemoteWriteOutflow {
 
       for batch in batch_set {
         let HttpBatch::Complete {
-          compressed_write_request,
+          compressed_write_requests,
           received_at,
           extra_headers,
         } = batch
@@ -444,7 +450,7 @@ impl HttpRemoteWriteOutflow {
           .clone()
           .send_request(
             SendRequest::Normal {
-              compressed_write_request,
+              compressed_write_requests,
               extra_headers,
               received_at,
             },
@@ -486,6 +492,7 @@ pub struct DefaultBatchRouter {
 impl DefaultBatchRouter {
   pub(crate) fn new(
     batch_max_samples: Option<u64>,
+    batch_max_size: Option<u64>,
     queue_policy: &QueuePolicy,
     scope: &Scope,
     shutdown: ComponentShutdown,
@@ -495,6 +502,7 @@ impl DefaultBatchRouter {
     Self {
       builder: Self::make_batch_builder(
         batch_max_samples,
+        batch_max_size,
         queue_policy,
         scope,
         shutdown,
@@ -507,6 +515,7 @@ impl DefaultBatchRouter {
   #[allow(clippy::needless_pass_by_value)] // Spurious
   pub(crate) fn make_batch_builder(
     batch_max_samples: Option<u64>,
+    batch_max_size: Option<u64>,
     queue_policy: &QueuePolicy,
     scope: &Scope,
     shutdown: ComponentShutdown,
@@ -521,7 +530,14 @@ impl DefaultBatchRouter {
     BatchBuilder::new(
       scope,
       queue_policy,
-      move || HttpBatch::new(batch_max_samples, extra_headers.clone(), finisher.clone()),
+      move || {
+        HttpBatch::new(
+          batch_max_samples,
+          batch_max_size.map(|s| s.try_into().unwrap()),
+          extra_headers.clone(),
+          finisher.clone(),
+        )
+      },
       shutdown,
     )
   }
@@ -547,11 +563,12 @@ pub enum HttpBatch {
   Building {
     samples: Vec<ParsedMetric>,
     max_samples: usize,
+    max_size: Option<usize>,
     extra_headers: Option<Arc<HeaderMap>>,
     finisher: Arc<dyn Fn(Vec<ParsedMetric>) -> Bytes + Send + Sync>,
   },
   Complete {
-    compressed_write_request: Bytes,
+    compressed_write_requests: Vec<Bytes>,
     received_at: Vec<Instant>,
     extra_headers: Option<Arc<HeaderMap>>,
   },
@@ -560,16 +577,50 @@ pub enum HttpBatch {
 impl HttpBatch {
   fn new(
     max_samples: usize,
+    max_size: Option<usize>,
     extra_headers: Option<Arc<HeaderMap>>,
     finisher: Arc<dyn Fn(Vec<ParsedMetric>) -> Bytes + Send + Sync>,
   ) -> Self {
     Self::Building {
       samples: Vec::with_capacity(max_samples),
       max_samples,
+      max_size,
       extra_headers,
       finisher,
     }
   }
+
+  fn maybe_split_requests(
+    finisher: &(dyn Fn(Vec<ParsedMetric>) -> Bytes + Send + Sync),
+    samples: Vec<ParsedMetric>,
+    max_size: Option<usize>,
+  ) -> (Vec<Bytes>, usize) {
+    // If max_size is set, we may need to split the request. To do this, we clone the samples
+    // so that we can split them in half if needed. This is not ideal as this is done for all
+    // requests.
+    let maybe_cloned_samples = max_size.map(|_| samples.clone());
+    let compressed_write_request = (finisher)(samples);
+    let (requests, size) = if max_size.is_some_and(|size| compressed_write_request.len() > size) {
+      // If the compressed request is larger than the max size, we need to split it.
+      // TODO(mattklein123): The requests could still be too large after splitting. We ignore this
+      // for now.
+      log::debug!(
+        "splitting remote write request of size {}",
+        compressed_write_request.len()
+      );
+      let mut first = maybe_cloned_samples.unwrap();
+      let rest = first.split_off(first.len() / 2);
+      let first = (finisher)(first);
+      let rest = (finisher)(rest);
+      let size = first.len() + rest.len();
+      (vec![first, rest], size)
+    } else {
+      let size = compressed_write_request.len();
+      (vec![compressed_write_request], size)
+    };
+
+    (requests, size)
+  }
 }
 
 impl Batch<ParsedMetric> for HttpBatch {
@@ -591,25 +642,27 @@ impl Batch<ParsedMetric> for HttpBatch {
   }
 
   fn finish(&mut self) -> usize {
-    let (samples, extra_headers, finisher) = match self {
+    let (samples, extra_headers, finisher, max_size) = match self {
       Self::Building {
         samples,
         extra_headers,
         finisher,
+        max_size,
         ..
       } => (
         std::mem::take(samples),
         std::mem::take(extra_headers),
         finisher,
+        *max_size,
       ),
       Self::Complete { .. } => unreachable!(),
     };
 
     let received_at = samples.iter().map(ParsedMetric::received_at).collect();
-    let compressed_write_request = (finisher)(samples);
-    let size = compressed_write_request.len();
+    let (compressed_write_requests, size) =
+      Self::maybe_split_requests(finisher.as_ref(), samples, max_size);
     *self = Self::Complete {
-      compressed_write_request,
+      compressed_write_requests,
       received_at,
       extra_headers,
     };
diff --git a/pulse-metrics/src/pipeline/outflow/otlp/mod.rs b/pulse-metrics/src/pipeline/outflow/otlp/mod.rs
index 96dc720..438c503 100644
--- a/pulse-metrics/src/pipeline/outflow/otlp/mod.rs
+++ b/pulse-metrics/src/pipeline/outflow/otlp/mod.rs
@@ -54,6 +54,7 @@ pub fn make_otlp_batch_router(
 
   Arc::new(DefaultBatchRouter::new(
     config.batch_max_samples,
+    None,
     &config.queue_policy,
     &stats.stats,
     shutdown,
diff --git a/pulse-metrics/src/pipeline/outflow/prom/mod.rs b/pulse-metrics/src/pipeline/outflow/prom/mod.rs
index 13b7d73..ce91fd2 100644
--- a/pulse-metrics/src/pipeline/outflow/prom/mod.rs
+++ b/pulse-metrics/src/pipeline/outflow/prom/mod.rs
@@ -68,6 +68,7 @@ pub fn make_prom_batch_router(
   } else {
     Arc::new(DefaultBatchRouter::new(
       config.batch_max_samples,
+      config.batch_max_size,
       &config.queue_policy,
       &stats.stats,
       shutdown,
@@ -147,6 +148,7 @@ impl LyftBatchRouter {
   ) -> Self {
     let generic = DefaultBatchRouter::make_batch_builder(
       batch_max_samples,
+      None,
       queue_policy,
       &scope.scope("general"),
       shutdown.clone(),
@@ -161,6 +163,7 @@ impl LyftBatchRouter {
       .map(|p| {
         DefaultBatchRouter::make_batch_builder(
           batch_max_samples,
+          None,
           queue_policy,
           &scope.scope("instance"),
           shutdown.clone(),
@@ -174,6 +177,7 @@ impl LyftBatchRouter {
       .map(|p| {
         DefaultBatchRouter::make_batch_builder(
           batch_max_samples,
+          None,
           queue_policy,
           &scope.scope("cloudwatch"),
           shutdown,
diff --git a/pulse-protobuf/proto/pulse/config/outflow/v1/prom_remote_write.proto b/pulse-protobuf/proto/pulse/config/outflow/v1/prom_remote_write.proto
index bb4da05..e4d5854 100644
--- a/pulse-protobuf/proto/pulse/config/outflow/v1/prom_remote_write.proto
+++ b/pulse-protobuf/proto/pulse/config/outflow/v1/prom_remote_write.proto
@@ -41,9 +41,15 @@ message PromRemoteWriteClientConfig {
   // Queue policy for the client.
   QueuePolicy queue_policy = 4;
 
-  // The maximum numer of samples per batch. Defaults to 1000.
+  // The maximum number of samples per batch. Defaults to 1000.
   optional uint64 batch_max_samples = 5;
 
+  // The maximum size of each batch in bytes, *after* compression. There is no default for this.
+  // This can be used of the TSDB has a strict maximum request size. Since it's impossible to know
+  // the size before compression, batch_max_samples should be tuned alongside this. If the limit is
+  // hit, the batch will be split in half into 2 requests so this is an expensive operation.
+  optional uint64 batch_max_size = 13;
+
   // Whether to send metadata (name and tags) only. If true, samples are dropped.
   bool metadata_only = 6;
 
diff --git a/pulse-protobuf/src/protos/opentelemetry/common.rs b/pulse-protobuf/src/protos/opentelemetry/common.rs
index bf8ce51..d250546 100644
--- a/pulse-protobuf/src/protos/opentelemetry/common.rs
+++ b/pulse-protobuf/src/protos/opentelemetry/common.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/opentelemetry/metrics.rs b/pulse-protobuf/src/protos/opentelemetry/metrics.rs
index fb20854..208491c 100644
--- a/pulse-protobuf/src/protos/opentelemetry/metrics.rs
+++ b/pulse-protobuf/src/protos/opentelemetry/metrics.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/opentelemetry/metrics_service.rs b/pulse-protobuf/src/protos/opentelemetry/metrics_service.rs
index 57ed9f7..cf0dbcb 100644
--- a/pulse-protobuf/src/protos/opentelemetry/metrics_service.rs
+++ b/pulse-protobuf/src/protos/opentelemetry/metrics_service.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/opentelemetry/resource.rs b/pulse-protobuf/src/protos/opentelemetry/resource.rs
index 72cb399..0da9f3f 100644
--- a/pulse-protobuf/src/protos/opentelemetry/resource.rs
+++ b/pulse-protobuf/src/protos/opentelemetry/resource.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/bootstrap/v1/bootstrap.rs b/pulse-protobuf/src/protos/pulse/config/bootstrap/v1/bootstrap.rs
index b8ca5b2..884ade3 100644
--- a/pulse-protobuf/src/protos/pulse/config/bootstrap/v1/bootstrap.rs
+++ b/pulse-protobuf/src/protos/pulse/config/bootstrap/v1/bootstrap.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/common/v1/common.rs b/pulse-protobuf/src/protos/pulse/config/common/v1/common.rs
index dbced1a..3873f2f 100644
--- a/pulse-protobuf/src/protos/pulse/config/common/v1/common.rs
+++ b/pulse-protobuf/src/protos/pulse/config/common/v1/common.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/common/v1/file_watcher.rs b/pulse-protobuf/src/protos/pulse/config/common/v1/file_watcher.rs
index daa0c5b..3032b41 100644
--- a/pulse-protobuf/src/protos/pulse/config/common/v1/file_watcher.rs
+++ b/pulse-protobuf/src/protos/pulse/config/common/v1/file_watcher.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/common/v1/retry.rs b/pulse-protobuf/src/protos/pulse/config/common/v1/retry.rs
index 40f6d53..e5e094a 100644
--- a/pulse-protobuf/src/protos/pulse/config/common/v1/retry.rs
+++ b/pulse-protobuf/src/protos/pulse/config/common/v1/retry.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/inflow/v1/inflow.rs b/pulse-protobuf/src/protos/pulse/config/inflow/v1/inflow.rs
index a2c4c28..5fe08fd 100644
--- a/pulse-protobuf/src/protos/pulse/config/inflow/v1/inflow.rs
+++ b/pulse-protobuf/src/protos/pulse/config/inflow/v1/inflow.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/inflow/v1/inflow_common.rs b/pulse-protobuf/src/protos/pulse/config/inflow/v1/inflow_common.rs
index 489def5..aca29f4 100644
--- a/pulse-protobuf/src/protos/pulse/config/inflow/v1/inflow_common.rs
+++ b/pulse-protobuf/src/protos/pulse/config/inflow/v1/inflow_common.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/inflow/v1/k8s_prom.rs b/pulse-protobuf/src/protos/pulse/config/inflow/v1/k8s_prom.rs
index c3e0fb8..e62194f 100644
--- a/pulse-protobuf/src/protos/pulse/config/inflow/v1/k8s_prom.rs
+++ b/pulse-protobuf/src/protos/pulse/config/inflow/v1/k8s_prom.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/inflow/v1/metric_generator.rs b/pulse-protobuf/src/protos/pulse/config/inflow/v1/metric_generator.rs
index c92c0c5..f9ccc49 100644
--- a/pulse-protobuf/src/protos/pulse/config/inflow/v1/metric_generator.rs
+++ b/pulse-protobuf/src/protos/pulse/config/inflow/v1/metric_generator.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/inflow/v1/otlp.rs b/pulse-protobuf/src/protos/pulse/config/inflow/v1/otlp.rs
index 220c56c..1da6365 100644
--- a/pulse-protobuf/src/protos/pulse/config/inflow/v1/otlp.rs
+++ b/pulse-protobuf/src/protos/pulse/config/inflow/v1/otlp.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/inflow/v1/prom_remote_write.rs b/pulse-protobuf/src/protos/pulse/config/inflow/v1/prom_remote_write.rs
index 056d74c..d57e9e5 100644
--- a/pulse-protobuf/src/protos/pulse/config/inflow/v1/prom_remote_write.rs
+++ b/pulse-protobuf/src/protos/pulse/config/inflow/v1/prom_remote_write.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/inflow/v1/wire.rs b/pulse-protobuf/src/protos/pulse/config/inflow/v1/wire.rs
index 873a91e..7a8a9f2 100644
--- a/pulse-protobuf/src/protos/pulse/config/inflow/v1/wire.rs
+++ b/pulse-protobuf/src/protos/pulse/config/inflow/v1/wire.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/outflow/v1/otlp.rs b/pulse-protobuf/src/protos/pulse/config/outflow/v1/otlp.rs
index 2dcb523..122960d 100644
--- a/pulse-protobuf/src/protos/pulse/config/outflow/v1/otlp.rs
+++ b/pulse-protobuf/src/protos/pulse/config/outflow/v1/otlp.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/outflow/v1/outflow.rs b/pulse-protobuf/src/protos/pulse/config/outflow/v1/outflow.rs
index 2b060b8..0a052af 100644
--- a/pulse-protobuf/src/protos/pulse/config/outflow/v1/outflow.rs
+++ b/pulse-protobuf/src/protos/pulse/config/outflow/v1/outflow.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/outflow/v1/outflow_common.rs b/pulse-protobuf/src/protos/pulse/config/outflow/v1/outflow_common.rs
index 93af0f1..cdb9660 100644
--- a/pulse-protobuf/src/protos/pulse/config/outflow/v1/outflow_common.rs
+++ b/pulse-protobuf/src/protos/pulse/config/outflow/v1/outflow_common.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/outflow/v1/prom_remote_write.rs b/pulse-protobuf/src/protos/pulse/config/outflow/v1/prom_remote_write.rs
index 356d8e6..8503337 100644
--- a/pulse-protobuf/src/protos/pulse/config/outflow/v1/prom_remote_write.rs
+++ b/pulse-protobuf/src/protos/pulse/config/outflow/v1/prom_remote_write.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
@@ -45,6 +45,8 @@ pub struct PromRemoteWriteClientConfig {
     pub queue_policy: ::protobuf::MessageField<super::queue_policy::QueuePolicy>,
     // @@protoc_insertion_point(field:pulse.config.outflow.v1.PromRemoteWriteClientConfig.batch_max_samples)
     pub batch_max_samples: ::std::option::Option<u64>,
+    // @@protoc_insertion_point(field:pulse.config.outflow.v1.PromRemoteWriteClientConfig.batch_max_size)
+    pub batch_max_size: ::std::option::Option<u64>,
     // @@protoc_insertion_point(field:pulse.config.outflow.v1.PromRemoteWriteClientConfig.metadata_only)
     pub metadata_only: bool,
     // @@protoc_insertion_point(field:pulse.config.outflow.v1.PromRemoteWriteClientConfig.auth)
@@ -76,7 +78,7 @@ impl PromRemoteWriteClientConfig {
     }
 
     fn generated_message_descriptor_data() -> ::protobuf::reflect::GeneratedMessageDescriptorData {
-        let mut fields = ::std::vec::Vec::with_capacity(12);
+        let mut fields = ::std::vec::Vec::with_capacity(13);
         let mut oneofs = ::std::vec::Vec::with_capacity(0);
         fields.push(::protobuf::reflect::rt::v2::make_simpler_field_accessor::<_, _>(
             "send_to",
@@ -103,6 +105,11 @@ impl PromRemoteWriteClientConfig {
             |m: &PromRemoteWriteClientConfig| { &m.batch_max_samples },
             |m: &mut PromRemoteWriteClientConfig| { &mut m.batch_max_samples },
         ));
+        fields.push(::protobuf::reflect::rt::v2::make_option_accessor::<_, _>(
+            "batch_max_size",
+            |m: &PromRemoteWriteClientConfig| { &m.batch_max_size },
+            |m: &mut PromRemoteWriteClientConfig| { &mut m.batch_max_size },
+        ));
         fields.push(::protobuf::reflect::rt::v2::make_simpler_field_accessor::<_, _>(
             "metadata_only",
             |m: &PromRemoteWriteClientConfig| { &m.metadata_only },
@@ -171,6 +178,9 @@ impl ::protobuf::Message for PromRemoteWriteClientConfig {
                 40 => {
                     self.batch_max_samples = ::std::option::Option::Some(is.read_uint64()?);
                 },
+                104 => {
+                    self.batch_max_size = ::std::option::Option::Some(is.read_uint64()?);
+                },
                 48 => {
                     self.metadata_only = is.read_bool()?;
                 },
@@ -221,6 +231,9 @@ impl ::protobuf::Message for PromRemoteWriteClientConfig {
         if let Some(v) = self.batch_max_samples {
             my_size += ::protobuf::rt::uint64_size(5, v);
         }
+        if let Some(v) = self.batch_max_size {
+            my_size += ::protobuf::rt::uint64_size(13, v);
+        }
         if self.metadata_only != false {
             my_size += 1 + 1;
         }
@@ -268,6 +281,9 @@ impl ::protobuf::Message for PromRemoteWriteClientConfig {
         if let Some(v) = self.batch_max_samples {
             os.write_uint64(5, v)?;
         }
+        if let Some(v) = self.batch_max_size {
+            os.write_uint64(13, v)?;
+        }
         if self.metadata_only != false {
             os.write_bool(6, self.metadata_only)?;
         }
@@ -311,6 +327,7 @@ impl ::protobuf::Message for PromRemoteWriteClientConfig {
         self.max_in_flight = ::std::option::Option::None;
         self.queue_policy.clear();
         self.batch_max_samples = ::std::option::Option::None;
+        self.batch_max_size = ::std::option::Option::None;
         self.metadata_only = false;
         self.auth.clear();
         self.request_headers.clear();
@@ -328,6 +345,7 @@ impl ::protobuf::Message for PromRemoteWriteClientConfig {
             max_in_flight: ::std::option::Option::None,
             queue_policy: ::protobuf::MessageField::none(),
             batch_max_samples: ::std::option::Option::None,
+            batch_max_size: ::std::option::Option::None,
             metadata_only: false,
             auth: ::protobuf::MessageField::none(),
             request_headers: ::std::vec::Vec::new(),
@@ -524,31 +542,33 @@ static file_descriptor_proto_data: &'static [u8] = b"\
     utflow.v1\x1a#pulse/config/common/v1/common.proto\x1a\"pulse/config/comm\
     on/v1/retry.proto\x1a*pulse/config/outflow/v1/queue_policy.proto\x1a,pul\
     se/config/outflow/v1/outflow_common.proto\x1a\x1egoogle/protobuf/duratio\
-    n.proto\x1a\x17validate/validate.proto\"\xa9\t\n\x1bPromRemoteWriteClien\
+    n.proto\x1a\x17validate/validate.proto\"\xe7\t\n\x1bPromRemoteWriteClien\
     tConfig\x12\x20\n\x07send_to\x18\x01\x20\x01(\tR\x06sendToB\x07\xfaB\x04\
     r\x02\x10\x01\x12L\n\x0frequest_timeout\x18\x02\x20\x01(\x0b2\x19.google\
     .protobuf.DurationR\x0erequestTimeoutB\x08\xfaB\x05\xaa\x01\x02*\0\x12'\
     \n\rmax_in_flight\x18\x03\x20\x01(\x04H\0R\x0bmaxInFlight\x88\x01\x01\
     \x12G\n\x0cqueue_policy\x18\x04\x20\x01(\x0b2$.pulse.config.outflow.v1.Q\
     ueuePolicyR\x0bqueuePolicy\x12/\n\x11batch_max_samples\x18\x05\x20\x01(\
-    \x04H\x01R\x0fbatchMaxSamples\x88\x01\x01\x12#\n\rmetadata_only\x18\x06\
-    \x20\x01(\x08R\x0cmetadataOnly\x12F\n\x04auth\x18\x07\x20\x01(\x0b22.pul\
-    se.config.outflow.v1.HttpRemoteWriteAuthConfigR\x04auth\x12O\n\x0freques\
-    t_headers\x18\x08\x20\x03(\x0b2&.pulse.config.outflow.v1.RequestHeaderR\
-    \x0erequestHeaders\x123\n\x13convert_metric_name\x18\t\x20\x01(\x08H\x02\
-    R\x11convertMetricName\x88\x01\x01\x12F\n\x0cretry_policy\x18\n\x20\x01(\
-    \x0b2#.pulse.config.common.v1.RetryPolicyR\x0bretryPolicy\x12y\n\x14lyft\
-    _specific_config\x18\x0b\x20\x01(\x0b2G.pulse.config.outflow.v1.PromRemo\
-    teWriteClientConfig.LyftSpecificConfigR\x12lyftSpecificConfig\x12E\n\x11\
-    pool_idle_timeout\x18\x0c\x20\x01(\x0b2\x19.google.protobuf.DurationR\
-    \x0fpoolIdleTimeout\x1a\xb9\x02\n\x12LyftSpecificConfig\x12=\n\x16genera\
-    l_storage_policy\x18\x01\x20\x01(\tR\x14generalStoragePolicyB\x07\xfaB\
-    \x04r\x02\x10\x01\x12J\n\x1finstance_metrics_storage_policy\x18\x02\x20\
-    \x01(\tH\0R\x1cinstanceMetricsStoragePolicy\x88\x01\x01\x12N\n!cloudwatc\
-    h_metrics_storage_policy\x18\x03\x20\x01(\tH\x01R\x1ecloudwatchMetricsSt\
-    oragePolicy\x88\x01\x01B\"\n\x20_instance_metrics_storage_policyB$\n\"_c\
-    loudwatch_metrics_storage_policyB\x10\n\x0e_max_in_flightB\x14\n\x12_bat\
-    ch_max_samplesB\x16\n\x14_convert_metric_nameb\x06proto3\
+    \x04H\x01R\x0fbatchMaxSamples\x88\x01\x01\x12)\n\x0ebatch_max_size\x18\r\
+    \x20\x01(\x04H\x02R\x0cbatchMaxSize\x88\x01\x01\x12#\n\rmetadata_only\
+    \x18\x06\x20\x01(\x08R\x0cmetadataOnly\x12F\n\x04auth\x18\x07\x20\x01(\
+    \x0b22.pulse.config.outflow.v1.HttpRemoteWriteAuthConfigR\x04auth\x12O\n\
+    \x0frequest_headers\x18\x08\x20\x03(\x0b2&.pulse.config.outflow.v1.Reque\
+    stHeaderR\x0erequestHeaders\x123\n\x13convert_metric_name\x18\t\x20\x01(\
+    \x08H\x03R\x11convertMetricName\x88\x01\x01\x12F\n\x0cretry_policy\x18\n\
+    \x20\x01(\x0b2#.pulse.config.common.v1.RetryPolicyR\x0bretryPolicy\x12y\
+    \n\x14lyft_specific_config\x18\x0b\x20\x01(\x0b2G.pulse.config.outflow.v\
+    1.PromRemoteWriteClientConfig.LyftSpecificConfigR\x12lyftSpecificConfig\
+    \x12E\n\x11pool_idle_timeout\x18\x0c\x20\x01(\x0b2\x19.google.protobuf.D\
+    urationR\x0fpoolIdleTimeout\x1a\xb9\x02\n\x12LyftSpecificConfig\x12=\n\
+    \x16general_storage_policy\x18\x01\x20\x01(\tR\x14generalStoragePolicyB\
+    \x07\xfaB\x04r\x02\x10\x01\x12J\n\x1finstance_metrics_storage_policy\x18\
+    \x02\x20\x01(\tH\0R\x1cinstanceMetricsStoragePolicy\x88\x01\x01\x12N\n!c\
+    loudwatch_metrics_storage_policy\x18\x03\x20\x01(\tH\x01R\x1ecloudwatchM\
+    etricsStoragePolicy\x88\x01\x01B\"\n\x20_instance_metrics_storage_policy\
+    B$\n\"_cloudwatch_metrics_storage_policyB\x10\n\x0e_max_in_flightB\x14\n\
+    \x12_batch_max_samplesB\x11\n\x0f_batch_max_sizeB\x16\n\x14_convert_metr\
+    ic_nameb\x06proto3\
 ";
 
 /// `FileDescriptorProto` object which was a source for this generated file
diff --git a/pulse-protobuf/src/protos/pulse/config/outflow/v1/queue_policy.rs b/pulse-protobuf/src/protos/pulse/config/outflow/v1/queue_policy.rs
index e05ae8b..50b87fa 100644
--- a/pulse-protobuf/src/protos/pulse/config/outflow/v1/queue_policy.rs
+++ b/pulse-protobuf/src/protos/pulse/config/outflow/v1/queue_policy.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/outflow/v1/wire.rs b/pulse-protobuf/src/protos/pulse/config/outflow/v1/wire.rs
index 3f6efd4..c228fe4 100644
--- a/pulse-protobuf/src/protos/pulse/config/outflow/v1/wire.rs
+++ b/pulse-protobuf/src/protos/pulse/config/outflow/v1/wire.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/processor/v1/aggregation.rs b/pulse-protobuf/src/protos/pulse/config/processor/v1/aggregation.rs
index b833fc9..0a99698 100644
--- a/pulse-protobuf/src/protos/pulse/config/processor/v1/aggregation.rs
+++ b/pulse-protobuf/src/protos/pulse/config/processor/v1/aggregation.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/processor/v1/buffer.rs b/pulse-protobuf/src/protos/pulse/config/processor/v1/buffer.rs
index a235e50..296ea7d 100644
--- a/pulse-protobuf/src/protos/pulse/config/processor/v1/buffer.rs
+++ b/pulse-protobuf/src/protos/pulse/config/processor/v1/buffer.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/processor/v1/cardinality_limiter.rs b/pulse-protobuf/src/protos/pulse/config/processor/v1/cardinality_limiter.rs
index dd17b07..8db1893 100644
--- a/pulse-protobuf/src/protos/pulse/config/processor/v1/cardinality_limiter.rs
+++ b/pulse-protobuf/src/protos/pulse/config/processor/v1/cardinality_limiter.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/processor/v1/cardinality_tracker.rs b/pulse-protobuf/src/protos/pulse/config/processor/v1/cardinality_tracker.rs
index 22cf664..e781eda 100644
--- a/pulse-protobuf/src/protos/pulse/config/processor/v1/cardinality_tracker.rs
+++ b/pulse-protobuf/src/protos/pulse/config/processor/v1/cardinality_tracker.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/processor/v1/drop.rs b/pulse-protobuf/src/protos/pulse/config/processor/v1/drop.rs
index 68040ec..5b36018 100644
--- a/pulse-protobuf/src/protos/pulse/config/processor/v1/drop.rs
+++ b/pulse-protobuf/src/protos/pulse/config/processor/v1/drop.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/processor/v1/elision.rs b/pulse-protobuf/src/protos/pulse/config/processor/v1/elision.rs
index f6a0389..8745f2e 100644
--- a/pulse-protobuf/src/protos/pulse/config/processor/v1/elision.rs
+++ b/pulse-protobuf/src/protos/pulse/config/processor/v1/elision.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/processor/v1/internode.rs b/pulse-protobuf/src/protos/pulse/config/processor/v1/internode.rs
index 1b3f368..ae90ae9 100644
--- a/pulse-protobuf/src/protos/pulse/config/processor/v1/internode.rs
+++ b/pulse-protobuf/src/protos/pulse/config/processor/v1/internode.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/processor/v1/mutate.rs b/pulse-protobuf/src/protos/pulse/config/processor/v1/mutate.rs
index ac5fab3..52cc6bd 100644
--- a/pulse-protobuf/src/protos/pulse/config/processor/v1/mutate.rs
+++ b/pulse-protobuf/src/protos/pulse/config/processor/v1/mutate.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/processor/v1/populate_cache.rs b/pulse-protobuf/src/protos/pulse/config/processor/v1/populate_cache.rs
index 7e9a9a2..6f12dff 100644
--- a/pulse-protobuf/src/protos/pulse/config/processor/v1/populate_cache.rs
+++ b/pulse-protobuf/src/protos/pulse/config/processor/v1/populate_cache.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/processor/v1/processor.rs b/pulse-protobuf/src/protos/pulse/config/processor/v1/processor.rs
index 221ed99..3b8171d 100644
--- a/pulse-protobuf/src/protos/pulse/config/processor/v1/processor.rs
+++ b/pulse-protobuf/src/protos/pulse/config/processor/v1/processor.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/processor/v1/regex.rs b/pulse-protobuf/src/protos/pulse/config/processor/v1/regex.rs
index a384fd8..65209d9 100644
--- a/pulse-protobuf/src/protos/pulse/config/processor/v1/regex.rs
+++ b/pulse-protobuf/src/protos/pulse/config/processor/v1/regex.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/config/processor/v1/sampler.rs b/pulse-protobuf/src/protos/pulse/config/processor/v1/sampler.rs
index baedf5c..dcbd977 100644
--- a/pulse-protobuf/src/protos/pulse/config/processor/v1/sampler.rs
+++ b/pulse-protobuf/src/protos/pulse/config/processor/v1/sampler.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/drop_tester/v1/drop_tester.rs b/pulse-protobuf/src/protos/pulse/drop_tester/v1/drop_tester.rs
index 871f422..4199034 100644
--- a/pulse-protobuf/src/protos/pulse/drop_tester/v1/drop_tester.rs
+++ b/pulse-protobuf/src/protos/pulse/drop_tester/v1/drop_tester.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/internode/v1/internode.rs b/pulse-protobuf/src/protos/pulse/internode/v1/internode.rs
index 646a572..0990f38 100644
--- a/pulse-protobuf/src/protos/pulse/internode/v1/internode.rs
+++ b/pulse-protobuf/src/protos/pulse/internode/v1/internode.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/internode/v1/metric.rs b/pulse-protobuf/src/protos/pulse/internode/v1/metric.rs
index 9be69af..c0fbd7e 100644
--- a/pulse-protobuf/src/protos/pulse/internode/v1/metric.rs
+++ b/pulse-protobuf/src/protos/pulse/internode/v1/metric.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-protobuf/src/protos/pulse/vrl_tester/v1/vrl_tester.rs b/pulse-protobuf/src/protos/pulse/vrl_tester/v1/vrl_tester.rs
index ed4bbf8..9d37b1a 100644
--- a/pulse-protobuf/src/protos/pulse/vrl_tester/v1/vrl_tester.rs
+++ b/pulse-protobuf/src/protos/pulse/vrl_tester/v1/vrl_tester.rs
@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.0
+// .proto file is parsed by protoc 32.1
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702
diff --git a/pulse-proxy/src/test/integration/prom.rs b/pulse-proxy/src/test/integration/prom.rs
index 4536123..5597686 100644
--- a/pulse-proxy/src/test/integration/prom.rs
+++ b/pulse-proxy/src/test/integration/prom.rs
@@ -117,6 +117,52 @@ async fn remote_write() {
   helper.shutdown().await;
 }
 
+fmt_reuse! {
+MAX_BATCH_SIZE = r#"
+  pipeline:
+    inflows:
+      prom_remote_write:
+        routes: ["outflow:prom"]
+        prom_remote_write:
+          bind: "inflow:prom"
+
+    outflows:
+      prom:
+        prom_remote_write:
+          send_to: "http://{fake_upstream}/api/v1/prom/write"
+          batch_max_size: 1024
+  "#;
+}
+
+#[tokio::test]
+async fn max_batch_size() {
+  let bind_resolver = HelperBindResolver::new(&["fake_upstream", "inflow:prom"], &[]).await;
+  let mut upstream = FakeHttpUpstream::new_prom(
+    "fake_upstream",
+    bind_resolver.clone(),
+    ParseConfig::default(),
+  )
+  .await;
+  let helper = Helper::new(
+    &fmt!(
+      MAX_BATCH_SIZE,
+      fake_upstream = bind_resolver.local_tcp_addr("fake_upstream"),
+    ),
+    bind_resolver.clone(),
+  )
+  .await;
+  let client = PromClient::new(bind_resolver.local_tcp_addr("inflow:prom")).await;
+
+  let metrics = (0 .. 150)
+    .map(|i| make_metric(&format!("metric_{i:03}"), &[], 1))
+    .collect_vec();
+  client.send(metrics).await;
+  assert_eq!(upstream.wait_for_metrics().await.1.len(), 75);
+  assert_eq!(upstream.wait_for_metrics().await.1.len(), 75);
+
+  helper.shutdown().await;
+}
+
 fmt_reuse! {
 LYFT_SPECIFIC_CONFIG = r#"
   pipeline: