[exporter/opensearch] Retry transient failures instead of dropping them as permanent

.chloggen/49208-opensearch-retry-transient-errors.yaml

@@ -0,0 +1,31 @@
+# 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. receiver/filelog)
+component: exporter/opensearch
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Retry transient OpenSearch failures instead of dropping them as permanent errors
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [49208]
+
+# (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: |
+  Indexer-level failures such as connection refused, timeouts, and DNS errors were wrapped as
+  permanent errors, so exporterhelper dropped the batch even with retry_on_failure enabled. They
+  are now surfaced as retryable, and per-item transport errors (net.OpError) are retried as well,
+  while encoding and bad-document errors remain permanent.
+
+# 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/opensearchexporter/log_bulk_indexer.go

@@ -7,6 +7,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"net"
 	"time"
 
 	"github.com/opensearch-project/opensearch-go/v4/opensearchapi"
@@ -47,7 +48,11 @@ func (lbi *logBulkIndexer) close(ctx context.Context) {
 
 func (lbi *logBulkIndexer) onIndexerError(_ context.Context, indexerErr error) {
 	if indexerErr != nil {
-		lbi.appendPermanentError(consumererror.NewPermanent(indexerErr))
+		// Indexer-level errors are transport/flush failures (connection refused,
+		// timeout, DNS). They are transient, so surface them as a retryable error
+		// and let exporterhelper's retry_on_failure resend the batch instead of
+		// dropping it as permanent.
+		lbi.errs = append(lbi.errs, indexerErr)
 	}
 }
 
@@ -128,8 +133,15 @@ func (lbi *logBulkIndexer) processItemFailure(resp opensearchapi.BulkRespItem, i
 		// Non-recoverable OpenSearch error while indexing document
 		lbi.appendPermanentError(responseAsError(resp))
 	default:
-		// Encoding error. We didn't even attempt to send the event
-		lbi.appendPermanentError(itemErr)
+		// A transport error (e.g. net.OpError) reported per item is transient and
+		// should be retried; anything else here is an encoding error we never sent,
+		// which is permanent.
+		var netErr *net.OpError
+		if errors.As(itemErr, &netErr) {
+			lbi.appendRetryLogError(itemErr, logs)
+		} else {
+			lbi.appendPermanentError(itemErr)
+		}
 	}
 }
 

exporter/opensearchexporter/log_bulk_indexer_test.go

@@ -5,10 +5,12 @@ package opensearchexporter
 
 import (
 	"errors"
+	"net"
 	"testing"
 	"time"
 
 	"github.com/opensearch-project/opensearch-go/v4/opensearchapi"
+	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
 )
@@ -60,6 +62,41 @@ func TestProcessItemFailure(t *testing.T) {
 	}
 }
 
