feat(prometheus_remote_write sink): add custom HTTP headers support

Add support for custom HTTP headers in the prometheus_remote_write sink
via the `request.headers` configuration option. This enables users to
add custom headers to outgoing requests for authentication, routing, or
other integration requirements with Prometheus-compatible backends.

Author: elohmeier

changelog.d/prometheus_remote_write_custom_headers.enhancement.md

@@ -0,0 +1,3 @@
+The `prometheus_remote_write` sink now supports custom HTTP headers via the `request.headers` configuration option. This allows users to add custom headers to outgoing requests, which is useful for authentication, routing, or other integration requirements with Prometheus-compatible backends.
+
+authors: elohmeier

src/sinks/prometheus/remote_write/config.rs

@@ -13,7 +13,10 @@ use crate::{
         UriParseSnafu,
         prelude::*,
         prometheus::PrometheusRemoteWriteAuth,
-        util::{auth::Auth, http::http_response_retry_logic},
+        util::{
+            auth::Auth,
+            http::{RequestConfig, http_response_retry_logic},
+        },
     },
 };
 
@@ -79,7 +82,7 @@ pub struct RemoteWriteConfig {
 
     #[configurable(derived)]
     #[serde(default)]
-    pub request: TowerRequestConfig,
+    pub request: RequestConfig,
 
     /// The tenant ID to send.
     ///
@@ -141,7 +144,7 @@ impl SinkConfig for RemoteWriteConfig {
     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.as_ref())?;
-        let request_settings = self.request.into_settings();
+        let request_settings = self.request.tower.into_settings();
         let buckets = self.buckets.clone();
         let quantiles = self.quantiles.clone();
         let default_namespace = self.default_namespace.clone();
@@ -178,11 +181,18 @@ impl SinkConfig for RemoteWriteConfig {
             None => None,
         };
 
+        // Split headers into static and template-based
+        let (static_headers, _template_headers) = self.request.split_headers();
+
+        // Validate that custom headers don't conflict with auth
+        crate::sinks::util::http::validate_headers(&static_headers)?;
+
         let healthcheck = healthcheck(
             client.clone(),
             endpoint.clone(),
             self.compression,
             auth.clone(),
+            static_headers.clone(),
         )
         .boxed();
 
@@ -191,6 +201,7 @@ impl SinkConfig for RemoteWriteConfig {
             client,
             auth,
             compression: self.compression,
+            headers: static_headers,
         };
         let service = ServiceBuilder::new()
             .settings(request_settings, http_response_retry_logic())
@@ -225,10 +236,19 @@ async fn healthcheck(
     endpoint: Uri,
     compression: Compression,
     auth: Option<Auth>,
+    headers: std::collections::BTreeMap<String, String>,
 ) -> crate::Result<()> {
     let body = bytes::Bytes::new();
-    let request =
-        build_request(http::Method::GET, &endpoint, compression, body, None, auth).await?;
+    let request = build_request(
+        http::Method::GET,
+        &endpoint,
+        compression,
+        body,
+        None,
+        auth,
+        headers,
+    )
+    .await?;
     let response = client.send(request).await?;
 
     match response.status() {

src/sinks/prometheus/remote_write/service.rs

@@ -1,4 +1,7 @@
-use std::task::{Context, Poll};
+use std::{
+    collections::BTreeMap,
+    task::{Context, Poll},
+};
 
 #[cfg(feature = "aws-core")]
 use aws_credential_types::provider::SharedCredentialsProvider;
@@ -34,6 +37,7 @@ pub(super) struct RemoteWriteService {
     pub(super) auth: Option<Auth>,
     pub(super) client: HttpClient,
     pub(super) compression: super::Compression,
+    pub(super) headers: BTreeMap<String, String>,
 }
 
 impl Service<RemoteWriteRequest> for RemoteWriteService {
@@ -51,6 +55,7 @@ impl Service<RemoteWriteRequest> for RemoteWriteService {
         let endpoint = self.endpoint.clone();
         let auth = self.auth.clone();
         let compression = self.compression;
+        let headers = self.headers.clone();
 
         Box::pin(async move {
             let metadata = std::mem::take(request.metadata_mut());
@@ -64,6 +69,7 @@ impl Service<RemoteWriteRequest> for RemoteWriteService {
                 request.request,
                 request.tenant_id.as_ref(),
                 auth,
+                headers,
             )
             .await?;
 
@@ -106,6 +112,7 @@ pub(super) async fn build_request(
     body: Bytes,
     tenant_id: Option<&String>,
     auth: Option<Auth>,
+    custom_headers: BTreeMap<String, String>,
 ) -> crate::Result<http::Request<hyper::Body>> {
     let mut builder = http::Request::builder()
         .method(method)
@@ -121,6 +128,11 @@ pub(super) async fn build_request(
         builder = builder.header(headers::X_SCOPE_ORGID, tenant_id);
     }
 
+    // Apply custom headers
+    for (name, value) in custom_headers {
+        builder = builder.header(name, value);
+    }
+
     let mut request = builder.body(body)?;
 
     if let Some(auth) = auth {

src/sinks/prometheus/remote_write/tests.rs

@@ -135,6 +135,38 @@ async fn sends_templated_x_scope_orgid_header() {
     assert_eq!(orgid.len(), 11);
 }
 
+#[tokio::test]
+async fn sends_custom_headers() {
+    let outputs = send_request(
+        indoc! {r#"
+                [request.headers]
+                X-Custom-Header = "custom-value"
+                X-Another-Header = "another-value"
+            "#},
+        vec![create_event("gauge-4".into(), 42.0)],
+    )
+    .await;
+
+    assert_eq!(outputs.len(), 1);
+    let (headers, req) = &outputs[0];
+
+    // Verify custom headers are present
+    assert_eq!(headers["x-custom-header"], "custom-value");
+    assert_eq!(headers["x-another-header"], "another-value");
+
+    // Verify standard headers are still present
+    assert_eq!(headers["x-prometheus-remote-write-version"], "0.1.0");
+    assert_eq!(headers["content-type"], "application/x-protobuf");
+
+    // Verify the metric data is correct
+    assert_eq!(req.timeseries.len(), 1);
+    assert_eq!(
+        req.timeseries[0].labels,
+        labels!("__name__" => "gauge-4", "production" => "true", "region" => "us-west-1")
+    );
+    assert_eq!(req.timeseries[0].samples[0].value, 42.0);
+}
+
 #[tokio::test]
 async fn retains_state_between_requests() {
     // This sink converts all incremental events to absolute, and

website/cue/reference/components/sinks/generated/prometheus_remote_write.cue

@@ -341,14 +341,8 @@ generated: components: sinks: prometheus_remote_write: configuration: {
 		}
 	}
 	request: {
-		description: """
-			Middleware settings for outbound requests.
-
-			Various settings can be configured, such as concurrency and rate limits, timeouts, and retry behavior.
-
-			Note that the retry backoff policy follows the Fibonacci sequence.
-			"""
-		required: false
+		description: "Outbound HTTP request settings."
+		required:    false
 		type: object: options: {
 			adaptive_concurrency: {
 				description: """
@@ -447,6 +441,23 @@ generated: components: sinks: prometheus_remote_write: configuration: {
 					uint: {}
 				}
 			}
+			headers: {
+				description: "Additional HTTP headers to add to every HTTP request."
+				required:    false
+				type: object: {
+					examples: [{
+						Accept:               "text/plain"
+						"X-Event-Level":      "{{level}}"
+						"X-Event-Timestamp":  "{{timestamp}}"
+						"X-My-Custom-Header": "A-Value"
+					}]
+					options: "*": {
+						description: "An HTTP request header and its value. Both header names and values support templating with event data."
+						required:    true
+						type: string: {}
+					}
+				}
+			}
 			rate_limit_duration_secs: {
 				description: "The time window used for the `rate_limit_num` option."
 				required:    false

website/cue/reference/components/sinks/prometheus_remote_write.cue

@@ -34,7 +34,7 @@ components: sinks: prometheus_remote_write: {
 				retry_initial_backoff_secs: 1
 				retry_max_duration_secs:    10
 				timeout_secs:               60
-				headers:                    false
+				headers:                    true
 			}
 			tls: {
 				enabled:                true