fix(elasticsearch sink): Elasticsearch sink with api_version set to "…

…auto"

  does not recognize the API version of ES6 as V6 (vectordotdev#17226)

src/sinks/elasticsearch/common.rs

@@ -276,9 +276,13 @@ async fn get_version(
     tls_settings: &TlsSettings,
     proxy_config: &ProxyConfig,
 ) -> crate::Result<usize> {
+    #[derive(Deserialize)]
+    struct Version {
+        number: Option<String>
+    }
     #[derive(Deserialize)]
     struct ClusterState {
-        version: Option<usize>,
+        version: Option<Version>
     }
 
     let client = HttpClient::new(tls_settings.clone(), proxy_config)?;
@@ -289,7 +293,7 @@ async fn get_version(
         region,
         request,
         client,
-        "/_cluster/state/version",
+        "/",
     )
     .await
     .map_err(|error| format!("Failed to get Elasticsearch API version: {}", error))?;
@@ -298,7 +302,17 @@ async fn get_version(
     let mut body = body::aggregate(body).await?;
     let body = body.copy_to_bytes(body.remaining());
     let ClusterState { version } = serde_json::from_slice(&body)?;
-    version.ok_or_else(||"Unexpected response from Elasticsearch endpoint `/_cluster/state/version`. Missing `version`. Consider setting `api_version` option.".into())
+    if let Some(version) = version {
+        if let Some(number) = version.number {
+            let v: Vec<&str> = number.split('.').collect();
+            if v.len() > 0 {
+                if let Ok(major_version) = v[0].parse::<usize>() {
+                    return Ok(major_version)
+                }
+            }
+        }
+    }
+    return Err("Unexpected response from Elasticsearch endpoint `/`. Consider setting `api_version` option.".into())
 }
 
 async fn get(