feat(vector sink): Add connection_ttl_secs config for load balancing

changelog.d/vector_sink_connection_ttl.feature.md

@@ -0,0 +1,3 @@
+The `vector` sink now supports a `connection_ttl_secs` configuration option to periodically close and re-establish connections. This enables proper load balancing by preventing long-lived HTTP/2 connections from sticking to a single backend instance. When set, the sink will automatically reconnect after the specified duration, allowing load balancers to distribute traffic across multiple backend instances.
+
+authors: Donny Xia

src/sinks/vector/config.rs

@@ -58,6 +58,18 @@ pub struct VectorConfig {
     #[serde(default)]
     compression: bool,
 
+    /// Time-to-live (TTL) for connections in seconds.
+    ///
+    /// After this duration, the connection will be closed and a new one will be established.
+    /// This is useful for load balancing scenarios where long-lived connections prevent
+    /// proper distribution of traffic across backend instances.
+    ///
+    /// If not set, connections will be reused indefinitely (subject to HTTP/2 keep-alive).
+    #[configurable(metadata(docs::examples = 300))]
+    #[configurable(metadata(docs::examples = 600))]
+    #[serde(default)]
+    pub(in crate::sinks::vector) connection_ttl_secs: Option<u64>,
+
     #[configurable(derived)]
     #[serde(default)]
     pub batch: BatchConfig<RealtimeEventBasedDefaultBatchSettings>,
@@ -98,6 +110,7 @@ fn default_config(address: &str) -> VectorConfig {
         version: None,
         address: address.to_owned(),
         compression: false,
+        connection_ttl_secs: None,
         batch: BatchConfig::default(),
         request: TowerRequestConfig::default(),
         tls: None,
@@ -111,18 +124,36 @@ impl SinkConfig for VectorConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSinkType, Healthcheck)> {
         let tls = MaybeTlsSettings::from_config(self.tls.as_ref(), false)?;
         let uri = with_default_scheme(&self.address, tls.is_tls())?;
+        let proxy_config = cx.proxy().clone();
 
-        let client = new_client(&tls, cx.proxy())?;
+        let client = new_client(&tls, &proxy_config)?;
 
         let healthcheck_uri = cx
             .healthcheck
             .uri
             .clone()
             .map(|uri| uri.uri)
             .unwrap_or_else(|| uri.clone());
-        let healthcheck_client = VectorService::new(client.clone(), healthcheck_uri, false);
-        let healthcheck = healthcheck(healthcheck_client, cx.healthcheck);
-        let service = VectorService::new(client, uri, self.compression);
+        let healthcheck_options = cx.healthcheck;
+        let healthcheck_client = VectorService::new(
+            client.clone(),
+            healthcheck_uri,
+            false,
+            None,
+            tls.clone(),
+            proxy_config.clone(),
+        );
+        let healthcheck = healthcheck(healthcheck_client, healthcheck_options);
+
+        let connection_ttl = self.connection_ttl_secs.map(std::time::Duration::from_secs);
+        let service = VectorService::new(
+            client,
+            uri,
+            self.compression,
+            connection_ttl,
+            tls,
+            proxy_config,
+        );
         let request_settings = self.request.into_settings();
         let batch_settings = self.batch.into_batcher_settings()?;
 

src/sinks/vector/mod.rs

@@ -191,6 +191,24 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_connection_ttl_config() {
+        // Test that connection_ttl_secs is parsed correctly
+        let config = r#"
+            address = "http://127.0.0.1:9000"
+            connection_ttl_secs = 300
+        "#;
+        let config: VectorConfig = toml::from_str(config).unwrap();
+        assert_eq!(config.connection_ttl_secs, Some(300));
+
+        // Test default (no TTL)
+        let config = r#"
+            address = "http://127.0.0.1:9000"
+        "#;
+        let config: VectorConfig = toml::from_str(config).unwrap();
+        assert_eq!(config.connection_ttl_secs, None);
+    }
+
     async fn get_received(
         rx: mpsc::Receiver<(Parts, Bytes)>,
         assert_parts: impl Fn(Parts),

src/sinks/vector/service.rs

@@ -1,4 +1,8 @@
-use std::task::{Context, Poll};
+use std::{
+    sync::{Arc, Mutex},
+    task::{Context, Poll},
+    time::{Duration, Instant},
+};
 
 use futures::{TryFutureExt, future::BoxFuture};
 use http::Uri;
@@ -16,17 +20,43 @@ use vector_lib::{
 use super::VectorSinkError;
 use crate::{
     Error,
+    config::ProxyConfig,
     event::{EventFinalizers, EventStatus, Finalizable},
+    http::build_proxy_connector,
     internal_events::EndpointBytesSent,
     proto::vector as proto_vector,
     sinks::util::uri,
+    tls::MaybeTlsSettings,
 };
 
-#[derive(Clone, Debug)]
+struct ClientState {
+    client: hyper::Client<ProxyConnector<HttpsConnector<HttpConnector>>, BoxBody>,
+    created_at: Instant,
+}
+
+#[derive(Clone)]
 pub struct VectorService {
     pub client: proto_vector::Client<HyperSvc>,
     pub protocol: String,
     pub endpoint: String,
+    uri: Uri,
+    compression: bool,
+    connection_ttl: Option<Duration>,
+    tls_settings: MaybeTlsSettings,
+    proxy_config: ProxyConfig,
+    client_state: Arc<Mutex<ClientState>>,
+}
+
+impl std::fmt::Debug for VectorService {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("VectorService")
+            .field("protocol", &self.protocol)
+            .field("endpoint", &self.endpoint)
+            .field("uri", &self.uri)
+            .field("compression", &self.compression)
+            .field("connection_ttl", &self.connection_ttl)
+            .finish_non_exhaustive()
+    }
 }
 
 pub struct VectorResponse {
@@ -71,20 +101,67 @@ impl VectorService {
         hyper_client: hyper::Client<ProxyConnector<HttpsConnector<HttpConnector>>, BoxBody>,
         uri: Uri,
         compression: bool,
+        connection_ttl: Option<Duration>,
+        tls_settings: MaybeTlsSettings,
+        proxy_config: ProxyConfig,
     ) -> Self {
         let (protocol, endpoint) = uri::protocol_endpoint(uri.clone());
         let mut proto_client = proto_vector::Client::new(HyperSvc {
-            uri,
-            client: hyper_client,
+            uri: uri.clone(),
+            client: hyper_client.clone(),
         });
 
         if compression {
             proto_client = proto_client.send_compressed(tonic::codec::CompressionEncoding::Gzip);
         }
+
+        let client_state = Arc::new(Mutex::new(ClientState {
+            client: hyper_client,
+            created_at: Instant::now(),
+        }));
+
         Self {
             client: proto_client,
             protocol,
             endpoint,
+            uri,
+            compression,
+            connection_ttl,
+            tls_settings,
+            proxy_config,
+            client_state,
+        }
+    }
+
+    fn check_and_recreate_client(&mut self) {
+        if let Some(ttl) = self.connection_ttl {
+            let mut state = self.client_state.lock().unwrap();
+            let elapsed = state.created_at.elapsed();
+
+            if elapsed >= ttl {
+                // Recreate the client
+                if let Ok(proxy) =
+                    build_proxy_connector(self.tls_settings.clone(), &self.proxy_config)
+                {
+                    let new_client = hyper::Client::builder().http2_only(true).build(proxy);
+
+                    state.client = new_client.clone();
+                    state.created_at = Instant::now();
+
+                    // Update the proto client with the new hyper client
+                    let mut proto_client = proto_vector::Client::new(HyperSvc {
+                        uri: self.uri.clone(),
+                        client: new_client,
+                    });
+
+                    if self.compression {
+                        proto_client =
+                            proto_client.send_compressed(tonic::codec::CompressionEncoding::Gzip);
+                    }
+
+                    self.client = proto_client;
+                }
+            }
         }
     }
 }
@@ -106,6 +183,9 @@ impl Service<VectorRequest> for VectorService {
 
     // Emission of internal events for errors and dropped events is handled upstream by the caller.
     fn call(&mut self, mut list: VectorRequest) -> Self::Future {
+        // Check if we need to recreate the client due to TTL expiration
+        self.check_and_recreate_client();
+
         let mut service = self.clone();
         let byte_size = list.request.encoded_len();
         let metadata = std::mem::take(list.metadata_mut());