From eb68a4af3d2fee5846c4187783424e7cf031a42d Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 13:51:55 -0400 Subject: [PATCH 01/16] Add HttpEndpoint type and migrate HTTP sinks to validated endpoint URIs --- src/sinks/appsignal/config.rs | 23 +-- src/sinks/appsignal/service.rs | 8 +- src/sinks/azure_monitor_logs/config.rs | 8 +- src/sinks/azure_monitor_logs/tests.rs | 6 +- src/sinks/datadog/metrics/config.rs | 36 ++-- src/sinks/datadog/metrics/request_builder.rs | 3 +- src/sinks/datadog/traces/config.rs | 30 ++- src/sinks/datadog/traces/request_builder.rs | 4 +- src/sinks/gcp/cloud_storage.rs | 13 +- src/sinks/gcp/pubsub.rs | 20 +- src/sinks/gcp/stackdriver/logs/config.rs | 4 +- src/sinks/gcs_common/config.rs | 8 +- src/sinks/gcs_common/service.rs | 51 ++--- src/sinks/honeycomb/config.rs | 21 +- src/sinks/honeycomb/service.rs | 7 +- src/sinks/influxdb/mod.rs | 16 +- src/sinks/prometheus/remote_write/config.rs | 7 +- src/sinks/prometheus/remote_write/service.rs | 5 +- src/sinks/sematext/metrics.rs | 4 +- src/sinks/util/mod.rs | 2 +- src/sinks/util/uri.rs | 197 +++++++++++++++++++ 21 files changed, 315 insertions(+), 158 deletions(-) diff --git a/src/sinks/appsignal/config.rs b/src/sinks/appsignal/config.rs index fea94af05b132..357da68f071b9 100644 --- a/src/sinks/appsignal/config.rs +++ b/src/sinks/appsignal/config.rs @@ -1,5 +1,5 @@ use futures::FutureExt; -use http::{Request, Uri, header::AUTHORIZATION}; +use http::{Request, header::AUTHORIZATION}; use hyper::Body; use tower::ServiceBuilder; use vector_lib::{ @@ -17,10 +17,11 @@ 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}, }, }, @@ -151,8 +152,8 @@ 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,16 +162,8 @@ async fn healthcheck(uri: Uri, push_api_key: String, client: HttpClient) -> crat } } -pub fn endpoint_uri(endpoint: &str, path: &str) -> crate::Result { - let uri = if endpoint.ends_with('/') { - format!("{endpoint}{path}") - } else { - format!("{endpoint}/{path}") - }; - match uri.parse::() { - Ok(u) => Ok(u), - Err(e) => Err(Box::new(BuildError::UriParseError { source: e })), - } +pub fn endpoint_uri(endpoint: &str, path: &str) -> crate::Result { + Ok(HttpEndpoint::parse(endpoint)?.append_path(path)?) } #[cfg(test)] diff --git a/src/sinks/appsignal/service.rs b/src/sinks/appsignal/service.rs index 622776d88c047..eba97fc98a622 100644 --- a/src/sinks/appsignal/service.rs +++ b/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, - 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()); diff --git a/src/sinks/azure_monitor_logs/config.rs b/src/sinks/azure_monitor_logs/config.rs index 6d68aed7efd75..5653650640a30 100644 --- a/src/sinks/azure_monitor_logs/config.rs +++ b/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 } diff --git a/src/sinks/azure_monitor_logs/tests.rs b/src/sinks/azure_monitor_logs/tests.rs index 077087e16119e..69867e9574330 100644 --- a/src/sinks/azure_monitor_logs/tests.rs +++ b/src/sinks/azure_monitor_logs/tests.rs @@ -12,7 +12,7 @@ use super::{ }; use crate::{ event::LogEvent, - sinks::{prelude::*, util::encoding::Encoder}, + sinks::{prelude::*, util::{encoding::Encoder, HttpEndpoint}}, test_util::{ components::{SINK_TAGS, run_and_assert_sink_compliance}, http::{always_200_response, spawn_blackhole_http_server}, @@ -37,7 +37,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 +184,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(); diff --git a/src/sinks/datadog/metrics/config.rs b/src/sinks/datadog/metrics/config.rs index a5fdccede6a14..9ac813d097fb3 100644 --- a/src/sinks/datadog/metrics/config.rs +++ b/src/sinks/datadog/metrics/config.rs @@ -1,5 +1,3 @@ -use http::Uri; -use snafu::ResultExt; use tower::ServiceBuilder; use vector_lib::{ config::proxy::ProxyConfig, configurable::configurable_component, stream::BatcherSettings, @@ -15,9 +13,12 @@ use crate::{ config::{AcknowledgementsConfig, Input, SinkConfig, SinkContext}, http::HttpClient, sinks::{ - Healthcheck, UriParseSnafu, VectorSink, + Healthcheck, VectorSink, datadog::{DatadogCommonConfig, LocalDatadogCommonConfig}, - util::{ServiceBuilderExt, SinkBatchSettings, TowerRequestConfig, batch::BatchConfig}, + util::{ + HttpEndpoint, ServiceBuilderExt, SinkBatchSettings, TowerRequestConfig, + batch::BatchConfig, + }, }, tls::{MaybeTlsSettings, TlsEnableableConfig}, }; @@ -137,13 +138,13 @@ impl DatadogMetricsCompression { /// Maps Datadog metric endpoints to their actual URI. pub struct DatadogMetricsEndpointConfiguration { - series_endpoint: Uri, - sketches_endpoint: Uri, + series_endpoint: HttpEndpoint, + sketches_endpoint: HttpEndpoint, } impl DatadogMetricsEndpointConfiguration { /// Creates a new `DatadogMEtricsEndpointConfiguration`. - pub const fn new(series_endpoint: Uri, sketches_endpoint: Uri) -> Self { + pub const fn new(series_endpoint: HttpEndpoint, sketches_endpoint: HttpEndpoint) -> Self { Self { series_endpoint, sketches_endpoint, @@ -151,7 +152,7 @@ impl DatadogMetricsEndpointConfiguration { } /// Gets the URI for the given Datadog metrics endpoint. - pub fn get_uri_for_endpoint(&self, endpoint: DatadogMetricsEndpoint) -> Uri { + pub fn get_uri_for_endpoint(&self, endpoint: DatadogMetricsEndpoint) -> HttpEndpoint { match endpoint { DatadogMetricsEndpoint::Series { .. } => self.series_endpoint.clone(), DatadogMetricsEndpoint::Sketches => self.sketches_endpoint.clone(), @@ -293,7 +294,7 @@ impl DatadogMetricsConfig { self.series_api_version, ); - let protocol = self.get_protocol(dd_common); + let protocol = self.get_protocol(dd_common)?; let sink = DatadogMetricsSink::new( service, request_builder, @@ -306,13 +307,13 @@ impl DatadogMetricsConfig { Ok(VectorSink::from_event_streamsink(sink)) } - fn get_protocol(&self, dd_common: &DatadogCommonConfig) -> String { - self.get_base_agent_endpoint(dd_common) - .parse::() - .unwrap() + fn get_protocol(&self, dd_common: &DatadogCommonConfig) -> crate::Result { + let endpoint = HttpEndpoint::parse(&self.get_base_agent_endpoint(dd_common))?; + Ok(endpoint + .as_uri() .scheme_str() .unwrap_or("http") - .to_string() + .to_string()) } } @@ -338,11 +339,8 @@ fn resolve_endpoint_batch_settings( Ok((series, sketches)) } -fn build_uri(host: &str, endpoint: &str) -> crate::Result { - let result = format!("{host}{endpoint}") - .parse::() - .context(UriParseSnafu)?; - Ok(result) +fn build_uri(host: &str, endpoint: &str) -> crate::Result { + Ok(HttpEndpoint::parse(host)?.append_path(endpoint)?) } #[cfg(test)] diff --git a/src/sinks/datadog/metrics/request_builder.rs b/src/sinks/datadog/metrics/request_builder.rs index 8d5136505b6dd..ec62610febef6 100644 --- a/src/sinks/datadog/metrics/request_builder.rs +++ b/src/sinks/datadog/metrics/request_builder.rs @@ -233,7 +233,8 @@ impl IncrementalRequestBuilder<((Option>, DatadogMetricsEndpoint), Vec< let (ddmetrics_metadata, request_metadata) = metadata; let uri = self .endpoint_configuration - .get_uri_for_endpoint(ddmetrics_metadata.endpoint); + .get_uri_for_endpoint(ddmetrics_metadata.endpoint) + .into_uri(); DatadogMetricsRequest { api_key: ddmetrics_metadata.api_key, diff --git a/src/sinks/datadog/traces/config.rs b/src/sinks/datadog/traces/config.rs index 4237c5a75ac5d..de747b44908c8 100644 --- a/src/sinks/datadog/traces/config.rs +++ b/src/sinks/datadog/traces/config.rs @@ -1,8 +1,6 @@ use std::sync::{Arc, Mutex}; -use http::Uri; use indoc::indoc; -use snafu::ResultExt; use tokio::sync::oneshot::{Sender, channel}; use tower::ServiceBuilder; use vector_lib::{ @@ -19,7 +17,7 @@ use crate::{ config::{GenerateConfig, Input, SinkConfig, SinkContext}, http::HttpClient, sinks::{ - Healthcheck, UriParseSnafu, VectorSink, + Healthcheck, VectorSink, datadog::{ DatadogCommonConfig, LocalDatadogCommonConfig, traces::{ @@ -28,7 +26,7 @@ use crate::{ }, }, util::{ - BatchConfig, Compression, SinkBatchSettings, TowerRequestConfig, + BatchConfig, Compression, HttpEndpoint, SinkBatchSettings, TowerRequestConfig, service::ServiceBuilderExt, }, }, @@ -94,12 +92,12 @@ pub enum DatadogTracesEndpoint { /// Store traces & APM stats endpoints actual URIs. #[derive(Clone)] pub struct DatadogTracesEndpointConfiguration { - traces_endpoint: Uri, - stats_endpoint: Uri, + traces_endpoint: HttpEndpoint, + stats_endpoint: HttpEndpoint, } impl DatadogTracesEndpointConfiguration { - pub fn get_uri_for_endpoint(&self, endpoint: DatadogTracesEndpoint) -> Uri { + pub fn get_uri_for_endpoint(&self, endpoint: DatadogTracesEndpoint) -> HttpEndpoint { match endpoint { DatadogTracesEndpoint::Traces => self.traces_endpoint.clone(), DatadogTracesEndpoint::APMStats => self.stats_endpoint.clone(), @@ -172,7 +170,7 @@ impl DatadogTracesConfig { request_builder, batcher_settings, shutdown, - self.get_protocol(dd_common), + self.get_protocol(dd_common)?, ); // Send the APM stats payloads independently of the sink framework. @@ -205,12 +203,13 @@ impl DatadogTracesConfig { Ok(HttpClient::new(tls_settings, proxy)?) } - fn get_protocol(&self, dd_common: &DatadogCommonConfig) -> String { - build_uri(&self.get_base_uri(dd_common), "") - .unwrap() + fn get_protocol(&self, dd_common: &DatadogCommonConfig) -> crate::Result { + let endpoint = HttpEndpoint::parse(&self.get_base_uri(dd_common))?; + Ok(endpoint + .as_uri() .scheme_str() .unwrap_or("http") - .to_string() + .to_string()) } } @@ -236,11 +235,8 @@ impl SinkConfig for DatadogTracesConfig { } } -fn build_uri(host: &str, endpoint: &str) -> crate::Result { - let result = format!("{host}{endpoint}") - .parse::() - .context(UriParseSnafu)?; - Ok(result) +fn build_uri(host: &str, endpoint: &str) -> crate::Result { + Ok(HttpEndpoint::parse(host)?.append_path(endpoint)?) } #[cfg(test)] diff --git a/src/sinks/datadog/traces/request_builder.rs b/src/sinks/datadog/traces/request_builder.rs index 063d055b96cad..5cc404e2801ae 100644 --- a/src/sinks/datadog/traces/request_builder.rs +++ b/src/sinks/datadog/traces/request_builder.rs @@ -197,7 +197,9 @@ pub fn build_request( body: payload, headers, finalizers: ddtraces_metadata.finalizers, - uri: endpoint_configuration.get_uri_for_endpoint(ddtraces_metadata.endpoint), + uri: endpoint_configuration + .get_uri_for_endpoint(ddtraces_metadata.endpoint) + .into_uri(), uncompressed_size: ddtraces_metadata.uncompressed_size, metadata: request_metadata, } diff --git a/src/sinks/gcp/cloud_storage.rs b/src/sinks/gcp/cloud_storage.rs index 161c6074e30e4..4982120078c0b 100644 --- a/src/sinks/gcp/cloud_storage.rs +++ b/src/sinks/gcp/cloud_storage.rs @@ -3,7 +3,6 @@ use std::{collections::HashMap, convert::TryFrom, io}; use bytes::Bytes; use chrono::{FixedOffset, Utc}; use http::{ - Uri, header::{HeaderName, HeaderValue}, }; use indoc::indoc; @@ -36,8 +35,8 @@ use crate::{ sink::GcsSink, }, util::{ - BulkSizeBasedDefaultBatchSettings, Compression, RequestBuilder, ServiceBuilderExt, - TowerRequestConfig, batch::BatchConfig, metadata::RequestMetadataBuilder, + BulkSizeBasedDefaultBatchSettings, Compression, HttpEndpoint, RequestBuilder, + ServiceBuilderExt, TowerRequestConfig, batch::BatchConfig, metadata::RequestMetadataBuilder, partitioner::KeyPartitioner, request_builder::EncodeResult, service::TowerRequestConfigDefaults, timezone_to_offset, }, @@ -267,7 +266,7 @@ impl GenerateConfig for GcsSinkConfig { impl SinkConfig for GcsSinkConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { let auth = self.auth.build(Scope::DevStorageReadWrite).await?; - let base_url = format!("{}/{}/", self.endpoint, self.bucket); + let base_url = HttpEndpoint::parse(&self.endpoint)?.append_path(&format!("{}/", self.bucket))?; let tls = TlsSettings::from_options(self.tls.as_ref())?; let client = HttpClient::new(tls, cx.proxy())?; let healthcheck = build_healthcheck( @@ -298,7 +297,7 @@ impl GcsSinkConfig { fn build_sink( &self, client: HttpClient, - base_url: String, + base_url: HttpEndpoint, auth: GcpAuthenticator, cx: SinkContext, ) -> crate::Result { @@ -308,7 +307,7 @@ impl GcsSinkConfig { let partitioner = self.key_partitioner()?; - let protocol = get_http_scheme_from_uri(&base_url.parse::().unwrap()); + let protocol = get_http_scheme_from_uri(base_url.as_uri()); let svc = ServiceBuilder::new() .settings(request, GcsRetryLogic::default()) @@ -536,7 +535,7 @@ mod tests { let sink = config .build_sink( client, - mock_endpoint.to_string(), + HttpEndpoint::parse(&mock_endpoint.to_string()).expect("valid mock endpoint"), GcpAuthenticator::None, context, ) diff --git a/src/sinks/gcp/pubsub.rs b/src/sinks/gcp/pubsub.rs index f61e9b6b179ca..72d8d42b19f24 100644 --- a/src/sinks/gcp/pubsub.rs +++ b/src/sinks/gcp/pubsub.rs @@ -5,7 +5,7 @@ use http::{Request, Uri}; use hyper::Body; use indoc::indoc; use serde_json::{Value, json}; -use snafu::{ResultExt, Snafu}; +use snafu::Snafu; use tokio_util::codec::Encoder as _; use vector_lib::configurable::configurable_component; @@ -16,10 +16,11 @@ use crate::{ gcp::{GcpAuthConfig, GcpAuthenticator, PUBSUB_URL, Scope}, http::HttpClient, sinks::{ - Healthcheck, UriParseSnafu, VectorSink, + Healthcheck, VectorSink, gcs_common::config::healthcheck_response, util::{ - BatchConfig, BoxedRawValue, JsonArrayBuffer, SinkBatchSettings, TowerRequestConfig, + BatchConfig, BoxedRawValue, HttpEndpoint, JsonArrayBuffer, SinkBatchSettings, + TowerRequestConfig, http::{BatchedHttpSink, HttpEventEncoder, HttpSink}, }, }, @@ -155,7 +156,7 @@ impl SinkConfig for PubsubConfig { struct PubsubSink { auth: GcpAuthenticator, - uri_base: String, + uri_base: HttpEndpoint, transformer: Transformer, encoder: Encoder<()>, } @@ -165,10 +166,10 @@ impl PubsubSink { // We only need to load the credentials if we are not targeting an emulator. let auth = config.auth.build(Scope::PubSub).await?; - let uri_base = format!( - "{}/v1/projects/{}/topics/{}", - config.endpoint, config.project, config.topic, - ); + let uri_base = HttpEndpoint::parse(&config.endpoint)?.append_path(&format!( + "/v1/projects/{}/topics/{}", + config.project, config.topic, + ))?; let transformer = config.encoding.transformer(); let serializer = config.encoding.build()?; @@ -183,8 +184,7 @@ impl PubsubSink { } fn uri(&self, suffix: &str) -> crate::Result { - let uri = format!("{}{}", self.uri_base, suffix); - let mut uri = uri.parse::().context(UriParseSnafu)?; + let mut uri = self.uri_base.append_path(suffix)?.into_uri(); self.auth.apply_uri(&mut uri); Ok(uri) } diff --git a/src/sinks/gcp/stackdriver/logs/config.rs b/src/sinks/gcp/stackdriver/logs/config.rs index f7f491a628c69..c6566b6f77f2b 100644 --- a/src/sinks/gcp/stackdriver/logs/config.rs +++ b/src/sinks/gcp/stackdriver/logs/config.rs @@ -24,7 +24,7 @@ use crate::{ gcs_common::config::healthcheck_response, prelude::*, util::{ - BoxedRawValue, RealtimeSizeBasedDefaultBatchSettings, + BoxedRawValue, HttpEndpoint, RealtimeSizeBasedDefaultBatchSettings, http::{HttpService, RetryStrategy, http_response_retry_logic}, service::TowerRequestConfigDefaults, }, @@ -310,7 +310,7 @@ impl SinkConfig for StackdriverConfig { let tls_settings = TlsSettings::from_options(self.tls.as_ref())?; let client = HttpClient::new(tls_settings, cx.proxy())?; - let uri: Uri = self.endpoint.parse()?; + let uri = HttpEndpoint::parse(&self.endpoint)?.into_uri(); let stackdriver_logs_service_request_builder = StackdriverLogsServiceRequestBuilder { uri: uri.clone(), diff --git a/src/sinks/gcs_common/config.rs b/src/sinks/gcs_common/config.rs index dcebb6bea9e77..f379d97d0d70c 100644 --- a/src/sinks/gcs_common/config.rs +++ b/src/sinks/gcs_common/config.rs @@ -1,7 +1,7 @@ use std::marker::PhantomData; use futures::FutureExt; -use http::{StatusCode, Uri}; +use http::StatusCode; use hyper::Body; use snafu::Snafu; use vector_lib::configurable::configurable_component; @@ -12,7 +12,7 @@ use crate::{ sinks::{ Healthcheck, HealthcheckError, gcs_common::service::GcsResponse, - util::retries::{RetryAction, RetryLogic}, + util::{HttpEndpoint, retries::{RetryAction, RetryLogic}}, }, }; @@ -108,11 +108,11 @@ pub enum GcsError { pub fn build_healthcheck( bucket: String, client: HttpClient, - base_url: String, + base_url: HttpEndpoint, auth: GcpAuthenticator, ) -> crate::Result { let healthcheck = async move { - let uri = base_url.parse::()?; + let uri = base_url.into_uri(); let mut request = http::Request::head(uri).body(Body::empty())?; auth.apply(&mut request); diff --git a/src/sinks/gcs_common/service.rs b/src/sinks/gcs_common/service.rs index 7c8dd4dd33622..4676a6b4adf43 100644 --- a/src/sinks/gcs_common/service.rs +++ b/src/sinks/gcs_common/service.rs @@ -3,7 +3,7 @@ use std::task::Poll; use bytes::Bytes; use futures::future::BoxFuture; use http::{ - Request, Uri, + Request, header::{HeaderName, HeaderValue}, }; use hyper::Body; @@ -17,17 +17,22 @@ use crate::{ event::{EventFinalizers, EventStatus, Finalizable}, gcp::GcpAuthenticator, http::{HttpClient, HttpError}, + sinks::util::HttpEndpoint, }; #[derive(Debug, Clone)] pub struct GcsService { client: HttpClient, - base_url: String, + base_url: HttpEndpoint, auth: GcpAuthenticator, } impl GcsService { - pub const fn new(client: HttpClient, base_url: String, auth: GcpAuthenticator) -> GcsService { + pub const fn new( + client: HttpClient, + base_url: HttpEndpoint, + auth: GcpAuthenticator, + ) -> GcsService { GcsService { client, base_url, @@ -115,9 +120,11 @@ impl Service for GcsService { let settings = request.settings; let metadata = request.metadata; - let uri = merge_url_and_key(&self.base_url, &request.key); - - let uri = uri.parse::().unwrap(); + let uri = self + .base_url + .append_path(&request.key) + .expect("partition key must be a valid URI path segment") + .into_uri(); let mut builder = Request::put(uri); let headers = builder.headers_mut().unwrap(); @@ -148,35 +155,3 @@ impl Service for GcsService { }) } } - -/// converts // to / between the base url and the key if necessary -fn merge_url_and_key(base_url: &str, key: &str) -> String { - let base_url = base_url.strip_suffix('/').unwrap_or(base_url); - let key = key.strip_prefix('/').unwrap_or(key); - format!("{base_url}/{key}") -} - -#[cfg(test)] -mod tests { - use crate::sinks::gcs_common::service::merge_url_and_key; - - #[test] - fn merge_base_url_and_key() { - assert_eq!( - "https://baseurl/key", - merge_url_and_key("https://baseurl/", "/key") - ); - assert_eq!( - "https://baseurl/key", - merge_url_and_key("https://baseurl/", "key") - ); - assert_eq!( - "https://baseurl/key", - merge_url_and_key("https://baseurl", "/key") - ); - assert_eq!( - "https://baseurl/key", - merge_url_and_key("https://baseurl", "key") - ); - } -} diff --git a/src/sinks/honeycomb/config.rs b/src/sinks/honeycomb/config.rs index e28fae6ebfe18..abc1431d4b4a3 100644 --- a/src/sinks/honeycomb/config.rs +++ b/src/sinks/honeycomb/config.rs @@ -2,7 +2,7 @@ use bytes::Bytes; use futures::FutureExt; -use http::{Request, StatusCode, Uri}; +use http::{Request, StatusCode}; use vector_lib::{configurable::configurable_component, sensitive_string::SensitiveString}; use vrl::value::Kind; @@ -15,7 +15,7 @@ use crate::{ sinks::{ prelude::*, util::{ - BatchConfig, BoxedRawValue, + BatchConfig, BoxedRawValue, HttpEndpoint, http::{HttpService, RetryStrategy, http_response_retry_logic}, }, }, @@ -153,18 +153,17 @@ impl SinkConfig for HoneycombConfig { } impl HoneycombConfig { - fn build_uri(&self) -> crate::Result { - let uri = format!( - "{}/1/batch/{}", - self.endpoint.trim_end_matches('/'), - self.dataset - ); - uri.parse::().map_err(Into::into) + fn build_uri(&self) -> crate::Result { + Ok(HttpEndpoint::parse(&self.endpoint)?.append_path(&format!("1/batch/{}", self.dataset))?) } } -async fn healthcheck(uri: Uri, api_key: SensitiveString, client: HttpClient) -> crate::Result<()> { - let request = Request::post(uri).header(HTTP_HEADER_HONEYCOMB, api_key.inner()); +async fn healthcheck( + uri: HttpEndpoint, + api_key: SensitiveString, + client: HttpClient, +) -> crate::Result<()> { + let request = Request::post(uri.as_uri()).header(HTTP_HEADER_HONEYCOMB, api_key.inner()); let body = crate::serde::json::to_bytes(&Vec::::new()) .unwrap() .freeze(); diff --git a/src/sinks/honeycomb/service.rs b/src/sinks/honeycomb/service.rs index a85f72b82f15a..a706fcee76cfc 100644 --- a/src/sinks/honeycomb/service.rs +++ b/src/sinks/honeycomb/service.rs @@ -1,7 +1,7 @@ //! Service implementation for the `honeycomb` sink. use bytes::Bytes; -use http::{Request, Uri}; +use http::Request; use snafu::ResultExt; use vector_lib::sensitive_string::SensitiveString; @@ -11,12 +11,13 @@ use crate::sinks::{ util::{ buffer::compression::Compression, http::{HttpRequest, HttpServiceRequestBuilder}, + HttpEndpoint, }, }; #[derive(Debug, Clone)] pub(super) struct HoneycombSvcRequestBuilder { - pub(super) uri: Uri, + pub(super) uri: HttpEndpoint, pub(super) api_key: SensitiveString, pub(super) compression: Compression, } @@ -24,7 +25,7 @@ pub(super) struct HoneycombSvcRequestBuilder { impl HttpServiceRequestBuilder<()> for HoneycombSvcRequestBuilder { fn build(&self, mut request: HttpRequest<()>) -> Result, crate::Error> { let mut builder = - Request::post(&self.uri).header(HTTP_HEADER_HONEYCOMB, self.api_key.inner()); + Request::post(self.uri.as_uri()).header(HTTP_HEADER_HONEYCOMB, self.api_key.inner()); if let Some(ce) = self.compression.content_encoding() { builder = builder.header("Content-Encoding".to_string(), ce.to_string()); diff --git a/src/sinks/influxdb/mod.rs b/src/sinks/influxdb/mod.rs index f49e7f98d02f9..ecae579951bee 100644 --- a/src/sinks/influxdb/mod.rs +++ b/src/sinks/influxdb/mod.rs @@ -7,7 +7,6 @@ use bytes::{BufMut, BytesMut}; use chrono::{DateTime, Utc}; use futures::FutureExt; use http::{StatusCode, Uri}; -use snafu::ResultExt; use tower::Service; use vector_lib::{ configurable::configurable_component, @@ -16,6 +15,7 @@ use vector_lib::{ }; use crate::http::HttpClient; +use crate::sinks::util::HttpEndpoint; pub(in crate::sinks) enum Field { /// string @@ -366,17 +366,13 @@ pub(in crate::sinks) fn encode_uri( } } - let mut url = if endpoint.ends_with('/') { - format!("{}{}?{}", endpoint, path, serializer.finish()) + let query = serializer.finish(); + let path_and_query = if query.is_empty() { + path.to_string() } else { - format!("{}/{}?{}", endpoint, path, serializer.finish()) + format!("{path}?{query}") }; - - if url.ends_with('?') { - url.pop(); - } - - Ok(url.parse::().context(super::UriParseSnafu)?) + Ok(HttpEndpoint::parse(endpoint)?.append_path(&path_and_query)?.into_uri()) } #[cfg(test)] diff --git a/src/sinks/prometheus/remote_write/config.rs b/src/sinks/prometheus/remote_write/config.rs index 25adb2ed9a47d..0b79b12452767 100644 --- a/src/sinks/prometheus/remote_write/config.rs +++ b/src/sinks/prometheus/remote_write/config.rs @@ -1,7 +1,6 @@ use std::{collections::BTreeMap, sync::Arc}; use http::{HeaderValue, Uri, header::AUTHORIZATION}; -use snafu::prelude::*; #[cfg(feature = "aws-core")] use super::Errors; @@ -12,13 +11,13 @@ use super::{ use crate::{ http::HttpClient, sinks::{ - UriParseSnafu, prelude::*, prometheus::PrometheusRemoteWriteAuth, util::{ auth::Auth, http::{OrderedHeaderName, RetryStrategy, http_response_retry_logic}, service::TowerRequestConfig, + HttpEndpoint, }, }, template::ConfinementConfig, @@ -197,7 +196,7 @@ impl SinkConfig for RemoteWriteConfig { }) .transpose()?; - let endpoint = self.endpoint.parse::().context(UriParseSnafu)?; + let endpoint = HttpEndpoint::parse(&self.endpoint)?; let tls_settings = TlsSettings::from_options(self.tls.as_ref())?; let request_settings = self.request.tower.into_settings(); let validated_headers = Arc::new(validate_headers( @@ -242,7 +241,7 @@ impl SinkConfig for RemoteWriteConfig { let healthcheck_endpoint = match cx.healthcheck.uri { Some(uri) => uri.uri, - None => endpoint.clone(), + None => endpoint.as_uri().clone(), }; let healthcheck = healthcheck( diff --git a/src/sinks/prometheus/remote_write/service.rs b/src/sinks/prometheus/remote_write/service.rs index 2ec1578e4dd82..dabe1713a8e9b 100644 --- a/src/sinks/prometheus/remote_write/service.rs +++ b/src/sinks/prometheus/remote_write/service.rs @@ -20,6 +20,7 @@ use crate::{ util::{ auth::Auth, http::{HttpResponse, OrderedHeaderName}, + HttpEndpoint, }, }, }; @@ -37,7 +38,7 @@ mod headers { #[derive(Clone)] pub(super) struct RemoteWriteService { - pub(super) endpoint: Uri, + pub(super) endpoint: HttpEndpoint, pub(super) auth: Option, pub(super) client: HttpClient, pub(super) compression: super::Compression, @@ -56,7 +57,7 @@ impl Service for RemoteWriteService { // Emission of internal events for errors and dropped events is handled upstream by the caller. fn call(&mut self, mut request: RemoteWriteRequest) -> Self::Future { let client = self.client.clone(); - let endpoint = self.endpoint.clone(); + let endpoint = self.endpoint.as_uri().clone(); let auth = self.auth.clone(); let compression = self.compression; let headers = Arc::clone(&self.headers); diff --git a/src/sinks/sematext/metrics.rs b/src/sinks/sematext/metrics.rs index 0da44dc28eb13..ef57a1cb8a57b 100644 --- a/src/sinks/sematext/metrics.rs +++ b/src/sinks/sematext/metrics.rs @@ -25,7 +25,7 @@ use crate::{ Healthcheck, HealthcheckError, VectorSink, influxdb::{Field, ProtocolVersion, encode_timestamp, encode_uri, influx_line_protocol}, util::{ - BatchConfig, EncodedEvent, SinkBatchSettings, TowerRequestConfig, + BatchConfig, EncodedEvent, HttpEndpoint, SinkBatchSettings, TowerRequestConfig, buffer::metrics::{MetricNormalize, MetricNormalizer, MetricSet, MetricsBuffer}, http::{HttpBatchService, HttpRetryLogic}, }, @@ -103,7 +103,7 @@ impl GenerateConfig for SematextMetricsConfig { } async fn healthcheck(endpoint: String, client: HttpClient) -> Result<()> { - let uri = format!("{endpoint}/health"); + let uri = HttpEndpoint::parse(&endpoint)?.append_path("health")?.into_uri(); let request = Request::get(uri) .body(Body::empty()) diff --git a/src/sinks/util/mod.rs b/src/sinks/util/mod.rs index 2dddb88bb7c68..81bc4afc81591 100644 --- a/src/sinks/util/mod.rs +++ b/src/sinks/util/mod.rs @@ -54,7 +54,7 @@ pub use service::{ }; pub use sink::{BatchSink, PartitionBatchSink, StreamSink}; use snafu::Snafu; -pub use uri::UriSerde; +pub use uri::{HttpEndpoint, HttpEndpointError, UriSerde}; use vector_lib::{TimeZone, json_size::JsonSize}; use crate::event::EventFinalizers; diff --git a/src/sinks/util/uri.rs b/src/sinks/util/uri.rs index 86477f99887c0..9eafce544393a 100644 --- a/src/sinks/util/uri.rs +++ b/src/sinks/util/uri.rs @@ -2,6 +2,7 @@ use std::{fmt, str::FromStr}; use http::uri::{Authority, PathAndQuery, Scheme, Uri}; use percent_encoding::percent_decode_str; +use snafu::{ResultExt, Snafu}; use vector_lib::configurable::configurable_component; use crate::http::Auth; @@ -196,6 +197,122 @@ pub fn protocol_endpoint(uri: Uri) -> (String, String) { ) } +/// Error returned when a configured endpoint cannot be used as an absolute HTTP URL. +#[derive(Debug, Snafu)] +pub enum HttpEndpointError { + #[snafu(display("endpoint `{endpoint}` is not a valid URI: {source}"))] + InvalidUri { + endpoint: String, + source: http::uri::InvalidUri, + }, + + #[snafu(display("endpoint `{endpoint}` has an invalid path `{path}`: {source}"))] + InvalidPath { + endpoint: String, + path: String, + source: http::uri::InvalidUri, + }, + + #[snafu(display("endpoint `{endpoint}` cannot be reassembled from its parts: {source}"))] + InvalidUriParts { + endpoint: String, + source: http::uri::InvalidUriParts, + }, + + #[snafu(display( + "endpoint must be an absolute http(s) URL, for example `https://example.com`; got `{endpoint}`" + ))] + NotAbsoluteHttp { endpoint: String }, +} + +/// A `Uri` proven to be an absolute `http`/`https` URL. +/// +/// Constructing an `HttpEndpoint` is the only way to obtain one: both +/// [`HttpEndpoint::new`] and [`HttpEndpoint::parse`] reject URIs without an +/// `http`/`https` scheme or without an authority. Sinks that issue requests +/// through `HttpClient` need this invariant, since `HttpClient` rejects such +/// URIs at request time, deferring a pure configuration error to runtime. +/// +/// Path composition goes through [`HttpEndpoint::append_path`], which +/// manipates `Uri` parts directly instead of string-concatenating and +/// re-parsing, so the scheme and authority are preserved and the result is +/// still an absolute `http(s)` URL. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct HttpEndpoint(Uri); + +impl HttpEndpoint { + /// Requires `uri` to be an absolute `http`/`https` URL. + pub fn new(uri: Uri) -> Result { + if matches!(uri.scheme_str(), Some("http" | "https")) && uri.authority().is_some() { + Ok(Self(uri)) + } else { + Err(HttpEndpointError::NotAbsoluteHttp { + endpoint: uri.to_string(), + }) + } + } + + /// Parses `endpoint` and requires it to be an absolute `http`/`https` URL. + pub fn parse(endpoint: &str) -> Result { + let uri = endpoint + .parse::() + .context(InvalidUriSnafu { endpoint })?; + Self::new(uri) + } + + /// Returns the underlying `Uri`. + pub const fn as_uri(&self) -> &Uri { + &self.0 + } + + /// Consumes the endpoint, returning the underlying `Uri`. + pub fn into_uri(self) -> Uri { + self.0 + } + + /// Appends `path` to this endpoint, preserving the scheme and authority. + /// + /// `path` may include a leading slash and a query. The existing query, if + /// any, is dropped (as with `UriSerde::append_path`), but the scheme and + /// authority are preserved and the result is still an absolute `http(s)` URL. + pub fn append_path(&self, path: &str) -> Result { + if path.is_empty() { + return Ok(self.clone()); + } + let mut parts = self.0.clone().into_parts(); + let base_path = parts + .path_and_query + .as_ref() + .map(PathAndQuery::path) + .unwrap_or_default(); + let joined = if base_path.is_empty() { + path.to_string() + } else if base_path.ends_with('/') { + format!("{base_path}{}", path.trim_start_matches('/')) + } else { + format!("{base_path}/{path}") + }; + parts.path_and_query = Some( + joined + .parse::() + .context(InvalidPathSnafu { + endpoint: self.0.to_string(), + path: joined, + })?, + ); + let uri = Uri::from_parts(parts).context(InvalidUriPartsSnafu { + endpoint: self.0.to_string(), + })?; + Self::new(uri) + } +} + +impl fmt::Display for HttpEndpoint { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.0.fmt(f) + } +} + #[cfg(test)] mod tests { use super::*; @@ -266,4 +383,84 @@ mod tests { ("gopher".into(), "gopher://example.net:123/path".into()) ); } + + #[test] + fn http_endpoint_accepts_absolute_http_urls() { + for endpoint in [ + "http://example.com", + "https://example.com", + "https://example.com:8088/services/collector", + "http://127.0.0.1:9000/endpoint?query=1", + "https://user:pass@example.com/path", + ] { + let endpoint = HttpEndpoint::parse(endpoint).expect("should accept absolute http(s) URL"); + assert!(matches!(endpoint.as_uri().scheme_str(), Some("http" | "https"))); + assert!(endpoint.as_uri().authority().is_some()); + } + } + + #[test] + fn http_endpoint_rejects_non_absolute_http_urls() { + for endpoint in [ + // No scheme: `http::Uri` parses these as authority-form or as a path. + "example.com:8088", + "localhost:8080", + "/services/collector", + "", + // Absolute, but not a scheme `HttpClient` can dial. + "gopher://example.com", + "unix:///var/run/vector.sock", + // Scheme but no authority. + "http:///path", + ] { + assert!( + matches!( + HttpEndpoint::parse(endpoint), + Err(HttpEndpointError::NotAbsoluteHttp { .. }) + | Err(HttpEndpointError::InvalidUri { .. }) + ), + "expected `{endpoint}` to be rejected" + ); + } + } + + #[test] + fn http_endpoint_reports_unparseable_endpoints() { + let error = HttpEndpoint::parse("http://exa mple.com").unwrap_err(); + assert!(matches!(error, HttpEndpointError::InvalidUri { .. })); + } + + #[test] + fn http_endpoint_append_path_joins_without_string_concatenation() { + let base = HttpEndpoint::parse("https://example.com").unwrap(); + + assert_eq!( + base.append_path("vector/events").unwrap().to_string(), + "https://example.com/vector/events" + ); + assert_eq!( + base.append_path("/api/v1/series").unwrap().to_string(), + "https://example.com/api/v1/series" + ); + assert_eq!( + HttpEndpoint::parse("https://example.com/") + .unwrap() + .append_path("vector/events") + .unwrap() + .to_string(), + "https://example.com/vector/events" + ); + // The query is carried in the appended path. + assert_eq!( + base.append_path("/write?db=mydb").unwrap().to_string(), + "https://example.com/write?db=mydb" + ); + // The scheme and authority survive appending. + let appended = HttpEndpoint::parse("https://user:pass@example.com:8088/base") + .unwrap() + .append_path("sub/path") + .unwrap(); + assert_eq!(appended.to_string(), "https://user:pass@example.com:8088/base/sub/path"); + assert!(matches!(appended.as_uri().scheme_str(), Some("https"))); + } } From 135c72434645291841b05dec9ab7f63f5b4b7ac3 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 14:48:26 -0400 Subject: [PATCH 02/16] feat(sinks): migrate sink endpoints to HttpEndpoint --- src/sinks/appsignal/config.rs | 36 +++++++++---- src/sinks/appsignal/integration_tests.rs | 7 ++- src/sinks/appsignal/tests.rs | 3 +- src/sinks/azure_logs_ingestion/config.rs | 12 ++--- src/sinks/azure_logs_ingestion/tests.rs | 12 ++--- src/sinks/azure_monitor_logs/tests.rs | 5 +- src/sinks/datadog/metrics/config.rs | 6 +-- src/sinks/datadog/traces/config.rs | 6 +-- src/sinks/gcp/cloud_storage.rs | 16 +++--- src/sinks/gcp/pubsub.rs | 10 ++-- src/sinks/gcp/stackdriver/logs/config.rs | 13 +++-- src/sinks/gcp/stackdriver/logs/tests.rs | 5 +- src/sinks/gcp/stackdriver/metrics/config.rs | 21 ++++---- src/sinks/gcp/stackdriver/metrics/tests.rs | 13 +++-- src/sinks/gcs_common/config.rs | 10 ++-- src/sinks/honeycomb/config.rs | 11 ++-- src/sinks/honeycomb/service.rs | 2 +- src/sinks/honeycomb/tests.rs | 4 +- src/sinks/humio/logs.rs | 10 ++-- src/sinks/humio/metrics.rs | 22 +++++--- src/sinks/influxdb/logs.rs | 8 +-- src/sinks/influxdb/metrics.rs | 9 ++-- src/sinks/influxdb/mod.rs | 51 ++++++++----------- src/sinks/prometheus/remote_write/config.rs | 11 ++-- .../remote_write/integration_tests.rs | 4 +- src/sinks/prometheus/remote_write/service.rs | 2 +- src/sinks/sematext/metrics.rs | 7 ++- src/sinks/splunk_hec/logs/config.rs | 22 ++++---- .../splunk_hec/logs/integration_tests.rs | 4 +- src/sinks/splunk_hec/logs/tests.rs | 4 +- src/sinks/splunk_hec/metrics/config.rs | 11 ++-- .../splunk_hec/metrics/integration_tests.rs | 4 +- src/sinks/splunk_hec/metrics/tests.rs | 4 +- src/sinks/util/uri.rs | 45 ++++++++++++---- src/sources/prometheus/remote_write.rs | 14 +++-- src/sources/splunk_hec/mod.rs | 4 +- .../components/sinks/generated/appsignal.cue | 2 +- .../sinks/generated/gcp_cloud_storage.cue | 2 +- .../components/sinks/generated/gcp_pubsub.cue | 2 +- .../components/sinks/generated/honeycomb.cue | 2 +- .../components/sinks/generated/humio_logs.cue | 2 +- .../sinks/generated/humio_metrics.cue | 2 +- 42 files changed, 254 insertions(+), 186 deletions(-) diff --git a/src/sinks/appsignal/config.rs b/src/sinks/appsignal/config.rs index 357da68f071b9..317bc7c09fd0a 100644 --- a/src/sinks/appsignal/config.rs +++ b/src/sinks/appsignal/config.rs @@ -1,3 +1,4 @@ +use derivative::Derivative; use futures::FutureExt; use http::{Request, header::AUTHORIZATION}; use hyper::Body; @@ -29,13 +30,15 @@ use crate::{ /// 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"))] @@ -74,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)] @@ -152,8 +155,13 @@ impl SinkConfig for AppsignalConfig { } } -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}")); +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() { @@ -162,13 +170,13 @@ async fn healthcheck(uri: HttpEndpoint, push_api_key: String, client: HttpClient } } -pub fn endpoint_uri(endpoint: &str, path: &str) -> crate::Result { - Ok(HttpEndpoint::parse(endpoint)?.append_path(path)?) +pub fn endpoint_uri(endpoint: &HttpEndpoint, path: &str) -> crate::Result { + Ok(endpoint.append_path(path)?) } #[cfg(test)] mod test { - use super::{AppsignalConfig, endpoint_uri}; + use super::{AppsignalConfig, HttpEndpoint, endpoint_uri}; #[test] fn generate_config() { @@ -177,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" @@ -186,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" diff --git a/src/sinks/appsignal/integration_tests.rs b/src/sinks/appsignal/integration_tests.rs index b6a651084266c..ad37e76c4e1e1 100644 --- a/src/sinks/appsignal/integration_tests.rs +++ b/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) -> (Vec, Receiver<(http::request: let (mut config, cx) = load_sink::(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(); diff --git a/src/sinks/appsignal/tests.rs b/src/sinks/appsignal/tests.rs index b7a04c30c1593..de74d1f3a758f 100644 --- a/src/sinks/appsignal/tests.rs +++ b/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(); diff --git a/src/sinks/azure_logs_ingestion/config.rs b/src/sinks/azure_logs_ingestion/config.rs index 89da496045d9a..1336375b77313 100644 --- a/src/sinks/azure_logs_ingestion/config.rs +++ b/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, 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 = self.auth.credential().await?; diff --git a/src/sinks/azure_logs_ingestion/tests.rs b/src/sinks/azure_logs_ingestion/tests.rs index 6126a98a8cffe..275f3a0d9811b 100644 --- a/src/sinks/azure_logs_ingestion/tests.rs +++ b/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, diff --git a/src/sinks/azure_monitor_logs/tests.rs b/src/sinks/azure_monitor_logs/tests.rs index 69867e9574330..6b806612e9a25 100644 --- a/src/sinks/azure_monitor_logs/tests.rs +++ b/src/sinks/azure_monitor_logs/tests.rs @@ -12,7 +12,10 @@ use super::{ }; use crate::{ event::LogEvent, - sinks::{prelude::*, util::{encoding::Encoder, HttpEndpoint}}, + sinks::{ + prelude::*, + util::{HttpEndpoint, encoding::Encoder}, + }, test_util::{ components::{SINK_TAGS, run_and_assert_sink_compliance}, http::{always_200_response, spawn_blackhole_http_server}, diff --git a/src/sinks/datadog/metrics/config.rs b/src/sinks/datadog/metrics/config.rs index 9ac813d097fb3..500993499cd39 100644 --- a/src/sinks/datadog/metrics/config.rs +++ b/src/sinks/datadog/metrics/config.rs @@ -309,11 +309,7 @@ impl DatadogMetricsConfig { fn get_protocol(&self, dd_common: &DatadogCommonConfig) -> crate::Result { let endpoint = HttpEndpoint::parse(&self.get_base_agent_endpoint(dd_common))?; - Ok(endpoint - .as_uri() - .scheme_str() - .unwrap_or("http") - .to_string()) + Ok(endpoint.as_uri().scheme_str().unwrap_or("http").to_string()) } } diff --git a/src/sinks/datadog/traces/config.rs b/src/sinks/datadog/traces/config.rs index de747b44908c8..7d78705c089ae 100644 --- a/src/sinks/datadog/traces/config.rs +++ b/src/sinks/datadog/traces/config.rs @@ -205,11 +205,7 @@ impl DatadogTracesConfig { fn get_protocol(&self, dd_common: &DatadogCommonConfig) -> crate::Result { let endpoint = HttpEndpoint::parse(&self.get_base_uri(dd_common))?; - Ok(endpoint - .as_uri() - .scheme_str() - .unwrap_or("http") - .to_string()) + Ok(endpoint.as_uri().scheme_str().unwrap_or("http").to_string()) } } diff --git a/src/sinks/gcp/cloud_storage.rs b/src/sinks/gcp/cloud_storage.rs index 4982120078c0b..a22d2520cc01c 100644 --- a/src/sinks/gcp/cloud_storage.rs +++ b/src/sinks/gcp/cloud_storage.rs @@ -2,9 +2,7 @@ use std::{collections::HashMap, convert::TryFrom, io}; use bytes::Bytes; use chrono::{FixedOffset, Utc}; -use http::{ - header::{HeaderName, HeaderValue}, -}; +use http::header::{HeaderName, HeaderValue}; use indoc::indoc; use snafu::{ResultExt, Snafu}; use tower::ServiceBuilder; @@ -36,9 +34,9 @@ use crate::{ }, util::{ BulkSizeBasedDefaultBatchSettings, Compression, HttpEndpoint, RequestBuilder, - ServiceBuilderExt, TowerRequestConfig, batch::BatchConfig, metadata::RequestMetadataBuilder, - partitioner::KeyPartitioner, request_builder::EncodeResult, - service::TowerRequestConfigDefaults, timezone_to_offset, + ServiceBuilderExt, TowerRequestConfig, batch::BatchConfig, + metadata::RequestMetadataBuilder, partitioner::KeyPartitioner, + request_builder::EncodeResult, service::TowerRequestConfigDefaults, timezone_to_offset, }, }, template::{ConfinementConfig, Template, TemplateParseError}, @@ -188,7 +186,7 @@ pub struct GcsSinkConfig { #[configurable(metadata(docs::examples = "http://localhost:9000"))] #[configurable(validation(format = "uri"))] #[serde(default = "default_endpoint")] - endpoint: String, + endpoint: HttpEndpoint, #[configurable(derived)] #[serde(default)] @@ -237,7 +235,7 @@ fn default_config(encoding: EncodingConfigWithFraming) -> GcsSinkConfig { encoding, compression: Compression::gzip_default(), batch: Default::default(), - endpoint: Default::default(), + endpoint: default_endpoint(), request: Default::default(), auth: Default::default(), tls: Default::default(), @@ -266,7 +264,7 @@ impl GenerateConfig for GcsSinkConfig { impl SinkConfig for GcsSinkConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { let auth = self.auth.build(Scope::DevStorageReadWrite).await?; - let base_url = HttpEndpoint::parse(&self.endpoint)?.append_path(&format!("{}/", self.bucket))?; + let base_url = self.endpoint.append_path(&format!("{}/", self.bucket))?; let tls = TlsSettings::from_options(self.tls.as_ref())?; let client = HttpClient::new(tls, cx.proxy())?; let healthcheck = build_healthcheck( diff --git a/src/sinks/gcp/pubsub.rs b/src/sinks/gcp/pubsub.rs index 72d8d42b19f24..531cc9df6b206 100644 --- a/src/sinks/gcp/pubsub.rs +++ b/src/sinks/gcp/pubsub.rs @@ -70,7 +70,7 @@ pub struct PubsubConfig { /// [pubsub_api]: https://cloud.google.com/pubsub/docs/reference/rest #[serde(default = "default_endpoint")] #[configurable(metadata(docs::examples = "https://us-central1-pubsub.googleapis.com"))] - pub endpoint: String, + pub endpoint: HttpEndpoint, #[serde(default, flatten)] pub auth: GcpAuthConfig, @@ -99,8 +99,8 @@ pub struct PubsubConfig { acknowledgements: AcknowledgementsConfig, } -fn default_endpoint() -> String { - PUBSUB_URL.to_string() +fn default_endpoint() -> HttpEndpoint { + HttpEndpoint::parse(PUBSUB_URL).expect("static default endpoint should be a valid http(s) URL") } impl GenerateConfig for PubsubConfig { @@ -166,7 +166,7 @@ impl PubsubSink { // We only need to load the credentials if we are not targeting an emulator. let auth = config.auth.build(Scope::PubSub).await?; - let uri_base = HttpEndpoint::parse(&config.endpoint)?.append_path(&format!( + let uri_base = config.endpoint.clone().append_path(&format!( "/v1/projects/{}/topics/{}", config.project, config.topic, ))?; @@ -295,7 +295,7 @@ mod integration_tests { PubsubConfig { project: PROJECT.into(), topic: topic.into(), - endpoint: gcp::PUBSUB_ADDRESS.clone(), + endpoint: HttpEndpoint::parse(&gcp::PUBSUB_ADDRESS).unwrap(), auth: GcpAuthConfig { skip_authentication: true, ..Default::default() diff --git a/src/sinks/gcp/stackdriver/logs/config.rs b/src/sinks/gcp/stackdriver/logs/config.rs index c6566b6f77f2b..9ecee67c5ea2c 100644 --- a/src/sinks/gcp/stackdriver/logs/config.rs +++ b/src/sinks/gcp/stackdriver/logs/config.rs @@ -50,11 +50,13 @@ impl TowerRequestConfigDefaults for StackdriverTowerRequestConfigDefaults { "gcp_stackdriver_logs", "Deliver logs to GCP's Cloud Operations suite." ))] -#[derive(Clone, Debug, Default)] +#[derive(Clone, Debug, Derivative)] +#[derivative(Default)] #[serde(deny_unknown_fields)] pub(super) struct StackdriverConfig { + #[derivative(Default(value = "default_endpoint()"))] #[serde(skip, default = "default_endpoint")] - pub(super) endpoint: String, + pub(super) endpoint: HttpEndpoint, #[serde(flatten)] pub(super) log_name: StackdriverLogName, @@ -121,8 +123,9 @@ pub(super) struct StackdriverConfig { pub confinement: ConfinementConfig, } -pub(super) fn default_endpoint() -> String { - "https://logging.googleapis.com/v2/entries:write".to_string() +pub(super) fn default_endpoint() -> HttpEndpoint { + HttpEndpoint::parse("https://logging.googleapis.com/v2/entries:write") + .expect("static default endpoint should be a valid http(s) URL") } // 10MB limit for entries.write: https://cloud.google.com/logging/quotas#api-limits @@ -310,7 +313,7 @@ impl SinkConfig for StackdriverConfig { let tls_settings = TlsSettings::from_options(self.tls.as_ref())?; let client = HttpClient::new(tls_settings, cx.proxy())?; - let uri = HttpEndpoint::parse(&self.endpoint)?.into_uri(); + let uri = self.endpoint.clone().into_uri(); let stackdriver_logs_service_request_builder = StackdriverLogsServiceRequestBuilder { uri: uri.clone(), diff --git a/src/sinks/gcp/stackdriver/logs/tests.rs b/src/sinks/gcp/stackdriver/logs/tests.rs index 0a0a3c0f51625..6e0a40fdb1edf 100644 --- a/src/sinks/gcp/stackdriver/logs/tests.rs +++ b/src/sinks/gcp/stackdriver/logs/tests.rs @@ -27,6 +27,7 @@ use crate::{ }, prelude::*, util::{ + HttpEndpoint, encoding::Encoder as _, http::{HttpRequest, HttpServiceRequestBuilder}, }, @@ -67,7 +68,7 @@ async fn component_spec_compliance() { // Metadata API, which we clearly don't have in unit tests. :) config.auth.credentials_path = None; config.auth.api_key = Some("fake".to_string().into()); - 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(); @@ -216,7 +217,7 @@ fn severity_remaps_strings() { #[tokio::test] async fn correct_request() { - let uri: Uri = default_endpoint().parse().unwrap(); + let uri: Uri = default_endpoint().into_uri(); let transformer = Transformer::default(); let encoder = StackdriverLogsEncoder::new( diff --git a/src/sinks/gcp/stackdriver/metrics/config.rs b/src/sinks/gcp/stackdriver/metrics/config.rs index ea2cc9d6b5293..8cb21f0cf00a1 100644 --- a/src/sinks/gcp/stackdriver/metrics/config.rs +++ b/src/sinks/gcp/stackdriver/metrics/config.rs @@ -14,6 +14,7 @@ use crate::{ HTTPRequestBuilderSnafu, gcp, prelude::*, util::{ + HttpEndpoint, http::{ HttpRequest, HttpService, HttpServiceRequestBuilder, RetryStrategy, http_response_retry_logic, @@ -35,10 +36,12 @@ impl TowerRequestConfigDefaults for StackdriverMetricsTowerRequestConfigDefaults "gcp_stackdriver_metrics", "Deliver metrics to GCP's Cloud Monitoring system." ))] -#[derive(Clone, Debug, Default)] +#[derive(Clone, Debug, Derivative)] +#[derivative(Default)] pub struct StackdriverConfig { + #[derivative(Default(value = "default_endpoint()"))] #[serde(skip, default = "default_endpoint")] - pub(super) endpoint: String, + pub(super) endpoint: HttpEndpoint, /// The project ID to which to publish metrics. /// @@ -88,8 +91,9 @@ fn default_metric_namespace_value() -> String { "namespace".to_string() } -fn default_endpoint() -> String { - "https://monitoring.googleapis.com".to_string() +fn default_endpoint() -> HttpEndpoint { + HttpEndpoint::parse("https://monitoring.googleapis.com") + .expect("static default endpoint should be a valid http(s) URL") } impl_generate_config_from_default!(StackdriverConfig); @@ -117,11 +121,10 @@ impl SinkConfig for StackdriverConfig { let request_limits = self.request.into_settings(); - let uri: Uri = format!( - "{}/v3/projects/{}/timeSeries", - self.endpoint, self.project_id - ) - .parse()?; + let uri = self + .endpoint + .append_path(&format!("/v3/projects/{}/timeSeries", self.project_id))? + .into_uri(); auth.spawn_regenerate_token(); diff --git a/src/sinks/gcp/stackdriver/metrics/tests.rs b/src/sinks/gcp/stackdriver/metrics/tests.rs index 504a0efb11d7a..e13102a35589c 100644 --- a/src/sinks/gcp/stackdriver/metrics/tests.rs +++ b/src/sinks/gcp/stackdriver/metrics/tests.rs @@ -6,7 +6,10 @@ use super::config::StackdriverConfig; use crate::{ config::SinkContext, gcp::GcpAuthConfig, - sinks::{prelude::*, util::test::build_test_server}, + sinks::{ + prelude::*, + util::{HttpEndpoint, test::build_test_server}, + }, test_util::{ addr::next_addr, components::{SINK_TAGS, run_and_assert_sink_compliance}, @@ -31,7 +34,7 @@ async fn component_spec_compliance() { // Metadata API, which we clearly don't have in unit tests. :) config.auth.credentials_path = None; config.auth.api_key = Some("fake".to_string().into()); - 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(); @@ -48,7 +51,7 @@ async fn component_spec_compliance() { async fn sends_metric() { let (_guard, in_addr) = next_addr(); let config = StackdriverConfig { - endpoint: format!("http://{in_addr}"), + endpoint: HttpEndpoint::parse(&format!("http://{in_addr}")).unwrap(), auth: GcpAuthConfig { api_key: None, credentials_path: None, @@ -108,7 +111,7 @@ async fn sends_multiple_metrics() { batch.max_events = Some(5); let config = StackdriverConfig { - endpoint: format!("http://{in_addr}"), + endpoint: HttpEndpoint::parse(&format!("http://{in_addr}")).unwrap(), auth: GcpAuthConfig { api_key: None, credentials_path: None, @@ -195,7 +198,7 @@ async fn does_not_aggregate_metrics() { batch.max_events = Some(5); let config = StackdriverConfig { - endpoint: format!("http://{in_addr}"), + endpoint: HttpEndpoint::parse(&format!("http://{in_addr}")).unwrap(), auth: GcpAuthConfig { api_key: None, credentials_path: None, diff --git a/src/sinks/gcs_common/config.rs b/src/sinks/gcs_common/config.rs index f379d97d0d70c..650ae3367d0ee 100644 --- a/src/sinks/gcs_common/config.rs +++ b/src/sinks/gcs_common/config.rs @@ -12,12 +12,16 @@ use crate::{ sinks::{ Healthcheck, HealthcheckError, gcs_common::service::GcsResponse, - util::{HttpEndpoint, retries::{RetryAction, RetryLogic}}, + util::{ + HttpEndpoint, + retries::{RetryAction, RetryLogic}, + }, }, }; -pub fn default_endpoint() -> String { - "https://storage.googleapis.com".to_string() +pub fn default_endpoint() -> HttpEndpoint { + HttpEndpoint::parse("https://storage.googleapis.com") + .expect("static default endpoint should be a valid http(s) URL") } /// GCS Predefined ACLs. diff --git a/src/sinks/honeycomb/config.rs b/src/sinks/honeycomb/config.rs index abc1431d4b4a3..f36e51bedc737 100644 --- a/src/sinks/honeycomb/config.rs +++ b/src/sinks/honeycomb/config.rs @@ -34,7 +34,7 @@ pub struct HoneycombConfig { docs::examples = "https://api.eu1.honeycomb.io", ))] #[configurable(validation(format = "uri"))] - pub(super) endpoint: String, + pub(super) endpoint: HttpEndpoint, /// The API key that is used to authenticate against Honeycomb. #[configurable(metadata(docs::examples = "${HONEYCOMB_API_KEY}"))] @@ -77,8 +77,9 @@ pub struct HoneycombConfig { pub retry_strategy: RetryStrategy, } -fn default_endpoint() -> String { - "https://api.honeycomb.io".to_string() +fn default_endpoint() -> HttpEndpoint { + HttpEndpoint::parse("https://api.honeycomb.io") + .expect("static default endpoint should be a valid http(s) URL") } #[derive(Clone, Copy, Debug, Default)] @@ -154,7 +155,9 @@ impl SinkConfig for HoneycombConfig { impl HoneycombConfig { fn build_uri(&self) -> crate::Result { - Ok(HttpEndpoint::parse(&self.endpoint)?.append_path(&format!("1/batch/{}", self.dataset))?) + Ok(self + .endpoint + .append_path(&format!("1/batch/{}", self.dataset))?) } } diff --git a/src/sinks/honeycomb/service.rs b/src/sinks/honeycomb/service.rs index a706fcee76cfc..b9a7d41b62b0a 100644 --- a/src/sinks/honeycomb/service.rs +++ b/src/sinks/honeycomb/service.rs @@ -9,9 +9,9 @@ use super::config::HTTP_HEADER_HONEYCOMB; use crate::sinks::{ HTTPRequestBuilderSnafu, util::{ + HttpEndpoint, buffer::compression::Compression, http::{HttpRequest, HttpServiceRequestBuilder}, - HttpEndpoint, }, }; diff --git a/src/sinks/honeycomb/tests.rs b/src/sinks/honeycomb/tests.rs index 6bd7f58fbf50a..eed3f47b52b3a 100644 --- a/src/sinks/honeycomb/tests.rs +++ b/src/sinks/honeycomb/tests.rs @@ -4,7 +4,7 @@ use futures::{future::ready, stream}; use super::config::HoneycombConfig; use crate::{ - sinks::prelude::*, + sinks::{prelude::*, util::HttpEndpoint}, test_util::{ components::{HTTP_SINK_TAGS, run_and_assert_sink_compliance}, http::{always_200_response, spawn_blackhole_http_server}, @@ -22,7 +22,7 @@ async fn component_spec_compliance() { let mut config: HoneycombConfig = serde_json::from_value(HoneycombConfig::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(); diff --git a/src/sinks/humio/logs.rs b/src/sinks/humio/logs.rs index 3e72bb8b63936..38dadac847f38 100644 --- a/src/sinks/humio/logs.rs +++ b/src/sinks/humio/logs.rs @@ -19,7 +19,7 @@ use crate::{ }, logs::config::HecLogsSinkConfig, }, - util::{BatchConfig, Compression, TowerRequestConfig}, + util::{BatchConfig, Compression, HttpEndpoint, TowerRequestConfig}, }, template::Template, tls::TlsConfig, @@ -51,7 +51,7 @@ pub struct HumioLogsConfig { docs::examples = "http://127.0.0.1", docs::examples = "https://example.com", ))] - pub endpoint: String, + pub endpoint: HttpEndpoint, /// The source of events sent to this sink. /// @@ -144,8 +144,8 @@ pub struct HumioLogsConfig { pub confinement: crate::template::ConfinementConfig, } -fn default_endpoint() -> String { - HOST.to_string() +fn default_endpoint() -> HttpEndpoint { + HttpEndpoint::parse(HOST).expect("static default endpoint should be a valid http(s) URL") } pub fn timestamp_nanos_key() -> Option { @@ -406,7 +406,7 @@ mod integration_tests { HumioLogsConfig { token: token.to_string().into(), - endpoint: humio_address(), + endpoint: HttpEndpoint::parse(&humio_address()).unwrap(), source: None, encoding: JsonSerializerConfig::default().into(), event_type: None, diff --git a/src/sinks/humio/metrics.rs b/src/sinks/humio/metrics.rs index 84b045849035b..14bff23f36036 100644 --- a/src/sinks/humio/metrics.rs +++ b/src/sinks/humio/metrics.rs @@ -23,7 +23,7 @@ use crate::{ sinks::{ Healthcheck, VectorSink, splunk_hec::common::SplunkHecDefaultBatchSettings, - util::{BatchConfig, Compression, TowerRequestConfig}, + util::{BatchConfig, Compression, HttpEndpoint, TowerRequestConfig}, }, template::Template, tls::TlsConfig, @@ -67,7 +67,7 @@ pub struct HumioMetricsConfig { docs::examples = "http://127.0.0.1", docs::examples = "https://example.com", ))] - pub(super) endpoint: String, + pub(super) endpoint: HttpEndpoint, /// The source of events sent to this sink. /// @@ -143,8 +143,8 @@ pub struct HumioMetricsConfig { pub confinement: crate::template::ConfinementConfig, } -fn default_endpoint() -> String { - HOST.to_string() +fn default_endpoint() -> HttpEndpoint { + HttpEndpoint::parse(HOST).expect("static default endpoint should be a valid http(s) URL") } impl GenerateConfig for HumioMetricsConfig { @@ -274,7 +274,10 @@ mod tests { "#}) .unwrap(); - assert_eq!("https://localhost:9200/".to_string(), config.endpoint); + assert_eq!( + HttpEndpoint::parse("https://localhost:9200/").unwrap(), + config.endpoint + ); let (config, _) = load_sink::(indoc! {r#" token = "atoken" batch.max_events = 1 @@ -282,7 +285,10 @@ mod tests { "#}) .unwrap(); - assert_eq!("https://localhost:9200/".to_string(), config.endpoint); + assert_eq!( + HttpEndpoint::parse("https://localhost:9200/").unwrap(), + config.endpoint + ); } #[tokio::test] @@ -296,7 +302,7 @@ mod tests { let (_guard, addr) = test_util::addr::next_addr(); // Swap out the endpoint so we can force send it // to our local server - config.endpoint = format!("http://{addr}"); + config.endpoint = HttpEndpoint::parse(&format!("http://{addr}")).unwrap(); let (sink, _) = config.build(cx).await.unwrap(); @@ -362,7 +368,7 @@ mod tests { let (_guard, addr) = test_util::addr::next_addr(); // Swap out the endpoint so we can force send it // to our local server - config.endpoint = format!("http://{addr}"); + config.endpoint = HttpEndpoint::parse(&format!("http://{addr}")).unwrap(); let (sink, _) = config.build(cx).await.unwrap(); diff --git a/src/sinks/influxdb/logs.rs b/src/sinks/influxdb/logs.rs index 3ade45aea4f75..a17771fee26cc 100644 --- a/src/sinks/influxdb/logs.rs +++ b/src/sinks/influxdb/logs.rs @@ -26,7 +26,7 @@ use crate::{ sinks::{ Healthcheck, VectorSink, util::{ - BatchConfig, Buffer, Compression, SinkBatchSettings, TowerRequestConfig, + BatchConfig, Buffer, Compression, HttpEndpoint, SinkBatchSettings, TowerRequestConfig, http::{BatchedHttpSink, HttpEventEncoder, HttpSink}, }, }, @@ -56,7 +56,7 @@ pub struct InfluxDbLogsConfig { /// /// This should be a full HTTP URI, including the scheme, host, and port. #[configurable(metadata(docs::examples = "http://localhost:8086"))] - pub endpoint: String, + pub endpoint: HttpEndpoint, /// The list of names of log fields that should be added as tags to each measurement. /// @@ -1013,7 +1013,7 @@ mod tests { // Swap out the host so we can force send it // to our local server let host = format!("http://{addr}"); - config.endpoint = host; + config.endpoint = HttpEndpoint::parse(&host).unwrap(); let (sink, _) = config.build(cx).await.unwrap(); @@ -1153,7 +1153,7 @@ mod integration_tests { let config = InfluxDbLogsConfig { measurement: Some(measure.clone()), - endpoint: endpoint.clone(), + endpoint: HttpEndpoint::parse(&endpoint).unwrap(), tags: Default::default(), version: Some(InfluxDbVersion::V2), database: None, diff --git a/src/sinks/influxdb/metrics.rs b/src/sinks/influxdb/metrics.rs index cacde167677ed..26ea076bac0a1 100644 --- a/src/sinks/influxdb/metrics.rs +++ b/src/sinks/influxdb/metrics.rs @@ -27,7 +27,7 @@ use crate::{ influxdb_settings, }, util::{ - BatchConfig, EncodedEvent, SinkBatchSettings, TowerRequestConfig, + BatchConfig, EncodedEvent, HttpEndpoint, SinkBatchSettings, TowerRequestConfig, buffer::metrics::{MetricNormalize, MetricNormalizer, MetricSet, MetricsBuffer}, encode_namespace, http::{HttpBatchService, HttpRetryLogic}, @@ -70,7 +70,7 @@ pub struct InfluxDbConfig { /// /// This should be a full HTTP URI, including the scheme, host, and port. #[configurable(metadata(docs::examples = "http://localhost:8086/"))] - pub endpoint: String, + pub endpoint: HttpEndpoint, /// The InfluxDB API version to use. /// @@ -1190,6 +1190,7 @@ mod integration_tests { onboarding_v1, onboarding_v2, query_v1, }, }, + sinks::util::HttpEndpoint, test_util::components::{HTTP_SINK_TAGS, run_and_assert_sink_compliance}, tls::{self, TlsConfig}, }; @@ -1218,7 +1219,7 @@ mod integration_tests { let cx = SinkContext::default(); let config = InfluxDbConfig { - endpoint: url.to_string(), + endpoint: HttpEndpoint::parse(url).unwrap(), version: Some(InfluxDbVersion::V1), database: Some(database.clone()), consistency: None, @@ -1313,7 +1314,7 @@ mod integration_tests { let cx = SinkContext::default(); let config = InfluxDbConfig { - endpoint, + endpoint: HttpEndpoint::parse(&endpoint).unwrap(), version: Some(InfluxDbVersion::V2), database: None, consistency: None, diff --git a/src/sinks/influxdb/mod.rs b/src/sinks/influxdb/mod.rs index ecae579951bee..44c9c71f5b4a9 100644 --- a/src/sinks/influxdb/mod.rs +++ b/src/sinks/influxdb/mod.rs @@ -137,14 +137,14 @@ pub enum InfluxDbSettings { } trait InfluxDbConnection: std::fmt::Debug { - fn write_uri(&self, endpoint: String) -> crate::Result; - fn healthcheck_uri(&self, endpoint: String) -> crate::Result; + fn write_uri(&self, endpoint: HttpEndpoint) -> crate::Result; + fn healthcheck_uri(&self, endpoint: HttpEndpoint) -> crate::Result; fn token(&self) -> SensitiveString; fn protocol_version(&self) -> ProtocolVersion; } impl InfluxDbConnection for InfluxDb1Settings { - fn write_uri(&self, endpoint: String) -> crate::Result { + fn write_uri(&self, endpoint: HttpEndpoint) -> crate::Result { encode_uri( &endpoint, "write", @@ -159,7 +159,7 @@ impl InfluxDbConnection for InfluxDb1Settings { ) } - fn healthcheck_uri(&self, endpoint: String) -> crate::Result { + fn healthcheck_uri(&self, endpoint: HttpEndpoint) -> crate::Result { encode_uri(&endpoint, "ping", &[]) } @@ -173,7 +173,7 @@ impl InfluxDbConnection for InfluxDb1Settings { } impl InfluxDbConnection for InfluxDb2Settings { - fn write_uri(&self, endpoint: String) -> crate::Result { + fn write_uri(&self, endpoint: HttpEndpoint) -> crate::Result { encode_uri( &endpoint, "api/v2/write", @@ -185,7 +185,7 @@ impl InfluxDbConnection for InfluxDb2Settings { ) } - fn healthcheck_uri(&self, endpoint: String) -> crate::Result { + fn healthcheck_uri(&self, endpoint: HttpEndpoint) -> crate::Result { encode_uri(&endpoint, "ping", &[]) } @@ -208,7 +208,7 @@ fn influxdb_settings(settings: InfluxDbSettings) -> Box // V1: https://docs.influxdata.com/influxdb/v1.7/tools/api/#ping-http-endpoint // V2: https://v2.docs.influxdata.com/v2.0/api/#operation/GetHealth fn healthcheck( - endpoint: String, + endpoint: HttpEndpoint, settings: InfluxDbSettings, mut client: HttpClient, ) -> crate::Result { @@ -354,7 +354,7 @@ pub(in crate::sinks) fn encode_timestamp(timestamp: Option>) -> i6 } pub(in crate::sinks) fn encode_uri( - endpoint: &str, + endpoint: &HttpEndpoint, path: &str, pairs: &[(&str, Option)], ) -> crate::Result { @@ -372,7 +372,7 @@ pub(in crate::sinks) fn encode_uri( } else { format!("{path}?{query}") }; - Ok(HttpEndpoint::parse(endpoint)?.append_path(&path_and_query)?.into_uri()) + Ok(endpoint.append_path(&path_and_query)?.into_uri()) } #[cfg(test)] @@ -605,7 +605,7 @@ mod tests { }; let uri = settings - .write_uri("http://localhost:8086".to_owned()) + .write_uri(HttpEndpoint::parse("http://localhost:8086").unwrap()) .unwrap(); assert_eq!( "http://localhost:8086/write?consistency=quorum&db=vector_db&rp=autogen&p=secret&u=writer&precision=ns", @@ -622,7 +622,7 @@ mod tests { }; let uri = settings - .write_uri("http://localhost:9999".to_owned()) + .write_uri(HttpEndpoint::parse("http://localhost:9999").unwrap()) .unwrap(); assert_eq!( "http://localhost:9999/api/v2/write?org=my-org&bucket=my-bucket&precision=ns", @@ -641,7 +641,7 @@ mod tests { }; let uri = settings - .healthcheck_uri("http://localhost:8086".to_owned()) + .healthcheck_uri(HttpEndpoint::parse("http://localhost:8086").unwrap()) .unwrap(); assert_eq!("http://localhost:8086/ping", uri.to_string()) } @@ -655,7 +655,7 @@ mod tests { }; let uri = settings - .healthcheck_uri("http://localhost:9999".to_owned()) + .healthcheck_uri(HttpEndpoint::parse("http://localhost:9999").unwrap()) .unwrap(); assert_eq!("http://localhost:9999/ping", uri.to_string()) } @@ -808,7 +808,7 @@ mod tests { #[test] fn test_encode_uri_valid() { let uri = encode_uri( - "http://localhost:9999", + &HttpEndpoint::parse("http://localhost:9999").unwrap(), "api/v2/write", &[ ("org", Some("my-org".to_owned())), @@ -823,7 +823,7 @@ mod tests { ); let uri = encode_uri( - "http://localhost:9999/", + &HttpEndpoint::parse("http://localhost:9999/").unwrap(), "api/v2/write", &[ ("org", Some("my-org".to_owned())), @@ -837,7 +837,7 @@ mod tests { ); let uri = encode_uri( - "http://localhost:9999", + &HttpEndpoint::parse("http://localhost:9999").unwrap(), "api/v2/write", &[ ("org", Some("Organization name".to_owned())), @@ -854,15 +854,7 @@ mod tests { #[test] fn test_encode_uri_invalid() { - encode_uri( - "localhost:9999", - "api/v2/write", - &[ - ("org", Some("my-org".to_owned())), - ("bucket", Some("my-bucket".to_owned())), - ], - ) - .unwrap_err(); + assert!(HttpEndpoint::parse("localhost:9999").is_err()); } } @@ -876,6 +868,7 @@ mod integration_tests { InfluxDb1Settings, InfluxDb2Settings, InfluxDbSettings, healthcheck, test_util::{BUCKET, ORG, TOKEN, address_v1, address_v2, next_database, onboarding_v2}, }, + sinks::util::HttpEndpoint, }; #[tokio::test] @@ -892,7 +885,7 @@ mod integration_tests { let proxy = ProxyConfig::default(); let client = HttpClient::new(None, &proxy).unwrap(); - healthcheck(endpoint, settings, client) + healthcheck(HttpEndpoint::parse(&endpoint).unwrap(), settings, client) .unwrap() .await .unwrap() @@ -912,7 +905,7 @@ mod integration_tests { let proxy = ProxyConfig::default(); let client = HttpClient::new(None, &proxy).unwrap(); - healthcheck(endpoint, settings, client) + healthcheck(HttpEndpoint::parse(&endpoint).unwrap(), settings, client) .unwrap() .await .unwrap(); @@ -932,7 +925,7 @@ mod integration_tests { let proxy = ProxyConfig::default(); let client = HttpClient::new(None, &proxy).unwrap(); - healthcheck(endpoint, settings, client) + healthcheck(HttpEndpoint::parse(&endpoint).unwrap(), settings, client) .unwrap() .await .unwrap(); @@ -952,7 +945,7 @@ mod integration_tests { let proxy = ProxyConfig::default(); let client = HttpClient::new(None, &proxy).unwrap(); - healthcheck(endpoint, settings, client) + healthcheck(HttpEndpoint::parse(&endpoint).unwrap(), settings, client) .unwrap() .await .unwrap(); diff --git a/src/sinks/prometheus/remote_write/config.rs b/src/sinks/prometheus/remote_write/config.rs index 0b79b12452767..d1a32222917cc 100644 --- a/src/sinks/prometheus/remote_write/config.rs +++ b/src/sinks/prometheus/remote_write/config.rs @@ -14,10 +14,10 @@ use crate::{ prelude::*, prometheus::PrometheusRemoteWriteAuth, util::{ + HttpEndpoint, auth::Auth, http::{OrderedHeaderName, RetryStrategy, http_response_retry_logic}, service::TowerRequestConfig, - HttpEndpoint, }, }, template::ConfinementConfig, @@ -51,7 +51,8 @@ pub struct RemoteWriteConfig { /// /// The endpoint should include the scheme and the path to write to. #[configurable(metadata(docs::examples = "https://localhost:8087/api/v1/write"))] - pub endpoint: String, + #[derivative(Default(value = "default_endpoint()"))] + pub endpoint: HttpEndpoint, /// The default namespace for any metrics sent. /// @@ -137,6 +138,10 @@ const fn default_compression() -> Compression { Compression::Snappy } +fn default_endpoint() -> HttpEndpoint { + HttpEndpoint::parse("https://localhost:8087/api/v1/write").unwrap() +} + impl_generate_config_from_default!(RemoteWriteConfig); /// Outbound HTTP request settings for the Prometheus remote write sink. @@ -196,7 +201,7 @@ impl SinkConfig for RemoteWriteConfig { }) .transpose()?; - let endpoint = HttpEndpoint::parse(&self.endpoint)?; + let endpoint = self.endpoint.clone(); let tls_settings = TlsSettings::from_options(self.tls.as_ref())?; let request_settings = self.request.tower.into_settings(); let validated_headers = Arc::new(validate_headers( diff --git a/src/sinks/prometheus/remote_write/integration_tests.rs b/src/sinks/prometheus/remote_write/integration_tests.rs index 25de1d2e3bac4..94f768895f9d9 100644 --- a/src/sinks/prometheus/remote_write/integration_tests.rs +++ b/src/sinks/prometheus/remote_write/integration_tests.rs @@ -9,6 +9,7 @@ use crate::{ sinks::{ influxdb::test_util::{cleanup_v1, format_timestamp, onboarding_v1, query_v1}, prometheus::remote_write::config::RemoteWriteConfig, + util::HttpEndpoint, }, test_util::components::{HTTP_SINK_TAGS, assert_sink_compliance}, tls::{self, TlsConfig}, @@ -34,7 +35,8 @@ async fn insert_metrics(url: &str) { let cx = SinkContext::default(); let config = RemoteWriteConfig { - endpoint: format!("{url}/api/v1/prom/write?db={database}"), + endpoint: HttpEndpoint::parse(&format!("{url}/api/v1/prom/write?db={database}")) + .unwrap(), tls: Some(TlsConfig { ca_file: Some(tls::TEST_PEM_CA_PATH.into()), ..Default::default() diff --git a/src/sinks/prometheus/remote_write/service.rs b/src/sinks/prometheus/remote_write/service.rs index dabe1713a8e9b..d085278868a2a 100644 --- a/src/sinks/prometheus/remote_write/service.rs +++ b/src/sinks/prometheus/remote_write/service.rs @@ -18,9 +18,9 @@ use crate::{ sinks::{ prelude::*, util::{ + HttpEndpoint, auth::Auth, http::{HttpResponse, OrderedHeaderName}, - HttpEndpoint, }, }, }; diff --git a/src/sinks/sematext/metrics.rs b/src/sinks/sematext/metrics.rs index ef57a1cb8a57b..853014a2ae811 100644 --- a/src/sinks/sematext/metrics.rs +++ b/src/sinks/sematext/metrics.rs @@ -103,7 +103,9 @@ impl GenerateConfig for SematextMetricsConfig { } async fn healthcheck(endpoint: String, client: HttpClient) -> Result<()> { - let uri = HttpEndpoint::parse(&endpoint)?.append_path("health")?.into_uri(); + let uri = HttpEndpoint::parse(&endpoint)? + .append_path("health")? + .into_uri(); let request = Request::get(uri) .body(Body::empty()) @@ -135,6 +137,7 @@ impl SinkConfig for SematextMetricsConfig { }; let healthcheck = healthcheck(endpoint.clone(), client.clone()).boxed(); + let endpoint = HttpEndpoint::parse(&endpoint)?; let sink = SematextMetricsService::new(self.clone(), write_uri(&endpoint)?, client)?; Ok((sink, healthcheck)) @@ -149,7 +152,7 @@ impl SinkConfig for SematextMetricsConfig { } } -fn write_uri(endpoint: &str) -> Result { +fn write_uri(endpoint: &HttpEndpoint) -> Result { encode_uri( endpoint, "write", diff --git a/src/sinks/splunk_hec/logs/config.rs b/src/sinks/splunk_hec/logs/config.rs index 4d0b6d5eb92b6..55a55f79092b3 100644 --- a/src/sinks/splunk_hec/logs/config.rs +++ b/src/sinks/splunk_hec/logs/config.rs @@ -17,7 +17,7 @@ use crate::{ build_healthcheck, build_http_batch_service, create_client, service::{HecService, HttpRequestBuilder}, }, - util::http::HttpRetryLogic, + util::{HttpEndpoint, http::HttpRetryLogic}, }, template::ConfinementConfig, }; @@ -48,7 +48,7 @@ pub struct HecLogsSinkConfig { docs::examples = "http://example.com" ))] #[configurable(validation(format = "uri"))] - pub endpoint: String, + pub endpoint: HttpEndpoint, /// Overrides the name of the log field used to retrieve the hostname to send to Splunk HEC. /// @@ -160,7 +160,7 @@ impl GenerateConfig for HecLogsSinkConfig { fn generate_config() -> serde_json::Value { serde_json::to_value(Self { default_token: "${VECTOR_SPLUNK_HEC_TOKEN}".to_owned().into(), - endpoint: "endpoint".to_owned(), + endpoint: HttpEndpoint::parse("http://example.com").unwrap(), host_key: None, indexed_fields: vec![], index: None, @@ -214,7 +214,7 @@ impl HecLogsSinkConfig { let client = create_client(self.tls.as_ref(), cx.proxy())?; let healthcheck = build_healthcheck( - self.endpoint.clone(), + self.endpoint.clone().into(), self.default_token.inner().to_owned(), client.clone(), ) @@ -275,7 +275,7 @@ impl HecLogsSinkConfig { let request_settings = self.request.into_settings(); let http_request_builder = Arc::new(HttpRequestBuilder::new( - self.endpoint.clone(), + self.endpoint.clone().into(), self.endpoint_target, self.default_token.inner().to_owned(), self.compression, @@ -365,7 +365,7 @@ mod tests { impl ValidatableComponent for HecLogsSinkConfig { fn validation_configuration() -> ValidationConfiguration { - let endpoint = "http://127.0.0.1:9001".to_string(); + let endpoint = HttpEndpoint::parse("http://127.0.0.1:9001").unwrap(); let mut batch = BatchConfig::default(); batch.max_events = Some(1); @@ -405,14 +405,14 @@ mod tests { confinement: ConfinementConfig::default(), }; - let endpoint = format!("{endpoint}/services/collector/raw"); + let endpoint = endpoint + .append_path("services/collector/raw") + .unwrap() + .into_uri(); let external_resource = ExternalResource::new( ResourceDirection::Push, - HttpResourceConfig::from_parts( - http::Uri::try_from(&endpoint).expect("should not fail to parse URI"), - None, - ), + HttpResourceConfig::from_parts(endpoint, None), config.encoding.clone(), ); diff --git a/src/sinks/splunk_hec/logs/integration_tests.rs b/src/sinks/splunk_hec/logs/integration_tests.rs index 475ee1c49460e..7d4c551f4d9ae 100644 --- a/src/sinks/splunk_hec/logs/integration_tests.rs +++ b/src/sinks/splunk_hec/logs/integration_tests.rs @@ -26,7 +26,7 @@ use crate::{ }, logs::config::HecLogsSinkConfig, }, - util::{BatchConfig, Compression, TowerRequestConfig}, + util::{BatchConfig, Compression, HttpEndpoint, TowerRequestConfig}, }, template::Template, test_util::{ @@ -120,7 +120,7 @@ async fn config( HecLogsSinkConfig { default_token: get_token().await.into(), - endpoint: splunk_hec_address(), + endpoint: HttpEndpoint::parse(&splunk_hec_address()).unwrap(), host_key: Some(OptionalTargetPath::event("host")), indexed_fields, index: None, diff --git a/src/sinks/splunk_hec/logs/tests.rs b/src/sinks/splunk_hec/logs/tests.rs index 96118577094f2..d23f3d6059fb1 100644 --- a/src/sinks/splunk_hec/logs/tests.rs +++ b/src/sinks/splunk_hec/logs/tests.rs @@ -22,7 +22,7 @@ use crate::{ logs::{config::HecLogsSinkConfig, encoder::HecLogsEncoder, sink::process_log}, }, util::{ - Compression, encoding::Encoder as _, processed_event::ProcessedEvent, + Compression, HttpEndpoint, encoding::Encoder as _, processed_event::ProcessedEvent, test::build_test_server, }, }, @@ -241,7 +241,7 @@ async fn splunk_passthrough_token() { let (_guard, addr) = next_addr(); let config = HecLogsSinkConfig { default_token: "token".to_string().into(), - endpoint: format!("http://{addr}"), + endpoint: HttpEndpoint::parse(&format!("http://{addr}")).unwrap(), host_key: None, indexed_fields: Vec::new(), index: None, diff --git a/src/sinks/splunk_hec/metrics/config.rs b/src/sinks/splunk_hec/metrics/config.rs index 5a939f3059ca7..e22c2c77611c7 100644 --- a/src/sinks/splunk_hec/metrics/config.rs +++ b/src/sinks/splunk_hec/metrics/config.rs @@ -21,7 +21,8 @@ use crate::{ service::{HecService, HttpRequestBuilder}, }, util::{ - BatchConfig, Compression, ServiceBuilderExt, TowerRequestConfig, http::HttpRetryLogic, + BatchConfig, Compression, HttpEndpoint, ServiceBuilderExt, TowerRequestConfig, + http::HttpRetryLogic, }, }, template::{ConfinedTemplate, Template}, @@ -65,7 +66,7 @@ pub struct HecMetricsSinkConfig { docs::examples = "http://example.com" ))] #[configurable(validation(format = "uri"))] - pub endpoint: String, + pub endpoint: HttpEndpoint, /// Overrides the name of the log field used to retrieve the hostname to send to Splunk HEC. /// @@ -128,7 +129,7 @@ impl GenerateConfig for HecMetricsSinkConfig { serde_json::to_value(Self { default_namespace: None, default_token: "${VECTOR_SPLUNK_HEC_TOKEN}".to_owned().into(), - endpoint: "http://localhost:8088".to_owned(), + endpoint: HttpEndpoint::parse("http://localhost:8088").unwrap(), host_key: config_host_key(), index: None, sourcetype: None, @@ -169,7 +170,7 @@ impl SinkConfig for HecMetricsSinkConfig { let client = create_client(self.tls.as_ref(), cx.proxy())?; let healthcheck = build_healthcheck( - self.endpoint.clone(), + self.endpoint.clone().into(), self.default_token.inner().to_owned(), client.clone(), ) @@ -232,7 +233,7 @@ impl HecMetricsSinkConfig { let request_settings = self.request.into_settings(); let http_request_builder = Arc::new(HttpRequestBuilder::new( - self.endpoint.clone(), + self.endpoint.clone().into(), EndpointTarget::default(), self.default_token.inner().to_owned(), self.compression, diff --git a/src/sinks/splunk_hec/metrics/integration_tests.rs b/src/sinks/splunk_hec/metrics/integration_tests.rs index bebde630994d1..21a01a1f7d2ea 100644 --- a/src/sinks/splunk_hec/metrics/integration_tests.rs +++ b/src/sinks/splunk_hec/metrics/integration_tests.rs @@ -17,7 +17,7 @@ use crate::{ splunk_hec::common::integration_test_helpers::{ get_token, splunk_api_address, splunk_hec_address, }, - util::{BatchConfig, Compression, TowerRequestConfig}, + util::{BatchConfig, Compression, HttpEndpoint, TowerRequestConfig}, }, template::Template, test_util::components::{ @@ -36,7 +36,7 @@ async fn config() -> HecMetricsSinkConfig { HecMetricsSinkConfig { default_namespace: None, default_token: get_token().await.into(), - endpoint: splunk_hec_address(), + endpoint: HttpEndpoint::parse(&splunk_hec_address()).unwrap(), host_key: OptionalValuePath::new("host"), index: None, sourcetype: None, diff --git a/src/sinks/splunk_hec/metrics/tests.rs b/src/sinks/splunk_hec/metrics/tests.rs index aa249f78c51fd..0cc8400606bb0 100644 --- a/src/sinks/splunk_hec/metrics/tests.rs +++ b/src/sinks/splunk_hec/metrics/tests.rs @@ -18,7 +18,7 @@ use crate::{ common::config_host_key, metrics::{config::HecMetricsSinkConfig, encoder::HecMetricsEncoder}, }, - util::{Compression, test::build_test_server}, + util::{Compression, HttpEndpoint, test::build_test_server}, }, template::{ConfinementConfig, Template}, test_util::addr::next_addr, @@ -352,7 +352,7 @@ async fn splunk_passthrough_token() { let (_guard, addr) = next_addr(); let config = HecMetricsSinkConfig { default_token: "token".to_owned().into(), - endpoint: format!("http://{addr}"), + endpoint: HttpEndpoint::parse(&format!("http://{addr}")).unwrap(), host_key: config_host_key(), index: None, sourcetype: None, diff --git a/src/sinks/util/uri.rs b/src/sinks/util/uri.rs index 9eafce544393a..8183a44878884 100644 --- a/src/sinks/util/uri.rs +++ b/src/sinks/util/uri.rs @@ -233,13 +233,33 @@ pub enum HttpEndpointError { /// through `HttpClient` need this invariant, since `HttpClient` rejects such /// URIs at request time, deferring a pure configuration error to runtime. /// +/// As a configuration type it deserializes from a string, so an invalid +/// endpoint is rejected at config load time with the config path in the error. +/// /// Path composition goes through [`HttpEndpoint::append_path`], which /// manipates `Uri` parts directly instead of string-concatenating and /// re-parsing, so the scheme and authority are preserved and the result is /// still an absolute `http(s)` URL. +#[configurable_component] +#[configurable(title = "An absolute http(s) URL.", description = "")] #[derive(Debug, Clone, PartialEq, Eq)] +#[serde(try_from = "String", into = "String")] pub struct HttpEndpoint(Uri); +impl TryFrom for HttpEndpoint { + type Error = HttpEndpointError; + + fn try_from(value: String) -> Result { + Self::parse(&value) + } +} + +impl From for String { + fn from(value: HttpEndpoint) -> Self { + value.to_string() + } +} + impl HttpEndpoint { /// Requires `uri` to be an absolute `http`/`https` URL. pub fn new(uri: Uri) -> Result { @@ -292,14 +312,10 @@ impl HttpEndpoint { } else { format!("{base_path}/{path}") }; - parts.path_and_query = Some( - joined - .parse::() - .context(InvalidPathSnafu { - endpoint: self.0.to_string(), - path: joined, - })?, - ); + parts.path_and_query = Some(joined.parse::().context(InvalidPathSnafu { + endpoint: self.0.to_string(), + path: joined, + })?); let uri = Uri::from_parts(parts).context(InvalidUriPartsSnafu { endpoint: self.0.to_string(), })?; @@ -393,8 +409,12 @@ mod tests { "http://127.0.0.1:9000/endpoint?query=1", "https://user:pass@example.com/path", ] { - let endpoint = HttpEndpoint::parse(endpoint).expect("should accept absolute http(s) URL"); - assert!(matches!(endpoint.as_uri().scheme_str(), Some("http" | "https"))); + let endpoint = + HttpEndpoint::parse(endpoint).expect("should accept absolute http(s) URL"); + assert!(matches!( + endpoint.as_uri().scheme_str(), + Some("http" | "https") + )); assert!(endpoint.as_uri().authority().is_some()); } } @@ -460,7 +480,10 @@ mod tests { .unwrap() .append_path("sub/path") .unwrap(); - assert_eq!(appended.to_string(), "https://user:pass@example.com:8088/base/sub/path"); + assert_eq!( + appended.to_string(), + "https://user:pass@example.com:8088/base/sub/path" + ); assert!(matches!(appended.as_uri().scheme_str(), Some("https"))); } } diff --git a/src/sources/prometheus/remote_write.rs b/src/sources/prometheus/remote_write.rs index c2019aa875b12..3b39194e1e8e9 100644 --- a/src/sources/prometheus/remote_write.rs +++ b/src/sources/prometheus/remote_write.rs @@ -220,7 +220,7 @@ mod test { use crate::{ SourceSender, config::{SinkConfig, SinkContext}, - sinks::prometheus::remote_write::RemoteWriteConfig, + sinks::{prometheus::remote_write::RemoteWriteConfig, util::HttpEndpoint}, test_util::{self, wait_for_tcp}, tls::MaybeTlsSettings, }; @@ -265,7 +265,8 @@ mod test { wait_for_tcp(address).await; let sink = RemoteWriteConfig { - endpoint: format!("{}://localhost:{}/", proto, address.port()), + endpoint: HttpEndpoint::parse(&format!("{}://localhost:{}/", proto, address.port())) + .unwrap(), tls: tls.map(|tls| tls.options), ..Default::default() }; @@ -461,7 +462,8 @@ mod test { wait_for_tcp(address).await; let sink = RemoteWriteConfig { - endpoint: format!("http://localhost:{}/", address.port()), + endpoint: HttpEndpoint::parse(&format!("http://localhost:{}/", address.port())) + .unwrap(), ..Default::default() }; let (sink, _) = sink @@ -689,7 +691,11 @@ mod test { wait_for_tcp(address).await; let sink = RemoteWriteConfig { - endpoint: format!("http://localhost:{}/api/v1/write", address.port()), + endpoint: HttpEndpoint::parse(&format!( + "http://localhost:{}/api/v1/write", + address.port() + )) + .unwrap(), ..Default::default() }; let (sink, _) = sink diff --git a/src/sources/splunk_hec/mod.rs b/src/sources/splunk_hec/mod.rs index de56e5afbb668..0ae060cf450da 100644 --- a/src/sources/splunk_hec/mod.rs +++ b/src/sources/splunk_hec/mod.rs @@ -2081,7 +2081,7 @@ mod tests { sinks::{ Healthcheck, VectorSink, splunk_hec::logs::config::HecLogsSinkConfig, - util::{BatchConfig, Compression, TowerRequestConfig}, + util::{BatchConfig, Compression, HttpEndpoint, TowerRequestConfig}, }, sources::splunk_hec::acknowledgements::{HecAckStatusRequest, HecAckStatusResponse}, test_util::{ @@ -2171,7 +2171,7 @@ mod tests { ) -> (VectorSink, Healthcheck) { HecLogsSinkConfig { default_token: TOKEN.to_owned().into(), - endpoint: format!("http://{address}"), + endpoint: HttpEndpoint::parse(&format!("http://{address}")).unwrap(), host_key: None, indexed_fields: vec![], index: None, diff --git a/website/cue/reference/components/sinks/generated/appsignal.cue b/website/cue/reference/components/sinks/generated/appsignal.cue index 41769e4ec2a41..786414bf49824 100644 --- a/website/cue/reference/components/sinks/generated/appsignal.cue +++ b/website/cue/reference/components/sinks/generated/appsignal.cue @@ -128,7 +128,7 @@ generated: components: sinks: appsignal: configuration: { description: "The URI for the AppSignal API to send data to." required: false type: string: { - default: "https://appsignal-endpoint.net" + default: "https://appsignal-endpoint.net/" examples: ["https://appsignal-endpoint.net"] } } diff --git a/website/cue/reference/components/sinks/generated/gcp_cloud_storage.cue b/website/cue/reference/components/sinks/generated/gcp_cloud_storage.cue index 0bbbb6b1141fc..7a49130aae884 100644 --- a/website/cue/reference/components/sinks/generated/gcp_cloud_storage.cue +++ b/website/cue/reference/components/sinks/generated/gcp_cloud_storage.cue @@ -680,7 +680,7 @@ generated: components: sinks: gcp_cloud_storage: configuration: { description: "API endpoint for Google Cloud Storage" required: false type: string: { - default: "https://storage.googleapis.com" + default: "https://storage.googleapis.com/" examples: ["http://localhost:9000"] } } diff --git a/website/cue/reference/components/sinks/generated/gcp_pubsub.cue b/website/cue/reference/components/sinks/generated/gcp_pubsub.cue index 3d57f6c43bf83..4c824e03e0cb2 100644 --- a/website/cue/reference/components/sinks/generated/gcp_pubsub.cue +++ b/website/cue/reference/components/sinks/generated/gcp_pubsub.cue @@ -548,7 +548,7 @@ generated: components: sinks: gcp_pubsub: configuration: { """ required: false type: string: { - default: "https://pubsub.googleapis.com" + default: "https://pubsub.googleapis.com/" examples: ["https://us-central1-pubsub.googleapis.com"] } } diff --git a/website/cue/reference/components/sinks/generated/honeycomb.cue b/website/cue/reference/components/sinks/generated/honeycomb.cue index 5c3acaed97ffa..14fa23ccab5ee 100644 --- a/website/cue/reference/components/sinks/generated/honeycomb.cue +++ b/website/cue/reference/components/sinks/generated/honeycomb.cue @@ -131,7 +131,7 @@ generated: components: sinks: honeycomb: configuration: { description: "Honeycomb's endpoint to send logs to" required: false type: string: { - default: "https://api.honeycomb.io" + default: "https://api.honeycomb.io/" examples: ["https://api.honeycomb.io", "https://api.eu1.honeycomb.io"] } } diff --git a/website/cue/reference/components/sinks/generated/humio_logs.cue b/website/cue/reference/components/sinks/generated/humio_logs.cue index f54dc7b4adfed..987ebaf24f51a 100644 --- a/website/cue/reference/components/sinks/generated/humio_logs.cue +++ b/website/cue/reference/components/sinks/generated/humio_logs.cue @@ -561,7 +561,7 @@ generated: components: sinks: humio_logs: configuration: { """ required: false type: string: { - default: "https://cloud.humio.com" + default: "https://cloud.humio.com/" examples: ["http://127.0.0.1", "https://example.com"] } } diff --git a/website/cue/reference/components/sinks/generated/humio_metrics.cue b/website/cue/reference/components/sinks/generated/humio_metrics.cue index 6621997a2f7a1..1c7ec12590c0f 100644 --- a/website/cue/reference/components/sinks/generated/humio_metrics.cue +++ b/website/cue/reference/components/sinks/generated/humio_metrics.cue @@ -120,7 +120,7 @@ generated: components: sinks: humio_metrics: configuration: { """ required: false type: string: { - default: "https://cloud.humio.com" + default: "https://cloud.humio.com/" examples: ["http://127.0.0.1", "https://example.com"] } } From 4fd530dcbf36e14e4ff814b79e6c83ec094544ba Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 14:58:20 -0400 Subject: [PATCH 03/16] chore(changelog): note sink endpoint validation --- changelog.d/sink_endpoint_absolute_urls.enhancement.md | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 changelog.d/sink_endpoint_absolute_urls.enhancement.md diff --git a/changelog.d/sink_endpoint_absolute_urls.enhancement.md b/changelog.d/sink_endpoint_absolute_urls.enhancement.md new file mode 100644 index 0000000000000..a62480089de8a --- /dev/null +++ b/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 From e45a1a414f58890a7fd0ee11ba4d701dfcc2d4e7 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 15:09:35 -0400 Subject: [PATCH 04/16] Require a non-empty host in HttpEndpoint validation --- src/sinks/util/uri.rs | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/src/sinks/util/uri.rs b/src/sinks/util/uri.rs index 8183a44878884..9f27995bb3e44 100644 --- a/src/sinks/util/uri.rs +++ b/src/sinks/util/uri.rs @@ -261,9 +261,15 @@ impl From for String { } impl HttpEndpoint { - /// Requires `uri` to be an absolute `http`/`https` URL. + /// Requires `uri` to be an absolute `http`/`https` URL with a host. + /// + /// The authority check alone is not enough: `http://:8080` parses as a + /// valid `http::Uri` with an authority but an empty host, so the host is + /// checked explicitly. pub fn new(uri: Uri) -> Result { - if matches!(uri.scheme_str(), Some("http" | "https")) && uri.authority().is_some() { + if matches!(uri.scheme_str(), Some("http" | "https")) + && uri.host().is_some_and(|host| !host.is_empty()) + { Ok(Self(uri)) } else { Err(HttpEndpointError::NotAbsoluteHttp { @@ -408,6 +414,9 @@ mod tests { "https://example.com:8088/services/collector", "http://127.0.0.1:9000/endpoint?query=1", "https://user:pass@example.com/path", + // IPv6 hosts are returned bracketed (`[::1]`) and must be accepted. + "http://[::1]:8080", + "https://[::1]/path", ] { let endpoint = HttpEndpoint::parse(endpoint).expect("should accept absolute http(s) URL"); @@ -432,6 +441,10 @@ mod tests { "unix:///var/run/vector.sock", // Scheme but no authority. "http:///path", + // Authority with a port but an empty host: `http::Uri` parses this + // with `authority() == Some` and `host() == Some("")`. + "http://:8080", + "http://:8080/path", ] { assert!( matches!( From b12d37949d0843d14c3ca32111d766b1d2a7b94f Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 15:33:24 -0400 Subject: [PATCH 05/16] Default a missing endpoint scheme to https in HttpEndpoint --- ...sink_endpoint_absolute_urls.enhancement.md | 2 +- src/sinks/influxdb/mod.rs | 5 -- src/sinks/util/uri.rs | 59 +++++++++++++++++-- 3 files changed, 54 insertions(+), 12 deletions(-) diff --git a/changelog.d/sink_endpoint_absolute_urls.enhancement.md b/changelog.d/sink_endpoint_absolute_urls.enhancement.md index a62480089de8a..1d9c396cdeedc 100644 --- a/changelog.d/sink_endpoint_absolute_urls.enhancement.md +++ b/changelog.d/sink_endpoint_absolute_urls.enhancement.md @@ -1,3 +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. +Sink `endpoint` options now require an absolute URL that includes a host. Endpoints without a scheme are defaulted to `https://` (for example `endpoint: "localhost:8080"` becomes `https://localhost:8080`). 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. Empty, host-less, or non-`http(s)` endpoints (for example `endpoint: ""`, `endpoint: "/path"`, or `endpoint: "ftp://example.com"`) are now rejected at configuration load with a clear error, including with `vector validate --no-environment`. 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 diff --git a/src/sinks/influxdb/mod.rs b/src/sinks/influxdb/mod.rs index 44c9c71f5b4a9..ed29c3e4c40d8 100644 --- a/src/sinks/influxdb/mod.rs +++ b/src/sinks/influxdb/mod.rs @@ -851,11 +851,6 @@ mod tests { "http://localhost:9999/api/v2/write?org=Organization+name&bucket=Bucket%3Dname" ); } - - #[test] - fn test_encode_uri_invalid() { - assert!(HttpEndpoint::parse("localhost:9999").is_err()); - } } #[cfg(feature = "influxdb-integration-tests")] diff --git a/src/sinks/util/uri.rs b/src/sinks/util/uri.rs index 9f27995bb3e44..8112e4409a104 100644 --- a/src/sinks/util/uri.rs +++ b/src/sinks/util/uri.rs @@ -279,10 +279,25 @@ impl HttpEndpoint { } /// Parses `endpoint` and requires it to be an absolute `http`/`https` URL. + /// + /// A missing scheme is defaulted to `https`, so `example.com:8080` becomes + /// `https://example.com:8080`. An explicit `http`/`https` scheme is + /// preserved. Endpoints that still lack a host after defaulting (for + /// example `/path`) are rejected. pub fn parse(endpoint: &str) -> Result { - let uri = endpoint - .parse::() - .context(InvalidUriSnafu { endpoint })?; + // Default a missing scheme to https. `http::Uri` cannot parse + // `host:port/path` without a scheme (it reads `host` as a scheme), so + // the scheme is added up front rather than relying on the parser to + // accept authority-form input. + let uri = if endpoint.contains("://") { + endpoint + .parse::() + .context(InvalidUriSnafu { endpoint })? + } else { + format!("https://{endpoint}") + .parse::() + .context(InvalidUriSnafu { endpoint })? + }; Self::new(uri) } @@ -417,6 +432,11 @@ mod tests { // IPv6 hosts are returned bracketed (`[::1]`) and must be accepted. "http://[::1]:8080", "https://[::1]/path", + // A missing scheme is defaulted to https. + "example.com", + "example.com:8088/services/collector", + "localhost:8080", + "[::1]:8080", ] { let endpoint = HttpEndpoint::parse(endpoint).expect("should accept absolute http(s) URL"); @@ -428,12 +448,38 @@ mod tests { } } + #[test] + fn http_endpoint_defaults_missing_scheme_to_https() { + for endpoint in [ + "example.com", + "example.com:8080", + "localhost:8080/path", + "[::1]:8080", + ] { + let endpoint = + HttpEndpoint::parse(endpoint).expect("should default a missing scheme to https"); + assert_eq!(endpoint.as_uri().scheme_str(), Some("https")); + assert!( + endpoint + .as_uri() + .host() + .is_some_and(|host| !host.is_empty()) + ); + } + // An explicit scheme is preserved. + assert_eq!( + HttpEndpoint::parse("http://example.com") + .unwrap() + .as_uri() + .scheme_str(), + Some("http") + ); + } + #[test] fn http_endpoint_rejects_non_absolute_http_urls() { for endpoint in [ - // No scheme: `http::Uri` parses these as authority-form or as a path. - "example.com:8088", - "localhost:8080", + // No scheme and no host: `http::Uri` parses these as a path. "/services/collector", "", // Absolute, but not a scheme `HttpClient` can dial. @@ -451,6 +497,7 @@ mod tests { HttpEndpoint::parse(endpoint), Err(HttpEndpointError::NotAbsoluteHttp { .. }) | Err(HttpEndpointError::InvalidUri { .. }) + | Err(HttpEndpointError::InvalidUriParts { .. }) ), "expected `{endpoint}` to be rejected" ); From 8cb4c9ec7ab39af1c029da5537169f81a9219adb Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 15:39:19 -0400 Subject: [PATCH 06/16] chore(changelog): narrow endpoint validation claim to load-time sinks --- changelog.d/sink_endpoint_absolute_urls.enhancement.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/changelog.d/sink_endpoint_absolute_urls.enhancement.md b/changelog.d/sink_endpoint_absolute_urls.enhancement.md index 1d9c396cdeedc..dac03644e35e4 100644 --- a/changelog.d/sink_endpoint_absolute_urls.enhancement.md +++ b/changelog.d/sink_endpoint_absolute_urls.enhancement.md @@ -1,3 +1,3 @@ -Sink `endpoint` options now require an absolute URL that includes a host. Endpoints without a scheme are defaulted to `https://` (for example `endpoint: "localhost:8080"` becomes `https://localhost:8080`). 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. Empty, host-less, or non-`http(s)` endpoints (for example `endpoint: ""`, `endpoint: "/path"`, or `endpoint: "ftp://example.com"`) are now rejected at configuration load with a clear error, including with `vector validate --no-environment`. 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. +Sink `endpoint` options now require an absolute URL that includes a host. Endpoints without a scheme are defaulted to `https://` (for example `endpoint: "localhost:8080"` becomes `https://localhost:8080`). 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. Empty, host-less, or non-`http(s)` endpoints (for example `endpoint: ""`, `endpoint: "/path"`, or `endpoint: "ftp://example.com"`) are now rejected with a clear error. This happens at configuration load for most sinks, including with `vector validate --no-environment`; for the `datadog_metrics`, `datadog_traces`, and `sematext` sinks it happens when the sink is built. 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 From 17d819d3f6c554bc63df01f7fcdef633c043ebf9 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 15:40:07 -0400 Subject: [PATCH 07/16] fix(pubsub sink): preserve :publish method suffix on topic path --- src/sinks/gcp/pubsub.rs | 4 +++- src/sinks/util/uri.rs | 40 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/src/sinks/gcp/pubsub.rs b/src/sinks/gcp/pubsub.rs index 531cc9df6b206..00897b8c75b10 100644 --- a/src/sinks/gcp/pubsub.rs +++ b/src/sinks/gcp/pubsub.rs @@ -184,7 +184,9 @@ impl PubsubSink { } fn uri(&self, suffix: &str) -> crate::Result { - let mut uri = self.uri_base.append_path(suffix)?.into_uri(); + // The suffix is a Google API method (for example `:publish`) that + // attaches directly to the topic path without a separator. + let mut uri = self.uri_base.append_raw_suffix(suffix)?.into_uri(); self.auth.apply_uri(&mut uri); Ok(uri) } diff --git a/src/sinks/util/uri.rs b/src/sinks/util/uri.rs index 8112e4409a104..fa7e229e37658 100644 --- a/src/sinks/util/uri.rs +++ b/src/sinks/util/uri.rs @@ -342,6 +342,32 @@ impl HttpEndpoint { })?; Self::new(uri) } + + /// Appends `suffix` directly to the path without inserting a separator. + /// + /// Unlike [`HttpEndpoint::append_path`], this does not add a `/`. It is for + /// API method suffixes that attach directly to a resource path, such as + /// Google's `:publish` convention. + pub fn append_raw_suffix(&self, suffix: &str) -> Result { + if suffix.is_empty() { + return Ok(self.clone()); + } + let mut parts = self.0.clone().into_parts(); + let base_path = parts + .path_and_query + .as_ref() + .map(PathAndQuery::path) + .unwrap_or_default(); + let joined = format!("{base_path}{suffix}"); + parts.path_and_query = Some(joined.parse::().context(InvalidPathSnafu { + endpoint: self.0.to_string(), + path: joined, + })?); + let uri = Uri::from_parts(parts).context(InvalidUriPartsSnafu { + endpoint: self.0.to_string(), + })?; + Self::new(uri) + } } impl fmt::Display for HttpEndpoint { @@ -546,4 +572,18 @@ mod tests { ); assert!(matches!(appended.as_uri().scheme_str(), Some("https"))); } + + #[test] + fn http_endpoint_append_raw_suffix_attaches_without_separator() { + let base = HttpEndpoint::parse("https://example.com/v1/projects/p/topics/t").unwrap(); + assert_eq!( + base.append_raw_suffix(":publish").unwrap().to_string(), + "https://example.com/v1/projects/p/topics/t:publish" + ); + // An empty suffix returns the endpoint unchanged. + assert_eq!( + base.append_raw_suffix("").unwrap().to_string(), + base.to_string() + ); + } } From 1b1f2d5791149472b2e3ffcadeee4629927fe66f Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 15:50:43 -0400 Subject: [PATCH 08/16] fix(sinks): reject malformed ports in HttpEndpoint --- src/sinks/util/uri.rs | 75 +++++++++++++++++++++++++++++++++++++------ 1 file changed, 65 insertions(+), 10 deletions(-) diff --git a/src/sinks/util/uri.rs b/src/sinks/util/uri.rs index fa7e229e37658..20a31684a4696 100644 --- a/src/sinks/util/uri.rs +++ b/src/sinks/util/uri.rs @@ -223,6 +223,9 @@ pub enum HttpEndpointError { "endpoint must be an absolute http(s) URL, for example `https://example.com`; got `{endpoint}`" ))] NotAbsoluteHttp { endpoint: String }, + + #[snafu(display("endpoint `{endpoint}` has an invalid port"))] + InvalidPort { endpoint: String }, } /// A `Uri` proven to be an absolute `http`/`https` URL. @@ -261,21 +264,27 @@ impl From for String { } impl HttpEndpoint { - /// Requires `uri` to be an absolute `http`/`https` URL with a host. + /// Requires `uri` to be an absolute `http`/`https` URL with a host and a + /// usable port. /// /// The authority check alone is not enough: `http://:8080` parses as a - /// valid `http::Uri` with an authority but an empty host, so the host is - /// checked explicitly. + /// valid `http::Uri` with an authority but an empty host, and + /// `http://localhost:notaport` parses with a nonempty host but a port that + /// cannot be dialed. Both are checked explicitly. pub fn new(uri: Uri) -> Result { - if matches!(uri.scheme_str(), Some("http" | "https")) - && uri.host().is_some_and(|host| !host.is_empty()) - { - Ok(Self(uri)) - } else { - Err(HttpEndpointError::NotAbsoluteHttp { + let has_valid_scheme_and_host = matches!(uri.scheme_str(), Some("http" | "https")) + && uri.host().is_some_and(|host| !host.is_empty()); + if !has_valid_scheme_and_host { + return Err(HttpEndpointError::NotAbsoluteHttp { endpoint: uri.to_string(), - }) + }); + } + if authority_has_invalid_port(&uri) { + return Err(HttpEndpointError::InvalidPort { + endpoint: uri.to_string(), + }); } + Ok(Self(uri)) } /// Parses `endpoint` and requires it to be an absolute `http`/`https` URL. @@ -376,6 +385,35 @@ impl fmt::Display for HttpEndpoint { } } +/// Returns `true` if the URI's authority contains a port that is not a valid +/// `u16`. +/// +/// `http::Uri` accepts non-numeric ports (for example +/// `http://localhost:notaport`), which `HttpClient` cannot dial. `Authority::port` +/// returns `None` for both a missing port and an invalid one, so the raw +/// authority is inspected instead. +fn authority_has_invalid_port(uri: &Uri) -> bool { + let Some(authority) = uri.authority() else { + return false; + }; + let auth = authority.as_str(); + // Strip any userinfo (everything up to the last `@`). + let host_port = auth + .rsplit_once('@') + .map(|(_, host_port)| host_port) + .unwrap_or(auth); + // An IPv6 host is bracketed; the port follows the closing `]`. + let host_end = host_port.rfind(']').map_or(0, |i| i + 1); + let Some(host_port) = host_port.get(host_end..) else { + return false; + }; + host_port.rfind(':').is_some_and(|i| { + host_port + .get(i + 1..) + .is_some_and(|port| port.parse::().is_err()) + }) +} + #[cfg(test)] mod tests { use super::*; @@ -517,6 +555,9 @@ mod tests { // with `authority() == Some` and `host() == Some("")`. "http://:8080", "http://:8080/path", + // A non-numeric port parses with a nonempty host but cannot be dialed. + "http://localhost:notaport", + "https://example.com:notaport/path", ] { assert!( matches!( @@ -524,6 +565,7 @@ mod tests { Err(HttpEndpointError::NotAbsoluteHttp { .. }) | Err(HttpEndpointError::InvalidUri { .. }) | Err(HttpEndpointError::InvalidUriParts { .. }) + | Err(HttpEndpointError::InvalidPort { .. }) ), "expected `{endpoint}` to be rejected" ); @@ -536,6 +578,19 @@ mod tests { assert!(matches!(error, HttpEndpointError::InvalidUri { .. })); } + #[test] + fn http_endpoint_rejects_malformed_ports() { + for endpoint in [ + "http://localhost:notaport", + "https://example.com:notaport/path", + ] { + assert!(matches!( + HttpEndpoint::parse(endpoint), + Err(HttpEndpointError::InvalidPort { .. }) + )); + } + } + #[test] fn http_endpoint_append_path_joins_without_string_concatenation() { let base = HttpEndpoint::parse("https://example.com").unwrap(); From 2c777aa126193782e3bc16c6f2d928877c61282b Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 16:33:24 -0400 Subject: [PATCH 09/16] fix(datadog sinks): normalize scheme-less endpoints in healthcheck --- src/sinks/datadog/mod.rs | 40 +++++++++++++++++++++++++++++++++++++--- 1 file changed, 37 insertions(+), 3 deletions(-) diff --git a/src/sinks/datadog/mod.rs b/src/sinks/datadog/mod.rs index c6c41bc4ff0e3..f540a1676aa03 100644 --- a/src/sinks/datadog/mod.rs +++ b/src/sinks/datadog/mod.rs @@ -11,7 +11,7 @@ use super::Healthcheck; use crate::{ common::datadog, http::{HttpClient, HttpError}, - sinks::HealthcheckError, + sinks::{HealthcheckError, util::HttpEndpoint}, }; #[cfg(feature = "sinks-datadog_events")] @@ -149,10 +149,13 @@ impl DatadogCommonConfig { /// Gets the API endpoint with a given suffix path. /// - /// If `endpoint` is not specified, we fallback to `site`. + /// If `endpoint` is not specified, we fallback to `site`. A missing scheme + /// is defaulted to `https`, so the healthcheck and data endpoints agree on + /// the scheme even for a scheme-less custom endpoint. fn get_api_endpoint(&self, path: &str) -> crate::Result { let base = datadog::get_api_base_endpoint(self.endpoint.as_deref(), self.site.as_str()); - [&base, path].join("").parse().map_err(Into::into) + let endpoint = HttpEndpoint::parse(&base)?.append_path(path)?; + Ok(endpoint.into_uri()) } } @@ -316,4 +319,35 @@ mod tests { let error = local.with_globals(global).unwrap_err(); assert_eq!(ConfigurationError::ApiKeyRequired, error); } + + #[test] + fn get_api_endpoint_defaults_missing_scheme_to_https() { + let config = DatadogCommonConfig { + endpoint: Some("localhost:8080".to_string()), + site: "datadoghq.com".to_string(), + default_api_key: SensitiveString::from("key".to_string()), + acknowledgements: Default::default(), + }; + assert_eq!( + config + .get_api_endpoint("/api/v1/validate") + .unwrap() + .to_string(), + "https://localhost:8080/api/v1/validate" + ); + // The default site-based endpoint keeps its scheme. + let default = DatadogCommonConfig { + endpoint: None, + site: "datadoghq.com".to_string(), + default_api_key: SensitiveString::from("key".to_string()), + acknowledgements: Default::default(), + }; + assert_eq!( + default + .get_api_endpoint("/api/v1/validate") + .unwrap() + .to_string(), + "https://api.datadoghq.com/api/v1/validate" + ); + } } From 110d222a568f6c62bf3887de399f4ea516a90e2a Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 16:33:29 -0400 Subject: [PATCH 10/16] fix(sinks): trim leading slash when appending to a non-root base path --- src/sinks/util/uri.rs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/src/sinks/util/uri.rs b/src/sinks/util/uri.rs index 20a31684a4696..dd76422fdf574 100644 --- a/src/sinks/util/uri.rs +++ b/src/sinks/util/uri.rs @@ -340,7 +340,7 @@ impl HttpEndpoint { } else if base_path.ends_with('/') { format!("{base_path}{}", path.trim_start_matches('/')) } else { - format!("{base_path}/{path}") + format!("{base_path}/{}", path.trim_start_matches('/')) }; parts.path_and_query = Some(joined.parse::().context(InvalidPathSnafu { endpoint: self.0.to_string(), @@ -626,6 +626,16 @@ mod tests { "https://user:pass@example.com:8088/base/sub/path" ); assert!(matches!(appended.as_uri().scheme_str(), Some("https"))); + // A non-root base path with a leading-slash appended path must not + // produce a double slash. + assert_eq!( + HttpEndpoint::parse("https://proxy/prefix") + .unwrap() + .append_path("/api/v1/series") + .unwrap() + .to_string(), + "https://proxy/prefix/api/v1/series" + ); } #[test] From 9633a5314a85d34fe6d7de0f72555594706f4b8e Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 16:34:36 -0400 Subject: [PATCH 11/16] Revert "chore(changelog): narrow endpoint validation claim to load-time sinks" This reverts commit 8cb4c9ec7ab39af1c029da5537169f81a9219adb. --- changelog.d/sink_endpoint_absolute_urls.enhancement.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/changelog.d/sink_endpoint_absolute_urls.enhancement.md b/changelog.d/sink_endpoint_absolute_urls.enhancement.md index dac03644e35e4..1d9c396cdeedc 100644 --- a/changelog.d/sink_endpoint_absolute_urls.enhancement.md +++ b/changelog.d/sink_endpoint_absolute_urls.enhancement.md @@ -1,3 +1,3 @@ -Sink `endpoint` options now require an absolute URL that includes a host. Endpoints without a scheme are defaulted to `https://` (for example `endpoint: "localhost:8080"` becomes `https://localhost:8080`). 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. Empty, host-less, or non-`http(s)` endpoints (for example `endpoint: ""`, `endpoint: "/path"`, or `endpoint: "ftp://example.com"`) are now rejected with a clear error. This happens at configuration load for most sinks, including with `vector validate --no-environment`; for the `datadog_metrics`, `datadog_traces`, and `sematext` sinks it happens when the sink is built. 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. +Sink `endpoint` options now require an absolute URL that includes a host. Endpoints without a scheme are defaulted to `https://` (for example `endpoint: "localhost:8080"` becomes `https://localhost:8080`). 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. Empty, host-less, or non-`http(s)` endpoints (for example `endpoint: ""`, `endpoint: "/path"`, or `endpoint: "ftp://example.com"`) are now rejected at configuration load with a clear error, including with `vector validate --no-environment`. 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 From 2b7b923c163b773cb66adf046be86ac13b44be1c Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 16:53:47 -0400 Subject: [PATCH 12/16] test(azure_logs_ingestion): expect trailing slash in endpoint string form --- src/sinks/azure_logs_ingestion/tests.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/sinks/azure_logs_ingestion/tests.rs b/src/sinks/azure_logs_ingestion/tests.rs index 275f3a0d9811b..f76175f6dfff6 100644 --- a/src/sinks/azure_logs_ingestion/tests.rs +++ b/src/sinks/azure_logs_ingestion/tests.rs @@ -65,7 +65,7 @@ fn basic_config_with_client_credentials() { assert_eq!( config.endpoint.to_string(), - "https://my-dce-5kyl.eastus-1.ingest.monitor.azure.com" + "https://my-dce-5kyl.eastus-1.ingest.monitor.azure.com/" ); assert_eq!( config.dcr_immutable_id, @@ -104,7 +104,7 @@ fn basic_config_with_managed_identity() { assert_eq!( config.endpoint.to_string(), - "https://my-dce-5kyl.eastus-1.ingest.monitor.azure.com" + "https://my-dce-5kyl.eastus-1.ingest.monitor.azure.com/" ); assert_eq!( config.dcr_immutable_id, From d30d42fba3b78080fa72d4bc47f60909456e1fe5 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 16:54:16 -0400 Subject: [PATCH 13/16] Fix examples --- website/generated/example-configs/sinks/appsignal/advanced.yaml | 2 +- .../example-configs/sinks/gcp_cloud_storage/advanced.yaml | 2 +- .../generated/example-configs/sinks/gcp_pubsub/advanced.yaml | 2 +- website/generated/example-configs/sinks/honeycomb/advanced.yaml | 2 +- .../generated/example-configs/sinks/humio_logs/advanced.yaml | 2 +- .../generated/example-configs/sinks/humio_metrics/advanced.yaml | 2 +- 6 files changed, 6 insertions(+), 6 deletions(-) diff --git a/website/generated/example-configs/sinks/appsignal/advanced.yaml b/website/generated/example-configs/sinks/appsignal/advanced.yaml index 0e4bf102dddd7..c4cd477f0e7dc 100644 --- a/website/generated/example-configs/sinks/appsignal/advanced.yaml +++ b/website/generated/example-configs/sinks/appsignal/advanced.yaml @@ -4,5 +4,5 @@ sinks: inputs: - my-source-or-transform-id compression: gzip - endpoint: https://appsignal-endpoint.net + endpoint: https://appsignal-endpoint.net/ push_api_key: 00000000-0000-0000-0000-000000000000 diff --git a/website/generated/example-configs/sinks/gcp_cloud_storage/advanced.yaml b/website/generated/example-configs/sinks/gcp_cloud_storage/advanced.yaml index 039f9322353d7..7b5772627f0ba 100644 --- a/website/generated/example-configs/sinks/gcp_cloud_storage/advanced.yaml +++ b/website/generated/example-configs/sinks/gcp_cloud_storage/advanced.yaml @@ -12,7 +12,7 @@ sinks: dangerously_allow_unconfined_template_resolution: false encoding: codec: json - endpoint: https://storage.googleapis.com + endpoint: https://storage.googleapis.com/ filename_append_uuid: true filename_time_format: "%s" key_prefix: date=%F/ diff --git a/website/generated/example-configs/sinks/gcp_pubsub/advanced.yaml b/website/generated/example-configs/sinks/gcp_pubsub/advanced.yaml index e8b3df1b39df3..6ac92cbedfa1e 100644 --- a/website/generated/example-configs/sinks/gcp_pubsub/advanced.yaml +++ b/website/generated/example-configs/sinks/gcp_pubsub/advanced.yaml @@ -5,6 +5,6 @@ sinks: - my-source-or-transform-id encoding: codec: json - endpoint: https://pubsub.googleapis.com + endpoint: https://pubsub.googleapis.com/ project: vector-123456 topic: this-is-a-topic diff --git a/website/generated/example-configs/sinks/honeycomb/advanced.yaml b/website/generated/example-configs/sinks/honeycomb/advanced.yaml index 369ab1be7f6fa..7c07185cf6f76 100644 --- a/website/generated/example-configs/sinks/honeycomb/advanced.yaml +++ b/website/generated/example-configs/sinks/honeycomb/advanced.yaml @@ -6,4 +6,4 @@ sinks: api_key: ${HONEYCOMB_API_KEY} compression: zstd dataset: my-honeycomb-dataset - endpoint: https://api.honeycomb.io + endpoint: https://api.honeycomb.io/ diff --git a/website/generated/example-configs/sinks/humio_logs/advanced.yaml b/website/generated/example-configs/sinks/humio_logs/advanced.yaml index a33148a1b2826..177bf528af295 100644 --- a/website/generated/example-configs/sinks/humio_logs/advanced.yaml +++ b/website/generated/example-configs/sinks/humio_logs/advanced.yaml @@ -7,7 +7,7 @@ sinks: dangerously_allow_unconfined_template_resolution: false encoding: codec: json - endpoint: https://cloud.humio.com + endpoint: https://cloud.humio.com/ event_type: json host_key: .host index: "{{ host }}" diff --git a/website/generated/example-configs/sinks/humio_metrics/advanced.yaml b/website/generated/example-configs/sinks/humio_metrics/advanced.yaml index 015724b221dc7..90dd2406fe85d 100644 --- a/website/generated/example-configs/sinks/humio_metrics/advanced.yaml +++ b/website/generated/example-configs/sinks/humio_metrics/advanced.yaml @@ -5,7 +5,7 @@ sinks: - my-source-or-transform-id compression: none dangerously_allow_unconfined_template_resolution: false - endpoint: https://cloud.humio.com + endpoint: https://cloud.humio.com/ event_type: json host_key: host host_tag: host From fca29310b72f9be54724dd4c611939a03f973b5c Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 17:29:37 -0400 Subject: [PATCH 14/16] fix(gcs sink): preserve significant leading slashes in object keys --- src/sinks/util/uri.rs | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/src/sinks/util/uri.rs b/src/sinks/util/uri.rs index dd76422fdf574..8c0228b06d54a 100644 --- a/src/sinks/util/uri.rs +++ b/src/sinks/util/uri.rs @@ -338,9 +338,9 @@ impl HttpEndpoint { let joined = if base_path.is_empty() { path.to_string() } else if base_path.ends_with('/') { - format!("{base_path}{}", path.trim_start_matches('/')) + format!("{base_path}{}", path.strip_prefix('/').unwrap_or(path)) } else { - format!("{base_path}/{}", path.trim_start_matches('/')) + format!("{base_path}/{}", path.strip_prefix('/').unwrap_or(path)) }; parts.path_and_query = Some(joined.parse::().context(InvalidPathSnafu { endpoint: self.0.to_string(), @@ -636,6 +636,16 @@ mod tests { .to_string(), "https://proxy/prefix/api/v1/series" ); + // Only the single boundary slash is removed; significant leading + // slashes in the appended path are preserved (GCS object keys). + assert_eq!( + HttpEndpoint::parse("https://storage.googleapis.com/bucket/") + .unwrap() + .append_path("//archive/") + .unwrap() + .to_string(), + "https://storage.googleapis.com/bucket//archive/" + ); } #[test] From b8080a6012e3622e414f0b49a05c6b7a9c703350 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 17:29:47 -0400 Subject: [PATCH 15/16] test(sinks): document rejection of multiple port separators --- src/sinks/util/uri.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/sinks/util/uri.rs b/src/sinks/util/uri.rs index 8c0228b06d54a..16e28a5026be1 100644 --- a/src/sinks/util/uri.rs +++ b/src/sinks/util/uri.rs @@ -558,6 +558,8 @@ mod tests { // A non-numeric port parses with a nonempty host but cannot be dialed. "http://localhost:notaport", "https://example.com:notaport/path", + // Multiple port separators are rejected by the URI parser. + "http://localhost:notaport:8080", ] { assert!( matches!( From 0576d67b3c1d93c00ce8c32a20ea0d362b4fb883 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 14 Aug 2026 17:29:49 -0400 Subject: [PATCH 16/16] fix(sinks): detect endpoint scheme only at the start --- src/sinks/util/uri.rs | 21 ++++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/src/sinks/util/uri.rs b/src/sinks/util/uri.rs index 16e28a5026be1..c5905baa21296 100644 --- a/src/sinks/util/uri.rs +++ b/src/sinks/util/uri.rs @@ -298,7 +298,7 @@ impl HttpEndpoint { // `host:port/path` without a scheme (it reads `host` as a scheme), so // the scheme is added up front rather than relying on the parser to // accept authority-form input. - let uri = if endpoint.contains("://") { + let uri = if has_scheme(endpoint) { endpoint .parse::() .context(InvalidUriSnafu { endpoint })? @@ -414,6 +414,23 @@ fn authority_has_invalid_port(uri: &Uri) -> bool { }) } +/// Returns `true` if `endpoint` starts with a URI scheme (`[a-zA-Z][a-zA-Z0-9+.-]*://`). +/// +/// The scheme must be at the very start: a `://` later in the path or query +/// (for example `localhost:8080/write?target=http://upstream`) is not a scheme +/// marker, so the endpoint is still defaulted to `https`. +fn has_scheme(endpoint: &str) -> bool { + let Some(scheme_end) = endpoint.find("://") else { + return false; + }; + let Some(scheme) = endpoint.get(..scheme_end) else { + return false; + }; + let mut chars = scheme.chars(); + matches!(chars.next(), Some(c) if c.is_ascii_alphabetic()) + && chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '+' | '-' | '.')) +} + #[cfg(test)] mod tests { use super::*; @@ -501,6 +518,8 @@ mod tests { "example.com:8088/services/collector", "localhost:8080", "[::1]:8080", + // A `://` later in the path or query is not a scheme marker. + "localhost:8080/write?target=http://upstream", ] { let endpoint = HttpEndpoint::parse(endpoint).expect("should accept absolute http(s) URL");