enhancement(prometheus scrape): support request headers

changelog.d/19656_prometheus_scrape_headers.enhancement.md

@@ -0,0 +1,3 @@
+The `prometheus_scrape` source now supports configuring HTTP request headers.
+
+authors: arfa79

src/sources/http_client/client.rs

@@ -37,8 +37,8 @@ use crate::{
         http::HttpMethod,
         http_client,
         http_client::{
-            GenericHttpClientInputs, HttpClientBuilder, build_url, call, default_interval,
-            default_timeout, warn_if_interval_too_low,
+            GenericHttpClientInputs, HttpClientBuilder, build_headers, build_url, call,
+            default_interval, default_timeout, warn_if_interval_too_low,
         },
     },
     tls::{TlsConfig, TlsSettings},
@@ -360,6 +360,7 @@ impl SourceConfig for HttpClientConfig {
         let decoder = self.get_decoding_config(Some(log_namespace)).build()?;
 
         let content_type = self.decoding.content_type(&self.framing).to_string();
+        let headers = build_headers(&self.headers)?;
 
         // Create context with the config for dynamic query parameter and body evaluation
         let context = HttpClientContext {
@@ -375,7 +376,7 @@ impl SourceConfig for HttpClientConfig {
             urls,
             interval: self.interval,
             timeout: self.timeout,
-            headers: self.headers.clone(),
+            headers,
             content_type,
             auth: self.auth.clone(),
             tls,

src/sources/prometheus/scrape.rs

@@ -18,8 +18,8 @@ use crate::{
         util::{
             http::HttpMethod,
             http_client::{
-                GenericHttpClientInputs, HttpClientBuilder, HttpClientContext, build_url, call,
-                default_interval, default_timeout, warn_if_interval_too_low,
+                GenericHttpClientInputs, HttpClientBuilder, HttpClientContext, build_headers,
+                build_url, call, default_interval, default_timeout, warn_if_interval_too_low,
             },
         },
     },
@@ -92,6 +92,16 @@ pub struct PrometheusScrapeConfig {
     #[configurable(metadata(docs::examples = "query_example()"))]
     query: QueryParameters,
 
+    /// Headers to apply to the scrape requests.
+    ///
+    /// One or more values for the same header can be provided.
+    #[serde(default)]
+    #[configurable(metadata(
+        docs::additional_props_description = "An HTTP request header and its value(s)."
+    ))]
+    #[configurable(metadata(docs::examples = "headers_example()"))]
+    headers: HashMap<String, Vec<String>>,
+
     #[configurable(derived)]
     tls: Option<TlsConfig>,
 
@@ -108,6 +118,12 @@ fn query_example() -> serde_json::Value {
     })
 }
 
