enhancement(http_server source): Configurable http response code

Cargo.toml

@@ -271,6 +271,7 @@ hashbrown = { version = "0.14.0", default-features = false, optional = true, fea
 headers = { version = "0.3.8", default-features = false }
 hostname = { version = "0.3.1", default-features = false }
 http = { version = "0.2.9", default-features = false }
+http-serde = "1.1.2"
 http-body = { version = "0.4.5", default-features = false }
 hyper = { version = "0.14.27", default-features = false, features = ["client", "runtime", "http1", "http2", "server", "stream"] }
 hyper-openssl = { version = "0.9.2", default-features = false }

lib/vector-config/Cargo.toml

@@ -26,6 +26,7 @@ snafu = { version = "0.7.5", default-features = false }
 toml = { version = "0.7.6", default-features = false }
 tracing = { version = "0.1.34", default-features = false }
 url = { version = "2.4.0", default-features = false, features = ["serde"] }
+http = { version = "0.2.9", default-features = false }
 vrl.workspace = true
 vector-config-common = { path = "../vector-config-common" }
 vector-config-macros = { path = "../vector-config-macros" }

lib/vector-config/src/http.rs

@@ -0,0 +1,35 @@
+use http::StatusCode;
+use serde_json::Value;
+use std::cell::RefCell;
+
+use crate::{
+    schema::{generate_number_schema, SchemaGenerator, SchemaObject},
+    Configurable, GenerateError, Metadata, ToValue,
+};
+
+impl ToValue for StatusCode {
+    fn to_value(&self) -> Value {
+        serde_json::to_value(self.as_u16()).expect("Could not convert HTTP status code to JSON")
+    }
+}
+
+impl Configurable for StatusCode {
+    fn referenceable_name() -> Option<&'static str> {
+        Some("http::StatusCode")
+    }
+
+    fn is_optional() -> bool {
+        true
+    }
+
+    fn metadata() -> Metadata {
+        let mut metadata = Metadata::default();
+        metadata.set_description("HTTP response status code");
+        metadata.set_default_value(StatusCode::OK);
+        metadata
+    }
+
+    fn generate_schema(_: &RefCell<SchemaGenerator>) -> Result<SchemaObject, GenerateError> {
+        Ok(generate_number_schema::<u16>())
+    }
+}

lib/vector-config/src/lib.rs

@@ -117,6 +117,7 @@ pub use self::configurable::{Configurable, ConfigurableRef, ToValue};
 mod errors;
 pub use self::errors::{BoundDirection, GenerateError};
 mod external;
+mod http;
 mod metadata;
 pub use self::metadata::Metadata;
 mod named;

src/sources/heroku_logs.rs

