diff --git a/Cargo.toml b/Cargo.toml
index 3d48ecf208b66..99686fcfd88ff 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -316,7 +316,7 @@ seahash = { version = "4.1.0", default-features = false }
 semver = { version = "1.0.20", default-features = false, features = ["serde", "std"], optional = true }
 smallvec = { version = "1", default-features = false, features = ["union", "serde"] }
 snafu = { version = "0.7.5", default-features = false, features = ["futures"] }
-snap = { version = "1.1.0", default-features = false, optional = true }
+snap = { version = "1.1.0", default-features = false }
 socket2 = { version = "0.5.5", default-features = false }
 stream-cancel = { version = "0.8.1", default-features = false }
 strip-ansi-escapes = { version = "0.2.0", default-features = false }
@@ -560,9 +560,9 @@ sources-splunk_hec = ["dep:roaring"]
 sources-statsd = ["sources-utils-net", "tokio-util/net"]
 sources-stdin = ["tokio-util/io"]
 sources-syslog = ["codecs-syslog", "sources-utils-net", "tokio-util/net"]
-sources-utils-http = ["dep:snap", "sources-utils-http-auth", "sources-utils-http-encoding", "sources-utils-http-error", "sources-utils-http-prelude"]
+sources-utils-http = ["sources-utils-http-auth", "sources-utils-http-encoding", "sources-utils-http-error", "sources-utils-http-prelude"]
 sources-utils-http-auth = ["sources-utils-http-error"]
-sources-utils-http-encoding = ["dep:snap", "sources-utils-http-error"]
+sources-utils-http-encoding = ["sources-utils-http-error"]
 sources-utils-http-error = []
 sources-utils-http-prelude = ["sources-utils-http", "sources-utils-http-auth", "sources-utils-http-encoding", "sources-utils-http-error"]
 sources-utils-http-query = []
@@ -715,7 +715,7 @@ sinks-nats = ["dep:async-nats", "dep:nkeys"]
 sinks-new_relic_logs = ["sinks-http"]
 sinks-new_relic = []
 sinks-papertrail = ["dep:syslog"]
-sinks-prometheus = ["dep:base64", "dep:prometheus-parser", "dep:snap"]
+sinks-prometheus = ["dep:base64", "dep:prometheus-parser"]
 sinks-pulsar = ["dep:apache-avro", "dep:pulsar", "dep:lru"]
 sinks-redis = ["dep:redis"]
 sinks-sematext = ["sinks-elasticsearch", "sinks-influxdb"]
diff --git a/lib/vector-stream/src/partitioned_batcher.rs b/lib/vector-stream/src/partitioned_batcher.rs
index 0d73cc5267098..b865216c1f764 100644
--- a/lib/vector-stream/src/partitioned_batcher.rs
+++ b/lib/vector-stream/src/partitioned_batcher.rs
@@ -15,7 +15,7 @@ use vector_core::{partition::Partitioner, time::KeyedTimer, ByteSizeOf};
 
 use crate::batcher::{
     config::BatchConfigParts,
-    data::BatchReduce,
+    data::BatchData,
     limiter::{ByteSizeOfItemSize, ItemBatchSize, SizeLimit},
     BatchConfig,
 };
@@ -155,16 +155,15 @@ impl BatcherSettings {
     }
 
     /// A batcher config using the `ItemBatchSize` trait to determine batch sizes.
-    /// The output is built with the supplied reducer function.
-    pub fn into_reducer_config<I, T, F, S>(
-        self,
+    /// The output is built with the supplied object implementing [`BatchData`].
+    pub fn as_reducer_config<I, T, B>(
+        &self,
         item_size: I,
-        reducer: F,
-    ) -> BatchConfigParts<SizeLimit<I>, BatchReduce<F, S>>
+        reducer: B,
+    ) -> BatchConfigParts<SizeLimit<I>, B>
     where
         I: ItemBatchSize<T>,
-        F: FnMut(&mut S, T),
-        S: Default,
+        B: BatchData<T>,
     {
         BatchConfigParts {
             batch_limiter: SizeLimit {
@@ -173,14 +172,14 @@ impl BatcherSettings {
                 current_size: 0,
                 item_size_calculator: item_size,
             },
-            batch_data: BatchReduce::new(reducer),
+            batch_data: reducer,
             timeout: self.timeout,
         }
     }
 }
 
 #[pin_project]
-pub struct PartitionedBatcher<St, Prt, KT, C, F>
+pub struct PartitionedBatcher<St, Prt, KT, C, F, B>
 where
     Prt: Partitioner,
 {
@@ -193,7 +192,7 @@ where
     /// The store of 'closed' batches. When this is not empty it will be
     /// preferentially flushed prior to consuming any new items from the
     /// underlying stream.
-    closed_batches: Vec<(Prt::Key, Vec<Prt::Item>)>,
+    closed_batches: Vec<(Prt::Key, B)>,
     /// The queue of pending batch expirations
     timer: KT,
     /// The partitioner for this `Batcher`
@@ -203,7 +202,7 @@ where
     stream: Fuse<St>,
 }
 
-impl<St, Prt, C, F> PartitionedBatcher<St, Prt, ExpirationQueue<Prt::Key>, C, F>
+impl<St, Prt, C, F, B> PartitionedBatcher<St, Prt, ExpirationQueue<Prt::Key>, C, F, B>
 where
     St: Stream<Item = Prt::Item>,
     Prt: Partitioner + Unpin,
@@ -226,7 +225,7 @@ where
 }
 
 #[cfg(test)]
-impl<St, Prt, KT, C, F> PartitionedBatcher<St, Prt, KT, C, F>
+impl<St, Prt, KT, C, F, B> PartitionedBatcher<St, Prt, KT, C, F, B>
 where
     St: Stream<Item = Prt::Item>,
     Prt: Partitioner + Unpin,
@@ -247,17 +246,17 @@ where
     }
 }
 
-impl<St, Prt, KT, C, F> Stream for PartitionedBatcher<St, Prt, KT, C, F>
+impl<St, Prt, KT, C, F, B> Stream for PartitionedBatcher<St, Prt, KT, C, F, B>
 where
     St: Stream<Item = Prt::Item>,
     Prt: Partitioner + Unpin,
     Prt::Key: Eq + Hash + Clone,
     Prt::Item: ByteSizeOf,
     KT: KeyedTimer<Prt::Key>,