+fn headers_example() -> serde_json::Value {
+    serde_json::json!({
+        "X-My-Header": ["value1", "value2"]
+    })
+}
+
 impl GenerateConfig for PrometheusScrapeConfig {
     fn generate_config() -> serde_json::Value {
         serde_json::to_value(Self {
@@ -118,6 +134,7 @@ impl GenerateConfig for PrometheusScrapeConfig {
             endpoint_tag: Some("endpoint".to_string()),
             honor_labels: false,
             query: HashMap::new(),
+            headers: HashMap::new(),
             tls: None,
             auth: None,
         })
@@ -136,6 +153,7 @@ impl SourceConfig for PrometheusScrapeConfig {
             .map(|r| r.map(|uri| build_url(&uri, &self.query)))
             .collect::<std::result::Result<Vec<Uri>, sources::BuildError>>()?;
         let tls = TlsSettings::from_options(self.tls.as_ref())?;
+        let headers = build_headers(&self.headers)?;
 
         let builder = PrometheusScrapeBuilder {
             honor_labels: self.honor_labels,
@@ -149,7 +167,7 @@ impl SourceConfig for PrometheusScrapeConfig {
             urls,
             interval: self.interval,
             timeout: self.timeout,
-            headers: HashMap::new(),
+            headers,
             content_type: "text/plain".to_string(),
             auth: self.auth.clone(),
             tls,
@@ -344,11 +362,14 @@ mod test {
     async fn test_prometheus_sets_headers() {
         let (_guard, in_addr) = next_addr();
 
-        let dummy_endpoint = warp::path!("metrics").and(warp::header::exact("Accept", "text/plain")).map(|| {
-            r#"
+        let dummy_endpoint = warp::path!("metrics")
+            .and(warp::header::exact("Accept", "application/openmetrics-text"))
+            .and(warp::header::exact("X-My-Header", "custom-value"))
+            .map(|| {
+                r#"
                     promhttp_metric_handler_requests_total{endpoint="http://example.com", instance="localhost:9999", code="200"} 100 1612411516789
                     "#
-        });
+            });
 
         tokio::spawn(warp::serve(dummy_endpoint).run(in_addr));
         wait_for_tcp(in_addr).await;
@@ -361,6 +382,13 @@ mod test {
             endpoint_tag: Some("endpoint".to_string()),
             honor_labels: true,
             query: HashMap::new(),
+            headers: HashMap::from([
+                (
+                    "Accept".to_string(),
+                    vec!["application/openmetrics-text".to_string()],
+                ),
+                ("X-My-Header".to_string(), vec!["custom-value".to_string()]),
+            ]),
             auth: None,
             tls: None,
         };
@@ -395,6 +423,7 @@ mod test {
             endpoint_tag: Some("endpoint".to_string()),
             honor_labels: true,
             query: HashMap::new(),
+            headers: HashMap::new(),
             auth: None,
             tls: None,
         };
@@ -447,6 +476,7 @@ mod test {
             endpoint_tag: Some("endpoint".to_string()),
             honor_labels: false,
             query: HashMap::new(),
+            headers: HashMap::new(),
             auth: None,
             tls: None,
         };
@@ -513,6 +543,7 @@ mod test {
             endpoint_tag: Some("endpoint".to_string()),
             honor_labels: true,
             query: HashMap::new(),
+            headers: HashMap::new(),
             auth: None,
             tls: None,
         };
@@ -582,6 +613,7 @@ mod test {
                     ]),
                 ),
             ]),
+            headers: HashMap::new(),
             auth: None,
             tls: None,
         };
@@ -682,6 +714,7 @@ mod test {
                 endpoint_tag: None,
                 honor_labels: false,
                 query: HashMap::new(),
+                headers: HashMap::new(),
                 interval: Duration::from_secs(1),
                 timeout: default_timeout(),
                 tls: None,
@@ -776,6 +809,7 @@ mod integration_tests {
             endpoint_tag: Some("endpoint".to_string()),
             honor_labels: false,
             query: HashMap::new(),
+            headers: HashMap::new(),
             auth: None,
             tls: None,
         };

src/sources/util/http_client.rs

@@ -15,8 +15,13 @@ use std::{collections::HashMap, future::ready, time::Duration};
 
 use bytes::Bytes;
 use futures_util::{FutureExt, StreamExt, TryFutureExt, stream};
-use http::{Uri, response::Parts};
+use http::{
+    HeaderMap, Uri,
+    header::{HeaderName, HeaderValue},
+    response::Parts,
+};
 use hyper::{Body, Request};
+use snafu::{ResultExt, Snafu};
 use tokio_stream::wrappers::IntervalStream;
 use vector_lib::{
     EstimatedJsonEncodedSizeOf, config::proxy::ProxyConfig, event::Event, json_size::JsonSize,
@@ -43,7 +48,7 @@ pub(crate) struct GenericHttpClientInputs {
     /// Timeout for the HTTP request.
     pub timeout: Duration,
     /// Map of Header+Value to apply to HTTP request.
-    pub headers: HashMap<String, Vec<String>>,
+    pub headers: HeaderMap<HeaderValue>,
     /// Content type of the HTTP request, determined by the source.
     pub content_type: String,
     pub auth: Option<Auth>,
@@ -52,6 +57,44 @@ pub(crate) struct GenericHttpClientInputs {
     pub shutdown: ShutdownSignal,
 }
 
+#[derive(Debug, Snafu)]
+pub(crate) enum HeaderError {
+    #[snafu(display("Invalid HTTP header name {name:?}: {source}"))]
+    InvalidName {
+        name: String,
+        source: http::header::InvalidHeaderName,
+    },
+    #[snafu(display("Invalid value for HTTP header {name:?}: {source}"))]
+    InvalidValue {
+        name: String,
+        source: http::header::InvalidHeaderValue,
+    },
+}
+
+pub(crate) fn build_headers(
+    headers: &HashMap<String, Vec<String>>,
+) -> Result<HeaderMap<HeaderValue>, HeaderError> {
+    let mut parsed = HeaderMap::new();
+
+    for (name, values) in headers {
+        let header_name =
+            HeaderName::from_bytes(name.as_bytes()).with_context(|_| InvalidNameSnafu {
+                name: name.to_owned(),
+            })?;
+
+        for value in values {
+            let mut header_value =
+                HeaderValue::from_bytes(value.as_bytes()).with_context(|_| InvalidValueSnafu {
+                    name: name.to_owned(),
+                })?;
+            header_value.set_sensitive(true);
+            parsed.append(header_name.clone(), header_value);
+        }
+    }
+
+    Ok(parsed)
+}
+
 /// The default interval to call the HTTP endpoint if none is configured.
 pub(crate) const fn default_interval() -> Duration {
     Duration::from_secs(15)
@@ -185,10 +228,8 @@ pub(crate) async fn call<
             };
 
             // add user specified headers
-            for (header, values) in &inputs.headers {
-                for value in values {
-                    builder = builder.header(header, value);
-                }
+            for (header, value) in &inputs.headers {
+                builder = builder.header(header, value);
             }
 
             // set ACCEPT header if not user specified
@@ -307,3 +348,46 @@ pub(crate) async fn call<
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn headers_are_normalized_and_sensitive() {
+        let headers = build_headers(&HashMap::from([(
+            "X-Private-Token".to_string(),
+            vec!["secret".to_string(), "another-secret".to_string()],
+        )]))
+        .unwrap();
+
+        let values = headers
+            .get_all("x-private-token")
+            .iter()
+            .collect::<Vec<_>>();
+        assert_eq!(values.len(), 2);
+        assert!(values.iter().all(|value| value.is_sensitive()));
+    }
+
+    #[test]
+    fn invalid_header_name_is_rejected() {
+        let error = build_headers(&HashMap::from([(
+            "invalid header".to_string(),
+            vec!["value".to_string()],
+        )]))
+        .unwrap_err();
+
+        assert!(error.to_string().contains("Invalid HTTP header name"));
+    }
+
+    #[test]
+    fn invalid_header_value_is_rejected() {
+        let error = build_headers(&HashMap::from([(
+            "x-header".to_string(),
+            vec!["invalid\nvalue".to_string()],
+        )]))
+        .unwrap_err();
+
+        assert!(error.to_string().contains("Invalid value for HTTP header"));
+    }
+}

website/cue/reference/components/sources/generated/prometheus_scrape.cue

@@ -211,6 +211,24 @@ generated: components: sources: prometheus_scrape: configuration: {
 		required: false
 		type: bool: default: false
 	}
+	headers: {
+		description: """
+			Headers to apply to the scrape requests.
+
+			One or more values for the same header can be provided.
+			"""
+		required: false
+		type: object: {
+			examples: [{
+				"X-My-Header": ["value1", "value2"]
+			}]
+			options: "*": {
+				description: "An HTTP request header and its value(s)."
+				required:    true
+				type: array: items: type: string: {}
+			}
+		}
+	}
 	instance_tag: {
 		description: """
 			The tag name added to each event representing the scraped instance's `host:port`.

website/generated/example-configs/sources/prometheus_scrape/advanced.yaml

@@ -3,6 +3,10 @@ sources:
     type: prometheus_scrape
     endpoints:
       - http://localhost:9090/metrics
+    headers:
+      X-My-Header:
+        - value1
+        - value2
     honor_labels: false
     query:
       match[]: