[exporter/elasticsearch] Fix regression retry::max_retries not applying correctly for HTTP request levels retries (open-telemetry#41142)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

`retry::max_retries` that gets passed to ES client is incorrectly
overwritten if `retry::max_retries` > 2. This causes HTTP requests to
give up retrying much sooner than expected, which affects scenario where
ES is down and data may be lost due to this bug.

This was a regression from open-telemetry#37333, released in v0.120.0

Re-enable integration test.

<!-- Issue number (e.g. open-telemetry#1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes open-telemetry#39670 
Fixes open-telemetry#40488

<!--Describe what testing was performed and which tests were added.-->
#### Testing

<!--Describe the documentation added.-->
#### Documentation

<!--Please delete paragraphs that you did not use before submitting.-->

---------

Co-authored-by: Andrzej Stencel <[email protected]>

Author: carsonip

.chloggen/elasticsearchexporter_fix-http-request-level-max-retries.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: bug_fix
+
+# 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: Fix regression `retry::max_retries` not applying correctly for HTTP request levels retries
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39670]
+
+# (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: The regression affected versions from v0.120.0 and might cause data loss due to prematurely stopping HTTP request level retries, e.g. when ES is unavailable, as it was cap to a maximum of 2 retries.
+
+# 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/esclient.go

@@ -129,6 +129,11 @@ func newElasticsearchClient(
 		componentHost:   host,
 	}
 
+	maxRetries := defaultMaxRetries
+	if config.Retry.MaxRetries != 0 {
+		maxRetries = config.Retry.MaxRetries
+	}
+
 	return elasticsearchv8.NewClient(elasticsearchv8.Config{
 		Transport: httpClient.Transport,
 
@@ -145,7 +150,7 @@ func newElasticsearchClient(
 		RetryOnError: func(_ *http.Request, err error) bool {
 			return !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded)
 		},
-		MaxRetries:   min(defaultMaxRetries, config.Retry.MaxRetries),
+		MaxRetries:   maxRetries,
 		RetryBackoff: createElasticsearchBackoffFunc(&config.Retry),
 
 		// configure sniffing

exporter/elasticsearchexporter/exporter_test.go

@@ -17,6 +17,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/elastic/go-docappender/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/tidwall/gjson"
@@ -496,23 +497,89 @@ func TestExporterLogs(t *testing.T) {
 		}
 	})
 
+	t.Run("retry http request batcher", func(t *testing.T) {
+		for _, maxRetries := range []int{0, 1, 11} {
+			t.Run(fmt.Sprintf("max retries %d", maxRetries), func(t *testing.T) {
+				t.Parallel()
+				expectedRetries := maxRetries
+				if maxRetries == 0 {
+					expectedRetries = defaultMaxRetries
+				}
+
+				var attempts atomic.Int64
+				rec := newBulkRecorder()
+				server := newESTestServer(t, func(_ []itemRequest) ([]itemResponse, error) {
+					// always return error, and assert that the number of attempts is expected, not more, not less.
+					attempts.Add(1)
+					return nil, &httpTestError{status: http.StatusServiceUnavailable, message: "oops"}
+				})
+
+				exporter := newTestLogsExporter(t, server.URL, func(cfg *Config) {
+					cfg.Retry.Enabled = true
+					cfg.Retry.RetryOnStatus = []int{http.StatusServiceUnavailable}
+					cfg.Retry.MaxRetries = maxRetries
+					cfg.Retry.InitialInterval = 1 * time.Millisecond
+					cfg.Retry.MaxInterval = 5 * time.Millisecond
+
+					// use sync bulk indexer
+					cfg.Batcher.Enabled = false
+					cfg.Batcher.enabledSet = true
+				})
+
+				logs := plog.NewLogs()
+				resourceLogs := logs.ResourceLogs().AppendEmpty()
+				scopeLogs := resourceLogs.ScopeLogs().AppendEmpty()
+				scopeLogs.LogRecords().AppendEmpty()
+				logs.MarkReadOnly()
+				err := exporter.ConsumeLogs(context.Background(), logs) // as sync bulk indexer is used, retries are finished on return
+				var errFlushFailed docappender.ErrorFlushFailed
+				require.ErrorAs(t, err, &errFlushFailed)
+
+				assert.Equal(t, 0, rec.countItems())
+				assert.Equal(t, int64(expectedRetries+1), attempts.Load()) // initial request + retries
+			})
+		}
+	})
+
 	t.Run("retry http request", func(t *testing.T) {
-		failures := 0
-		rec := newBulkRecorder()
-		server := newESTestServer(t, func(docs []itemRequest) ([]itemResponse, error) {
-			if failures == 0 {
-				failures++
-				return nil, &httpTestError{status: http.StatusTooManyRequests, message: "oops"}
-			}
+		for _, maxRetries := range []int{0, 1, 11} {
+			t.Run(fmt.Sprintf("max retries %d", maxRetries), func(t *testing.T) {
+				t.Parallel()
+				expectedRetries := maxRetries
+				if maxRetries == 0 {
+					expectedRetries = defaultMaxRetries
+				}
 
-			rec.Record(docs)
-			return itemsAllOK(docs)
-		})
+				var attempts atomic.Int64
+				rec := newBulkRecorder()
+				server := newESTestServer(t, func(_ []itemRequest) ([]itemResponse, error) {
+					// always return error, and assert that the number of attempts is expected, not more, not less.
+					attempts.Add(1)
+					return nil, &httpTestError{status: http.StatusServiceUnavailable, message: "oops"}
+				})
 
-		exporter := newTestLogsExporter(t, server.URL)
-		mustSendLogRecords(t, exporter, plog.NewLogRecord())
+				exporter := newTestLogsExporter(t, server.URL, func(cfg *Config) {
+					cfg.Retry.Enabled = true
+					cfg.Retry.RetryOnStatus = []int{http.StatusServiceUnavailable}
+					cfg.Retry.MaxRetries = maxRetries
+					cfg.Retry.InitialInterval = 1 * time.Millisecond
+					cfg.Retry.MaxInterval = 5 * time.Millisecond
+
+					// use async bulk indexer
+					cfg.Batcher.enabledSet = false
+				})
+				mustSendLogRecords(t, exporter, plog.NewLogRecord()) // as sync bulk indexer is used, retries are not guaranteed to finish
 
-		rec.WaitItems(1)
+				assert.Eventually(t, func() bool {
+					return int64(expectedRetries+1) == attempts.Load()
+				}, time.Second, 5*time.Millisecond)
+
+				// assert that it does not retry in async more than expected
+				time.Sleep(20 * time.Millisecond)
+				assert.Equal(t, int64(expectedRetries+1), attempts.Load())
+				assert.Equal(t, 0, rec.countItems())
+			})
+		}
 	})
 
 	t.Run("no retry", func(t *testing.T) {

exporter/elasticsearchexporter/integrationtest/datareceiver.go

@@ -50,6 +50,18 @@ const (
 	TestTracesIndex = "traces-test-idx"
 )
 
+type errElasticsearch struct {
+	httpStatus    int
+	httpDocStatus int
+}
+
+func (e errElasticsearch) Error() string {
+	if e.httpStatus != http.StatusOK {
+		return fmt.Sprintf("Simulated Elasticsearch returned HTTP status %d", e.httpStatus)
+	}
+	return fmt.Sprintf("Simulated Elasticsearch returned document status %d", e.httpDocStatus)
+}
+
 type esDataReceiver struct {
 	testbed.DataReceiverBase
 	receiver          receiver.Logs
@@ -139,23 +151,21 @@ func (es *esDataReceiver) GenConfigYAMLStr() string {
   elasticsearch:
     endpoints: [%s]
     logs_index: %s
-    logs_dynamic_index:
-      enabled: false
     metrics_index: %s
-    metrics_dynamic_index:
-      enabled: false
     traces_index: %s
-    traces_dynamic_index:
-      enabled: false
     sending_queue:
       enabled: true
+      block_on_overflow: true
     mapping:
       mode: otel
     retry:
       enabled: true
       initial_interval: 100ms
-      max_interval: 1s
-      max_requests: 10000`,
+      max_interval: 500ms
+      max_retries: 10000
+      retry_on_status: [429, 503]
+    timeout: 10m
+`,
 		es.endpoint, TestLogsIndex, TestMetricsIndex, TestTracesIndex,
 	)
 
@@ -303,13 +313,22 @@ func (es *mockESReceiver) Start(ctx context.Context, host component.Host) error
 				case TestTracesIndex:
 					consumeErr = es.tracesConsumer.ConsumeTraces(context.Background(), emptyTrace)
 				}
+				var errES errElasticsearch
 				if consumeErr != nil {
+					if !errors.As(consumeErr, &errES) {
+						// panic to surface test logic error because we only expect error of type errElasticsearch
+						panic("unknown consume error")
+					}
+					if errES.httpStatus != http.StatusOK {
+						w.WriteHeader(errES.httpStatus)
+						return
+					}
 					response.HasErrors = true
-					item.Status = http.StatusTooManyRequests
+					item.Status = errES.httpDocStatus
 					item.Error.Type = "simulated_es_error"
 					item.Error.Reason = consumeErr.Error()
+					itemMap[k] = item
 				}
-				itemMap[k] = item
 			}
 		}
 		if jsonErr := json.NewEncoder(w).Encode(response); jsonErr != nil {

exporter/elasticsearchexporter/integrationtest/exporter_test.go

@@ -4,8 +4,8 @@
 package integrationtest
 
 import (
-	"errors"
 	"fmt"
+	"net/http"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -17,7 +17,6 @@ import (
 )
 
 func TestExporter(t *testing.T) {
-	t.Skip("flaky test: https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/40488")
 	for _, eventType := range []string{"logs", "metrics", "traces"} {
 		for _, tc := range []struct {
 			name string
@@ -32,33 +31,36 @@ func TestExporter(t *testing.T) {
 			// the collector allows durability testing of the ES exporter
 			// based on the OTEL config used for testing.
 			restartCollector bool
-			mockESFailure    bool
+			mockESErr        error
 		}{
 			{name: "basic"},
-			{name: "es_intermittent_failure", mockESFailure: true},
+			{name: "es_intermittent_http_error", mockESErr: errElasticsearch{httpStatus: http.StatusServiceUnavailable}},
+			{name: "es_intermittent_doc_error", mockESErr: errElasticsearch{httpStatus: http.StatusOK, httpDocStatus: http.StatusTooManyRequests}},
 
 			{name: "batcher_enabled", batcherEnabled: ptrTo(true)},
-			{name: "batcher_enabled_es_intermittent_failure", batcherEnabled: ptrTo(true), mockESFailure: true},
+			{name: "batcher_enabled_es_intermittent_http_error", batcherEnabled: ptrTo(true), mockESErr: errElasticsearch{httpStatus: http.StatusServiceUnavailable}},
+			{name: "batcher_enabled_es_intermittent_doc_error", batcherEnabled: ptrTo(true), mockESErr: errElasticsearch{httpStatus: http.StatusOK, httpDocStatus: http.StatusTooManyRequests}},
 			{name: "batcher_disabled", batcherEnabled: ptrTo(false)},
-			{name: "batcher_disabled_es_intermittent_failure", batcherEnabled: ptrTo(false), mockESFailure: true},
+			{name: "batcher_disabled_es_intermittent_http_error", batcherEnabled: ptrTo(false), mockESErr: errElasticsearch{httpStatus: http.StatusServiceUnavailable}},
+			{name: "batcher_disabled_es_intermittent_doc_error", batcherEnabled: ptrTo(false), mockESErr: errElasticsearch{httpStatus: http.StatusOK, httpDocStatus: http.StatusTooManyRequests}},
 
 			/* TODO: Below tests should be enabled after https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/30792 is fixed
 			{name: "collector_restarts", restartCollector: true},
-			{name: "collector_restart_with_es_intermittent_failure", mockESFailure: true, restartCollector: true},
+			{name: "collector_restart_with_es_intermittent_failure", mockESErr: true, restartCollector: true},
 			*/
 		} {
 			t.Run(fmt.Sprintf("%s/%s", eventType, tc.name), func(t *testing.T) {
 				var opts []dataReceiverOption
 				if tc.batcherEnabled != nil {
 					opts = append(opts, withBatcherEnabled(*tc.batcherEnabled))
 				}
-				runner(t, eventType, tc.restartCollector, tc.mockESFailure, opts...)
+				runner(t, eventType, tc.restartCollector, tc.mockESErr, opts...)
 			})
 		}
 	}
 }
 
-func runner(t *testing.T, eventType string, restartCollector, mockESFailure bool, opts ...dataReceiverOption) {
+func runner(t *testing.T, eventType string, restartCollector bool, mockESErr error, opts ...dataReceiverOption) {
 	t.Helper()
 
 	var (
@@ -102,7 +104,7 @@ func runner(t *testing.T, eventType string, restartCollector, mockESFailure bool
 		&testbed.CorrectnessResults{},
 		testbed.WithDecisionFunc(func() error {
 			if esFailing.Load() {
-				return errors.New("simulated ES failure")
+				return mockESErr
 			}
 			return nil
 		}),
@@ -118,7 +120,7 @@ func runner(t *testing.T, eventType string, restartCollector, mockESFailure bool
 	tc.Sleep(2 * time.Second)
 
 	// Fail ES if required and send load.
-	if mockESFailure {
+	if mockESErr != nil {
 		esFailing.Store(true)
 		tc.Sleep(2 * time.Second)
 	}