add coonection config for Elastcisearch sink

Author: vparfonov

src/sinks/elasticsearch/common.rs

@@ -23,6 +23,7 @@ use crate::{
     tls::TlsSettings,
     transforms::metric_to_log::MetricToLog,
 };
+use crate::sinks::util::http::ConnectionConfig;
 
 #[derive(Debug, Clone)]
 pub struct ElasticsearchCommon {
@@ -190,6 +191,7 @@ impl ElasticsearchCommon {
                         &request,
                         &tls_settings,
                         proxy_config,
+                        config.connection
                     )
                     .await
                     {
@@ -341,6 +343,7 @@ async fn get_version(
     request: &RequestConfig,
     tls_settings: &TlsSettings,
     proxy_config: &ProxyConfig,
+    connection_config: Option<ConnectionConfig>,
 ) -> crate::Result<usize> {
     #[derive(Deserialize)]
     struct Version {
@@ -351,7 +354,7 @@ async fn get_version(
         version: Option<Version>,
     }
 
-    let client = HttpClient::new(tls_settings.clone(), proxy_config)?;
+    let client = HttpClient::new_with_connection_config(tls_settings.clone(), proxy_config, connection_config)?;
     let response = get(
         base_url,
         auth,

src/sinks/elasticsearch/config.rs

@@ -35,6 +35,7 @@ use vector_lib::lookup::event_path;
 use vector_lib::lookup::lookup_v2::ConfigValuePath;
 use vector_lib::schema::Requirement;
 use vrl::value::Kind;
+use crate::sinks::util::http::ConnectionConfig;
 
 /// The field name for the timestamp required by data stream mode
 pub const DATA_STREAM_TIMESTAMP_KEY: &str = "@timestamp";
@@ -218,6 +219,16 @@ pub struct ElasticsearchConfig {
     )]
     #[configurable(derived)]
     pub acknowledgements: AcknowledgementsConfig,
+
+
+    /// Connection-level settings for the underlying HTTP client.
+    ///
+    /// This allows configuring parameters like connection idle timeout and
+    /// maximum idle connections per host. Useful when running behind load
+    /// balancers with strict idle policies.
+    #[configurable(derived)]
+    #[serde(default)]
+    pub connection: Option<ConnectionConfig>,
 }
 
 fn default_doc_type() -> String {
@@ -255,6 +266,7 @@ impl Default for ElasticsearchConfig {
             data_stream: None,
             metrics: None,
             acknowledgements: Default::default(),
+            connection: None,
         }
     }
 }
@@ -541,7 +553,7 @@ impl SinkConfig for ElasticsearchConfig {
         let commons = ElasticsearchCommon::parse_many(self, cx.proxy()).await?;
         let common = commons[0].clone();
 
-        let client = HttpClient::new(common.tls_settings.clone(), cx.proxy())?;
+        let client = HttpClient::new_with_connection_config(common.tls_settings.clone(), cx.proxy(), self.connection)?;
 
         let request_limits = self.request.tower.into_settings();
 

src/sinks/http/config.rs

@@ -168,7 +168,7 @@ impl From<HttpMethod> for Method {
 impl HttpSinkConfig {
     fn build_http_client(&self, cx: &SinkContext) -> crate::Result<HttpClient> {
         let tls = TlsSettings::from_options(self.tls.as_ref())?;
-        Ok(HttpClient::new(tls, cx.proxy())?)
+        Ok(HttpClient::new_with_connection_config(tls, cx.proxy(), self.connection.clone())?)
     }
 
     pub(super) fn build_encoder(&self) -> crate::Result<Encoder<Framer>> {