-    C: BatchConfig<Prt::Item, Batch = Vec<Prt::Item>>,
+    C: BatchConfig<Prt::Item, Batch = B>,
     F: Fn() -> C + Send,
 {
-    type Item = (Prt::Key, Vec<Prt::Item>);
+    type Item = (Prt::Key, B);
 
     fn size_hint(&self) -> (usize, Option<usize>) {
         self.stream.size_hint()
@@ -270,20 +269,18 @@ where
                 return Poll::Ready(this.closed_batches.pop());
             }
             match this.stream.as_mut().poll_next(cx) {
-                Poll::Pending => {
-                    match this.timer.poll_expired(cx) {
-                        // Unlike normal streams, `DelayQueue` can return `None`
-                        // here but still be usable later if more entries are added.
-                        Poll::Pending | Poll::Ready(None) => return Poll::Pending,
-                        Poll::Ready(Some(item_key)) => {
-                            let mut batch = this
-                                .batches
-                                .remove(&item_key)
-                                .expect("batch should exist if it is set to expire");
-                            this.closed_batches.push((item_key, batch.take_batch()));
-                        }
+                Poll::Pending => match this.timer.poll_expired(cx) {
+                    // Unlike normal streams, `DelayQueue` can return `None`
+                    // here but still be usable later if more entries are added.
+                    Poll::Pending | Poll::Ready(None) => return Poll::Pending,
+                    Poll::Ready(Some(item_key)) => {
+                        let mut batch = this
+                            .batches
+                            .remove(&item_key)
+                            .expect("batch should exist if it is set to expire");
+                        this.closed_batches.push((item_key, batch.take_batch()));
                     }
-                }
+                },
                 Poll::Ready(None) => {
                     // Now that the underlying stream is closed, we need to
                     // clear out our batches, including all expiration
diff --git a/src/sinks/prometheus/remote_write.rs b/src/sinks/prometheus/remote_write.rs
deleted file mode 100644
index 043254630b63e..0000000000000
--- a/src/sinks/prometheus/remote_write.rs
+++ /dev/null
@@ -1,790 +0,0 @@
-use std::io::Read;
-use std::sync::Arc;
-use std::task;
-
-#[cfg(feature = "aws-core")]
-use aws_credential_types::provider::SharedCredentialsProvider;
-#[cfg(feature = "aws-core")]
-use aws_types::region::Region;
-use bytes::{Bytes, BytesMut};
-use futures::{future::BoxFuture, stream, FutureExt, SinkExt};
-use http::{Request, Uri};
-use prost::Message;
-use snafu::{ResultExt, Snafu};
-use tower::Service;
-use vector_config::configurable_component;
-use vector_core::{ByteSizeOf, EstimatedJsonEncodedSizeOf};
-
-use super::collector::{self, MetricCollector as _};
-use crate::{
-    config::{self, AcknowledgementsConfig, Input, SinkConfig},
-    event::{Event, Metric},
-    http::HttpClient,
-    internal_events::{EndpointBytesSent, TemplateRenderingError},
-    sinks::{
-        self,
-        prometheus::PrometheusRemoteWriteAuth,
-        util::{
-            auth::Auth,
-            batch::BatchConfig,
-            buffer::metrics::{MetricNormalize, MetricNormalizer, MetricSet, MetricsBuffer},
-            http::HttpRetryLogic,
-            uri, EncodedEvent, PartitionBuffer, PartitionInnerBuffer, SinkBatchSettings,
-            TowerRequestConfig,
-        },
-    },
-    template::Template,
-    tls::{TlsConfig, TlsSettings},
-};
-
-#[derive(Clone, Copy, Debug, Default)]
-pub struct PrometheusRemoteWriteDefaultBatchSettings;
-
-impl SinkBatchSettings for PrometheusRemoteWriteDefaultBatchSettings {
-    const MAX_EVENTS: Option<usize> = Some(1_000);
-    const MAX_BYTES: Option<usize> = None;
-    const TIMEOUT_SECS: f64 = 1.0;
-}
-
-#[derive(Debug, Snafu)]
-enum Errors {
-    #[snafu(display(r#"Prometheus remote_write sink cannot accept "set" metrics"#))]
-    SetMetricInvalid,
-    #[cfg(feature = "aws-core")]
-    #[snafu(display("aws.region required when AWS authentication is in use"))]
-    AwsRegionRequired,
-}
-
-/// Configuration for the `prometheus_remote_write` sink.
-#[configurable_component(sink(
-    "prometheus_remote_write",
-    "Deliver metric data to a Prometheus remote write endpoint."
-))]
-#[derive(Clone, Debug, Default)]
-#[serde(deny_unknown_fields)]
-pub struct RemoteWriteConfig {
-    /// The endpoint to send data to.
-    ///
-    /// The endpoint should include the scheme and the path to write to.
-    #[configurable(metadata(docs::examples = "https://localhost:8087/api/v1/write"))]
-    pub endpoint: String,
-
-    /// The default namespace for any metrics sent.
-    ///
-    /// This namespace is only used if a metric has no existing namespace. When a namespace is
-    /// present, it is used as a prefix to the metric name, and separated with an underscore (`_`).
-    ///
-    /// It should follow the Prometheus [naming conventions][prom_naming_docs].
-    ///
-    /// [prom_naming_docs]: https://prometheus.io/docs/practices/naming/#metric-names
-    #[configurable(metadata(docs::examples = "service"))]
-    #[configurable(metadata(docs::advanced))]
-    pub default_namespace: Option<String>,
-
-    /// Default buckets to use for aggregating [distribution][dist_metric_docs] metrics into histograms.
-    ///
-    /// [dist_metric_docs]: https://vector.dev/docs/about/under-the-hood/architecture/data-model/metric/#distribution
-    #[serde(default = "super::default_histogram_buckets")]
-    #[configurable(metadata(docs::advanced))]
-    pub buckets: Vec<f64>,
-
-    /// Quantiles to use for aggregating [distribution][dist_metric_docs] metrics into a summary.
-    ///
-    /// [dist_metric_docs]: https://vector.dev/docs/about/under-the-hood/architecture/data-model/metric/#distribution
-    #[serde(default = "super::default_summary_quantiles")]
-    #[configurable(metadata(docs::advanced))]
-    pub quantiles: Vec<f64>,
-
-    #[configurable(derived)]
-    #[serde(default)]
-    pub batch: BatchConfig<PrometheusRemoteWriteDefaultBatchSettings>,
-
-    #[configurable(derived)]
-    #[serde(default)]
-    pub request: TowerRequestConfig,
-
-    /// The tenant ID to send.
-    ///
-    /// If set, a header named `X-Scope-OrgID` is added to outgoing requests with the value of this setting.
-    ///
-    /// This may be used by Cortex or other remote services to identify the tenant making the request.
-    #[serde(default)]
-    #[configurable(metadata(docs::examples = "my-domain"))]
-    #[configurable(metadata(docs::advanced))]
-    pub tenant_id: Option<Template>,
-
-    #[configurable(derived)]
-    pub tls: Option<TlsConfig>,
-
-    #[configurable(derived)]
-    pub auth: Option<PrometheusRemoteWriteAuth>,
-
-    #[cfg(feature = "aws-core")]
-    #[configurable(derived)]
-    #[configurable(metadata(docs::advanced))]
-    pub aws: Option<crate::aws::RegionOrEndpoint>,
-
-    #[configurable(derived)]
-    #[serde(
-        default,
-        deserialize_with = "crate::serde::bool_or_struct",
-        skip_serializing_if = "crate::serde::skip_serializing_if_default"
-    )]
-    pub acknowledgements: AcknowledgementsConfig,
-
-    #[configurable(derived)]
-    #[configurable(metadata(docs::advanced))]
-    #[serde(default)]
-    pub compression: Compression,
-}
-
-impl_generate_config_from_default!(RemoteWriteConfig);
-
-/// Supported compression types for Prometheus Remote Write.
-#[configurable_component]
-#[derive(Clone, Copy, Debug, Derivative)]
-#[derivative(Default)]
-#[serde(rename_all = "lowercase")]
-pub enum Compression {
-    /// Snappy.
-    #[derivative(Default)]
-    Snappy,
-
-    /// Gzip.
-    Gzip,
-
-    /// Zstandard.
-    Zstd,
-}
-
-const fn convert_compression_to_content_encoding(compression: Compression) -> &'static str {
-    match compression {
-        Compression::Snappy => "snappy",
-        Compression::Gzip => "gzip",
-        Compression::Zstd => "zstd",
-    }
-}
-
-#[async_trait::async_trait]
-#[typetag::serde(name = "prometheus_remote_write")]
-impl SinkConfig for RemoteWriteConfig {
-    async fn build(
-        &self,
-        cx: config::SinkContext,
-    ) -> crate::Result<(sinks::VectorSink, sinks::Healthcheck)> {
-        let endpoint = self.endpoint.parse::<Uri>().context(sinks::UriParseSnafu)?;
-        let tls_settings = TlsSettings::from_options(&self.tls)?;
-        let batch = self.batch.into_batch_settings()?;
-        let request_settings = self.request.unwrap_with(&TowerRequestConfig::default());
-        let buckets = self.buckets.clone();
-        let quantiles = self.quantiles.clone();
-
-        let client = HttpClient::new(tls_settings, cx.proxy())?;
-        let tenant_id = self.tenant_id.clone();
-
-        let auth = match &self.auth {
-            Some(PrometheusRemoteWriteAuth::Basic { user, password }) => {
-                Some(Auth::Basic(crate::http::Auth::Basic {
-                    user: user.clone(),
-                    password: password.clone().into(),
-                }))
-            }
-            Some(PrometheusRemoteWriteAuth::Bearer { token }) => {
-                Some(Auth::Basic(crate::http::Auth::Bearer {
-                    token: token.clone(),
-                }))
-            }
-            #[cfg(feature = "aws-core")]
-            Some(PrometheusRemoteWriteAuth::Aws(aws_auth)) => {
-                let region = self
-                    .aws
-                    .as_ref()
-                    .map(|config| config.region())
-                    .ok_or(Errors::AwsRegionRequired)?
-                    .ok_or(Errors::AwsRegionRequired)?;
-                Some(Auth::Aws {
-                    credentials_provider: aws_auth.credentials_provider(region.clone()).await?,
-                    region,
-                })
-            }
-            None => None,
-        };
-
-        let http_request_builder = Arc::new(HttpRequestBuilder {
-            endpoint: endpoint.clone(),
-            auth,
-            compression: self.compression,
-        });
-
-        let healthcheck = healthcheck(client.clone(), Arc::clone(&http_request_builder)).boxed();
-        let service = RemoteWriteService {
-            default_namespace: self.default_namespace.clone(),
-            client,
-            buckets,
-            quantiles,
-            http_request_builder,
-            compression: self.compression,
-        };
-
-        let sink = {
-            let buffer = PartitionBuffer::new(MetricsBuffer::new(batch.size));
-            let mut normalizer = MetricNormalizer::<PrometheusMetricNormalize>::default();
-
-            request_settings
-                .partition_sink(HttpRetryLogic, service, buffer, batch.timeout)
-                .with_flat_map(move |event: Event| {
-                    let byte_size = event.size_of();
-                    let json_size = event.estimated_json_encoded_size_of();
-
-                    stream::iter(normalizer.normalize(event.into_metric()).map(|event| {
-                        let tenant_id = tenant_id.as_ref().and_then(|template| {
-                            template
-                                .render_string(&event)
-                                .map_err(|error| {
-                                    emit!(TemplateRenderingError {
-                                        error,
-                                        field: Some("tenant_id"),
-                                        drop_event: true,
-                                    })
-                                })
-                                .ok()
-                        });
-                        let key = PartitionKey { tenant_id };
-                        Ok(EncodedEvent::new(
-                            PartitionInnerBuffer::new(event, key),
-                            byte_size,
-                            json_size,
-                        ))
-                    }))
-                })
-                .sink_map_err(
-                    |error| error!(message = "Prometheus remote_write sink error.", %error),
-                )
-        };
-
-        #[allow(deprecated)]
-        Ok((sinks::VectorSink::from_event_sink(sink), healthcheck))
-    }
-
-    fn input(&self) -> Input {
-        Input::metric()
-    }
-
-    fn acknowledgements(&self) -> &AcknowledgementsConfig {
-        &self.acknowledgements
-    }
-}
-
-#[derive(Clone, Eq, Hash, PartialEq)]
-struct PartitionKey {
-    tenant_id: Option<String>,
-}
-
-async fn healthcheck(
-    client: HttpClient,
-    http_request_builder: Arc<HttpRequestBuilder>,
-) -> crate::Result<()> {
-    let body = bytes::Bytes::new();
-    let request = http_request_builder
-        .build_request(http::Method::GET, body.into(), None)
-        .await?;
-    let response = client.send(request).await?;
-
-    match response.status() {
-        http::StatusCode::OK => Ok(()),
-        other => Err(sinks::HealthcheckError::UnexpectedStatus { status: other }.into()),
-    }
-}
-
-#[derive(Default)]
-pub struct PrometheusMetricNormalize;
-
-impl MetricNormalize for PrometheusMetricNormalize {
-    fn normalize(&mut self, state: &mut MetricSet, metric: Metric) -> Option<Metric> {
-        state.make_absolute(metric)
-    }
-}
-
-#[derive(Clone)]
-struct RemoteWriteService {
-    default_namespace: Option<String>,
-    client: HttpClient,
-    buckets: Vec<f64>,
-    quantiles: Vec<f64>,
-    http_request_builder: Arc<HttpRequestBuilder>,
-    compression: Compression,
-}
-
-impl RemoteWriteService {
-    fn encode_events(&self, metrics: Vec<Metric>) -> Bytes {
-        let mut time_series = collector::TimeSeries::new();
-        for metric in metrics {
-            time_series.encode_metric(
-                self.default_namespace.as_deref(),
-                &self.buckets,
-                &self.quantiles,
-                &metric,
-            );
-        }
-        let request = time_series.finish();
-
-        let mut out = BytesMut::with_capacity(request.encoded_len());
-        request.encode(&mut out).expect("Out of memory");
-        out.freeze()
-    }
-}
-
-impl Service<PartitionInnerBuffer<Vec<Metric>, PartitionKey>> for RemoteWriteService {
-    type Response = http::Response<Bytes>;
-    type Error = crate::Error;
-    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
-
-    // Emission of an internal event in case of errors is handled upstream by the caller.
-    fn poll_ready(&mut self, _task: &mut task::Context<'_>) -> task::Poll<Result<(), Self::Error>> {
-        task::Poll::Ready(Ok(()))
-    }
-
-    // Emission of internal events for errors and dropped events is handled upstream by the caller.
-    fn call(&mut self, buffer: PartitionInnerBuffer<Vec<Metric>, PartitionKey>) -> Self::Future {
-        let (events, key) = buffer.into_parts();
-        let body = self.encode_events(events);
-        let body = compress_block(self.compression, body);
-
-        let client = self.client.clone();
-        let request_builder = Arc::clone(&self.http_request_builder);
-
-        Box::pin(async move {
-            let request = request_builder
-                .build_request(http::Method::POST, body, key.tenant_id)
-                .await?;
-
-            let (protocol, endpoint) = uri::protocol_endpoint(request.uri().clone());
-
-            let response = client.send(request).await?;
-            let (parts, body) = response.into_parts();
-            let body = hyper::body::to_bytes(body).await?;
-
-            emit!(EndpointBytesSent {
-                byte_size: body.len(),
-                protocol: &protocol,
-                endpoint: &endpoint
-            });
-
-            Ok(hyper::Response::from_parts(parts, body))
-        })
-    }
-}
-
-pub struct HttpRequestBuilder {
-    pub endpoint: Uri,
-    pub auth: Option<Auth>,
-    pub compression: Compression,
-}
-
-impl HttpRequestBuilder {
-    pub async fn build_request(
-        &self,
-        method: http::Method,
-        body: Vec<u8>,
-        tenant_id: Option<String>,
-    ) -> Result<Request<hyper::Body>, crate::Error> {
-        let content_encoding = convert_compression_to_content_encoding(self.compression);
-
-        let mut builder = http::Request::builder()
-            .method(method)
-            .uri(self.endpoint.clone())
-            .header("X-Prometheus-Remote-Write-Version", "0.1.0")
-            .header("Content-Encoding", content_encoding)
-            .header("Content-Type", "application/x-protobuf");
-
-        if let Some(tenant_id) = &tenant_id {
-            builder = builder.header("X-Scope-OrgID", tenant_id);
-        }
-
-        let mut request: Request<Bytes> = builder.body(body.into()).unwrap();
-        if let Some(auth) = &self.auth {
-            match auth {
-                Auth::Basic(http_auth) => http_auth.apply(&mut request),
-                #[cfg(feature = "aws-core")]
-                Auth::Aws {
-                    credentials_provider: provider,
-                    region,
-                } => sign_request(&mut request, provider, &Some(region.clone())).await?,
-            }
-        }
-
-        let (parts, body) = request.into_parts();
-        let request: Request<hyper::Body> = hyper::Request::from_parts(parts, body.into());
-
-        Ok(request)
-    }
-}
-
-fn compress_block(compression: Compression, data: Bytes) -> Vec<u8> {
-    match compression {
-        Compression::Snappy => snap::raw::Encoder::new()
-            .compress_vec(&data)
-            .expect("snap compression failed, please report"),
-        Compression::Gzip => {
-            let mut buf = Vec::new();
-            flate2::read::GzEncoder::new(data.as_ref(), flate2::Compression::default())
-                .read_to_end(&mut buf)
-                .expect("gzip compression failed, please report");
-            buf
-        }
-        Compression::Zstd => {
-            zstd::encode_all(data.as_ref(), 0).expect("zstd compression failed, please report")
-        }
-    }
-}
-
-#[cfg(feature = "aws-core")]
-async fn sign_request(
-    request: &mut http::Request<Bytes>,
-    credentials_provider: &SharedCredentialsProvider,
-    region: &Option<Region>,
-) -> crate::Result<()> {
-    crate::aws::sign_request("aps", request, credentials_provider, region).await
-}
-
-#[cfg(test)]
-mod tests {
-    use futures::StreamExt;
-    use http::HeaderMap;
-    use indoc::indoc;
-    use prometheus_parser::proto;
-    use vector_core::metric_tags;
-
-    use super::*;
-    use crate::{
-        config::SinkContext,
-        event::{MetricKind, MetricValue},
-        sinks::util::test::build_test_server,
-        test_util::{
-            self,
-            components::{assert_sink_compliance, HTTP_SINK_TAGS},
-        },
-    };
-
-    #[test]
-    fn generate_config() {
-        crate::test_util::test_generate_config::<RemoteWriteConfig>();
-    }
-
-    macro_rules! labels {
-        ( $( $name:expr => $value:expr ),* ) => {
-            vec![ $( proto::Label {
-                name: $name.to_string(),
-                value: $value.to_string()
-            }, )* ]
-        }
-    }
-
-    #[tokio::test]
-    async fn sends_request() {
-        let outputs = send_request("", vec![create_event("gauge-2".into(), 32.0)]).await;
-        assert_eq!(outputs.len(), 1);
-        let (headers, req) = &outputs[0];
-
-        assert!(!headers.contains_key("x-scope-orgid"));
-
-        assert_eq!(req.timeseries.len(), 1);
-        assert_eq!(
-            req.timeseries[0].labels,
-            labels!("__name__" => "gauge-2", "production" => "true", "region" => "us-west-1")
-        );
-        assert_eq!(req.timeseries[0].samples.len(), 1);
-        assert_eq!(req.timeseries[0].samples[0].value, 32.0);
-        assert_eq!(req.metadata.len(), 1);
-        assert_eq!(req.metadata[0].r#type, proto::MetricType::Gauge as i32);
-        assert_eq!(req.metadata[0].metric_family_name, "gauge-2");
-    }
-
-    #[tokio::test]
-    async fn sends_authenticated_request() {
-        let outputs = send_request(
-            indoc! {r#"
-                tenant_id = "tenant-%Y"
-                [auth]
-                strategy = "basic"
-                user = "user"
-                password = "password"
-            "#},
-            vec![create_event("gauge-2".into(), 32.0)],
-        )
-        .await;
-
-        assert_eq!(outputs.len(), 1);
-        let (_headers, req) = &outputs[0];
-
-        assert_eq!(req.timeseries.len(), 1);
-        assert_eq!(
-            req.timeseries[0].labels,
-            labels!("__name__" => "gauge-2", "production" => "true", "region" => "us-west-1")
-        );
-        assert_eq!(req.timeseries[0].samples.len(), 1);
-        assert_eq!(req.timeseries[0].samples[0].value, 32.0);
-        assert_eq!(req.metadata.len(), 1);
-        assert_eq!(req.metadata[0].r#type, proto::MetricType::Gauge as i32);
-        assert_eq!(req.metadata[0].metric_family_name, "gauge-2");
-    }
-
-    #[tokio::test]
-    async fn sends_authenticated_aws_request() {
-        let outputs = send_request(
-            indoc! {r#"
-                tenant_id = "tenant-%Y"
-                [aws]
-                region = "foo"
-                [auth]
-                strategy = "aws"
-                access_key_id = "foo"
-                secret_access_key = "bar"
-            "#},
-            vec![create_event("gauge-2".into(), 32.0)],
-        )
-        .await;
-
-        assert_eq!(outputs.len(), 1);
-        let (headers, _req) = &outputs[0];
-
-        let auth = headers["authorization"]
-            .to_str()
-            .expect("Missing AWS authorization header");
-        assert!(auth.starts_with("AWS4-HMAC-SHA256"));
-    }
-
-    #[tokio::test]
-    async fn sends_x_scope_orgid_header() {
-        let outputs = send_request(
-            r#"tenant_id = "tenant""#,
-            vec![create_event("gauge-3".into(), 12.0)],
-        )
-        .await;
-
-        assert_eq!(outputs.len(), 1);
-        let (headers, _) = &outputs[0];
-        assert_eq!(headers["x-scope-orgid"], "tenant");
-    }
-
-    #[tokio::test]
-    async fn sends_templated_x_scope_orgid_header() {
-        let outputs = send_request(
-            r#"tenant_id = "tenant-%Y""#,
-            vec![create_event("gauge-3".into(), 12.0)],
-        )
-        .await;
-
-        assert_eq!(outputs.len(), 1);
-        let (headers, _) = &outputs[0];
-        let orgid = headers["x-scope-orgid"]
-            .to_str()
-            .expect("Missing x-scope-orgid header");
-        assert!(orgid.starts_with("tenant-20"));
-        assert_eq!(orgid.len(), 11);
-    }
-
-    #[tokio::test]
-    async fn retains_state_between_requests() {
-        // This sink converts all incremental events to absolute, and
-        // should accumulate their totals between batches.
-        let outputs = send_request(
-            r#"batch.max_events = 1"#,
-            vec![
-                create_inc_event("counter-1".into(), 12.0),
-                create_inc_event("counter-2".into(), 13.0),
-                create_inc_event("counter-1".into(), 14.0),
-            ],
-        )
-        .await;
-
-        assert_eq!(outputs.len(), 3);
-
-        let check_output = |index: usize, name: &str, value: f64| {
-            let (_, req) = &outputs[index];
-            assert_eq!(req.timeseries.len(), 1);
-            assert_eq!(req.timeseries[0].labels, labels!("__name__" => name));
-            assert_eq!(req.timeseries[0].samples.len(), 1);
-            assert_eq!(req.timeseries[0].samples[0].value, value);
-        };
-        check_output(0, "counter-1", 12.0);
-        check_output(1, "counter-2", 13.0);
-        check_output(2, "counter-1", 26.0);
-    }
-
-    async fn send_request(
-        config: &str,
-        events: Vec<Event>,
-    ) -> Vec<(HeaderMap, proto::WriteRequest)> {
-        assert_sink_compliance(&HTTP_SINK_TAGS, async {
-            let addr = test_util::next_addr();
-            let (rx, trigger, server) = build_test_server(addr);
-            tokio::spawn(server);
-
-            let config = format!("endpoint = \"http://{}/write\"\n{}", addr, config);
-            let config: RemoteWriteConfig = toml::from_str(&config).unwrap();
-            let cx = SinkContext::default();
-
-            let (sink, _) = config.build(cx).await.unwrap();
-            sink.run_events(events).await.unwrap();
-
-            drop(trigger);
-
-            rx.map(|(parts, body)| {
-                assert_eq!(parts.method, "POST");
-                assert_eq!(parts.uri.path(), "/write");
-                let headers = parts.headers;
-                assert_eq!(headers["x-prometheus-remote-write-version"], "0.1.0");
-                assert_eq!(headers["content-encoding"], "snappy");
-                assert_eq!(headers["content-type"], "application/x-protobuf");
-
-                if config.auth.is_some() {
-                    assert!(headers.contains_key("authorization"));
-                }
-
-                let decoded = snap::raw::Decoder::new()
-                    .decompress_vec(&body)
-                    .expect("Invalid snappy compressed data");
-                let request =
-                    proto::WriteRequest::decode(Bytes::from(decoded)).expect("Invalid protobuf");
-                (headers, request)
-            })
-            .collect::<Vec<_>>()
-            .await
-        })
-        .await
-    }
-
-    pub(super) fn create_event(name: String, value: f64) -> Event {
-        Metric::new(name, MetricKind::Absolute, MetricValue::Gauge { value })
-            .with_tags(Some(metric_tags!(
-                "region" => "us-west-1",
-                "production" => "true",
-            )))
-            .with_timestamp(Some(chrono::Utc::now()))
-            .into()
-    }
-
-    fn create_inc_event(name: String, value: f64) -> Event {
-        Metric::new(
-            name,
-            MetricKind::Incremental,
-            MetricValue::Counter { value },
-        )
-        .with_timestamp(Some(chrono::Utc::now()))
-        .into()
-    }
-}
-
-#[cfg(all(test, feature = "prometheus-integration-tests"))]
-mod integration_tests {
-    use std::{collections::HashMap, ops::Range};
-
-    use serde_json::Value;
-
-    use super::{tests::*, *};
-    use crate::{
-        config::{SinkConfig, SinkContext},
-        event::{metric::MetricValue, Event},
-        sinks::influxdb::test_util::{cleanup_v1, format_timestamp, onboarding_v1, query_v1},
-        test_util::components::{assert_sink_compliance, HTTP_SINK_TAGS},
-        tls::{self, TlsConfig},
-    };
-
-    const HTTP_URL: &str = "http://influxdb-v1:8086";
-    const HTTPS_URL: &str = "https://influxdb-v1-tls:8087";
-
-    #[tokio::test]
-    async fn insert_metrics_over_http() {
-        insert_metrics(HTTP_URL).await;
-    }
-
-    #[tokio::test]
-    async fn insert_metrics_over_https() {
-        insert_metrics(HTTPS_URL).await;
-    }
-
-    async fn insert_metrics(url: &str) {
-        assert_sink_compliance(&HTTP_SINK_TAGS, async {
-            let database = onboarding_v1(url).await;
-
-            let cx = SinkContext::default();
-
-            let config = RemoteWriteConfig {
-                endpoint: format!("{}/api/v1/prom/write?db={}", url, database),
-                tls: Some(TlsConfig {
-                    ca_file: Some(tls::TEST_PEM_CA_PATH.into()),
-                    ..Default::default()
-                }),
-                ..Default::default()
-            };
-            let events = create_events(0..5, |n| n * 11.0);
-
-            let (sink, _) = config.build(cx).await.expect("error building config");
-            sink.run_events(events.clone()).await.unwrap();
-
-            let result = query(url, &format!("show series on {}", database)).await;
-
-            let values = &result["results"][0]["series"][0]["values"];
-            assert_eq!(values.as_array().unwrap().len(), 5);
-
-            for event in events {
-                let metric = event.into_metric();
-                let result = query(
-                    url,
-                    &format!(r#"SELECT * FROM "{}".."{}""#, database, metric.name()),
-                )
-                .await;
-
-                let metrics = decode_metrics(&result["results"][0]["series"][0]);
-                assert_eq!(metrics.len(), 1);
-                let output = &metrics[0];
-
-                match metric.value() {
-                    MetricValue::Gauge { value } => {
-                        assert_eq!(output["value"], Value::Number((*value as u32).into()))
-                    }
-                    _ => panic!("Unhandled metric value, fix the test"),
-                }
-                for (tag, value) in metric.tags().unwrap().iter_single() {
-                    assert_eq!(output[tag], Value::String(value.to_string()));
-                }
-                let timestamp =
-                    format_timestamp(metric.timestamp().unwrap(), chrono::SecondsFormat::Millis);
-                assert_eq!(output["time"], Value::String(timestamp));
-            }
-
-            cleanup_v1(url, &database).await;
-        })
-        .await
-    }
-
-    async fn query(url: &str, query: &str) -> Value {
-        let result = query_v1(url, query).await;
-        let text = result.text().await.unwrap();
-        serde_json::from_str(&text).expect("error when parsing InfluxDB response JSON")
-    }
-
-    fn decode_metrics(data: &Value) -> Vec<HashMap<String, Value>> {
-        let data = data.as_object().expect("Data is not an object");
-        let columns = data["columns"].as_array().expect("Columns is not an array");
-        data["values"]
-            .as_array()
-            .expect("Values is not an array")
-            .iter()
-            .map(|values| {
-                columns
-                    .iter()
-                    .zip(values.as_array().unwrap().iter())
-                    .map(|(column, value)| (column.as_str().unwrap().to_owned(), value.clone()))
-                    .collect()
-            })
-            .collect()
-    }
-
-    fn create_events(name_range: Range<i32>, value: impl Fn(f64) -> f64) -> Vec<Event> {
-        name_range
-            .map(move |num| create_event(format!("metric_{}", num), value(num as f64)))
-            .collect()
-    }
-}
diff --git a/src/sinks/prometheus/remote_write/config.rs b/src/sinks/prometheus/remote_write/config.rs
new file mode 100644
index 0000000000000..8f9f065642bdf
--- /dev/null
+++ b/src/sinks/prometheus/remote_write/config.rs
@@ -0,0 +1,229 @@
+use http::Uri;
+use snafu::prelude::*;
+
+use crate::{
+    http::HttpClient,
+    sinks::{
+        prelude::*,
+        prometheus::PrometheusRemoteWriteAuth,
+        util::{auth::Auth, http::http_response_retry_logic},
+        UriParseSnafu,
+    },
+};
+
+use super::{
+    service::{build_request, RemoteWriteService},
+    sink::{PrometheusRemoteWriteDefaultBatchSettings, RemoteWriteSink},
+};
+
+#[cfg(feature = "aws-core")]
+use super::Errors;
+
+/// The batch config for remote write.
+#[configurable_component]
+#[derive(Clone, Copy, Debug, Derivative)]
+#[derivative(Default)]
+pub struct RemoteWriteBatchConfig {
+    #[configurable(derived)]
+    #[serde(flatten)]
+    pub batch_settings: BatchConfig<PrometheusRemoteWriteDefaultBatchSettings>,
+
+    /// Whether or not to aggregate metrics within a batch.
+    #[serde(default = "crate::serde::default_true")]
+    #[derivative(Default(value = "true"))]
+    pub aggregate: bool,
+}
+
+/// Configuration for the `prometheus_remote_write` sink.
+#[configurable_component(sink(
+    "prometheus_remote_write",
+    "Deliver metric data to a Prometheus remote write endpoint."
+))]
+#[derive(Clone, Debug, Derivative)]
+#[derivative(Default)]
+#[serde(deny_unknown_fields)]
+pub struct RemoteWriteConfig {
+    /// The endpoint to send data to.
+    ///
+    /// The endpoint should include the scheme and the path to write to.
+    #[configurable(metadata(docs::examples = "https://localhost:8087/api/v1/write"))]
+    pub endpoint: String,
+
+    /// The default namespace for any metrics sent.
+    ///
+    /// This namespace is only used if a metric has no existing namespace. When a namespace is
+    /// present, it is used as a prefix to the metric name, and separated with an underscore (`_`).
+    ///
+    /// It should follow the Prometheus [naming conventions][prom_naming_docs].
+    ///
+    /// [prom_naming_docs]: https://prometheus.io/docs/practices/naming/#metric-names
+    #[configurable(metadata(docs::examples = "service"))]
+    #[configurable(metadata(docs::advanced))]
+    pub default_namespace: Option<String>,
+
+    /// Default buckets to use for aggregating [distribution][dist_metric_docs] metrics into histograms.
+    ///
+    /// [dist_metric_docs]: https://vector.dev/docs/about/under-the-hood/architecture/data-model/metric/#distribution
+    #[serde(default = "crate::sinks::prometheus::default_histogram_buckets")]
+    #[configurable(metadata(docs::advanced))]
+    pub buckets: Vec<f64>,
+
+    /// Quantiles to use for aggregating [distribution][dist_metric_docs] metrics into a summary.
+    ///
+    /// [dist_metric_docs]: https://vector.dev/docs/about/under-the-hood/architecture/data-model/metric/#distribution
+    #[serde(default = "crate::sinks::prometheus::default_summary_quantiles")]
+    #[configurable(metadata(docs::advanced))]
+    pub quantiles: Vec<f64>,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub batch: RemoteWriteBatchConfig,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub request: TowerRequestConfig,
+
+    /// The tenant ID to send.
+    ///
+    /// If set, a header named `X-Scope-OrgID` is added to outgoing requests with the value of this setting.
+    ///
+    /// This may be used by Cortex or other remote services to identify the tenant making the request.
+    #[serde(default)]
+    #[configurable(metadata(docs::examples = "my-domain"))]
+    #[configurable(metadata(docs::advanced))]
+    pub tenant_id: Option<Template>,
+
+    #[configurable(derived)]
+    pub tls: Option<TlsConfig>,
+
+    #[configurable(derived)]
+    pub auth: Option<PrometheusRemoteWriteAuth>,
+
+    #[cfg(feature = "aws-config")]
+    #[configurable(derived)]
+    #[configurable(metadata(docs::advanced))]
+    pub aws: Option<crate::aws::RegionOrEndpoint>,
+
+    #[configurable(derived)]
+    #[serde(
+        default,
+        deserialize_with = "crate::serde::bool_or_struct",
+        skip_serializing_if = "crate::serde::skip_serializing_if_default"
+    )]
+    pub acknowledgements: AcknowledgementsConfig,
+
+    #[configurable(derived)]
+    #[configurable(metadata(docs::advanced))]
+    #[serde(default = "default_compression")]
+    #[derivative(Default(value = "default_compression()"))]
+    pub compression: Compression,
+}
+
+const fn default_compression() -> Compression {
+    Compression::Snappy
+}
+
+impl_generate_config_from_default!(RemoteWriteConfig);
+
+#[async_trait::async_trait]
+#[typetag::serde(name = "prometheus_remote_write")]
+impl SinkConfig for RemoteWriteConfig {
+    fn acknowledgements(&self) -> &AcknowledgementsConfig {
+        &self.acknowledgements
+    }
+
+    async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
+        let endpoint = self.endpoint.parse::<Uri>().context(UriParseSnafu)?;
+        let tls_settings = TlsSettings::from_options(&self.tls)?;
+        let request_settings = self.request.unwrap_with(&TowerRequestConfig::default());
+        let buckets = self.buckets.clone();
+        let quantiles = self.quantiles.clone();
+        let default_namespace = self.default_namespace.clone();
+
+        let client = HttpClient::new(tls_settings, cx.proxy())?;
+
+        let auth = match &self.auth {
+            Some(PrometheusRemoteWriteAuth::Basic { user, password }) => {
+                Some(Auth::Basic(crate::http::Auth::Basic {
+                    user: user.clone(),
+                    password: password.clone().into(),
+                }))
+            }
+            Some(PrometheusRemoteWriteAuth::Bearer { token }) => {
+                Some(Auth::Basic(crate::http::Auth::Bearer {
+                    token: token.clone(),
+                }))
+            }
+            #[cfg(feature = "aws-core")]
+            Some(PrometheusRemoteWriteAuth::Aws(aws_auth)) => {
+                let region = self
+                    .aws
+                    .as_ref()
+                    .map(|config| config.region())
+                    .ok_or(Errors::AwsRegionRequired)?
+                    .ok_or(Errors::AwsRegionRequired)?;
+                Some(Auth::Aws {
+                    credentials_provider: aws_auth.credentials_provider(region.clone()).await?,
+                    region,
+                })
+            }
+            None => None,
+        };
+
+        let healthcheck = healthcheck(
+            client.clone(),
+            endpoint.clone(),
+            self.compression,
+            auth.clone(),
+        )
+        .boxed();
+
+        let service = RemoteWriteService {
+            endpoint,
+            client,
+            auth,
+            compression: self.compression,
+        };
+        let service = ServiceBuilder::new()
+            .settings(request_settings, http_response_retry_logic())
+            .service(service);
+
+        let sink = RemoteWriteSink {
+            tenant_id: self.tenant_id.clone(),
+            compression: self.compression,
+            aggregate: self.batch.aggregate,
+            batch_settings: self
+                .batch
+                .batch_settings
+                .validate()?
+                .into_batcher_settings()?,
+            buckets,
+            quantiles,
+            default_namespace,
+            service,
+        };
+
+        Ok((VectorSink::from_event_streamsink(sink), healthcheck))
+    }
+
+    fn input(&self) -> Input {
+        Input::metric()
+    }
+}
+
+async fn healthcheck(
+    client: HttpClient,
+    endpoint: Uri,
+    compression: Compression,
+    auth: Option<Auth>,
+) -> crate::Result<()> {
+    let body = bytes::Bytes::new();
+    let request =
+        build_request(http::Method::GET, &endpoint, compression, body, None, auth).await?;
+    let response = client.send(request).await?;
+
+    match response.status() {
+        http::StatusCode::OK => Ok(()),
+        other => Err(HealthcheckError::UnexpectedStatus { status: other }.into()),
+    }
+}
diff --git a/src/sinks/prometheus/remote_write/integration_tests.rs b/src/sinks/prometheus/remote_write/integration_tests.rs
new file mode 100644
index 0000000000000..07f106053fdad
--- /dev/null
+++ b/src/sinks/prometheus/remote_write/integration_tests.rs
@@ -0,0 +1,109 @@
+use std::{collections::HashMap, ops::Range};
+
+use serde_json::Value;
+
+use super::{tests::*, *};
+use crate::{
+    config::{SinkConfig, SinkContext},
+    event::{metric::MetricValue, Event},
+    sinks::influxdb::test_util::{cleanup_v1, format_timestamp, onboarding_v1, query_v1},
+    test_util::components::{assert_sink_compliance, HTTP_SINK_TAGS},
+    tls::{self, TlsConfig},
+};
+
+const HTTP_URL: &str = "http://influxdb-v1:8086";
+const HTTPS_URL: &str = "https://influxdb-v1-tls:8087";
+
+#[tokio::test]
+async fn insert_metrics_over_http() {
+    insert_metrics(HTTP_URL).await;
+}
+
+#[tokio::test]
+async fn insert_metrics_over_https() {
+    insert_metrics(HTTPS_URL).await;
+}
+
+async fn insert_metrics(url: &str) {
+    assert_sink_compliance(&HTTP_SINK_TAGS, async {
+        let database = onboarding_v1(url).await;
+
+        let cx = SinkContext::default();
+
+        let config = RemoteWriteConfig {
+            endpoint: format!("{}/api/v1/prom/write?db={}", url, database),
+            tls: Some(TlsConfig {
+                ca_file: Some(tls::TEST_PEM_CA_PATH.into()),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        let events = create_events(0..5, |n| n * 11.0);
+
+        let (sink, _) = config.build(cx).await.expect("error building config");
+        sink.run_events(events.clone()).await.unwrap();
+
+        let result = query(url, &format!("show series on {}", database)).await;
+
+        let values = &result["results"][0]["series"][0]["values"];
+        assert_eq!(values.as_array().unwrap().len(), 5);
+
+        for event in events {
+            let metric = event.into_metric();
+            let result = query(
+                url,
+                &format!(r#"SELECT * FROM "{}".."{}""#, database, metric.name()),
+            )
+            .await;
+
+            let metrics = decode_metrics(&result["results"][0]["series"][0]);
+            assert_eq!(metrics.len(), 1);
+            let output = &metrics[0];
+
+            match metric.value() {
+                MetricValue::Gauge { value } => {
+                    assert_eq!(output["value"], Value::Number((*value as u32).into()))
+                }
+                _ => panic!("Unhandled metric value, fix the test"),
+            }
+            for (tag, value) in metric.tags().unwrap().iter_single() {
+                assert_eq!(output[tag], Value::String(value.to_string()));
+            }
+            let timestamp =
+                format_timestamp(metric.timestamp().unwrap(), chrono::SecondsFormat::Millis);
+            assert_eq!(output["time"], Value::String(timestamp));
+        }
+
+        cleanup_v1(url, &database).await;
+    })
+    .await
+}
+
+async fn query(url: &str, query: &str) -> Value {
+    let result = query_v1(url, query).await;
+    let text = result.text().await.unwrap();
+    serde_json::from_str(&text).expect("error when parsing InfluxDB response JSON")
+}
+
+fn decode_metrics(data: &Value) -> Vec<HashMap<String, Value>> {
+    let data = data.as_object().expect("Data is not an object");
+    let columns = data["columns"].as_array().expect("Columns is not an array");
+    data["values"]
+        .as_array()
+        .expect("Values is not an array")
+        .iter()
+        .map(|values| {
+            columns
+                .iter()
+                .zip(values.as_array().unwrap().iter())
+                .map(|(column, value)| (column.as_str().unwrap().to_owned(), value.clone()))
+                .collect()
+        })
+        .collect()
+}
+
+fn create_events(name_range: Range<i32>, value: impl Fn(f64) -> f64) -> Vec<Event> {
+    name_range
+        .map(move |num| create_event(format!("metric_{}", num), value(num as f64)))
+        .collect()
+}
diff --git a/src/sinks/prometheus/remote_write/mod.rs b/src/sinks/prometheus/remote_write/mod.rs
new file mode 100644
index 0000000000000..3442c88de812f
--- /dev/null
+++ b/src/sinks/prometheus/remote_write/mod.rs
@@ -0,0 +1,48 @@
+//! The Prometheus Remote Write [`vector_core::sink::VectorSink`].
+//! Contains the [`vector_core::sink::VectorSink`] instance that is responsible
+//! for taking a stream of [`vector_core::event::Event`] and forwarding
+//! them to a server via the [Prometheus Remote Write protocol][remote_write].
+//!
+//! [remote_write]: https://prometheus.io/docs/concepts/remote_write_spec/
+
+use snafu::prelude::*;
+use vector_core::event::Metric;
+
+use crate::sinks::{
+    prelude::*,
+    util::buffer::metrics::{MetricNormalize, MetricSet},
+};
+
+mod config;
+mod request_builder;
+mod service;
+mod sink;
+
+#[cfg(test)]
+mod tests;
+
+#[cfg(all(test, feature = "prometheus-integration-tests"))]
+mod integration_tests;
+
+pub use config::RemoteWriteConfig;
+
+#[derive(Debug, Snafu)]
+enum Errors {
+    #[cfg(feature = "aws-core")]
+    #[snafu(display("aws.region required when AWS authentication is in use"))]
+    AwsRegionRequired,
+}
+
+#[derive(Clone, Eq, Hash, PartialEq)]
+struct PartitionKey {
+    tenant_id: Option<String>,
+}
+
+#[derive(Default)]
+pub struct PrometheusMetricNormalize;
+
+impl MetricNormalize for PrometheusMetricNormalize {
+    fn normalize(&mut self, state: &mut MetricSet, metric: Metric) -> Option<Metric> {
+        state.make_absolute(metric)
+    }
+}
diff --git a/src/sinks/prometheus/remote_write/request_builder.rs b/src/sinks/prometheus/remote_write/request_builder.rs
new file mode 100644
index 0000000000000..a2e156c42cfb8
--- /dev/null
+++ b/src/sinks/prometheus/remote_write/request_builder.rs
@@ -0,0 +1,131 @@
+use std::io;
+
+use bytes::{Bytes, BytesMut};
+use prost::Message;
+use vector_core::{config::telemetry, event::Metric};
+
+use crate::sinks::{
+    prelude::*,
+    prometheus::{collector, collector::MetricCollector as _},
+};
+
+use super::{sink::EventCollection, PartitionKey};
+
+pub(crate) struct RemoteWriteEncoder {
+    pub(super) default_namespace: Option<String>,
+    pub(super) buckets: Vec<f64>,
+    pub(super) quantiles: Vec<f64>,
+}
+
+impl encoding::Encoder<Vec<Metric>> for RemoteWriteEncoder {
+    fn encode_input(
+        &self,
+        input: Vec<Metric>,
+        writer: &mut dyn io::Write,
+    ) -> io::Result<(usize, GroupedCountByteSize)> {
+        let mut byte_size = telemetry().create_request_count_byte_size();
+
+        let mut time_series = collector::TimeSeries::new();
+        let len = input.len();
+        for metric in input {
+            byte_size.add_event(&metric, metric.estimated_json_encoded_size_of());
+
+            time_series.encode_metric(
+                self.default_namespace.as_deref(),
+                &self.buckets,
+                &self.quantiles,
+                &metric,
+            );
+        }
+        let request = time_series.finish();
+
+        let mut out = BytesMut::with_capacity(request.encoded_len());
+        request.encode(&mut out).expect("Out of memory");
+        let body = out.freeze();
+
+        write_all(writer, len, body.as_ref())?;
+
+        Ok((body.len(), byte_size))
+    }
+}
+
+#[derive(Clone)]
+pub(super) struct RemoteWriteRequest {
+    pub(super) request: Bytes,
+    pub(super) tenant_id: Option<String>,
+    finalizers: EventFinalizers,
+    metadata: RequestMetadata,
+}
+
+impl Finalizable for RemoteWriteRequest {
+    fn take_finalizers(&mut self) -> EventFinalizers {
+        std::mem::take(&mut self.finalizers)
+    }
+}
+
+impl MetaDescriptive for RemoteWriteRequest {
+    fn get_metadata(&self) -> &RequestMetadata {
+        &self.metadata
+    }
+
+    fn metadata_mut(&mut self) -> &mut RequestMetadata {
+        &mut self.metadata
+    }
+}
+
+pub(super) struct RemoteWriteMetadata {
+    finalizers: EventFinalizers,
+    tenant_id: Option<String>,
+}
+
+pub(super) struct RemoteWriteRequestBuilder {
+    pub(super) compression: Compression,
+    pub(super) encoder: RemoteWriteEncoder,
+}
+
+impl RequestBuilder<(PartitionKey, EventCollection)> for RemoteWriteRequestBuilder {
+    type Metadata = RemoteWriteMetadata;
+    type Events = Vec<Metric>;
+    type Encoder = RemoteWriteEncoder;
+    type Payload = Bytes;
+    type Request = RemoteWriteRequest;
+    type Error = io::Error;
+
+    fn compression(&self) -> Compression {
+        self.compression
+    }
+
+    fn encoder(&self) -> &Self::Encoder {
+        &self.encoder
+    }
+
+    fn split_input(
+        &self,
+        input: (PartitionKey, EventCollection),
+    ) -> (Self::Metadata, RequestMetadataBuilder, Self::Events) {
+        let (key, events) = input;
+        let finalizers = events.finalizers;
+        let metrics = events.events.into_metrics();
+        let builder = RequestMetadataBuilder::from_events(&metrics);
+        let metadata = RemoteWriteMetadata {
+            finalizers,
+            tenant_id: key.tenant_id,
+        };
+
+        (metadata, builder, metrics)
+    }
+
+    fn build_request(
+        &self,
+        metadata: Self::Metadata,
+        request_metadata: RequestMetadata,
+        payload: EncodeResult<Self::Payload>,
+    ) -> Self::Request {
+        RemoteWriteRequest {
+            request: payload.into_payload(),
+            finalizers: metadata.finalizers,
+            tenant_id: metadata.tenant_id,
+            metadata: request_metadata,
+        }
+    }
+}
diff --git a/src/sinks/prometheus/remote_write/service.rs b/src/sinks/prometheus/remote_write/service.rs
new file mode 100644
index 0000000000000..fb459d7c4bd8e
--- /dev/null
+++ b/src/sinks/prometheus/remote_write/service.rs
@@ -0,0 +1,139 @@
+use std::task::{Context, Poll};
+
+#[cfg(feature = "aws-core")]
+use aws_credential_types::provider::SharedCredentialsProvider;
+#[cfg(feature = "aws-core")]
+use aws_types::region::Region;
+
+use bytes::Bytes;
+use http::Uri;
+
+use super::request_builder::RemoteWriteRequest;
+use crate::{
+    http::HttpClient,
+    internal_events::EndpointBytesSent,
+    sinks::{
+        prelude::*,
+        util::{auth::Auth, http::HttpResponse},
+    },
+};
+
+/// Constants for header strings.
+mod headers {
+    pub(super) const X_PROMETHEUS_REMOTE_WRITE_VERSION: &str = "X-Prometheus-Remote-Write-Version";
+    pub(super) const CONTENT_ENCODING: &str = "Content-Encoding";
+    pub(super) const CONTENT_TYPE: &str = "Content-Type";
+    pub(super) const X_SCOPE_ORGID: &str = "X-Scope-OrgID";
+
+    pub(super) const VERSION: &str = "0.1.0";
+    pub(super) const APPLICATION_X_PROTOBUF: &str = "application/x-protobuf";
+}
+
+#[derive(Clone)]
+pub(super) struct RemoteWriteService {
+    pub(super) endpoint: Uri,
+    pub(super) auth: Option<Auth>,
+    pub(super) client: HttpClient,
+    pub(super) compression: super::Compression,
+}
+
+impl Service<RemoteWriteRequest> for RemoteWriteService {
+    type Response = HttpResponse;
+    type Error = crate::Error;
+    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
+
+    fn poll_ready(&mut self, _task: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    // Emission of internal events for errors and dropped events is handled upstream by the caller.
+    fn call(&mut self, mut request: RemoteWriteRequest) -> Self::Future {
+        let client = self.client.clone();
+        let endpoint = self.endpoint.clone();
+        let auth = self.auth.clone();
+        let compression = self.compression;
+
+        Box::pin(async move {
+            let metadata = std::mem::take(request.metadata_mut());
+            let json_size = metadata.into_events_estimated_json_encoded_byte_size();
+            let raw_byte_size = request.request.len();
+
+            let http_request = build_request(
+                http::Method::POST,
+                &endpoint,
+                compression,
+                request.request,
+                request.tenant_id.as_ref(),
+                auth,
+            )
+            .await?;
+
+            let response = client.send(http_request).await?;
+            let (parts, body) = response.into_parts();
+            let body = hyper::body::to_bytes(body).await?;
+            let http_response = hyper::Response::from_parts(parts, body);
+
+            if http_response.status().is_success() {
+                // We can't rely on the framework to emit this because we need to specify the additional `endpoint` tag.
+                emit!(EndpointBytesSent {
+                    byte_size: raw_byte_size,
+                    protocol: "http",
+                    endpoint: &endpoint.to_string(),
+                });
+            }
+
+            Ok(HttpResponse {
+                events_byte_size: json_size,
+                http_response,
+                raw_byte_size,
+            })
+        })
+    }
+}
+
+#[cfg(feature = "aws-core")]
+async fn sign_request(
+    request: &mut http::Request<Bytes>,
+    credentials_provider: &SharedCredentialsProvider,
+    region: &Option<Region>,
+) -> crate::Result<()> {
+    crate::aws::sign_request("aps", request, credentials_provider, region).await
+}
+
+pub(super) async fn build_request(
+    method: http::Method,
+    endpoint: &Uri,
+    compression: Compression,
+    body: Bytes,
+    tenant_id: Option<&String>,
+    auth: Option<Auth>,
+) -> crate::Result<http::Request<hyper::Body>> {
+    let mut builder = http::Request::builder()
+        .method(method)
+        .uri(endpoint)
+        .header(headers::X_PROMETHEUS_REMOTE_WRITE_VERSION, headers::VERSION)
+        .header(headers::CONTENT_TYPE, headers::APPLICATION_X_PROTOBUF);
+
+    if let Some(content_encoding) = compression.content_encoding() {
+        builder = builder.header(headers::CONTENT_ENCODING, content_encoding);
+    }
+
+    if let Some(tenant_id) = tenant_id {
+        builder = builder.header(headers::X_SCOPE_ORGID, tenant_id);
+    }
+
+    let mut request = builder.body(body)?;
+
+    if let Some(auth) = auth {
+        match auth {
+            Auth::Basic(http_auth) => http_auth.apply(&mut request),
+            #[cfg(feature = "aws-core")]
+            Auth::Aws {
+                credentials_provider: provider,
+                region,
+            } => sign_request(&mut request, &provider, &Some(region.clone())).await?,
+        }
+    }
+
+    Ok(request.map(hyper::Body::from))
+}
diff --git a/src/sinks/prometheus/remote_write/sink.rs b/src/sinks/prometheus/remote_write/sink.rs
new file mode 100644
index 0000000000000..1551f846ea92b
--- /dev/null
+++ b/src/sinks/prometheus/remote_write/sink.rs
@@ -0,0 +1,233 @@
+use std::fmt;
+
+use vector_common::byte_size_of::ByteSizeOf;
+use vector_core::event::Metric;
+use vector_stream::batcher::{data::BatchData, limiter::ByteSizeOfItemSize};
+
+use crate::sinks::{prelude::*, util::buffer::metrics::MetricSet};
+
+use super::{
+    request_builder::{RemoteWriteEncoder, RemoteWriteRequest, RemoteWriteRequestBuilder},
+    PartitionKey, PrometheusMetricNormalize,
+};
+
+pub(super) struct RemoteWriteMetric {
+    pub(super) metric: Metric,
+    tenant_id: Option<String>,
+}
+
+impl Finalizable for RemoteWriteMetric {
+    fn take_finalizers(&mut self) -> EventFinalizers {
+        self.metric.take_finalizers()
+    }
+}
+
+impl GetEventCountTags for RemoteWriteMetric {
+    fn get_tags(&self) -> TaggedEventsSent {
+        self.metric.get_tags()
+    }
+}
+
+impl EstimatedJsonEncodedSizeOf for RemoteWriteMetric {
+    fn estimated_json_encoded_size_of(&self) -> JsonSize {
+        self.metric.estimated_json_encoded_size_of()
+    }
+}
+
+impl ByteSizeOf for RemoteWriteMetric {
+    fn allocated_bytes(&self) -> usize {
+        self.metric.allocated_bytes()
+    }
+}
+
+#[derive(Clone, Copy, Debug, Default)]
+pub struct PrometheusRemoteWriteDefaultBatchSettings;
+
+impl SinkBatchSettings for PrometheusRemoteWriteDefaultBatchSettings {
+    const MAX_EVENTS: Option<usize> = Some(1_000);
+    const MAX_BYTES: Option<usize> = None;
+    const TIMEOUT_SECS: f64 = 1.0;
+}
+
+pub(super) struct PrometheusTenantIdPartitioner;
+
+impl Partitioner for PrometheusTenantIdPartitioner {
+    type Item = RemoteWriteMetric;
+    type Key = PartitionKey;
+
+    fn partition(&self, item: &Self::Item) -> Self::Key {
+        PartitionKey {
+            tenant_id: item.tenant_id.clone(),
+        }
+    }
+}
+
+pub(super) enum BatchedMetrics {
+    Aggregated(MetricSet),
+    Unaggregated(Vec<Metric>),
+}
+
+impl BatchedMetrics {
+    pub(super) fn into_metrics(self) -> Vec<Metric> {
+        match self {
+            BatchedMetrics::Aggregated(metrics) => metrics.into_metrics(),
+            BatchedMetrics::Unaggregated(metrics) => metrics,
+        }
+    }
+
+    pub(super) fn insert_update(&mut self, metric: Metric) {
+        match self {
+            BatchedMetrics::Aggregated(metrics) => metrics.insert_update(metric),
+            BatchedMetrics::Unaggregated(metrics) => metrics.push(metric),
+        }
+    }
+
+    pub(super) fn len(&self) -> usize {
+        match self {
+            BatchedMetrics::Aggregated(metrics) => metrics.len(),
+            BatchedMetrics::Unaggregated(metrics) => metrics.len(),
+        }
+    }
+}
+
+pub(super) struct EventCollection {
+    pub(super) finalizers: EventFinalizers,
+    pub(super) events: BatchedMetrics,
+    pub(super) events_byte_size: usize,
+    pub(super) events_json_byte_size: GroupedCountByteSize,
+}
+
+impl EventCollection {
+    /// Creates a new event collection that will either aggregate the incremental metrics
+    /// or store all the metrics, depending on the value of the `aggregate` parameter.
+    fn new(aggregate: bool) -> Self {
+        Self {
+            finalizers: Default::default(),
+            events: if aggregate {
+                BatchedMetrics::Aggregated(Default::default())
+            } else {
+                BatchedMetrics::Unaggregated(Default::default())
+            },
+            events_byte_size: Default::default(),
+            events_json_byte_size: telemetry().create_request_count_byte_size(),
+        }
+    }
+
+    const fn is_aggregated(&self) -> bool {
+        matches!(self.events, BatchedMetrics::Aggregated(_))
+    }
+}
+
+impl BatchData<RemoteWriteMetric> for EventCollection {
+    type Batch = Self;
+
+    fn len(&self) -> usize {
+        self.events.len()
+    }
+
+    fn take_batch(&mut self) -> Self::Batch {
+        let mut new = Self::new(self.is_aggregated());
+        std::mem::swap(self, &mut new);
+        new
+    }
+
+    fn push_item(&mut self, mut item: RemoteWriteMetric) {
+        self.finalizers
+            .merge(item.metric.metadata_mut().take_finalizers());
+        self.events_byte_size += item.size_of();
+        self.events_json_byte_size
+            .add_event(&item.metric, item.estimated_json_encoded_size_of());
+        self.events.insert_update(item.metric);
+    }
+}
+
+pub(super) struct RemoteWriteSink<S> {
+    pub(super) tenant_id: Option<Template>,
+    pub(super) batch_settings: BatcherSettings,
+    pub(super) aggregate: bool,
+    pub(super) compression: super::Compression,
+    pub(super) default_namespace: Option<String>,
+    pub(super) buckets: Vec<f64>,
+    pub(super) quantiles: Vec<f64>,
+    pub(super) service: S,
+}
+
+impl<S> RemoteWriteSink<S>
+where
+    S: Service<RemoteWriteRequest> + Send + 'static,
+    S::Future: Send + 'static,
+    S::Response: DriverResponse + Send + 'static,
+    S::Error: fmt::Debug + Into<crate::Error> + Send,
+{
+    async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
+        let request_builder = RemoteWriteRequestBuilder {
+            compression: self.compression,
+            encoder: RemoteWriteEncoder {
+                default_namespace: self.default_namespace.clone(),
+                buckets: self.buckets.clone(),
+                quantiles: self.quantiles.clone(),
+            },
+        };
+
+        let batch_settings = self.batch_settings;
+        let tenant_id = self.tenant_id.clone();
+        let service = self.service;
+
+        input
+            .filter_map(|event| future::ready(event.try_into_metric()))
+            .normalized_with_default::<PrometheusMetricNormalize>()
+            .filter_map(move |event| {
+                future::ready(make_remote_write_event(tenant_id.as_ref(), event))
+            })
+            .batched_partitioned(PrometheusTenantIdPartitioner, || {
+                batch_settings
+                    .as_reducer_config(ByteSizeOfItemSize, EventCollection::new(self.aggregate))
+            })
+            .request_builder(default_request_builder_concurrency_limit(), request_builder)
+            .filter_map(|request| async move {
+                match request {
+                    Err(e) => {
+                        error!("Failed to build Remote Write request: {:?}.", e);
+                        None
+                    }
+                    Ok(req) => Some(req),
+                }
+            })
+            .into_driver(service)
+            .run()
+            .await
+    }
+}
+
+#[async_trait]
+impl<S> StreamSink<Event> for RemoteWriteSink<S>
+where
+    S: Service<RemoteWriteRequest> + Send + 'static,
+    S::Future: Send + 'static,
+    S::Response: DriverResponse + Send + 'static,
+    S::Error: fmt::Debug + Into<crate::Error> + Send,
+{
+    async fn run(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
+        self.run_inner(input).await
+    }
+}
+
+fn make_remote_write_event(
+    tenant_id: Option<&Template>,
+    metric: Metric,
+) -> Option<RemoteWriteMetric> {
+    let tenant_id = tenant_id.and_then(|template| {
+        template
+            .render_string(&metric)
+            .map_err(|error| {
+                emit!(TemplateRenderingError {
+                    error,
+                    field: Some("tenant_id"),
+                    drop_event: true,
+                })
+            })
+            .ok()
+    });
+
+    Some(RemoteWriteMetric { metric, tenant_id })
+}
diff --git a/src/sinks/prometheus/remote_write/tests.rs b/src/sinks/prometheus/remote_write/tests.rs
new file mode 100644
index 0000000000000..190d3723bce6d
--- /dev/null
+++ b/src/sinks/prometheus/remote_write/tests.rs
@@ -0,0 +1,290 @@
+use bytes::Bytes;
+use futures::StreamExt;
+use http::HeaderMap;
+use indoc::indoc;
+use prometheus_parser::proto;
+use prost::Message;
+use vector_core::metric_tags;
+
+use super::*;
+use crate::{
+    config::SinkContext,
+    event::{MetricKind, MetricValue},
+    sinks::{prometheus::remote_write::RemoteWriteConfig, util::test::build_test_server},
+    test_util::{
+        self,
+        components::{assert_sink_compliance, HTTP_SINK_TAGS},
+    },
+};
+
+#[test]
+fn generate_config() {
+    crate::test_util::test_generate_config::<RemoteWriteConfig>();
+}
+
+macro_rules! labels {
+        ( $( $name:expr => $value:expr ),* ) => {
+            vec![ $( proto::Label {
+                name: $name.to_string(),
+                value: $value.to_string()
+            }, )* ]
+        }
+    }
+
+#[tokio::test]
+async fn sends_request() {
+    let outputs = send_request("", vec![create_event("gauge-2".into(), 32.0)]).await;
+    assert_eq!(outputs.len(), 1);
+    let (headers, req) = &outputs[0];
+
+    assert!(!headers.contains_key("x-scope-orgid"));
+
+    assert_eq!(req.timeseries.len(), 1);
+    assert_eq!(
+        req.timeseries[0].labels,
+        labels!("__name__" => "gauge-2", "production" => "true", "region" => "us-west-1")
+    );
+    assert_eq!(req.timeseries[0].samples.len(), 1);
+    assert_eq!(req.timeseries[0].samples[0].value, 32.0);
+    assert_eq!(req.metadata.len(), 1);
+    assert_eq!(req.metadata[0].r#type, proto::MetricType::Gauge as i32);
+    assert_eq!(req.metadata[0].metric_family_name, "gauge-2");
+}
+
+#[tokio::test]
+async fn sends_authenticated_request() {
+    let outputs = send_request(
+        indoc! {r#"
+                tenant_id = "tenant-%Y"
+                [auth]
+                strategy = "basic"
+                user = "user"
+                password = "password"
+            "#},
+        vec![create_event("gauge-2".into(), 32.0)],
+    )
+    .await;
+
+    assert_eq!(outputs.len(), 1);
+    let (_headers, req) = &outputs[0];
+
+    assert_eq!(req.timeseries.len(), 1);
+    assert_eq!(
+        req.timeseries[0].labels,
+        labels!("__name__" => "gauge-2", "production" => "true", "region" => "us-west-1")
+    );
+    assert_eq!(req.timeseries[0].samples.len(), 1);
+    assert_eq!(req.timeseries[0].samples[0].value, 32.0);
+    assert_eq!(req.metadata.len(), 1);
+    assert_eq!(req.metadata[0].r#type, proto::MetricType::Gauge as i32);
+    assert_eq!(req.metadata[0].metric_family_name, "gauge-2");
+}
+
+#[cfg(feature = "aws-config")]
+#[tokio::test]
+async fn sends_authenticated_aws_request() {
+    let outputs = send_request(
+        indoc! {r#"
+                tenant_id = "tenant-%Y"
+                [aws]
+                region = "foo"
+                [auth]
+                strategy = "aws"
+                access_key_id = "foo"
+                secret_access_key = "bar"
+            "#},
+        vec![create_event("gauge-2".into(), 32.0)],
+    )
+    .await;
+
+    assert_eq!(outputs.len(), 1);
+    let (headers, _req) = &outputs[0];
+
+    let auth = headers["authorization"]
+        .to_str()
+        .expect("Missing AWS authorization header");
+    assert!(auth.starts_with("AWS4-HMAC-SHA256"));
+}
+
+#[tokio::test]
+async fn sends_x_scope_orgid_header() {
+    let outputs = send_request(
+        r#"tenant_id = "tenant""#,
+        vec![create_event("gauge-3".into(), 12.0)],
+    )
+    .await;
+
+    assert_eq!(outputs.len(), 1);
+    let (headers, _) = &outputs[0];
+    assert_eq!(headers["x-scope-orgid"], "tenant");
+}
+
+#[tokio::test]
+async fn sends_templated_x_scope_orgid_header() {
+    let outputs = send_request(
+        r#"tenant_id = "tenant-%Y""#,
+        vec![create_event("gauge-3".into(), 12.0)],
+    )
+    .await;
+
+    assert_eq!(outputs.len(), 1);
+    let (headers, _) = &outputs[0];
+    let orgid = headers["x-scope-orgid"]
+        .to_str()
+        .expect("Missing x-scope-orgid header");
+    assert!(orgid.starts_with("tenant-20"));
+    assert_eq!(orgid.len(), 11);
+}
+
+#[tokio::test]
+async fn retains_state_between_requests() {
+    // This sink converts all incremental events to absolute, and
+    // should accumulate their totals between batches.
+    let outputs = send_request(
+        r#"batch.max_events = 1"#,
+        vec![
+            create_inc_event("counter-1".into(), 12.0),
+            create_inc_event("counter-2".into(), 13.0),
+            create_inc_event("counter-1".into(), 14.0),
+        ],
+    )
+    .await;
+
+    assert_eq!(outputs.len(), 3);
+
+    let check_output = |index: usize, name: &str, value: f64| {
+        let (_, req) = &outputs[index];
+        assert_eq!(req.timeseries.len(), 1);
+        assert_eq!(req.timeseries[0].labels, labels!("__name__" => name));
+        assert_eq!(req.timeseries[0].samples.len(), 1);
+        assert_eq!(req.timeseries[0].samples[0].value, value);
+    };
+    check_output(0, "counter-1", 12.0);
+    check_output(1, "counter-2", 13.0);
+    check_output(2, "counter-1", 26.0);
+}
+
+#[tokio::test]
+async fn aggregates_batches() {
+    let outputs = send_request(
+        r#"batch.max_events = 3"#,
+        vec![
+            create_inc_event("counter-1".into(), 12.0),
+            create_inc_event("counter-1".into(), 14.0),
+            create_inc_event("counter-2".into(), 13.0),
+            create_inc_event("counter-2".into(), 14.0),
+        ],
+    )
+    .await;
+
+    assert_eq!(outputs.len(), 1);
+
+    let (_, req) = &outputs[0];
+    assert_eq!(req.timeseries.len(), 2);
+    assert_eq!(req.timeseries[0].labels, labels!("__name__" => "counter-1"));
+    assert_eq!(req.timeseries[0].samples.len(), 1);
+    assert_eq!(req.timeseries[0].samples[0].value, 26.0);
+
+    assert_eq!(req.timeseries[1].labels, labels!("__name__" => "counter-2"));
+    assert_eq!(req.timeseries[1].samples.len(), 1);
+    assert_eq!(req.timeseries[1].samples[0].value, 27.0);
+}
+
+#[tokio::test]
+async fn doesnt_aggregate_batches() {
+    let outputs = send_request(
+        indoc! {
+            r#"
+            batch.max_events = 3
+            batch.aggregate = false
+            "#
+        },
+        vec![
+            create_inc_event("counter-1".into(), 12.0),
+            create_inc_event("counter-1".into(), 14.0),
+            create_inc_event("counter-2".into(), 13.0),
+            create_inc_event("counter-2".into(), 14.0),
+        ],
+    )
+    .await;
+
+    assert_eq!(outputs.len(), 2);
+
+    // The first three metrics are in the first batch.
+    let (_, req) = &outputs[0];
+    assert_eq!(req.timeseries.len(), 2);
+    assert_eq!(req.timeseries[0].labels, labels!("__name__" => "counter-1"));
+    assert_eq!(req.timeseries[0].samples.len(), 2);
+    assert_eq!(req.timeseries[0].samples[0].value, 12.0);
+    assert_eq!(req.timeseries[0].samples[1].value, 26.0);
+
+    assert_eq!(req.timeseries[1].labels, labels!("__name__" => "counter-2"));
+    assert_eq!(req.timeseries[1].samples.len(), 1);
+    assert_eq!(req.timeseries[1].samples[0].value, 13.0);
+
+    // The last metric is in the last batch.
+    let (_, req) = &outputs[1];
+    assert_eq!(req.timeseries[0].labels, labels!("__name__" => "counter-2"));
+    assert_eq!(req.timeseries[0].samples.len(), 1);
+    assert_eq!(req.timeseries[0].samples[0].value, 27.0);
+}
+
+async fn send_request(config: &str, events: Vec<Event>) -> Vec<(HeaderMap, proto::WriteRequest)> {
+    assert_sink_compliance(&HTTP_SINK_TAGS, async {
+        let addr = test_util::next_addr();
+        let (rx, trigger, server) = build_test_server(addr);
+        tokio::spawn(server);
+
+        let config = format!("endpoint = \"http://{}/write\"\n{}", addr, config);
+        let config: RemoteWriteConfig = toml::from_str(&config).unwrap();
+        let cx = SinkContext::default();
+
+        let (sink, _) = config.build(cx).await.unwrap();
+        sink.run_events(events).await.unwrap();
+
+        drop(trigger);
+
+        rx.map(|(parts, body)| {
+            assert_eq!(parts.method, "POST");
+            assert_eq!(parts.uri.path(), "/write");
+            let headers = parts.headers;
+            assert_eq!(headers["x-prometheus-remote-write-version"], "0.1.0");
+            assert_eq!(headers["content-encoding"], "snappy");
+            assert_eq!(headers["content-type"], "application/x-protobuf");
+
+            if config.auth.is_some() {
+                assert!(headers.contains_key("authorization"));
+            }
+
+            let decoded = snap::raw::Decoder::new()
+                .decompress_vec(&body)
+                .expect("Invalid snappy compressed data");
+            let request =
+                proto::WriteRequest::decode(Bytes::from(decoded)).expect("Invalid protobuf");
+            (headers, request)
+        })
+        .collect::<Vec<_>>()
+        .await
+    })
+    .await
+}
+
+pub(super) fn create_event(name: String, value: f64) -> Event {
+    Metric::new(name, MetricKind::Absolute, MetricValue::Gauge { value })
+        .with_tags(Some(metric_tags!(
+            "region" => "us-west-1",
+            "production" => "true",
+        )))
+        .with_timestamp(Some(chrono::Utc::now()))
+        .into()
+}
+
+fn create_inc_event(name: String, value: f64) -> Event {
+    Metric::new(
+        name,
+        MetricKind::Incremental,
+        MetricValue::Counter { value },
+    )
+    .with_timestamp(Some(chrono::Utc::now()))
+    .into()
+}
diff --git a/src/sinks/util/buffer/compression.rs b/src/sinks/util/buffer/compression.rs
index 11796b046104a..943232ad25be5 100644
--- a/src/sinks/util/buffer/compression.rs
+++ b/src/sinks/util/buffer/compression.rs
@@ -37,6 +37,11 @@ pub enum Compression {
     ///
     /// [zstd]: https://facebook.github.io/zstd/
     Zstd(CompressionLevel),
+
+    /// [Snappy][snappy] compression.
+    ///
+    /// [snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+    Snappy,
 }
 
 impl Compression {
@@ -70,6 +75,7 @@ impl Compression {
             Self::Gzip(_) => Some("gzip"),
             Self::Zlib(_) => Some("deflate"),
             Self::Zstd(_) => Some("zstd"),
+            Self::Snappy => Some("snappy"),
         }
     }
 
@@ -78,6 +84,7 @@ impl Compression {
             Self::Gzip(_) => Some("gzip"),
             Self::Zlib(_) => Some("deflate"),
             Self::Zstd(_) => Some("zstd"),
+            Self::Snappy => Some("snappy"),
             _ => None,
         }
     }
@@ -88,6 +95,7 @@ impl Compression {
             Self::Gzip(_) => "log.gz",
             Self::Zlib(_) => "log.zz",
             Self::Zstd(_) => "log.zst",
+            Self::Snappy => "log.snappy",
         }
     }
 
@@ -97,12 +105,13 @@ impl Compression {
             Compression::Gzip(_) => 9,
             Compression::Zlib(_) => 9,
             Compression::Zstd(_) => 21,
+            Compression::Snappy => 0,
         }
     }
 
     pub const fn compression_level(self) -> CompressionLevel {
         match self {
-            Self::None => CompressionLevel::None,
+            Self::None | Self::Snappy => CompressionLevel::None,
             Self::Gzip(level) | Self::Zlib(level) | Self::Zstd(level) => level,
         }
     }
@@ -117,6 +126,7 @@ impl fmt::Display for Compression {
             Compression::Zstd(ref level) => {
                 write!(f, "zstd({})", ZstdCompressionLevel::from(*level))
             }
+            Compression::Snappy => write!(f, "snappy"),
         }
     }
 }
@@ -144,6 +154,7 @@ impl<'de> de::Deserialize<'de> for Compression {
                     "gzip" => Ok(Compression::gzip_default()),
                     "zlib" => Ok(Compression::zlib_default()),
                     "zstd" => Ok(Compression::zstd_default()),
+                    "snappy" => Ok(Compression::Snappy),
                     _ => Err(de::Error::invalid_value(
                         de::Unexpected::Str(s),
                         &r#""none" or "gzip" or "zlib" or "zstd""#,
@@ -187,9 +198,13 @@ impl<'de> de::Deserialize<'de> for Compression {
                     "gzip" => Ok(Compression::Gzip(level.unwrap_or_default())),
                     "zlib" => Ok(Compression::Zlib(level.unwrap_or_default())),
                     "zstd" => Ok(Compression::Zstd(level.unwrap_or_default())),
+                    "snappy" => match level {
+                        Some(_) => Err(de::Error::unknown_field("level", &[])),
+                        None => Ok(Compression::Snappy),
+                    },
                     algorithm => Err(de::Error::unknown_variant(
                         algorithm,
-                        &["none", "gzip", "zlib", "zstd"],
+                        &["none", "gzip", "zlib", "zstd", "snappy"],
                     )),
                 }?;
 
@@ -252,6 +267,7 @@ impl ser::Serialize for Compression {
                     serializer.serialize_str("zstd")
                 }
             }
+            Compression::Snappy => serializer.serialize_str("snappy"),
         }
     }
 }
@@ -314,11 +330,18 @@ impl Configurable for Compression {
             "[zstd]: https://facebook.github.io/zstd/",
         );
 
+        let snappy_string_subschema = generate_string_schema(
+            "Snappy",
+            Some("[Snappy][snappy] compression."),
+            "[snappy]: https://github.com/google/snappy/blob/main/docs/README.md",
+        );
+
         let mut all_string_oneof_subschema = generate_one_of_schema(&[
             none_string_subschema,
             gzip_string_subschema,
             zlib_string_subschema,
             zstd_string_subschema,
+            snappy_string_subschema,
         ]);
         apply_base_metadata(&mut all_string_oneof_subschema, string_metadata);
 
@@ -509,6 +532,7 @@ mod test {
             (r#""none""#, Compression::None),
             (r#""gzip""#, Compression::Gzip(CompressionLevel::default())),
             (r#""zlib""#, Compression::Zlib(CompressionLevel::default())),
+            (r#""snappy""#, Compression::Snappy),
             (r#"{"algorithm": "none"}"#, Compression::None),
             (
                 r#"{"algorithm": "gzip"}"#,
@@ -551,7 +575,7 @@ mod test {
             ),
             (
                 r#"{"algorithm": "b42"}"#,
-                r#"unknown variant `b42`, expected one of `none`, `gzip`, `zlib`, `zstd` at line 1 column 20"#,
+                r#"unknown variant `b42`, expected one of `none`, `gzip`, `zlib`, `zstd`, `snappy` at line 1 column 20"#,
             ),
             (
                 r#"{"algorithm": "none", "level": "default"}"#,
@@ -581,6 +605,10 @@ mod test {
                 r#"{"algorithm": "zstd", "level": 22}"#,
                 r#"invalid value `22`, expected value in range [0, 21] at line 1 column 34"#,
             ),
+            (
+                r#"{"algorithm": "snappy", "level": 3}"#,
+                r#"unknown field `level`, there are no fields at line 1 column 35"#,
+            ),
         ];
         for (source, result) in fixtures_invalid.iter() {
             let deserialized: Result<Compression, _> = serde_json::from_str(source);
diff --git a/src/sinks/util/buffer/metrics/normalize.rs b/src/sinks/util/buffer/metrics/normalize.rs
index afd39f8c31c73..16812fcd26333 100644
--- a/src/sinks/util/buffer/metrics/normalize.rs
+++ b/src/sinks/util/buffer/metrics/normalize.rs
@@ -1,5 +1,4 @@
-use std::collections::HashMap;
-
+use indexmap::IndexMap;
 use vector_core::event::{
     metric::{MetricData, MetricSeries},
     EventMetadata, Metric, MetricKind,
@@ -83,10 +82,11 @@ type MetricEntry = (MetricData, EventMetadata);
 
 /// Metric storage for use with normalization.
 ///
-/// This is primarily a wrapper around `HashMap` with convenience methods to make it easier to perform
+/// This is primarily a wrapper around [`IndexMap`] (to ensure insertion order
+/// is maintained) with convenience methods to make it easier to perform
 /// normalization-specific operations.
-#[derive(Clone, Default)]
-pub struct MetricSet(HashMap<MetricSeries, MetricEntry>);
+#[derive(Clone, Default, Debug)]
+pub struct MetricSet(IndexMap<MetricSeries, MetricEntry>);
 
 impl MetricSet {
     /// Creates an empty `MetricSet` with the specified capacity.
@@ -94,7 +94,7 @@ impl MetricSet {
     /// The metric set will be able to hold at least `capacity` elements without reallocating. If `capacity` is 0, the
     /// metric set will not allocate.
     pub fn with_capacity(capacity: usize) -> Self {
-        Self(HashMap::with_capacity(capacity))
+        Self(IndexMap::with_capacity(capacity))
     }
 
     /// Returns the number of elements in the set.
diff --git a/src/sinks/util/buffer/mod.rs b/src/sinks/util/buffer/mod.rs
index 296e14b02f9c8..f0ab3dd6af8fb 100644
--- a/src/sinks/util/buffer/mod.rs
+++ b/src/sinks/util/buffer/mod.rs
@@ -5,6 +5,7 @@ use flate2::write::{GzEncoder, ZlibEncoder};
 
 use super::{
     batch::{err_event_too_large, Batch, BatchSize, PushResult},
+    snappy::SnappyEncoder,
     zstd::ZstdEncoder,
 };
 
@@ -32,6 +33,7 @@ pub enum InnerBuffer {
     Gzip(GzEncoder<bytes::buf::Writer<BytesMut>>),
     Zlib(ZlibEncoder<bytes::buf::Writer<BytesMut>>),
     Zstd(ZstdEncoder<bytes::buf::Writer<BytesMut>>),
+    Snappy(SnappyEncoder<bytes::buf::Writer<BytesMut>>),
 }
 
 impl Buffer {
@@ -62,6 +64,7 @@ impl Buffer {
                     ZstdEncoder::new(writer, level.into())
                         .expect("Zstd encoder should not fail on init."),
                 ),
+                Compression::Snappy => InnerBuffer::Snappy(SnappyEncoder::new(writer)),
             }
         })
     }
@@ -81,6 +84,7 @@ impl Buffer {
             InnerBuffer::Zstd(inner) => {
                 inner.write_all(input).unwrap();
             }
+            InnerBuffer::Snappy(inner) => inner.write_all(input).unwrap(),
         }
     }
 
@@ -92,6 +96,7 @@ impl Buffer {
                 InnerBuffer::Gzip(inner) => inner.get_ref().get_ref().is_empty(),
                 InnerBuffer::Zlib(inner) => inner.get_ref().get_ref().is_empty(),
                 InnerBuffer::Zstd(inner) => inner.get_ref().get_ref().is_empty(),
+                InnerBuffer::Snappy(inner) => inner.is_empty(),
             })
             .unwrap_or(true)
     }
@@ -142,6 +147,10 @@ impl Batch for Buffer {
                 .finish()
                 .expect("This can't fail because the inner writer is a Vec")
                 .into_inner(),
+            Some(InnerBuffer::Snappy(inner)) => inner
+                .finish()
+                .expect("This can't fail because the inner writer is a Vec")
+                .into_inner(),
             None => BytesMut::new(),
         }
     }
diff --git a/src/sinks/util/builder.rs b/src/sinks/util/builder.rs
index 7e5fb0d8ca954..da74aeb20f6bf 100644
--- a/src/sinks/util/builder.rs
+++ b/src/sinks/util/builder.rs
@@ -44,11 +44,11 @@ pub trait SinkBuilderExt: Stream {
     /// The stream will yield batches of events, with their partition key, when either a batch fills
     /// up or times out. [`Partitioner`] operates on a per-event basis, and has access to the event
     /// itself, and so can access any and all fields of an event.
-    fn batched_partitioned<P, C, F>(
+    fn batched_partitioned<P, C, F, B>(
         self,
         partitioner: P,
         settings: F,
-    ) -> PartitionedBatcher<Self, P, ExpirationQueue<P::Key>, C, F>
+    ) -> PartitionedBatcher<Self, P, ExpirationQueue<P::Key>, C, F, B>
     where
         Self: Stream<Item = P::Item> + Sized,
         P: Partitioner + Unpin,
diff --git a/src/sinks/util/compressor.rs b/src/sinks/util/compressor.rs
index b918893a39bd2..f6d3c01ccca87 100644
--- a/src/sinks/util/compressor.rs
+++ b/src/sinks/util/compressor.rs
@@ -3,13 +3,14 @@ use std::io;
 use bytes::{BufMut, BytesMut};
 use flate2::write::{GzEncoder, ZlibEncoder};
 
-use super::{zstd::ZstdEncoder, Compression};
+use super::{snappy::SnappyEncoder, zstd::ZstdEncoder, Compression};
 
 enum Writer {
     Plain(bytes::buf::Writer<BytesMut>),
     Gzip(GzEncoder<bytes::buf::Writer<BytesMut>>),
     Zlib(ZlibEncoder<bytes::buf::Writer<BytesMut>>),
     Zstd(ZstdEncoder<bytes::buf::Writer<BytesMut>>),
+    Snappy(SnappyEncoder<bytes::buf::Writer<BytesMut>>),
 }
 
 impl Writer {
@@ -19,6 +20,7 @@ impl Writer {
             Writer::Gzip(inner) => inner.get_ref().get_ref(),
             Writer::Zlib(inner) => inner.get_ref().get_ref(),
             Writer::Zstd(inner) => inner.get_ref().get_ref(),
+            Writer::Snappy(inner) => inner.get_ref().get_ref(),
         }
     }
 }
@@ -35,6 +37,7 @@ impl From<Compression> for Writer {
                     .expect("Zstd encoder should not fail on init.");
                 Writer::Zstd(encoder)
             }
+            Compression::Snappy => Writer::Snappy(SnappyEncoder::new(writer)),
         }
     }
 }
@@ -47,6 +50,7 @@ impl io::Write for Writer {
             Writer::Gzip(writer) => writer.write(buf),
             Writer::Zlib(writer) => writer.write(buf),
             Writer::Zstd(writer) => writer.write(buf),
+            Writer::Snappy(writer) => writer.write(buf),
         }
     }
 
@@ -56,6 +60,7 @@ impl io::Write for Writer {
             Writer::Gzip(writer) => writer.flush(),
             Writer::Zlib(writer) => writer.flush(),
             Writer::Zstd(writer) => writer.flush(),
+            Writer::Snappy(writer) => writer.flush(),
         }
     }
 }