@@ -175,6 +175,7 @@ impl SourceConfig for LogplexConfig {
             self.address,
             "events",
             HttpMethod::Post,
+            StatusCode::OK,
             true,
             &self.tls,
             &self.auth,

src/sources/http_server.rs

@@ -9,6 +9,7 @@ use codecs::{
 };
 
 use http::{StatusCode, Uri};
+use http_serde;
 use lookup::{lookup_v2::OptionalValuePath, owned_value_path, path};
 use tokio_util::codec::Decoder as _;
 use vector_config::configurable_component;
@@ -129,6 +130,13 @@ pub struct SimpleHttpConfig {
     #[serde(default = "default_http_method")]
     method: HttpMethod,
 
+    /// Specifies the HTTP response status code that will be returned on successful requests.
+    #[configurable(metadata(docs::examples = "202"))]
+    #[configurable(metadata(docs::numeric_type = "u16"))]
+    #[serde(with = "http_serde::status_code")]
+    #[serde(default = "default_http_response_code")]
+    response_code: StatusCode,
+
     #[configurable(derived)]
     tls: Option<TlsEnableableConfig>,
 
@@ -242,6 +250,7 @@ impl Default for SimpleHttpConfig {
             path: default_path(),
             path_key: default_path_key(),
             method: default_http_method(),
+            response_code: default_http_response_code(),
             strict_path: true,
             framing: None,
             decoding: Some(default_decoding()),
@@ -289,6 +298,10 @@ fn default_path_key() -> OptionalValuePath {
     OptionalValuePath::from(owned_value_path!("path"))
 }
 
+fn default_http_response_code() -> StatusCode {
+    StatusCode::OK
+}
+
 /// Removes duplicates from the list, and logs a `warn!()` for each duplicate removed.
 fn remove_duplicates(mut list: Vec<String>, list_name: &str) -> Vec<String> {
     list.sort();
@@ -328,6 +341,7 @@ impl SourceConfig for SimpleHttpConfig {
             self.address,
             self.path.as_str(),
             self.method,
+            self.response_code,
             self.strict_path,
             &self.tls,
             &self.auth,
@@ -478,7 +492,7 @@ mod tests {
         Compression,
     };
     use futures::Stream;
-    use http::{HeaderMap, Method};
+    use http::{HeaderMap, Method, StatusCode};
     use lookup::lookup_v2::OptionalValuePath;
     use similar_asserts::assert_eq;
 
@@ -529,6 +543,7 @@ mod tests {
                 headers,
                 encoding: None,
                 query_parameters,
+                response_code: StatusCode::OK,
                 tls: None,
                 auth: None,
                 strict_path,

src/sources/prometheus/remote_write.rs

@@ -80,6 +80,7 @@ impl SourceConfig for PrometheusRemoteWriteConfig {
             self.address,
             "",
             HttpMethod::Post,
+            StatusCode::OK,
             true,
             &self.tls,
             &self.auth,

src/sources/util/http/prelude.rs

@@ -63,6 +63,7 @@ pub trait HttpSource: Clone + Send + Sync + 'static {
         address: SocketAddr,
         path: &str,
         method: HttpMethod,
+        response_code: StatusCode,
         strict_path: bool,
         tls: &Option<TlsEnableableConfig>,
         auth: &Option<HttpSourceAuthConfig>,
@@ -152,7 +153,7 @@ pub trait HttpSource: Clone + Send + Sync + 'static {
                                 events
                             });
 
-                        handle_request(events, acknowledgements, cx.out.clone())
+                        handle_request(events, acknowledgements, response_code, cx.out.clone())
                     },
                 )
                 .with(warp::trace(move |_info| span.clone()));
@@ -205,6 +206,7 @@ impl warp::reject::Reject for RejectShuttingDown {}
 async fn handle_request(
     events: Result<Vec<Event>, ErrorMessage>,
     acknowledgements: bool,
+    response_code: StatusCode,
     mut out: SourceSender,
 ) -> Result<impl warp::Reply, Rejection> {
     match events {
@@ -219,7 +221,7 @@ async fn handle_request(
                     emit!(StreamClosedError { count });
                     warp::reject::custom(RejectShuttingDown)
                 })
-                .and_then(|_| handle_batch_status(receiver))
+                .and_then(|_| handle_batch_status(response_code, receiver))
                 .await
         }
         Err(error) => {
@@ -230,12 +232,13 @@ async fn handle_request(
 }
 
 async fn handle_batch_status(
+    success_response_code: StatusCode,
     receiver: Option<BatchStatusReceiver>,
 ) -> Result<impl warp::Reply, Rejection> {
     match receiver {
-        None => Ok(warp::reply()),
+        None => Ok(success_response_code),
         Some(receiver) => match receiver.await {
-            BatchStatus::Delivered => Ok(warp::reply()),
+            BatchStatus::Delivered => Ok(success_response_code),
             BatchStatus::Errored => Err(warp::reject::custom(ErrorMessage::new(
                 StatusCode::INTERNAL_SERVER_ERROR,
                 "Error delivering contents to sink".into(),

website/cue/reference/components/sources/base/http.cue

@@ -333,6 +333,16 @@ base: components: sources: http: configuration: {
 			items: type: string: examples: ["application", "source"]
 		}
 	}
+	response_code: {
+		description: "Specifies the HTTP response status code on request success."
+		required:    false
+		type: u16: {
+			default: 200
+			examples: [
+				"202",
+			]
+		}
+	}
 	strict_path: {
 		description: """
 			Whether or not to treat the configured `path` as an absolute path.

website/cue/reference/components/sources/base/http_server.cue

@@ -333,6 +333,16 @@ base: components: sources: http_server: configuration: {
 			items: type: string: examples: ["application", "source"]
 		}
 	}
+	response_code: {
+		description: "Specifies the HTTP response status code on request success."
+		required:    false
+		type: u16: {
+			default: 200
+			examples: [
+				"202",
+			]
+		}
+	}
 	strict_path: {
 		description: """
 			Whether or not to treat the configured `path` as an absolute path.