[exporter/elasticsearch] Add retry.retry_on_status config

.chloggen/elasticsearchexporter_retry-on-status.yaml

@@ -0,0 +1,30 @@
+# 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: Add retry.retry_on_status config and default to [429]
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32584]
+
+# (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: |
+  Previously the status codes that will trigger retries are 500, 502, 503, 504, 429, and may lead to duplicates.
+  Introduce a breaking change such that it now defaults to 429 only.
+  To opt-in to the previous behavior, configure retry.retry_on_status accordingly.
+
+# 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

@@ -58,6 +58,7 @@ This exporter supports sending OpenTelemetry logs and traces to [Elasticsearch](
   - `max_requests` (default=3): Number of HTTP request retries.
   - `initial_interval` (default=100ms): Initial waiting time if a HTTP request failed.
   - `max_interval` (default=1m): Max waiting time if a HTTP request failed.
+  - `retry_on_status` (default=[429]): Status codes that trigger request or document level retries. Request level retry and document level retry status codes are shared and cannot be configured separately.
 - `mapping`: Events are encoded to JSON. The `mapping` allows users to
   configure additional mapping rules.
   - `mode` (default=none): The fields naming mode. valid modes are:

exporter/elasticsearchexporter/config.go

@@ -151,6 +151,9 @@ type RetrySettings struct {
 
 	// MaxInterval configures the max waiting time if consecutive requests failed.
 	MaxInterval time.Duration `mapstructure:"max_interval"`
+
+	// RetryOnStatus configures the status codes that trigger request or document level retries.
+	RetryOnStatus []int `mapstructure:"retry_on_status"`
 }
 
 type MappingsSettings struct {

exporter/elasticsearchexporter/config_test.go

@@ -61,6 +61,7 @@ func TestLoad_DeprecatedIndexConfigOption(t *testing.T) {
 			MaxRequests:     5,
 			InitialInterval: 100 * time.Millisecond,
 			MaxInterval:     1 * time.Minute,
+			RetryOnStatus:   []int{429},
 		},
 		Mapping: MappingsSettings{
 			Mode:  "none",
@@ -136,6 +137,7 @@ func TestLoadConfig(t *testing.T) {
 					MaxRequests:     5,
 					InitialInterval: 100 * time.Millisecond,
 					MaxInterval:     1 * time.Minute,
+					RetryOnStatus:   []int{429, 500},
 				},
 				Mapping: MappingsSettings{
 					Mode:  "none",
@@ -186,6 +188,7 @@ func TestLoadConfig(t *testing.T) {
 					MaxRequests:     5,
 					InitialInterval: 100 * time.Millisecond,
 					MaxInterval:     1 * time.Minute,
+					RetryOnStatus:   []int{429, 500},
 				},
 				Mapping: MappingsSettings{
 					Mode:  "none",

exporter/elasticsearchexporter/elasticsearch_bulk.go

@@ -103,7 +103,7 @@ func newElasticsearchClient(logger *zap.Logger, config *Config) (*esClientCurren
 		Header:    headers,
 
 		// configure retry behavior
-		RetryOnStatus:        retryOnStatus,
+		RetryOnStatus:        config.Retry.RetryOnStatus,
 		DisableRetry:         retryDisabled,
 		EnableRetryOnTimeout: config.Retry.Enabled,
 		//RetryOnError:  retryOnError, // should be used from esclient version 8 onwards
@@ -175,7 +175,10 @@ func createElasticsearchBackoffFunc(config *RetrySettings) func(int) time.Durati
 	}
 }
 
-func shouldRetryEvent(status int) bool {
+func shouldRetryEvent(status int, retryOnStatus []int) bool {
+	if retryOnStatus == nil {
+		return false
+	}
 	for _, retryable := range retryOnStatus {
 		if status == retryable {
 			return true
@@ -184,15 +187,15 @@ func shouldRetryEvent(status int) bool {
 	return false
 }
 
-func pushDocuments(ctx context.Context, logger *zap.Logger, index string, document []byte, bulkIndexer esBulkIndexerCurrent, maxAttempts int) error {
+func pushDocuments(ctx context.Context, logger *zap.Logger, index string, document []byte, bulkIndexer esBulkIndexerCurrent, maxAttempts int, retryOnStatus []int) error {
 	attempts := 1
 	body := bytes.NewReader(document)
 	item := esBulkIndexerItem{Action: createAction, Index: index, Body: body}
 	// Setup error handler. The handler handles the per item response status based on the
 	// selective ACKing in the bulk response.
 	item.OnFailure = func(ctx context.Context, item esBulkIndexerItem, resp esBulkIndexerResponseItem, err error) {
 		switch {
-		case attempts < maxAttempts && shouldRetryEvent(resp.Status):
+		case attempts < maxAttempts && shouldRetryEvent(resp.Status, retryOnStatus):
 			logger.Debug("Retrying to index",
 				zap.String("name", index),
 				zap.Int("attempt", attempts),

exporter/elasticsearchexporter/factory.go

@@ -51,6 +51,7 @@ func createDefaultConfig() component.Config {
 			MaxRequests:     3,
 			InitialInterval: 100 * time.Millisecond,
 			MaxInterval:     1 * time.Minute,
+			RetryOnStatus:   []int{429},
 		},
 		Mapping: MappingsSettings{
 			Mode:  "none",

exporter/elasticsearchexporter/integrationtest/datareceiver.go

@@ -137,7 +137,7 @@ func newMockESReceiver(params receiver.CreateSettings, cfg *config, next consume
 
 				if err := next.ConsumeLogs(context.Background(), logs); err != nil {
 					response.HasErrors = true
-					item.Status = http.StatusInternalServerError
+					item.Status = 429
 					item.Error.Type = "simulated_es_error"
 					item.Error.Reason = err.Error()
 				}

exporter/elasticsearchexporter/logs_exporter.go

@@ -23,14 +23,13 @@ type elasticsearchLogsExporter struct {
 	logstashFormat LogstashFormatSettings
 	dynamicIndex   bool
 	maxAttempts    int
+	retryOnStatus  []int
 
 	client      *esClientCurrent
 	bulkIndexer esBulkIndexerCurrent
 	model       mappingModel
 }
 
-var retryOnStatus = []int{500, 502, 503, 504, 429}
-
 const createAction = "create"
 
 func newLogsExporter(logger *zap.Logger, cfg *Config) (*elasticsearchLogsExporter, error) {
@@ -71,6 +70,7 @@ func newLogsExporter(logger *zap.Logger, cfg *Config) (*elasticsearchLogsExporte
 		index:          indexStr,
 		dynamicIndex:   cfg.LogsDynamicIndex.Enabled,
 		maxAttempts:    maxAttempts,
+		retryOnStatus:  cfg.Retry.RetryOnStatus,
 		model:          model,
 		logstashFormat: cfg.LogstashFormat,
 	}
@@ -129,5 +129,5 @@ func (e *elasticsearchLogsExporter) pushLogRecord(ctx context.Context, resource
 	if err != nil {
 		return fmt.Errorf("Failed to encode log event: %w", err)
 	}
-	return pushDocuments(ctx, e.logger, fIndex, document, e.bulkIndexer, e.maxAttempts)
+	return pushDocuments(ctx, e.logger, fIndex, document, e.bulkIndexer, e.maxAttempts, e.retryOnStatus)
 }

exporter/elasticsearchexporter/logs_exporter_test.go

@@ -328,7 +328,7 @@ func TestExporter_PushEvent(t *testing.T) {
 		server := newESTestServer(t, func(docs []itemRequest) ([]itemResponse, error) {
 			if failures == 0 {
 				failures++
-				return nil, &httpTestError{message: "oops"}
+				return nil, &httpTestError{status: 429, message: "oops"}
 			}
 
 			rec.Record(docs)
@@ -508,7 +508,7 @@ func withTestExporterConfig(fns ...func(*Config)) func(string) *Config {
 }
 
 func mustSend(t *testing.T, exporter *elasticsearchLogsExporter, contents string) {
-	err := pushDocuments(context.TODO(), zap.L(), exporter.index, []byte(contents), exporter.bulkIndexer, exporter.maxAttempts)
+	err := pushDocuments(context.TODO(), zap.L(), exporter.index, []byte(contents), exporter.bulkIndexer, exporter.maxAttempts, exporter.retryOnStatus)
 	require.NoError(t, err)
 }
 

exporter/elasticsearchexporter/testdata/config.yaml

@@ -19,6 +19,9 @@ elasticsearch/trace:
     bytes: 10485760
   retry:
     max_requests: 5
+    retry_on_status:
+      - 429
+      - 500
 elasticsearch/log:
   tls:
     insecure: false
@@ -38,6 +41,9 @@ elasticsearch/log:
     bytes: 10485760
   retry:
     max_requests: 5
+    retry_on_status:
+      - 429
+      - 500
   sending_queue:
     enabled: true
 elasticsearch/logstash_format:

exporter/elasticsearchexporter/trace_exporter.go

@@ -23,6 +23,7 @@ type elasticsearchTracesExporter struct {
 	logstashFormat LogstashFormatSettings
 	dynamicIndex   bool
 	maxAttempts    int
+	retryOnStatus  []int
 
 	client      *esClientCurrent
 	bulkIndexer esBulkIndexerCurrent
@@ -63,6 +64,7 @@ func newTracesExporter(logger *zap.Logger, cfg *Config) (*elasticsearchTracesExp
 		index:          cfg.TracesIndex,
 		dynamicIndex:   cfg.TracesDynamicIndex.Enabled,
 		maxAttempts:    maxAttempts,
+		retryOnStatus:  cfg.Retry.RetryOnStatus,
 		model:          model,
 		logstashFormat: cfg.LogstashFormat,
 	}, nil
@@ -122,5 +124,5 @@ func (e *elasticsearchTracesExporter) pushTraceRecord(ctx context.Context, resou
 	if err != nil {
 		return fmt.Errorf("Failed to encode trace record: %w", err)
 	}
-	return pushDocuments(ctx, e.logger, fIndex, document, e.bulkIndexer, e.maxAttempts)
+	return pushDocuments(ctx, e.logger, fIndex, document, e.bulkIndexer, e.maxAttempts, e.retryOnStatus)
 }

exporter/elasticsearchexporter/traces_exporter_test.go

@@ -272,7 +272,7 @@ func TestExporter_PushTraceRecord(t *testing.T) {
 		server := newESTestServer(t, func(docs []itemRequest) ([]itemResponse, error) {
 			if failures == 0 {
 				failures++
-				return nil, &httpTestError{message: "oops"}
+				return nil, &httpTestError{status: 429, message: "oops"}
 			}
 
 			rec.Record(docs)
@@ -463,7 +463,7 @@ func withTestTracesExporterConfig(fns ...func(*Config)) func(string) *Config {
 }
 
 func mustSendTraces(t *testing.T, exporter *elasticsearchTracesExporter, contents string) {
-	err := pushDocuments(context.TODO(), zap.L(), exporter.index, []byte(contents), exporter.bulkIndexer, exporter.maxAttempts)
+	err := pushDocuments(context.TODO(), zap.L(), exporter.index, []byte(contents), exporter.bulkIndexer, exporter.maxAttempts, exporter.retryOnStatus)
 	require.NoError(t, err)
 }