+func TestOnIndexerErrorIsRetryable(t *testing.T) {
+	lbi := &logBulkIndexer{}
+	// A transport failure surfaces through the bulk indexer's OnError callback.
+	lbi.onIndexerError(t.Context(), &net.OpError{Op: "dial", Err: errors.New("connection refused")})
+	err := lbi.joinedError()
+	if err == nil {
+		t.Fatal("expected an error")
+	}
+	if consumererror.IsPermanent(err) {
+		t.Error("indexer-level transport error must be retryable, not permanent (otherwise retry_on_failure silently drops the batch)")
+	}
+}
+
+func TestProcessItemFailureTransportErrorRetryable(t *testing.T) {
+	// A per-item transport error (net.OpError) with no HTTP status is transient.
+	lbi := &logBulkIndexer{}
+	lbi.processItemFailure(opensearchapi.BulkRespItem{Status: 0}, &net.OpError{Op: "read", Err: errors.New("i/o timeout")}, plog.NewLogs())
+	if len(lbi.errs) != 1 {
+		t.Fatalf("expected 1 error, got %d", len(lbi.errs))
+	}
+	if consumererror.IsPermanent(lbi.errs[0]) {
+		t.Error("per-item transport error must be retryable, not permanent")
+	}
+
+	// A genuine encoding error (never sent) stays permanent.
+	lbi2 := &logBulkIndexer{}
+	lbi2.processItemFailure(opensearchapi.BulkRespItem{Status: 0}, errors.New("encode failed"), plog.NewLogs())
+	if len(lbi2.errs) != 1 {
+		t.Fatalf("expected 1 error, got %d", len(lbi2.errs))
+	}
+	if !consumererror.IsPermanent(lbi2.errs[0]) {
+		t.Error("encoding error must remain permanent")
+	}
+}
+
 func TestNewLogBulkIndexerWithPipeline(t *testing.T) {
 	tests := []struct {
 		name     string

exporter/opensearchexporter/metadata.yaml

@@ -15,4 +15,6 @@ tests:
   expect_consumer_error: true
   config:
     http:
-      endpoint: https://opensearch.example.com:9200
+      endpoint: https://opensearch.example.com:9200
+    retry_on_failure:
+      enabled: false

exporter/opensearchexporter/trace_bulk_indexer.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"net"
 	"slices"
 	"time"
 
@@ -49,7 +50,11 @@ func (tbi *traceBulkIndexer) close(ctx context.Context) {
 
 func (tbi *traceBulkIndexer) onIndexerError(_ context.Context, indexerErr error) {
 	if indexerErr != nil {
-		tbi.appendPermanentError(consumererror.NewPermanent(indexerErr))
+		// Indexer-level errors are transport/flush failures (connection refused,
+		// timeout, DNS). They are transient, so surface them as a retryable error
+		// and let exporterhelper's retry_on_failure resend the batch instead of
+		// dropping it as permanent.
+		tbi.errs = append(tbi.errs, indexerErr)
 	}
 }
 
@@ -130,8 +135,15 @@ func (tbi *traceBulkIndexer) processItemFailure(resp opensearchapi.BulkRespItem,
 		// Non-recoverable OpenSearch error while indexing document
 		tbi.appendPermanentError(responseAsError(resp))
 	default:
-		// Encoding error. We didn't even attempt to send the event
-		tbi.appendPermanentError(itemErr)
+		// A transport error (e.g. net.OpError) reported per item is transient and
+		// should be retried; anything else here is an encoding error we never sent,
+		// which is permanent.
+		var netErr *net.OpError
+		if errors.As(itemErr, &netErr) {
+			tbi.appendRetryTraceError(itemErr, traces)
+		} else {
+			tbi.appendPermanentError(itemErr)
+		}
 	}
 }
 

exporter/opensearchexporter/trace_bulk_indexer_test.go

@@ -5,10 +5,12 @@ package opensearchexporter
 
 import (
 	"errors"
+	"net"
 	"testing"
 	"time"
 
 	"github.com/opensearch-project/opensearch-go/v4/opensearchapi"
+	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 )
@@ -60,6 +62,38 @@ func TestTraceProcessItemFailure(t *testing.T) {
 	}
 }
 
+func TestTraceOnIndexerErrorIsRetryable(t *testing.T) {
+	tbi := &traceBulkIndexer{}
+	tbi.onIndexerError(t.Context(), &net.OpError{Op: "dial", Err: errors.New("connection refused")})
+	err := tbi.joinedError()
+	if err == nil {
+		t.Fatal("expected an error")
+	}
+	if consumererror.IsPermanent(err) {
+		t.Error("indexer-level transport error must be retryable, not permanent")
+	}
+}
+
+func TestTraceProcessItemFailureTransportErrorRetryable(t *testing.T) {
+	tbi := &traceBulkIndexer{}
+	tbi.processItemFailure(opensearchapi.BulkRespItem{Status: 0}, &net.OpError{Op: "read", Err: errors.New("i/o timeout")}, ptrace.NewTraces())
+	if len(tbi.errs) != 1 {
+		t.Fatalf("expected 1 error, got %d", len(tbi.errs))
+	}
+	if consumererror.IsPermanent(tbi.errs[0]) {
+		t.Error("per-item transport error must be retryable, not permanent")
+	}
+
+	tbi2 := &traceBulkIndexer{}
+	tbi2.processItemFailure(opensearchapi.BulkRespItem{Status: 0}, errors.New("encode failed"), ptrace.NewTraces())
+	if len(tbi2.errs) != 1 {
+		t.Fatalf("expected 1 error, got %d", len(tbi2.errs))
+	}
+	if !consumererror.IsPermanent(tbi2.errs[0]) {
+		t.Error("encoding error must remain permanent")
+	}
+}
+
 func TestNewTraceBulkIndexerWithPipeline(t *testing.T) {
 	tests := []struct {
 		name     string