enhancement(sinks): validate sink endpoints as absolute http(s) URLs

changelog.d/sink_endpoint_absolute_urls.enhancement.md

@@ -0,0 +1,3 @@
+Sink `endpoint` options now require an absolute `http://` or `https://` URL that includes a host. Previously, partial or empty endpoints (for example `endpoint: ""` or `endpoint: "localhost:8080"` without a scheme) were accepted at configuration load and only failed when the sink attempted to send data, or were silently completed with a default scheme and host. Such configurations now fail validation, including with `vector validate --no-environment`, with a clear error. This affects the `appsignal`, `azure_logs_ingestion`, `azure_monitor_logs`, `datadog_metrics`, `datadog_traces`, `gcp_cloud_storage`, `gcp_pubsub`, `gcp_stackdriver_logs`, `gcp_stackdriver_metrics`, `honeycomb`, `humio`, `influxdb`, `prometheus_remote_write`, `sematext`, and `splunk_hec` sinks.
+
+authors: thomasqueirozb

src/sinks/appsignal/config.rs

@@ -1,5 +1,6 @@
+use derivative::Derivative;
 use futures::FutureExt;
-use http::{Request, Uri, header::AUTHORIZATION};
+use http::{Request, header::AUTHORIZATION};
 use hyper::Body;
 use tower::ServiceBuilder;
 use vector_lib::{
@@ -17,24 +18,27 @@ use crate::{
     codecs::Transformer,
     http::HttpClient,
     sinks::{
-        BuildError, Healthcheck, HealthcheckError, VectorSink,
+        Healthcheck, HealthcheckError, VectorSink,
         prelude::{SinkConfig, SinkContext},
         util::{
-            BatchConfig, Compression, ServiceBuilderExt, SinkBatchSettings, TowerRequestConfig,
+            BatchConfig, Compression, HttpEndpoint, ServiceBuilderExt, SinkBatchSettings,
+            TowerRequestConfig,
             http::{HttpStatusRetryLogic, RetryStrategy},
         },
     },
 };
 
 /// Configuration for the `appsignal` sink.
 #[configurable_component(sink("appsignal", "Deliver log and metric event data to AppSignal."))]
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Debug, Derivative)]
+#[derivative(Default)]
 pub(super) struct AppsignalConfig {
     /// The URI for the AppSignal API to send data to.
     #[configurable(validation(format = "uri"))]
     #[configurable(metadata(docs::examples = "https://appsignal-endpoint.net"))]
+    #[derivative(Default(value = "default_endpoint()"))]
     #[serde(default = "default_endpoint")]
-    pub(super) endpoint: String,
+    pub(super) endpoint: HttpEndpoint,
 
     /// A valid app-level AppSignal Push API key.
     #[configurable(metadata(docs::examples = "00000000-0000-0000-0000-000000000000"))]
@@ -73,8 +77,8 @@ pub(super) struct AppsignalConfig {
     retry_strategy: RetryStrategy,
 }
 
-pub(super) fn default_endpoint() -> String {
-    "https://appsignal-endpoint.net".to_string()
+pub(super) fn default_endpoint() -> HttpEndpoint {
+    HttpEndpoint::parse("https://appsignal-endpoint.net").unwrap()
 }
 
 #[derive(Clone, Copy, Debug, Default)]
@@ -151,8 +155,13 @@ impl SinkConfig for AppsignalConfig {
     }
 }
 
-async fn healthcheck(uri: Uri, push_api_key: String, client: HttpClient) -> crate::Result<()> {
-    let request = Request::get(uri).header(AUTHORIZATION, format!("Bearer {push_api_key}"));
+async fn healthcheck(
+    uri: HttpEndpoint,
+    push_api_key: String,
+    client: HttpClient,
+) -> crate::Result<()> {
+    let request =
+        Request::get(uri.as_uri()).header(AUTHORIZATION, format!("Bearer {push_api_key}"));
     let response = client.send(request.body(Body::empty()).unwrap()).await?;
 
     match response.status() {
@@ -161,21 +170,13 @@ async fn healthcheck(uri: Uri, push_api_key: String, client: HttpClient) -> crat
     }
 }
 
-pub fn endpoint_uri(endpoint: &str, path: &str) -> crate::Result<Uri> {
-    let uri = if endpoint.ends_with('/') {
-        format!("{endpoint}{path}")
-    } else {
-        format!("{endpoint}/{path}")
-    };
-    match uri.parse::<Uri>() {
-        Ok(u) => Ok(u),
-        Err(e) => Err(Box::new(BuildError::UriParseError { source: e })),
-    }
+pub fn endpoint_uri(endpoint: &HttpEndpoint, path: &str) -> crate::Result<HttpEndpoint> {
+    Ok(endpoint.append_path(path)?)
 }
 
 #[cfg(test)]
 mod test {
-    use super::{AppsignalConfig, endpoint_uri};
+    use super::{AppsignalConfig, HttpEndpoint, endpoint_uri};
 
     #[test]
     fn generate_config() {
@@ -184,7 +185,10 @@ mod test {
 
     #[test]
     fn endpoint_uri_with_path() {
-        let uri = endpoint_uri("https://appsignal-endpoint.net", "vector/events");
+        let uri = endpoint_uri(
+            &HttpEndpoint::parse("https://appsignal-endpoint.net").unwrap(),
+            "vector/events",
+        );
         assert_eq!(
             uri.expect("Not a valid URI").to_string(),
             "https://appsignal-endpoint.net/vector/events"
@@ -193,7 +197,10 @@ mod test {
 
     #[test]
     fn endpoint_uri_with_trailing_slash() {
-        let uri = endpoint_uri("https://appsignal-endpoint.net/", "vector/events");
+        let uri = endpoint_uri(
+            &HttpEndpoint::parse("https://appsignal-endpoint.net/").unwrap(),
+            "vector/events",
+        );
         assert_eq!(
             uri.expect("Not a valid URI").to_string(),
             "https://appsignal-endpoint.net/vector/events"

src/sinks/appsignal/integration_tests.rs

@@ -11,7 +11,10 @@ use crate::{
     config::SinkConfig,
     sinks::{
         appsignal::config::AppsignalConfig,
-        util::test::{build_test_server_status, load_sink},
+        util::{
+            HttpEndpoint,
+            test::{build_test_server_status, load_sink},
+        },
     },
     test_util::{
         addr::next_addr,
@@ -32,7 +35,7 @@ async fn start_test(events: Vec<Event>) -> (Vec<Event>, Receiver<(http::request:
     let (mut config, cx) = load_sink::<AppsignalConfig>(config.as_str()).unwrap();
     let (_guard, addr) = next_addr();
     // Set the endpoint to a local server so we can fetch the sent events later
-    config.endpoint = format!("http://{addr}");
+    config.endpoint = HttpEndpoint::parse(&format!("http://{addr}")).unwrap();
 
     let (sink, _) = config.build(cx).await.unwrap();
 

src/sinks/appsignal/service.rs

@@ -5,7 +5,7 @@ use futures::{
     future,
     future::{BoxFuture, Ready},
 };
-use http::{Request, StatusCode, Uri, header::AUTHORIZATION};
+use http::{Request, StatusCode, header::AUTHORIZATION};
 use hyper::Body;
 use tower::{Service, ServiceExt};
 use vector_lib::{
@@ -18,7 +18,7 @@ use vector_lib::{
 use super::request_builder::AppsignalRequest;
 use crate::{
     http::HttpClient,
-    sinks::util::{Compression, http::HttpBatchService, sink::Response},
+    sinks::util::{Compression, HttpEndpoint, http::HttpBatchService, sink::Response},
 };
 
 #[derive(Clone)]
@@ -33,14 +33,14 @@ pub(super) struct AppsignalService {
 impl AppsignalService {
     pub fn new(
         http_client: HttpClient<Body>,
-        endpoint: Uri,
+        endpoint: HttpEndpoint,
         push_api_key: SensitiveString,
         compression: Compression,
     ) -> Self {
         let batch_service = HttpBatchService::new(http_client, move |req| {
             let req: AppsignalRequest = req;
 
-            let mut request = Request::post(&endpoint)
+            let mut request = Request::post(endpoint.as_uri())
                 .header("Content-Type", "application/json")
                 .header(AUTHORIZATION, format!("Bearer {}", push_api_key.inner()))
                 .header("Content-Length", req.payload.len());

src/sinks/appsignal/tests.rs

@@ -7,6 +7,7 @@ use vector_lib::{
 use super::config::AppsignalConfig;
 use crate::{
     config::{SinkConfig, SinkContext},
+    sinks::util::HttpEndpoint,
     test_util::{
         components::{HTTP_SINK_TAGS, run_and_assert_sink_compliance},
         http::{always_200_response, spawn_blackhole_http_server},
@@ -19,7 +20,7 @@ async fn component_spec_compliance() {
 
     let mut config: AppsignalConfig =
         serde_json::from_value(AppsignalConfig::generate_config()).expect("config should be valid");
-    config.endpoint = mock_endpoint.to_string();
+    config.endpoint = HttpEndpoint::parse(&mock_endpoint.to_string()).unwrap();
 
     let context = SinkContext::default();
     let (sink, _healthcheck) = config.build(context).await.unwrap();

src/sinks/azure_logs_ingestion/config.rs

@@ -11,7 +11,7 @@ use crate::{
         azure_common::config::AzureAuthentication,
         prelude::*,
         util::{
-            RealtimeSizeBasedDefaultBatchSettings, UriSerde,
+            HttpEndpoint, RealtimeSizeBasedDefaultBatchSettings,
             http::{HttpStatusRetryLogic, RetryStrategy},
         },
     },
@@ -47,7 +47,7 @@ pub struct AzureLogsIngestionConfig {
     #[configurable(metadata(
         docs::examples = "https://my-dce-5kyl.eastus-1.ingest.monitor.azure.com"
     ))]
-    pub endpoint: String,
+    pub endpoint: HttpEndpoint,
 
     /// The [Data collection rule immutable ID][dcr_immutable_id] for the Data collection endpoint.
     ///
@@ -114,7 +114,7 @@ pub struct AzureLogsIngestionConfig {
 impl Default for AzureLogsIngestionConfig {
     fn default() -> Self {
         Self {
-            endpoint: Default::default(),
+            endpoint: HttpEndpoint::parse("http://localhost:8080").unwrap(),
             dcr_immutable_id: Default::default(),
             stream_name: Default::default(),
             auth: Default::default(),
@@ -135,14 +135,14 @@ impl AzureLogsIngestionConfig {
     pub(super) async fn build_inner(
         &self,
         cx: SinkContext,
-        endpoint: UriSerde,
+        endpoint: HttpEndpoint,
         dcr_immutable_id: String,
         stream_name: String,
         credential: Arc<dyn TokenCredential>,
         token_scope: String,
         timestamp_field: String,
     ) -> crate::Result<(VectorSink, Healthcheck)> {
-        let endpoint = endpoint.with_default_parts().uri;
+        let endpoint = endpoint.into_uri();
         let protocol = get_http_scheme_from_uri(&endpoint).to_string();
 
         let batch_settings = self
@@ -191,7 +191,7 @@ impl_generate_config_from_default!(AzureLogsIngestionConfig);
 #[typetag::serde(name = "azure_logs_ingestion")]
 impl SinkConfig for AzureLogsIngestionConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
-        let endpoint: UriSerde = self.endpoint.parse()?;
+        let endpoint = self.endpoint.clone();
 
         let credential: Arc<dyn TokenCredential> = self.auth.credential().await?;
 

src/sinks/azure_logs_ingestion/tests.rs

@@ -14,7 +14,7 @@ use super::config::AzureLogsIngestionConfig;
 
 use crate::{
     event::LogEvent,
-    sinks::prelude::*,
+    sinks::{prelude::*, util::HttpEndpoint},
     test_util::{
         components::{SINK_TAGS, run_and_assert_sink_compliance},
         http::spawn_blackhole_http_server,
@@ -64,7 +64,7 @@ fn basic_config_with_client_credentials() {
         .expect("Config parsing failed");
 
     assert_eq!(
-        config.endpoint,
+        config.endpoint.to_string(),
         "https://my-dce-5kyl.eastus-1.ingest.monitor.azure.com"
     );
     assert_eq!(
@@ -103,7 +103,7 @@ fn basic_config_with_managed_identity() {
         .expect("Config parsing failed");
 
     assert_eq!(
-        config.endpoint,
+        config.endpoint.to_string(),
         "https://my-dce-5kyl.eastus-1.ingest.monitor.azure.com"
     );
     assert_eq!(
@@ -176,7 +176,7 @@ async fn correct_request() {
     let (sink, healthcheck) = config
         .build_inner(
             context,
-            mock_endpoint.try_into().unwrap(),
+            HttpEndpoint::new(mock_endpoint).unwrap(),
             config.dcr_immutable_id.clone(),
             config.stream_name.clone(),
             credential,
@@ -288,7 +288,7 @@ async fn mock_healthcheck_with_400_response() {
     let (_sink, healthcheck) = config
         .build_inner(
             context,
-            mock_endpoint.try_into().unwrap(),
+            HttpEndpoint::new(mock_endpoint).unwrap(),
             config.dcr_immutable_id.clone(),
             config.stream_name.clone(),
             credential,
@@ -355,7 +355,7 @@ async fn mock_healthcheck_with_403_response() {
     let (_sink, healthcheck) = config
         .build_inner(
             context,
-            mock_endpoint.try_into().unwrap(),
+            HttpEndpoint::new(mock_endpoint).unwrap(),
             config.dcr_immutable_id.clone(),
             config.stream_name.clone(),
             credential,

src/sinks/azure_monitor_logs/config.rs

@@ -17,7 +17,7 @@ use crate::{
     sinks::{
         prelude::*,
         util::{
-            RealtimeSizeBasedDefaultBatchSettings, UriSerde,
+            HttpEndpoint, RealtimeSizeBasedDefaultBatchSettings,
             http::{HttpStatusRetryLogic, RetryStrategy},
         },
     },
@@ -161,9 +161,9 @@ impl AzureMonitorLogsConfig {
     pub(super) async fn build_inner(
         &self,
         cx: SinkContext,
-        endpoint: UriSerde,
+        endpoint: HttpEndpoint,
     ) -> crate::Result<(VectorSink, Healthcheck)> {
-        let endpoint = endpoint.with_default_parts().uri;
+        let endpoint = endpoint.into_uri();
         let protocol = get_http_scheme_from_uri(&endpoint).to_string();
 
         let batch_settings = self
@@ -216,7 +216,7 @@ impl_generate_config_from_default!(AzureMonitorLogsConfig);
 #[typetag::serde(name = "azure_monitor_logs")]
 impl SinkConfig for AzureMonitorLogsConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
-        let endpoint = format!("https://{}.{}", self.customer_id, self.host).parse()?;
+        let endpoint = HttpEndpoint::parse(&format!("https://{}.{}", self.customer_id, self.host))?;
         self.build_inner(cx, endpoint).await
     }
 

src/sinks/azure_monitor_logs/tests.rs

@@ -12,7 +12,10 @@ use super::{
 };
 use crate::{
     event::LogEvent,
-    sinks::{prelude::*, util::encoding::Encoder},
+    sinks::{
+        prelude::*,
+        util::{HttpEndpoint, encoding::Encoder},
+    },
     test_util::{
         components::{SINK_TAGS, run_and_assert_sink_compliance},
         http::{always_200_response, spawn_blackhole_http_server},
@@ -37,7 +40,7 @@ async fn component_spec_compliance() {
 
     let context = SinkContext::default();
     let (sink, _healthcheck) = config
-        .build_inner(context, mock_endpoint.try_into().unwrap())
+        .build_inner(context, HttpEndpoint::new(mock_endpoint).unwrap())
         .await
         .unwrap();
 
@@ -184,7 +187,7 @@ async fn correct_request() {
 
     let context = SinkContext::default();
     let (sink, _healthcheck) = config
-        .build_inner(context, mock_endpoint.try_into().unwrap())
+        .build_inner(context, HttpEndpoint::new(mock_endpoint).unwrap())
         .await
         .unwrap();