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;
 
@@ -506,6 +520,7 @@ mod tests {
         path_key: &'a str,
         path: &'a str,
         method: &'a str,
+        response_code: StatusCode,
         strict_path: bool,
         status: EventStatus,
         acknowledgements: bool,
@@ -529,6 +544,7 @@ mod tests {
                 headers,
                 encoding: None,
                 query_parameters,
+                response_code: response_code,
                 tls: None,
                 auth: None,
                 strict_path,
@@ -639,6 +655,7 @@ mod tests {
                 "http_path",
                 "/",
                 "POST",
+                StatusCode::OK,
                 true,
                 EventStatus::Delivered,
                 true,
@@ -694,6 +711,7 @@ mod tests {
                 "http_path",
                 "/",
                 "POST",
+                StatusCode::OK,
                 true,
                 EventStatus::Delivered,
                 true,
@@ -737,6 +755,7 @@ mod tests {
                 "http_path",
                 "/",
                 "POST",
+                StatusCode::OK,
                 true,
                 EventStatus::Delivered,
                 true,
@@ -771,6 +790,7 @@ mod tests {
                 "http_path",
                 "/",
                 "POST",
+                StatusCode::OK,
                 true,
                 EventStatus::Delivered,
                 true,
@@ -821,6 +841,7 @@ mod tests {
                 "http_path",
                 "/",
                 "POST",
+                StatusCode::OK,
                 true,
                 EventStatus::Delivered,
                 true,
@@ -864,6 +885,7 @@ mod tests {
                 "http_path",
                 "/",
                 "POST",
+                StatusCode::OK,
                 true,
                 EventStatus::Delivered,
                 true,
@@ -913,6 +935,7 @@ mod tests {
                 "http_path",
                 "/",
                 "POST",
+                StatusCode::OK,
                 true,
                 EventStatus::Delivered,
                 true,
@@ -1003,6 +1026,7 @@ mod tests {
                 "http_path",
                 "/",
                 "POST",
+                StatusCode::OK,
                 true,
                 EventStatus::Delivered,
                 true,
@@ -1044,6 +1068,7 @@ mod tests {
                 "http_path",
                 "/",
                 "POST",
+                StatusCode::OK,
                 true,
                 EventStatus::Delivered,
                 true,
@@ -1094,6 +1119,7 @@ mod tests {
                 "http_path",
                 "/",
                 "POST",
+                StatusCode::OK,
                 true,
                 EventStatus::Delivered,
                 true,
@@ -1126,6 +1152,7 @@ mod tests {
                 "vector_http_path",
                 "/event/path",
                 "POST",
+                StatusCode::OK,
                 true,
                 EventStatus::Delivered,
                 true,
@@ -1170,6 +1197,7 @@ mod tests {
                 "vector_http_path",
                 "/event",
                 "POST",
+                StatusCode::OK,
                 false,
                 EventStatus::Delivered,
                 true,
@@ -1239,6 +1267,7 @@ mod tests {
             "vector_http_path",
             "/",
             "POST",
+            StatusCode::OK,
             true,
             EventStatus::Delivered,
             true,
@@ -1253,6 +1282,39 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn http_status_code() {
+        assert_source_compliance(&HTTP_PUSH_SOURCE_TAGS, async move {
+            let (rx, addr) = source(
+                vec![],
+                vec![],
+                "http_path",
+                "/",
+                "POST",
+                StatusCode::ACCEPTED,
+                true,
+                EventStatus::Delivered,
+                true,
+                None,
+                None,
+            )
+            .await;
+
+            spawn_collect_n(
+                async move {
+                    assert_eq!(
+                        StatusCode::ACCEPTED,
+                        send(addr, "{\"key1\":\"value1\"}").await
+                    );
+                },
+                rx,
+                1,
+            )
+            .await;
+        })
+        .await;
+    }
+
     #[tokio::test]
     async fn http_delivery_failure() {
         assert_source_compliance(&HTTP_PUSH_SOURCE_TAGS, async {
@@ -1262,6 +1324,7 @@ mod tests {
                 "http_path",
                 "/",
                 "POST",
+                StatusCode::OK,
                 true,
                 EventStatus::Rejected,
                 true,
@@ -1291,6 +1354,7 @@ mod tests {
                 "http_path",
                 "/",
                 "POST",
+                StatusCode::OK,
                 true,
                 EventStatus::Rejected,
                 false,
@@ -1322,6 +1386,7 @@ mod tests {
             "http_path",
             "/",
             "GET",
+            StatusCode::OK,
             true,
             EventStatus::Delivered,
             true,

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,