fix(otelconsumer): do not retry 401 errors from Elasticsearch (elastic#50261)

* fix(otelconsumer): treat 401 errors from Elasticsearch as permanent

* refactor: prevent tight coupling with docappender library

I'm not sure if it makes things better (less coupling) or worse (less explicit).

* libbeat: mage check

---------

Co-authored-by: Craig MacKenzie <craig.mackenzie@elastic.co>

Author: 2 people

libbeat/otel/otelconsumer/otelconsumer.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http"
 	"os"
 	"runtime"
 	"time"
@@ -47,6 +48,12 @@ const (
 	esDocumentIDAttribute = "elasticsearch.document_id"
 )
 
+// statusCodeError is satisfied by errors that carry an HTTP status code,
+// such as docappender.ErrorFlushFailed errors returned from the OTelCol Elasticsearch exporter.
+type statusCodeError interface {
+	StatusCode() int
+}
+
 type otelConsumer struct {
 	observer       outputs.Observer
 	logsConsumer   consumer.Logs
@@ -193,13 +200,20 @@ func (out *otelConsumer) logsPublish(ctx context.Context, batch publisher.Batch)
 
 	err := out.logsConsumer.ConsumeLogs(otelctx.NewConsumerContext(ctx, out.beatInfo), pLogs)
 	if err != nil {
+		// Work around the fact that Elasticsearch exporter returns 401 as a non-permanent error.
+		isAuthorizationError := false
+		var statusErr statusCodeError
+		if errors.As(err, &statusErr) {
+			isAuthorizationError = statusErr.StatusCode() == http.StatusUnauthorized
+		}
+
 		// Permanent errors shouldn't be retried. This tipically means
 		// the data cannot be serialized by the exporter that is attached
 		// to the pipeline or when the destination refuses the data because
 		// it cannot decode it. Retrying in this case is useless.
 		//
 		// See https://github.com/open-telemetry/opentelemetry-collector/blob/1c47d89/receiver/doc.go#L23-L40
-		if consumererror.IsPermanent(err) {
+		if consumererror.IsPermanent(err) || isAuthorizationError {
 			st.PermanentErrors(len(events))
 			batch.Drop()
 		} else {

libbeat/otel/otelconsumer/otelconsumer_test.go

@@ -240,6 +240,32 @@ func TestPublish(t *testing.T) {
 		assert.Equal(t, outest.BatchRetry, batch.Signals[0].Tag)
 	})
 
+	t.Run("drops batch on 401 Unauthorized error", func(t *testing.T) {
+		batch := outest.NewBatch(event1, event2, event3)
+
+		otelConsumer := makeOtelConsumer(t, func(ctx context.Context, ld plog.Logs) error {
+			return &testStatusCodeError{statusCode: 401, msg: "flush failed (401): unauthorized"}
+		})
+
+		err := otelConsumer.Publish(ctx, batch)
+		assert.NoError(t, err)
+		assert.Len(t, batch.Signals, 1)
+		assert.Equal(t, outest.BatchDrop, batch.Signals[0].Tag)
+	})
+
+	t.Run("retries batch on non-401 status code error", func(t *testing.T) {
+		batch := outest.NewBatch(event1, event2, event3)
+
+		otelConsumer := makeOtelConsumer(t, func(ctx context.Context, ld plog.Logs) error {
+			return &testStatusCodeError{statusCode: 500, msg: "flush failed (500): internal server error"}
+		})
+
+		err := otelConsumer.Publish(ctx, batch)
+		assert.NoError(t, err)
+		assert.Len(t, batch.Signals, 1)
+		assert.Equal(t, outest.BatchRetry, batch.Signals[0].Tag)
+	})
+
 	t.Run("sets the elasticsearchexporter doc id attribute from metadata", func(t *testing.T) {
 		batch := outest.NewBatch(event4)
 
@@ -364,3 +390,11 @@ func checkEventsActive(reg *monitoring.Registry) int64 {
 	outputSnapshot := monitoring.CollectFlatSnapshot(reg, monitoring.Full, true)
 	return outputSnapshot.Ints["events.active"]
 }
+
+type testStatusCodeError struct {
+	statusCode int
+	msg        string
+}
+
+func (e *testStatusCodeError) Error() string   { return e.msg }
+func (e *testStatusCodeError) StatusCode() int { return e.statusCode }