Non fatal error & better error desc

Signed-off-by: Kruno Tomola Fabro <krunotf@gmail.com>

src/sinks/elasticsearch/common.rs

@@ -131,7 +131,7 @@ impl ElasticsearchCommon {
                 ElasticsearchApiVersion::V7 => 7,
                 ElasticsearchApiVersion::V8 => 8,
                 ElasticsearchApiVersion::Auto => {
-                    get_version(
+                    match get_version(
                         &base_url,
                         &http_auth,
                         &aws_auth,
@@ -140,7 +140,20 @@ impl ElasticsearchCommon {
                         &tls_settings,
                         proxy_config,
                     )
-                    .await?
+                    .await
+                    {
+                        Ok(version) => version,
+                        // This error should be fatal, but for now we only emit it as a warning
+                        // to make the transition smoother.
+                        Err(error) => {
+                            warn!(message = "Failed to determine Elasticsearch version. Please set option `api_version`.", %error);
+                            // For now, estimate version.
+                            match config.suppress_type_name {
+                                Some(true) => 8,
+                                _ => 6,
+                            }
+                        }
+                    }
                 }
             };
             *version = Some(ver);
@@ -252,7 +265,7 @@ async fn get_version(
 ) -> crate::Result<usize> {
     #[derive(Deserialize)]
     struct ClusterState {
-        version: usize,
+        version: Option<usize>,
     }
 
     let client = HttpClient::new(tls_settings.clone(), proxy_config)?;
@@ -272,7 +285,7 @@ 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)?;
-    Ok(version)
+    version.ok_or_else(||"Unexpected response from Elasticsearch endpoint /_cluster/state/version. Missing `version`. Consider setting `api_version` option.".into())
 }
 
 async fn get(