fix(elasticsearch sink): Revert add api_version option (#15006)

Revert "enhancement(elasticsearch sink): Add `api_version` option (#14918)"

This reverts commit 4e45127.

Author: jszwedko

src/sinks/axiom.rs

@@ -9,7 +9,7 @@ use crate::{
         SinkContext,
     },
     sinks::{
-        elasticsearch::{ElasticsearchApiVersion, ElasticsearchAuth, ElasticsearchConfig},
+        elasticsearch::{ElasticsearchAuth, ElasticsearchConfig},
         util::{http::RequestConfig, Compression},
         Healthcheck, VectorSink,
     },
@@ -100,7 +100,6 @@ impl SinkConfig for AxiomConfig {
             query: Some(query),
             tls: self.tls.clone(),
             request,
-            api_version: ElasticsearchApiVersion::V6,
             ..Default::default()
         };
 

src/sinks/elasticsearch/common.rs

@@ -2,16 +2,12 @@ use std::collections::HashMap;
 
 use aws_types::credentials::SharedCredentialsProvider;
 use aws_types::region::Region;
-use bytes::{Buf, Bytes};
-use http::{Response, StatusCode, Uri};
-use hyper::{body, Body};
-use serde::Deserialize;
+use bytes::Bytes;
+use http::{StatusCode, Uri};
 use snafu::ResultExt;
-use vector_core::config::proxy::ProxyConfig;
 
 use super::{
-    request_builder::ElasticsearchRequestBuilder, ElasticsearchApiVersion, ElasticsearchEncoder,
-    InvalidHostSnafu, Request,
+    request_builder::ElasticsearchRequestBuilder, ElasticsearchEncoder, InvalidHostSnafu, Request,
 };
 use crate::{
     http::{Auth, HttpClient, MaybeAuth},
@@ -42,12 +38,7 @@ pub struct ElasticsearchCommon {
 }
 
 impl ElasticsearchCommon {
-    pub async fn parse_config(
-        config: &ElasticsearchConfig,
-        endpoint: &str,
-        proxy_config: &ProxyConfig,
-        version: &mut Option<usize>,
-    ) -> crate::Result<Self> {
+    pub async fn parse_config(config: &ElasticsearchConfig, endpoint: &str) -> crate::Result<Self> {
         // Test the configured host, but ignore the result
         let uri = format!("{}/_test", endpoint);
         let uri = uri
@@ -87,6 +78,16 @@ impl ElasticsearchCommon {
 
         let mode = config.common_mode()?;
 
+        let doc_type = config.doc_type.clone().unwrap_or_else(|| "_doc".into());
+        let request_builder = ElasticsearchRequestBuilder {
+            compression: config.compression,
+            encoder: ElasticsearchEncoder {
+                transformer: config.encoding.clone(),
+                doc_type,
+                suppress_type_name: config.suppress_type_name,
+            },
+        };
+
         let tower_request = config
             .request
             .tower
@@ -102,13 +103,11 @@ impl ElasticsearchCommon {
             query_params.insert("pipeline".into(), pipeline.into());
         }
 
-        let bulk_url = {
-            let mut query = url::form_urlencoded::Serializer::new(String::new());
-            for (p, v) in &query_params {
-                query.append_pair(&p[..], &v[..]);
-            }
-            format!("{}/_bulk?{}", base_url, query.finish())
-        };
+        let mut query = url::form_urlencoded::Serializer::new(String::new());
+        for (p, v) in &query_params {
+            query.append_pair(&p[..], &v[..]);
+        }
+        let bulk_url = format!("{}/_bulk?{}", base_url, query.finish());
         let bulk_uri = bulk_url.parse::<Uri>().unwrap();
 
         let tls_settings = TlsSettings::from_options(&config.tls)?;
@@ -123,46 +122,6 @@ impl ElasticsearchCommon {
 
         let region = config.aws.as_ref().and_then(|config| config.region());
 
-        let version = if let Some(version) = *version {
-            version
-        } else {
-            let ver = match config.api_version {
-                ElasticsearchApiVersion::V6 => 6,
-                ElasticsearchApiVersion::V7 => 7,
-                ElasticsearchApiVersion::V8 => 8,
-                ElasticsearchApiVersion::Auto => {
-                    get_version(
-                        &base_url,
-                        &http_auth,
-                        &aws_auth,
-                        &region,
-                        &request,
-                        &tls_settings,
-                        proxy_config,
-                    )
-                    .await?
-                }
-            };
-            *version = Some(ver);
-            ver
-        };
-
-        let doc_type = config.doc_type.clone().unwrap_or_else(|| "_doc".into());
-        let suppress_type_name = if let Some(suppress_type_name) = config.suppress_type_name {
-            warn!(message = "DEPRECATION, use of deprecated option `suppress_type_name`. Please use `api_version` option instead.");
-            suppress_type_name
-        } else {
-            version >= 7
-        };
-        let request_builder = ElasticsearchRequestBuilder {
-            compression: config.compression,
-            encoder: ElasticsearchEncoder {
-                transformer: config.encoding.clone(),
-                doc_type,
-                suppress_type_name,
-            },
-        };
-
         Ok(Self {
             http_auth,
             base_url,
@@ -179,17 +138,11 @@ impl ElasticsearchCommon {
     }
 
     /// Parses endpoints into a vector of ElasticsearchCommons. The resulting vector is guaranteed to not be empty.
-    pub async fn parse_many(
-        config: &ElasticsearchConfig,
-        proxy_config: &ProxyConfig,
-    ) -> crate::Result<Vec<Self>> {
-        let mut version = None;
+    pub async fn parse_many(config: &ElasticsearchConfig) -> crate::Result<Vec<Self>> {
         if let Some(endpoint) = config.endpoint.as_ref() {
             warn!(message = "DEPRECATION, use of deprecated option `endpoint`. Please use `endpoints` option instead.");
             if config.endpoints.is_empty() {
-                Ok(vec![
-                    Self::parse_config(config, endpoint, proxy_config, &mut version).await?,
-                ])
+                Ok(vec![Self::parse_config(config, endpoint).await?])
             } else {
                 Err(ParseError::EndpointsExclusive.into())
             }
@@ -198,8 +151,7 @@ impl ElasticsearchCommon {
         } else {
             let mut commons = Vec::new();
             for endpoint in config.endpoints.iter() {
-                commons
-                    .push(Self::parse_config(config, endpoint, proxy_config, &mut version).await?);
+                commons.push(Self::parse_config(config, endpoint).await?);
             }
             Ok(commons)
         }
@@ -208,25 +160,30 @@ impl ElasticsearchCommon {
     /// Parses a single endpoint, else panics.
     #[cfg(test)]
     pub async fn parse_single(config: &ElasticsearchConfig) -> crate::Result<Self> {
-        let mut commons =
-            Self::parse_many(config, crate::config::SinkContext::new_test().proxy()).await?;
-        assert_eq!(commons.len(), 1);
+        let mut commons = Self::parse_many(config).await?;
+        assert!(commons.len() == 1);
         Ok(commons.remove(0))
     }
 
     pub async fn healthcheck(self, client: HttpClient) -> crate::Result<()> {
-        match get(
-            &self.base_url,
-            &self.http_auth,
-            &self.aws_auth,
-            &self.region,
-            &self.request,
-            client,
-            "/_cluster/health",
-        )
-        .await?
-        .status()
-        {
+        let mut builder = Request::get(format!("{}/_cluster/health", self.base_url));
+
+        if let Some(authorization) = &self.http_auth {
+            builder = authorization.apply_builder(builder);
+        }
+
+        for (header, value) in &self.request.headers {
+            builder = builder.header(&header[..], &value[..]);
+        }
+
+        let mut request = builder.body(Bytes::new())?;
+
+        if let Some(credentials_provider) = &self.aws_auth {
+            sign_request(&mut request, credentials_provider, &self.region).await?;
+        }
+        let response = client.send(request.map(hyper::Body::from)).await?;
+
+        match response.status() {
             StatusCode::OK => Ok(()),
             status => Err(HealthcheckError::UnexpectedStatus { status }.into()),
         }
@@ -240,67 +197,3 @@ pub async fn sign_request(
 ) -> crate::Result<()> {
     crate::aws::sign_request("es", request, credentials_provider, region).await
 }
-
-async fn get_version(
-    base_url: &str,
-    http_auth: &Option<Auth>,
-    aws_auth: &Option<SharedCredentialsProvider>,
-    region: &Option<Region>,
-    request: &RequestConfig,
-    tls_settings: &TlsSettings,
-    proxy_config: &ProxyConfig,
-) -> crate::Result<usize> {
-    #[derive(Deserialize)]
-    struct ClusterState {
-        version: usize,
-    }
-
-    let client = HttpClient::new(tls_settings.clone(), proxy_config)?;
-    let response = get(
-        base_url,
-        http_auth,
-        aws_auth,
-        region,
-        request,
-        client,
-        "/_cluster/state/version",
-    )
-    .await
-    .map_err(|error| format!("Failed to get Elasticsearch API version: {}", error))?;
-
-    let (_, body) = response.into_parts();
-    let mut body = body::aggregate(body).await?;
-    let body = body.copy_to_bytes(body.remaining());
-    let ClusterState { version } = serde_json::from_slice(&body)?;
-    Ok(version)
-}
-
-async fn get(
-    base_url: &str,
-    http_auth: &Option<Auth>,
-    aws_auth: &Option<SharedCredentialsProvider>,
-    region: &Option<Region>,
-    request: &RequestConfig,
-    client: HttpClient,
-    path: &str,
-) -> crate::Result<Response<Body>> {
-    let mut builder = Request::get(format!("{}{}", base_url, path));
-
-    if let Some(authorization) = &http_auth {
-        builder = authorization.apply_builder(builder);
-    }
-
-    for (header, value) in &request.headers {
-        builder = builder.header(&header[..], &value[..]);
-    }
-
-    let mut request = builder.body(Bytes::new())?;
-
-    if let Some(credentials_provider) = aws_auth {
-        sign_request(&mut request, credentials_provider, region).await?;
-    }
-    client
-        .send(request.map(hyper::Body::from))
-        .await
-        .map_err(Into::into)
-}

src/sinks/elasticsearch/config.rs

@@ -20,8 +20,8 @@ use crate::{
             retry::ElasticsearchRetryLogic,
             service::{ElasticsearchService, HttpRequestBuilder},
             sink::ElasticsearchSink,
-            BatchActionTemplateSnafu, ElasticsearchApiVersion, ElasticsearchAuth,
-            ElasticsearchCommon, ElasticsearchCommonMode, ElasticsearchMode, IndexTemplateSnafu,
+            BatchActionTemplateSnafu, ElasticsearchAuth, ElasticsearchCommon,
+            ElasticsearchCommonMode, ElasticsearchMode, IndexTemplateSnafu,
         },
         util::{
             http::RequestConfig, service::HealthConfig, BatchConfig, Compression,
@@ -61,19 +61,13 @@ pub struct ElasticsearchConfig {
     /// set this option since Elasticsearch has removed it.
     pub doc_type: Option<String>,
 
-    /// The API version of Elasticsearch.
-    #[serde(default)]
-    pub api_version: ElasticsearchApiVersion,
-
     /// Whether or not to send the `type` field to Elasticsearch.
     ///
     /// `type` field was deprecated in Elasticsearch 7.x and removed in Elasticsearch 8.x.
     ///
     /// If enabled, the `doc_type` option will be ignored.
-    ///
-    /// This option has been deprecated, the `api_version` option should be used instead.
-    #[configurable(deprecated)]
-    pub suppress_type_name: Option<bool>,
+    #[serde(default)]
+    pub suppress_type_name: bool,
 
     /// The name of the event key that should map to Elasticsearch’s [`_id` field][es_id].
     ///
@@ -381,7 +375,7 @@ impl DataStreamConfig {
 #[async_trait::async_trait]
 impl SinkConfig for ElasticsearchConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
-        let commons = ElasticsearchCommon::parse_many(self, cx.proxy()).await?;
+        let commons = ElasticsearchCommon::parse_many(self).await?;
         let common = commons[0].clone();
 
         let client = HttpClient::new(common.tls_settings.clone(), cx.proxy())?;
@@ -489,28 +483,4 @@ mod tests {
         )
         .unwrap();
     }
-
-    #[test]
-    fn parse_version() {
-        let config = toml::from_str::<ElasticsearchConfig>(
-            r#"
-            endpoints = [""]
-            api_version = "v7"
-        "#,
-        )
-        .unwrap();
-        assert_eq!(config.api_version, ElasticsearchApiVersion::V7);
-    }
-
-    #[test]
-    fn parse_version_auto() {
-        let config = toml::from_str::<ElasticsearchConfig>(
-            r#"
-            endpoints = [""]
-            api_version = "auto"
-        "#,
-        )
-        .unwrap();
-        assert_eq!(config.api_version, ElasticsearchApiVersion::Auto);
-    }
 }