@@ -98,6 +103,7 @@ impl Compressor {
             Writer::Gzip(writer) => writer.finish()?,
             Writer::Zlib(writer) => writer.finish()?,
             Writer::Zstd(writer) => writer.finish()?,
+            Writer::Snappy(writer) => writer.finish()?,
         }
         .into_inner();
 
@@ -125,6 +131,9 @@ impl Compressor {
             Writer::Zstd(writer) => writer
                 .finish()
                 .expect("zstd writer should not fail to finish"),
+            Writer::Snappy(writer) => writer
+                .finish()
+                .expect("snappy writer should not fail to finish"),
         }
         .into_inner()
     }
diff --git a/src/sinks/util/http.rs b/src/sinks/util/http.rs
index eb67ca9772ba3..635618664fcd6 100644
--- a/src/sinks/util/http.rs
+++ b/src/sinks/util/http.rs
@@ -653,8 +653,8 @@ impl ByteSizeOf for HttpRequest {
 /// Response type for use in the `Service` implementation of HTTP stream sinks.
 pub struct HttpResponse {
     pub http_response: Response<Bytes>,
-    events_byte_size: GroupedCountByteSize,
-    raw_byte_size: usize,
+    pub events_byte_size: GroupedCountByteSize,
+    pub raw_byte_size: usize,
 }
 
 impl DriverResponse for HttpResponse {
diff --git a/src/sinks/util/mod.rs b/src/sinks/util/mod.rs
index 1fcae4842cd7f..f5c09b97f380f 100644
--- a/src/sinks/util/mod.rs
+++ b/src/sinks/util/mod.rs
@@ -16,6 +16,7 @@ pub mod request_builder;
 pub mod retries;
 pub mod service;
 pub mod sink;
+pub mod snappy;
 pub mod socket_bytes_sink;
 pub mod statistic;
 pub mod tcp;
diff --git a/src/sinks/util/snappy.rs b/src/sinks/util/snappy.rs
new file mode 100644
index 0000000000000..a46edd41ebf97
--- /dev/null
+++ b/src/sinks/util/snappy.rs
@@ -0,0 +1,88 @@
+//! An encoder for [Snappy] compression.
+//! Whilst there does exist a [Writer] implementation for Snappy, this compresses
+//! using the [Snappy frame format][frame], which is not quite what we want. So
+//! instead this encoder buffers the data in a [`Vec`] until the end. The `raw`
+//! compressor is then used to compress the data and writes it to the provided
+//! writer.
+//!
+//! [Snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+//! [Writer]: https://docs.rs/snap/latest/snap/write/struct.FrameEncoder.html
+//! [frame]: https://github.com/google/snappy/blob/master/framing_format.txt
+
+use std::io;
+
+use snap::raw::Encoder;
+
+pub struct SnappyEncoder<W: io::Write> {
+    writer: W,
+    buffer: Vec<u8>,
+}
+
+impl<W: io::Write> SnappyEncoder<W> {
+    pub fn new(writer: W) -> Self {
+        Self {
+            writer,
+            buffer: Vec::new(),
+        }
+    }
+
+    pub fn finish(mut self) -> io::Result<W> {
+        let mut encoder = Encoder::new();
+        let compressed = encoder.compress_vec(&self.buffer)?;
+
+        self.writer.write_all(&compressed)?;
+
+        Ok(self.writer)
+    }
+
+    pub const fn get_ref(&self) -> &W {
+        &self.writer
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.buffer.is_empty()
+    }
+}
+
+impl<W: io::Write> io::Write for SnappyEncoder<W> {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.buffer.extend_from_slice(buf);
+        Ok(buf.len())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+impl<W: io::Write + std::fmt::Debug> std::fmt::Debug for SnappyEncoder<W> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("SnappyEncoder")
+            .field("inner", &self.get_ref())
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::io::Write;
+
+    use bytes::{BufMut, BytesMut};
+
+    use super::*;
+
+    #[test]
+    fn is_empty() {
+        let writer = BytesMut::with_capacity(64).writer();
+        let mut encoder = SnappyEncoder::new(writer);
+
+        encoder.write_all(b"I am a potato").unwrap();
+
+        // Because we are buffering the results until the end, the writer will be
+        // empty, but our buffer won't be. The `is_empty` function is provided to
+        // allow us to determine if data has been written to the encoder without having
+        // to check the writer.
+        assert!(encoder.get_ref().get_ref().is_empty());
+        assert!(!encoder.is_empty());
+    }
+}
diff --git a/src/sinks/vector/sink.rs b/src/sinks/vector/sink.rs
index 8d6527326d5b4..0375302b95b6c 100644
--- a/src/sinks/vector/sink.rs
+++ b/src/sinks/vector/sink.rs
@@ -6,7 +6,7 @@ use prost::Message;
 use tower::Service;
 use vector_common::request_metadata::GroupedCountByteSize;
 use vector_core::{config::telemetry, ByteSizeOf, EstimatedJsonEncodedSizeOf};
-use vector_stream::{BatcherSettings, DriverResponse};
+use vector_stream::{batcher::data::BatchReduce, BatcherSettings, DriverResponse};
 
 use super::service::VectorRequest;
 use crate::{
@@ -68,14 +68,14 @@ where
                     wrapper: EventWrapper::from(event),
                 }
             })
