feat(http source): Add custom response header configuration

src/sources/http_server.rs

@@ -98,6 +98,11 @@ pub struct SimpleHttpConfig {
     #[configurable(metadata(docs::examples = "X-*"))]
     #[configurable(metadata(docs::examples = "*"))]
     headers: Vec<String>,
+
+    /// Custom response headers to be added to the HTTP response
+    #[serde(default)]    
+    #[configurable(metadata(docs::examples = "example_custom_response_headers()"))]
+    custom_response_headers: HashMap<String, String>,
 
     /// A list of URL query parameters to include in the log event.
     ///
@@ -170,6 +175,12 @@ pub struct SimpleHttpConfig {
     keepalive: KeepaliveConfig,
 }
 
+fn example_custom_response_headers() -> HashMap<String, String> {
+    HashMap::<_, _>::from_iter([
+        ("Access-Control-Allow-Origin", "my-cool-server"),
+    ])
+}
+
 impl SimpleHttpConfig {
     /// Builds the `schema::Definition` for this source using the provided `LogNamespace`.
     fn schema_definition(&self, log_namespace: LogNamespace) -> Definition {
@@ -265,6 +276,7 @@ impl Default for SimpleHttpConfig {
             address: "0.0.0.0:8080".parse().unwrap(),
             encoding: None,
             headers: Vec::new(),
+            custom_response_headers: HashMap::new(),
             query_parameters: Vec::new(),
             tls: None,
             auth: None,
@@ -355,6 +367,7 @@ impl SourceConfig for SimpleHttpConfig {
 
         let source = SimpleHttpSource {
             headers: build_param_matcher(&remove_duplicates(self.headers.clone(), "headers"))?,
+            custom_response_headers: self.custom_response_headers.clone(),
             query_parameters: remove_duplicates(self.query_parameters.clone(), "query_parameters"),
             path_key: self.path_key.clone(),
             host_key: self.host_key.clone(),
@@ -403,6 +416,7 @@ impl SourceConfig for SimpleHttpConfig {
 #[derive(Clone)]
 struct SimpleHttpSource {
     headers: Vec<HttpConfigParamKind>,
+    custom_response_headers: HashMap<String, String>,
     query_parameters: Vec<String>,
     path_key: OptionalValuePath,
     host_key: OptionalValuePath,
@@ -544,6 +558,20 @@ impl HttpSource for SimpleHttpSource {
     fn enable_source_ip(&self) -> bool {
         self.host_key.path.is_some()
     }
+
+    /// Enriches the warp::reply::Reply with custom headers
+    ///
+    /// This method adds the custom headers specified in the configuration
+    /// to the HTTP response.
+    fn enrich_reply<T : warp::Reply>(
+        &self,
+        mut reply: T
+    ) -> warp::http::response::Builder {
+        for (key, value) in &self.custom_response_headers {
+            reply = warp::reply::with_headers(reply, key, value);
+        }
+        reply
+    }
 }
 
 #[cfg(test)]

src/sources/util/http/prelude.rs

@@ -70,6 +70,11 @@ pub trait HttpSource: Clone + Send + Sync + 'static {
         decode(encoding_header, body)
     }
 
+    // This function can be defined to enrich warp::Replies.
+    fn enrich_reply<T : warp::Reply>(&self, reply: T) -> T {
+        reply
+    }
+
     #[allow(clippy::too_many_arguments)]
     fn run(
         self,
@@ -186,7 +191,7 @@ pub trait HttpSource: Clone + Send + Sync + 'static {
                     });
                     Err(r)
                 }
-            });
+            }).map(|reply| &self.enrich_reply(reply));
 
             let span = Span::current();
             let make_svc = make_service_fn(move |conn: &MaybeTlsIncomingStream<TcpStream>| {