[exporter/elasticsearch] use v8 client, drop support for EOL versions of ES (#37333)

#### Description

Update to go-elasticsearch v8, which breaks support for older versions
of Elasticsearch v7. The non-EOL v7.17.x versions of Elasticsearch are
still supported by the exporter. Add an explicit statement to the README
that EOL versions are not supported by the exporter.

Updating to the v8 client allows us to turn on OTel tracing
instrumentation for the go-elasticsearch client.

#### Link to tracking issue

Closes
#32454

#### Testing

- Tested with 8.latest, confirmed working
- Tested with 7.17.lasted, confirmed working
- Tested with 7.10 (EOL), confirmed that it fails with an error message:

> 2025-02-05T13:23:42.788+0800 error
elasticsearchexporter/bulkindexer.go:337 bulk indexer flush error
{"kind": "exporter", "data_type": "metrics", "name": "elasticsearch",
"error": "failed to execute the request: the client noticed that the
server is not Elasticsearch and we do not support this unknown product"}

(7.10 predates the product check protocol that was added to
go-elasticsearch v8.)

#### Documentation

Updated README.

Author: axw

.chloggen/elasticsearchexporter-goelasticsearchv8.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: breaking
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: elasticsearchexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Use go-elasticsearch/v8, require minimum version of ES 7.17.x or 8.x
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32454]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/elasticsearchexporter/README.md

@@ -16,6 +16,10 @@
 
 This exporter supports sending logs, metrics and traces to [Elasticsearch](https://www.elastic.co/elasticsearch).
 
+The Exporter is API-compatible with Elasticsearch 7.17.x and 8.x. Certain features of the exporter,
+such as the `otel` mapping mode, may require newer versions of Elasticsearch. Limited effort will
+be made to support EOL versions of Elasticsearch -- see https://www.elastic.co/support/eol.
+
 ## Configuration options
 
 Exactly one of the following settings is required:

exporter/elasticsearchexporter/bulkindexer.go

@@ -15,7 +15,7 @@ import (
 	"time"
 
 	"github.com/elastic/go-docappender/v2"
-	"github.com/elastic/go-elasticsearch/v7"
+	"github.com/elastic/go-elasticsearch/v8/esapi"
 	"go.opentelemetry.io/collector/config/configcompression"
 	"go.uber.org/zap"
 )
@@ -55,14 +55,14 @@ type bulkIndexerSession interface {
 
 const defaultMaxRetries = 2
 
-func newBulkIndexer(logger *zap.Logger, client *elasticsearch.Client, config *Config) (bulkIndexer, error) {
+func newBulkIndexer(logger *zap.Logger, client esapi.Transport, config *Config) (bulkIndexer, error) {
 	if config.Batcher.Enabled != nil {
 		return newSyncBulkIndexer(logger, client, config), nil
 	}
 	return newAsyncBulkIndexer(logger, client, config)
 }
 
-func bulkIndexerConfig(client *elasticsearch.Client, config *Config) docappender.BulkIndexerConfig {
+func bulkIndexerConfig(client esapi.Transport, config *Config) docappender.BulkIndexerConfig {
 	var maxDocRetries int
 	if config.Retry.Enabled {
 		maxDocRetries = defaultMaxRetries
@@ -84,7 +84,7 @@ func bulkIndexerConfig(client *elasticsearch.Client, config *Config) docappender
 	}
 }
 
-func newSyncBulkIndexer(logger *zap.Logger, client *elasticsearch.Client, config *Config) *syncBulkIndexer {
+func newSyncBulkIndexer(logger *zap.Logger, client esapi.Transport, config *Config) *syncBulkIndexer {
 	return &syncBulkIndexer{
 		config:       bulkIndexerConfig(client, config),
 		flushTimeout: config.Timeout,
@@ -176,7 +176,7 @@ func (s *syncBulkIndexerSession) Flush(ctx context.Context) error {
 	}
 }
 
-func newAsyncBulkIndexer(logger *zap.Logger, client *elasticsearch.Client, config *Config) (*asyncBulkIndexer, error) {
+func newAsyncBulkIndexer(logger *zap.Logger, client esapi.Transport, config *Config) (*asyncBulkIndexer, error) {
 	numWorkers := config.NumWorkers
 	if numWorkers == 0 {
 		numWorkers = runtime.NumCPU()

exporter/elasticsearchexporter/bulkindexer_test.go

@@ -13,7 +13,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/elastic/go-elasticsearch/v7"
+	"github.com/elastic/go-elasticsearch/v8"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/config/confighttp"
@@ -293,15 +293,11 @@ func TestAsyncBulkIndexer_logRoundTrip(t *testing.T) {
 			runBulkIndexerOnce(t, &tt.config, client)
 
 			records := logObserver.AllUntimed()
-			assert.Len(t, records, 2)
+			require.Len(t, records, 1)
 
-			assert.Equal(t, "/", records[0].ContextMap()["path"])
-			assert.Nil(t, records[0].ContextMap()["request_body"])
+			assert.Equal(t, "/_bulk", records[0].ContextMap()["path"])
+			assert.Equal(t, "{\"create\":{\"_index\":\"foo\"}}\n{\"foo\": \"bar\"}\n", records[0].ContextMap()["request_body"])
 			assert.JSONEq(t, successResp, records[0].ContextMap()["response_body"].(string))
-
-			assert.Equal(t, "/_bulk", records[1].ContextMap()["path"])
-			assert.Equal(t, "{\"create\":{\"_index\":\"foo\"}}\n{\"foo\": \"bar\"}\n", records[1].ContextMap()["request_body"])
-			assert.JSONEq(t, successResp, records[1].ContextMap()["response_body"].(string))
 		})
 	}
 }
@@ -327,8 +323,9 @@ func TestSyncBulkIndexer_flushBytes(t *testing.T) {
 				reqCnt.Add(1)
 			}
 			return &http.Response{
-				Header: http.Header{"X-Elastic-Product": []string{"Elasticsearch"}},
-				Body:   io.NopCloser(strings.NewReader(successResp)),
+				Header:     http.Header{"X-Elastic-Product": []string{"Elasticsearch"}},
+				Body:       io.NopCloser(strings.NewReader(successResp)),
+				StatusCode: http.StatusOK,
 			}, nil
 		},
 	}})

exporter/elasticsearchexporter/esclient.go

@@ -5,12 +5,14 @@ package elasticsearchexporter // import "github.com/open-telemetry/opentelemetry
 
 import (
 	"context"
+	"errors"
 	"io"
 	"net/http"
 	"time"
 
 	"github.com/cenkalti/backoff/v4"
-	"github.com/elastic/go-elasticsearch/v7"
+	elasticsearchv8 "github.com/elastic/go-elasticsearch/v8"
+	"github.com/elastic/go-elasticsearch/v8/esapi"
 	"github.com/klauspost/compress/gzip"
 	"go.opentelemetry.io/collector/component"
 	"go.uber.org/zap"
@@ -82,14 +84,14 @@ func (cl *clientLogger) ResponseBodyEnabled() bool {
 	return cl.logResponseBody
 }
 
-// newElasticsearchClient returns a new elasticsearch.Client
+// newElasticsearchClient returns a new esapi.Transport.
 func newElasticsearchClient(
 	ctx context.Context,
 	config *Config,
 	host component.Host,
 	telemetry component.TelemetrySettings,
 	userAgent string,
-) (*elasticsearch.Client, error) {
+) (esapi.Transport, error) {
 	httpClient, err := config.ClientConfig.ToClient(ctx, host, telemetry)
 	if err != nil {
 		return nil, err
@@ -105,18 +107,13 @@ func newElasticsearchClient(
 		return nil, err
 	}
 
-	esLogger := clientLogger{
+	esLogger := &clientLogger{
 		Logger:          telemetry.Logger,
 		logRequestBody:  config.LogRequestBody,
 		logResponseBody: config.LogResponseBody,
 	}
 
-	maxRetries := defaultMaxRetries
-	if config.Retry.MaxRetries != 0 {
-		maxRetries = config.Retry.MaxRetries
-	}
-
-	return elasticsearch.NewClient(elasticsearch.Config{
+	return elasticsearchv8.NewClient(elasticsearchv8.Config{
 		Transport: httpClient.Transport,
 
 		// configure connection setup
@@ -127,11 +124,12 @@ func newElasticsearchClient(
 		Header:    headers,
 
 		// configure retry behavior
-		RetryOnStatus:        config.Retry.RetryOnStatus,
-		DisableRetry:         !config.Retry.Enabled,
-		EnableRetryOnTimeout: config.Retry.Enabled,
-		// RetryOnError:  retryOnError, // should be used from esclient version 8 onwards
-		MaxRetries:   maxRetries,
+		RetryOnStatus: config.Retry.RetryOnStatus,
+		DisableRetry:  !config.Retry.Enabled,
+		RetryOnError: func(_ *http.Request, err error) bool {
+			return !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded)
+		},
+		MaxRetries:   min(defaultMaxRetries, config.Retry.MaxRetries),
 		RetryBackoff: createElasticsearchBackoffFunc(&config.Retry),
 
 		// configure sniffing
@@ -141,7 +139,11 @@ func newElasticsearchClient(
 		// configure internal metrics reporting and logging
 		EnableMetrics:     false, // TODO
 		EnableDebugLogger: false, // TODO
-		Logger:            &esLogger,
+		Instrumentation: elasticsearchv8.NewOpenTelemetryInstrumentation(
+			telemetry.TracerProvider,
+			false, /* captureSearchBody */
+		),
+		Logger: esLogger,
 	})
 }
 

exporter/elasticsearchexporter/exporter_test.go

@@ -1881,6 +1881,7 @@ func TestExporterBatcher(t *testing.T) {
 	exporter := newUnstartedTestLogsExporter(t, "http://testing.invalid", func(cfg *Config) {
 		cfg.Batcher = BatcherConfig{Enabled: &batcherEnabled}
 		cfg.Auth = &configauth.Authentication{AuthenticatorID: testauthID}
+		cfg.Retry.Enabled = false
 	})
 	err := exporter.Start(context.Background(), &mockHost{
 		extensions: map[component.ID]component.Component{

exporter/elasticsearchexporter/go.mod

@@ -4,8 +4,8 @@ go 1.22.0
 
 require (
 	github.com/cenkalti/backoff/v4 v4.3.0
-	github.com/elastic/go-docappender/v2 v2.4.0
-	github.com/elastic/go-elasticsearch/v7 v7.17.10
+	github.com/elastic/go-docappender/v2 v2.3.3
+	github.com/elastic/go-elasticsearch/v8 v8.17.0
 	github.com/elastic/go-structform v0.0.12
 	github.com/klauspost/compress v1.17.11
 	github.com/lestrrat-go/strftime v1.1.0
@@ -34,7 +34,6 @@ require (
 	github.com/armon/go-radix v1.0.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/elastic/elastic-transport-go/v8 v8.6.0 // indirect
-	github.com/elastic/go-elasticsearch/v8 v8.17.0 // indirect
 	github.com/elastic/go-sysinfo v1.7.1 // indirect
 	github.com/elastic/go-windows v1.0.1 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect

exporter/elasticsearchexporter/go.sum

@@ -46,7 +46,6 @@ require (
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/ebitengine/purego v0.8.2 // indirect
 	github.com/elastic/elastic-transport-go/v8 v8.6.0 // indirect
-	github.com/elastic/go-elasticsearch/v7 v7.17.10 // indirect
 	github.com/elastic/go-elasticsearch/v8 v8.17.0 // indirect
 	github.com/elastic/go-grok v0.3.1 // indirect
 	github.com/elastic/go-structform v0.0.12 // indirect