-            .batched(self.batch_settings.into_reducer_config(
+            .batched(self.batch_settings.as_reducer_config(
                 |data: &EventData| data.wrapper.encoded_len(),
-                |event_collection: &mut EventCollection, item: EventData| {
+                BatchReduce::new(|event_collection: &mut EventCollection, item: EventData| {
                     event_collection.finalizers.merge(item.finalizers);
                     event_collection.events.push(item.wrapper);
                     event_collection.events_byte_size += item.byte_size;
                     event_collection.events_json_byte_size += item.json_byte_size;
-                },
+                }),
             ))
             .map(|event_collection| {
                 let builder = RequestMetadataBuilder::new(
diff --git a/src/sources/prometheus/remote_write.rs b/src/sources/prometheus/remote_write.rs
index 6fe08a32c7b51..9a2e4356975b6 100644
--- a/src/sources/prometheus/remote_write.rs
+++ b/src/sources/prometheus/remote_write.rs
@@ -155,11 +155,7 @@ mod test {
     use crate::{
         config::{SinkConfig, SinkContext},
         sinks::prometheus::remote_write::RemoteWriteConfig,
-        test_util::{
-            self,
-            components::{assert_source_compliance, HTTP_PUSH_SOURCE_TAGS},
-            wait_for_tcp,
-        },
+        test_util::{self, wait_for_tcp},
         tls::MaybeTlsSettings,
         SourceSender,
     };
@@ -180,54 +176,51 @@ mod test {
     }
 
     async fn receives_metrics(tls: Option<TlsEnableableConfig>) {
-        assert_source_compliance(&HTTP_PUSH_SOURCE_TAGS, async {
-            let address = test_util::next_addr();
-            let (tx, rx) = SourceSender::new_test_finalize(EventStatus::Delivered);
-
-            let proto = MaybeTlsSettings::from_config(&tls, true)
-                .unwrap()
-                .http_protocol_name();
-            let source = PrometheusRemoteWriteConfig {
-                address,
-                auth: None,
-                tls: tls.clone(),
-                acknowledgements: SourceAcknowledgementsConfig::default(),
-            };
-            let source = source
-                .build(SourceContext::new_test(tx, None))
-                .await
-                .unwrap();
-            tokio::spawn(source);
-            wait_for_tcp(address).await;
-
-            let sink = RemoteWriteConfig {
-                endpoint: format!("{}://localhost:{}/", proto, address.port()),
-                tls: tls.map(|tls| tls.options),
-                ..Default::default()
-            };
-            let (sink, _) = sink
-                .build(SinkContext::default())
-                .await
-                .expect("Error building config.");
-
-            let events = make_events();
-            let events_copy = events.clone();
-            let mut output = test_util::spawn_collect_ready(
-                async move {
-                    sink.run_events(events_copy).await.unwrap();
-                },
-                rx,
-                1,
-            )
-            .await;
-
-            // The MetricBuffer used by the sink may reorder the metrics, so
-            // put them back into order before comparing.
-            output.sort_unstable_by_key(|event| event.as_metric().name().to_owned());
+        let address = test_util::next_addr();
+        let (tx, rx) = SourceSender::new_test_finalize(EventStatus::Delivered);
 
-            vector_common::assert_event_data_eq!(events, output);
-        })
+        let proto = MaybeTlsSettings::from_config(&tls, true)
+            .unwrap()
+            .http_protocol_name();
+        let source = PrometheusRemoteWriteConfig {
+            address,
+            auth: None,
+            tls: tls.clone(),
+            acknowledgements: SourceAcknowledgementsConfig::default(),
+        };
+        let source = source
+            .build(SourceContext::new_test(tx, None))
+            .await
+            .unwrap();
+        tokio::spawn(source);
+        wait_for_tcp(address).await;
+
+        let sink = RemoteWriteConfig {
+            endpoint: format!("{}://localhost:{}/", proto, address.port()),
+            tls: tls.map(|tls| tls.options),
+            ..Default::default()
+        };
+        let (sink, _) = sink
+            .build(SinkContext::default())
+            .await
+            .expect("Error building config.");
+
+        let events = make_events();
+        let events_copy = events.clone();
+        let mut output = test_util::spawn_collect_ready(
+            async move {
+                sink.run_events(events_copy).await.unwrap();
+            },
+            rx,
+            1,
+        )
         .await;
+
+        // The MetricBuffer used by the sink may reorder the metrics, so
+        // put them back into order before comparing.
+        output.sort_unstable_by_key(|event| event.as_metric().name().to_owned());
+
+        vector_common::assert_event_data_eq!(events, output);
     }
 
     fn make_events() -> Vec<Event> {
@@ -278,69 +271,66 @@ mod test {
     /// we accept the metric, but take the last label in the list.
     #[tokio::test]
     async fn receives_metrics_duplicate_labels() {
-        assert_source_compliance(&HTTP_PUSH_SOURCE_TAGS, async {
-            let address = test_util::next_addr();
-            let (tx, rx) = SourceSender::new_test_finalize(EventStatus::Delivered);
-
-            let source = PrometheusRemoteWriteConfig {
-                address,
-                auth: None,
-                tls: None,
-                acknowledgements: SourceAcknowledgementsConfig::default(),
-            };
-            let source = source
-                .build(SourceContext::new_test(tx, None))
-                .await
-                .unwrap();
-            tokio::spawn(source);
-            wait_for_tcp(address).await;
-
-            let sink = RemoteWriteConfig {
-                endpoint: format!("http://localhost:{}/", address.port()),
-                ..Default::default()
-            };
-            let (sink, _) = sink
-                .build(SinkContext::default())
-                .await
-                .expect("Error building config.");
-
-            let timestamp = Utc::now().trunc_subsecs(3);
-
-            let events = vec![Metric::new(
-                "gauge_2",
-                MetricKind::Absolute,
-                MetricValue::Gauge { value: 41.0 },
-            )
-            .with_timestamp(Some(timestamp))
-            .with_tags(Some(metric_tags! {
-                "code" => "200".to_string(),
-                "code" => "success".to_string(),
-            }))
-            .into()];
-
-            let expected = vec![Metric::new(
-                "gauge_2",
-                MetricKind::Absolute,
-                MetricValue::Gauge { value: 41.0 },
-            )
-            .with_timestamp(Some(timestamp))
-            .with_tags(Some(metric_tags! {
-                "code" => "success".to_string(),
-            }))
-            .into()];
-
-            let output = test_util::spawn_collect_ready(
-                async move {
-                    sink.run_events(events).await.unwrap();
-                },
-                rx,
-                1,
-            )
-            .await;
+        let address = test_util::next_addr();
+        let (tx, rx) = SourceSender::new_test_finalize(EventStatus::Delivered);
 
-            vector_common::assert_event_data_eq!(expected, output);
-        })
+        let source = PrometheusRemoteWriteConfig {
+            address,
+            auth: None,
+            tls: None,
+            acknowledgements: SourceAcknowledgementsConfig::default(),
+        };
+        let source = source
+            .build(SourceContext::new_test(tx, None))
+            .await
+            .unwrap();
+        tokio::spawn(source);
+        wait_for_tcp(address).await;
+
+        let sink = RemoteWriteConfig {
+            endpoint: format!("http://localhost:{}/", address.port()),
+            ..Default::default()
+        };
+        let (sink, _) = sink
+            .build(SinkContext::default())
+            .await
+            .expect("Error building config.");
+
+        let timestamp = Utc::now().trunc_subsecs(3);
+
+        let events = vec![Metric::new(
+            "gauge_2",
+            MetricKind::Absolute,
+            MetricValue::Gauge { value: 41.0 },
+        )
+        .with_timestamp(Some(timestamp))
+        .with_tags(Some(metric_tags! {
+            "code" => "200".to_string(),
+            "code" => "success".to_string(),
+        }))
+        .into()];
+
+        let expected = vec![Metric::new(
+            "gauge_2",
+            MetricKind::Absolute,
+            MetricValue::Gauge { value: 41.0 },
+        )
+        .with_timestamp(Some(timestamp))
+        .with_tags(Some(metric_tags! {
+            "code" => "success".to_string(),
+        }))
+        .into()];
+
+        let output = test_util::spawn_collect_ready(
+            async move {
+                sink.run_events(events).await.unwrap();
+            },
+            rx,
+            1,
+        )
         .await;
+
+        vector_common::assert_event_data_eq!(expected, output);
     }
 }
 
diff --git a/website/cue/reference/components/sinks/base/appsignal.cue b/website/cue/reference/components/sinks/base/appsignal.cue
index 68904ee5046e6..48b3e4b518f7c 100644
--- a/website/cue/reference/components/sinks/base/appsignal.cue
+++ b/website/cue/reference/components/sinks/base/appsignal.cue
@@ -78,6 +78,11 @@ base: components: sinks: appsignal: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/aws_cloudwatch_logs.cue b/website/cue/reference/components/sinks/base/aws_cloudwatch_logs.cue
index fd8bebd59bdc0..162b8dbe4853f 100644
--- a/website/cue/reference/components/sinks/base/aws_cloudwatch_logs.cue
+++ b/website/cue/reference/components/sinks/base/aws_cloudwatch_logs.cue
@@ -180,6 +180,11 @@ base: components: sinks: aws_cloudwatch_logs: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/aws_cloudwatch_metrics.cue b/website/cue/reference/components/sinks/base/aws_cloudwatch_metrics.cue
index daa94b379ffda..16e90e50e5193 100644
--- a/website/cue/reference/components/sinks/base/aws_cloudwatch_metrics.cue
+++ b/website/cue/reference/components/sinks/base/aws_cloudwatch_metrics.cue
@@ -177,6 +177,11 @@ base: components: sinks: aws_cloudwatch_metrics: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/aws_kinesis_firehose.cue b/website/cue/reference/components/sinks/base/aws_kinesis_firehose.cue
index 3e4ac7c1d325b..c5ba35bba8dea 100644
--- a/website/cue/reference/components/sinks/base/aws_kinesis_firehose.cue
+++ b/website/cue/reference/components/sinks/base/aws_kinesis_firehose.cue
@@ -180,6 +180,11 @@ base: components: sinks: aws_kinesis_firehose: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/aws_kinesis_streams.cue b/website/cue/reference/components/sinks/base/aws_kinesis_streams.cue
index a2a7a659b8b46..1e4e3387b2581 100644
--- a/website/cue/reference/components/sinks/base/aws_kinesis_streams.cue
+++ b/website/cue/reference/components/sinks/base/aws_kinesis_streams.cue
@@ -180,6 +180,11 @@ base: components: sinks: aws_kinesis_streams: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/aws_s3.cue b/website/cue/reference/components/sinks/base/aws_s3.cue
index 5e6e0fef81491..55ad8f643d934 100644
--- a/website/cue/reference/components/sinks/base/aws_s3.cue
+++ b/website/cue/reference/components/sinks/base/aws_s3.cue
@@ -264,6 +264,11 @@ base: components: sinks: aws_s3: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/axiom.cue b/website/cue/reference/components/sinks/base/axiom.cue
index a20276b2ee26b..2de88265fc8f8 100644
--- a/website/cue/reference/components/sinks/base/axiom.cue
+++ b/website/cue/reference/components/sinks/base/axiom.cue
@@ -43,6 +43,11 @@ base: components: sinks: axiom: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/azure_blob.cue b/website/cue/reference/components/sinks/base/azure_blob.cue
index 34d92a850d390..482f927432613 100644
--- a/website/cue/reference/components/sinks/base/azure_blob.cue
+++ b/website/cue/reference/components/sinks/base/azure_blob.cue
@@ -127,6 +127,11 @@ base: components: sinks: azure_blob: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/clickhouse.cue b/website/cue/reference/components/sinks/base/clickhouse.cue
index acd2cafb2c588..52c00efdd4b71 100644
--- a/website/cue/reference/components/sinks/base/clickhouse.cue
+++ b/website/cue/reference/components/sinks/base/clickhouse.cue
@@ -122,6 +122,11 @@ base: components: sinks: clickhouse: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/datadog_logs.cue b/website/cue/reference/components/sinks/base/datadog_logs.cue
index 1858aab1659eb..6e2961f8b39af 100644
--- a/website/cue/reference/components/sinks/base/datadog_logs.cue
+++ b/website/cue/reference/components/sinks/base/datadog_logs.cue
@@ -76,6 +76,11 @@ base: components: sinks: datadog_logs: configuration: {
 				[gzip]: https://www.gzip.org/
 				"""
 			none: "No compression."
+			snappy: """
+				[Snappy][snappy] compression.
+
+				[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+				"""
 			zlib: """
 				[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/datadog_traces.cue b/website/cue/reference/components/sinks/base/datadog_traces.cue
index edf6ec6d65bc7..1564e8f720f2d 100644
--- a/website/cue/reference/components/sinks/base/datadog_traces.cue
+++ b/website/cue/reference/components/sinks/base/datadog_traces.cue
@@ -76,6 +76,11 @@ base: components: sinks: datadog_traces: configuration: {
 				[gzip]: https://www.gzip.org/
 				"""
 			none: "No compression."
+			snappy: """
+				[Snappy][snappy] compression.
+
+				[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+				"""
 			zlib: """
 				[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/elasticsearch.cue b/website/cue/reference/components/sinks/base/elasticsearch.cue
index 9e0041fcdd068..1e039927ff1bc 100644
--- a/website/cue/reference/components/sinks/base/elasticsearch.cue
+++ b/website/cue/reference/components/sinks/base/elasticsearch.cue
@@ -279,6 +279,11 @@ base: components: sinks: elasticsearch: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/gcp_cloud_storage.cue b/website/cue/reference/components/sinks/base/gcp_cloud_storage.cue
index d28a12a3693c7..8bcf9922c23a0 100644
--- a/website/cue/reference/components/sinks/base/gcp_cloud_storage.cue
+++ b/website/cue/reference/components/sinks/base/gcp_cloud_storage.cue
@@ -151,6 +151,11 @@ base: components: sinks: gcp_cloud_storage: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/http.cue b/website/cue/reference/components/sinks/base/http.cue
index 4f2733fba9158..43a14b957d084 100644
--- a/website/cue/reference/components/sinks/base/http.cue
+++ b/website/cue/reference/components/sinks/base/http.cue
@@ -122,6 +122,11 @@ base: components: sinks: http: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/humio_logs.cue b/website/cue/reference/components/sinks/base/humio_logs.cue
index 74fe040a84fcb..6529d8294f36a 100644
--- a/website/cue/reference/components/sinks/base/humio_logs.cue
+++ b/website/cue/reference/components/sinks/base/humio_logs.cue
@@ -75,6 +75,11 @@ base: components: sinks: humio_logs: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/humio_metrics.cue b/website/cue/reference/components/sinks/base/humio_metrics.cue
index 2d39e4dca0b90..c45509ce09b10 100644
--- a/website/cue/reference/components/sinks/base/humio_metrics.cue
+++ b/website/cue/reference/components/sinks/base/humio_metrics.cue
@@ -75,6 +75,11 @@ base: components: sinks: humio_metrics: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/new_relic.cue b/website/cue/reference/components/sinks/base/new_relic.cue
index 1adcd87c1e0d8..535650bb074d4 100644
--- a/website/cue/reference/components/sinks/base/new_relic.cue
+++ b/website/cue/reference/components/sinks/base/new_relic.cue
@@ -92,6 +92,11 @@ base: components: sinks: new_relic: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/prometheus_remote_write.cue b/website/cue/reference/components/sinks/base/prometheus_remote_write.cue
index 4f825fdac5752..8858c0f99ed74 100644
--- a/website/cue/reference/components/sinks/base/prometheus_remote_write.cue
+++ b/website/cue/reference/components/sinks/base/prometheus_remote_write.cue
@@ -190,9 +190,14 @@ base: components: sinks: prometheus_remote_write: configuration: {
 		}
 	}
 	batch: {
-		description: "Event batching behavior."
+		description: "The batch config for remote write."
 		required:    false
 		type: object: options: {
+			aggregate: {
+				description: "Whether or not to aggregate metrics within a batch."
+				required:    false
+				type: bool: default: true
+			}
 			max_bytes: {
 				description: """
 					The maximum size of a batch that is processed by a sink.
@@ -234,14 +239,36 @@ base: components: sinks: prometheus_remote_write: configuration: {
 		}
 	}
 	compression: {
-		description: "Supported compression types for Prometheus Remote Write."
-		required:    false
+		description: """
+			Compression configuration.
+
+			All compression algorithms use the default compression level unless otherwise specified.
+			"""
+		required: false
 		type: string: {
 			default: "snappy"
 			enum: {
-				gzip:   "Gzip."
-				snappy: "Snappy."
-				zstd:   "Zstandard."
+				gzip: """
+					[Gzip][gzip] compression.
+
+					[gzip]: https://www.gzip.org/
+					"""
+				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
+				zlib: """
+					[Zlib][zlib] compression.
+
+					[zlib]: https://zlib.net/
+					"""
+				zstd: """
+					[Zstandard][zstd] compression.
+
+					[zstd]: https://facebook.github.io/zstd/
+					"""
 			}
 		}
 	}
diff --git a/website/cue/reference/components/sinks/base/splunk_hec_logs.cue b/website/cue/reference/components/sinks/base/splunk_hec_logs.cue
index f105e81aad584..cfb3fc32268aa 100644
--- a/website/cue/reference/components/sinks/base/splunk_hec_logs.cue
+++ b/website/cue/reference/components/sinks/base/splunk_hec_logs.cue
@@ -116,6 +116,11 @@ base: components: sinks: splunk_hec_logs: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/splunk_hec_metrics.cue b/website/cue/reference/components/sinks/base/splunk_hec_metrics.cue
index 96c8517576f39..820454b8e867d 100644
--- a/website/cue/reference/components/sinks/base/splunk_hec_metrics.cue
+++ b/website/cue/reference/components/sinks/base/splunk_hec_metrics.cue
@@ -102,6 +102,11 @@ base: components: sinks: splunk_hec_metrics: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.
 
diff --git a/website/cue/reference/components/sinks/base/webhdfs.cue b/website/cue/reference/components/sinks/base/webhdfs.cue
index e85dc6de61bf1..a757159037863 100644
--- a/website/cue/reference/components/sinks/base/webhdfs.cue
+++ b/website/cue/reference/components/sinks/base/webhdfs.cue
@@ -75,6 +75,11 @@ base: components: sinks: webhdfs: configuration: {
 					[gzip]: https://www.gzip.org/
 					"""
 				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
+
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
 				zlib: """
 					[Zlib][zlib] compression.