enhancement(loki sink): validate endpoint via HttpEndpoint (#26195)

Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>

Author: thomasqueirozb

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`, `datadog_events`, `datadog_logs`, `datadog_metrics`, `datadog_traces`, `elasticsearch`, `gcp_cloud_storage`, `gcp_pubsub`, `gcp_stackdriver_logs`, `gcp_stackdriver_metrics`, `honeycomb`, `humio`, `influxdb`, `prometheus_remote_write`, `sematext`, `splunk_hec`, and `webhdfs` 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`, `datadog_events`, `datadog_logs`, `datadog_metrics`, `datadog_traces`, `elasticsearch`, `gcp_cloud_storage`, `gcp_pubsub`, `gcp_stackdriver_logs`, `gcp_stackdriver_metrics`, `honeycomb`, `humio`, `influxdb`, `loki`, `prometheus_remote_write`, `sematext`, `splunk_hec`, and `webhdfs` sinks.
 
 authors: thomasqueirozb

src/sinks/loki/config.rs

@@ -3,6 +3,7 @@ use std::collections::HashMap;
 use vrl::value::Kind;
 
 use super::{
+    append_loki_path,
     healthcheck::healthcheck,
     sink::{LokiSink, confine_template_keys},
 };
@@ -12,7 +13,7 @@ use crate::{
     schema,
     sinks::{
         prelude::*,
-        util::{UriSerde, service::TowerRequestSettings},
+        util::{HttpEndpoint, service::TowerRequestSettings},
     },
     template::{ConfinementConfig, Template, UnconfinedTemplate},
 };
@@ -34,7 +35,7 @@ pub struct LokiConfig {
     ///
     /// The `path` value is appended to this.
     #[configurable(metadata(docs::examples = "http://localhost:3100"))]
-    pub endpoint: UriSerde,
+    pub endpoint: HttpEndpoint,
 
     /// The path to use in the URL of the Loki instance.
     #[serde(default = "default_loki_path")]
@@ -246,6 +247,11 @@ impl SinkConfig for LokiConfig {
 
 #[derive(Clone, Debug)]
 pub struct ValidatedLokiSink {
+    /// The push URL: `endpoint` with `path` appended. Used by `LokiService`.
+    pub(super) endpoint: HttpEndpoint,
+    /// The credential-free configured endpoint, including any user-supplied base path.
+    /// The healthcheck appends `ready` or `/` to this URL.
+    pub(super) base_endpoint: HttpEndpoint,
     pub(super) auth: Option<Auth>,
     pub(super) request_limits: TowerRequestSettings,
     pub(super) transformer: Transformer,
@@ -270,7 +276,11 @@ impl ValidatedSink for LokiConfig {
             }
         }
 
-        let auth = self.auth.choose_one(&self.endpoint.auth)?;
+        // Extract basic-auth credentials embedded in the endpoint URL and strip
+        // the userinfo, so credentials are sent as an `Authorization` header
+        // rather than in the request URL.
+        let (base_endpoint, endpoint_auth) = self.endpoint.clone().extract_basic_auth()?;
+        let auth = self.auth.choose_one(&endpoint_auth)?;
 
         let request_limits = match self.out_of_order_action {
             OutOfOrderAction::Accept => self.request.into_settings(),
@@ -302,17 +312,15 @@ impl ValidatedSink for LokiConfig {
             "structured_metadata[key]",
         )?;
 
-        if !matches!(self.endpoint.uri.scheme_str(), Some("http" | "https"))
-            || self.endpoint.uri.authority().is_none()
-        {
-            return Err("Loki endpoint must be an absolute http(s) URL".into());
-        }
-
-        self.endpoint.append_path(&self.path)?;
+        // The push URL appends `path`; the healthcheck appends `ready`/`` to the
+        // configured base endpoint, so both are retained.
+        let endpoint = append_loki_path(&base_endpoint, &self.path)?;
 
         let batch_settings = self.batch.into_batcher_settings()?;
 
         Ok(ValidatedLokiSink {
+            endpoint,
+            base_endpoint,
             auth,
             request_limits,
             transformer,
@@ -330,14 +338,14 @@ impl ValidatedSink for LokiConfig {
     ) -> crate::Result<(VectorSink, crate::sinks::Healthcheck)> {
         let client = self.build_client(cx)?;
 
-        let config = LokiConfig {
-            auth: validated.auth.clone(),
-            ..self.clone()
-        };
+        let sink = LokiSink::from_validated(self, validated.clone(), client.clone())?;
 
-        let sink = LokiSink::from_validated(config.clone(), validated.clone(), client.clone())?;
-
-        let healthcheck = healthcheck(config, client).boxed();
+        let healthcheck = healthcheck(
+            validated.base_endpoint.clone(),
+            validated.auth.clone(),
+            client,
+        )
+        .boxed();
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 }
@@ -434,6 +442,26 @@ mod tests {
         );
     }
 
+    #[test]
+    fn deserialize_rejects_non_http_endpoint() {
+        let result = serde_yaml::from_str::<LokiConfig>(
+            r#"
+            endpoint: "ftp://localhost:3100"
+            labels:
+              test_name: "placeholder"
+            encoding:
+              codec: json
+            "#,
+        );
+
+        assert!(
+            result.is_err(),
+            "non-http endpoints must fail at config load"
+        );
+        let message = result.unwrap_err().to_string();
+        assert!(message.contains("ftp://localhost:3100"), "{message}");
+    }
+
     #[test]
     fn validate_returns_usable_values() {
         let config: LokiConfig = serde_yaml::from_str(
@@ -454,10 +482,38 @@ mod tests {
     }
 
     #[test]
-    fn validate_rejects_non_http_endpoint() {
+    fn validate_extracts_endpoint_basic_auth() {
+        let config: LokiConfig = serde_yaml::from_str(
+            r#"
+            endpoint: "http://user:pass@localhost:3100"
+            labels:
+              test_name: "placeholder"
+            encoding:
+              codec: json
+            "#,
+        )
+        .unwrap();
+
+        let validated = config.validate().expect("validation should succeed");
+        assert!(
+            validated.auth.is_some(),
+            "credentials embedded in the endpoint must be extracted as auth"
+        );
+        assert!(
+            !validated.endpoint.to_string().contains('@'),
+            "userinfo must be stripped from the endpoint retained for build"
+        );
+    }
+
+    #[test]
+    fn validate_rejects_conflicting_auth() {
         let config: LokiConfig = serde_yaml::from_str(
             r#"
-            endpoint: "ftp://example.com"
+            endpoint: "http://user:pass@localhost:3100"
+            auth:
+              strategy: "basic"
+              user: "other"
+              password: "other"
             labels:
               test_name: "placeholder"
             encoding:
@@ -468,7 +524,7 @@ mod tests {
 
         assert!(
             config.validate().is_err(),
-            "a non-http(s) endpoint must be rejected before it reaches the HTTP client"
+            "explicit auth and endpoint-embedded auth must not both be configured"
         );
     }
 

src/sinks/loki/healthcheck.rs

@@ -1,30 +1,43 @@
-use super::config::LokiConfig;
-use crate::http::HttpClient;
+use super::append_loki_path;
+
+use crate::{
+    http::{Auth, HttpClient},
+    sinks::util::HttpEndpoint,
+};
 
 async fn fetch_status(
-    endpoint: &str,
-    config: &LokiConfig,
+    base_endpoint: &HttpEndpoint,
+    path: &str,
+    auth: &Option<Auth>,
     client: &HttpClient,
 ) -> crate::Result<http::StatusCode> {
-    let endpoint = config.endpoint.append_path(endpoint)?;
+    let endpoint = append_loki_path(base_endpoint, path)?;
 
-    let mut req = http::Request::get(endpoint.uri)
+    let mut req = http::Request::get(endpoint.as_uri())
         .body(hyper::Body::empty())
         .expect("Building request never fails.");
 
-    if let Some(auth) = &config.auth {
+    if let Some(auth) = auth {
         auth.apply(&mut req);
     }
 
     Ok(client.send(req).await?.status())
 }
 
-pub async fn healthcheck(config: LokiConfig, client: HttpClient) -> crate::Result<()> {
-    let status = match fetch_status("ready", &config, &client).await? {
+pub async fn healthcheck(
+    base_endpoint: HttpEndpoint,
+    auth: Option<Auth>,
+    client: HttpClient,
+) -> crate::Result<()> {
+    let status = match fetch_status(&base_endpoint, "ready", &auth, &client).await? {
         // Issue https://github.com/vectordotdev/vector/issues/6463
         http::StatusCode::NOT_FOUND => {
             debug!("Endpoint `/ready` not found. Retrying healthcheck with top level query.");
-            fetch_status("", &config, &client).await?
+            // Probe the normalized base path with one trailing slash (`/loki/`,
+            // not `/loki`), matching the pre-`HttpEndpoint` behavior. Reverse
+            // proxies commonly redirect `/loki` to `/loki/`, and the healthcheck
+            // rejects non-200 responses rather than following them.
+            fetch_status(&base_endpoint, "/", &auth, &client).await?
         }
         status => status,
     };

src/sinks/loki/mod.rs

@@ -10,6 +10,29 @@
 //!
 //! If an event produces no labels, this can happen if the template
 //! does not match, we will add a default label `{agent="vector"}`.
+
+use crate::sinks::util::HttpEndpoint;
+
+/// Appends a Loki path while retaining the boundary normalization used by
+/// `UriSerde::append_path`: all trailing slashes on the endpoint and all
+/// leading slashes on the path collapse to one separator.
+fn append_loki_path(endpoint: &HttpEndpoint, path: &str) -> crate::Result<HttpEndpoint> {
+    let mut parts = endpoint.as_uri().clone().into_parts();
+    let base_path = parts
+        .path_and_query
+        .as_ref()
+        .map(http::uri::PathAndQuery::path)
+        .unwrap_or_default();
+    let joined = format!(
+        "{}/{}",
+        base_path.trim_end_matches('/'),
+        path.trim_start_matches('/')
+    );
+    parts.path_and_query = Some(joined.parse()?);
+
+    Ok(HttpEndpoint::new(http::Uri::from_parts(parts)?)?)
+}
+
 mod config;
 mod event;
 mod healthcheck;

src/sinks/loki/service.rs

@@ -7,7 +7,7 @@ use tracing::Instrument;
 
 use crate::{
     http::{Auth, HttpClient},
-    sinks::{prelude::*, util::UriSerde},
+    sinks::{prelude::*, util::HttpEndpoint},
 };
 
 #[derive(Clone)]
@@ -85,20 +85,18 @@ impl MetaDescriptive for LokiRequest {
 
 #[derive(Debug, Clone)]
 pub struct LokiService {
-    endpoint: UriSerde,
+    endpoint: HttpEndpoint,
     client: HttpClient,
+    auth: Option<Auth>,
 }
 
 impl LokiService {
-    pub fn new(
-        client: HttpClient,
-        endpoint: UriSerde,
-        path: String,
-        auth: Option<Auth>,
-    ) -> crate::Result<Self> {
-        let endpoint = endpoint.append_path(&path)?.with_auth(auth);
-
-        Ok(Self { client, endpoint })
+    pub const fn new(client: HttpClient, endpoint: HttpEndpoint, auth: Option<Auth>) -> Self {
+        Self {
+            client,
+            endpoint,
+            auth,
+        }
     }
 }
 
@@ -116,7 +114,8 @@ impl Service<LokiRequest> for LokiService {
             Compression::Snappy => "application/x-protobuf",
             _ => "application/json",
         };
-        let mut req = http::Request::post(&self.endpoint.uri).header("Content-Type", content_type);
+        let mut req =
+            http::Request::post(self.endpoint.as_uri()).header("Content-Type", content_type);
 
         let metadata = request.get_metadata().clone();
 
@@ -131,7 +130,7 @@ impl Service<LokiRequest> for LokiService {
         let body = hyper::Body::from(request.payload);
         let mut req = req.body(body).unwrap();
 
-        if let Some(auth) = &self.endpoint.auth {
+        if let Some(auth) = &self.auth {
             auth.apply(&mut req);
         }
 

src/sinks/loki/sink.rs

@@ -493,29 +493,28 @@ impl LokiSink {
     #[allow(clippy::missing_const_for_fn)] // const cannot run destructor
     pub fn new(config: LokiConfig, client: HttpClient) -> crate::Result<Self> {
         let validated = config.validate()?;
-        Self::from_validated(config, validated, client)
+        Self::from_validated(&config, validated, client)
     }
 
     /// Builds the sink from already-validated state. No pure validation is
     /// repeated here: auth, request limits, encoder, confined templates and
     /// batch settings all come from `validated`.
     #[allow(clippy::missing_const_for_fn)] // const cannot run destructor
     pub fn from_validated(
-        config: LokiConfig,
+        config: &LokiConfig,
         validated: ValidatedLokiSink,
         client: HttpClient,
     ) -> crate::Result<Self> {
         let compression = config.compression;
 
-        let protocol = get_http_scheme_from_uri(&config.endpoint.uri);
+        let protocol = get_http_scheme_from_uri(validated.endpoint.as_uri());
         let service = tower::ServiceBuilder::new()
             .settings(validated.request_limits.clone(), LokiRetryLogic)
             .service(LokiService::new(
                 client,
-                config.endpoint,
-                config.path,
-                config.auth,
-            )?);
+                validated.endpoint.clone(),
+                validated.auth.clone(),
+            ));
 
         let batch_encoder = match config.compression {
             Compression::Snappy => LokiBatchEncoder(LokiBatchEncoding::Protobuf),