Avoid the exporter for being stuck when telemetry data is bigger than batch.max_size

Signed-off-by: Israel Blancas <[email protected]>

Author: iblancasa

.chloggen/12893.yaml

@@ -0,0 +1,25 @@
+# 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. otlpreceiver)
+component: exporterhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Prevents the exporter for being stuck when telemetry data is bigger than batch.max_size"
+
+# One or more tracking issues or pull requests related to the change
+issues: [12893]
+
+# (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:
+
+# 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: []

exporter/exporterhelper/internal/partialsuccess/partial_success.go

@@ -0,0 +1,17 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package partialsuccess // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/partialsuccess"
+
+import "fmt"
+
+var _ error = &PartialSuccessError{}
+
+type PartialSuccessError struct {
+	FailureCount int
+	Reason       string
+}
+
+func (e *PartialSuccessError) Error() string {
+	return fmt.Sprintf("partial success: %d failed: %s", e.FailureCount, e.Reason)
+}

exporter/exporterhelper/internal/queuebatch/multi_batcher.go

@@ -6,6 +6,8 @@ import (
 	"context"
 	"sync"
 
+	"go.uber.org/zap"
+
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
@@ -17,6 +19,7 @@ type batcherSettings[T any] struct {
 	partitioner Partitioner[T]
 	next        sender.SendFunc[T]
 	maxWorkers  int
+	logger      *zap.Logger
 }
 
 type multiBatcher struct {
@@ -26,6 +29,7 @@ type multiBatcher struct {
 	sizer       request.Sizer[request.Request]
 	partitioner Partitioner[request.Request]
 	consumeFunc sender.SendFunc[request.Request]
+	logger      *zap.Logger
 
 	singleShard *shardBatcher
 	shards      sync.Map
@@ -41,17 +45,19 @@ func newMultiBatcher(bCfg BatchConfig, bSet batcherSettings[request.Request]) *m
 			workerPool <- struct{}{}
 		}
 	}
+
 	mb := &multiBatcher{
 		cfg:         bCfg,
 		wp:          workerPool,
 		sizerType:   bSet.sizerType,
 		sizer:       bSet.sizer,
 		partitioner: bSet.partitioner,
 		consumeFunc: bSet.next,
+		logger:      bSet.logger,
 	}
 
 	if bSet.partitioner == nil {
-		mb.singleShard = newShard(mb.cfg, mb.sizerType, mb.sizer, mb.wp, mb.consumeFunc)
+		mb.singleShard = newShard(mb.cfg, mb.sizerType, mb.sizer, mb.wp, mb.consumeFunc, mb.logger.With(zap.String("shard", "single")))
 	}
 	return mb
 }
@@ -67,7 +73,7 @@ func (mb *multiBatcher) getShard(ctx context.Context, req request.Request) *shar
 	if found {
 		return s.(*shardBatcher)
 	}
-	newS := newShard(mb.cfg, mb.sizerType, mb.sizer, mb.wp, mb.consumeFunc)
+	newS := newShard(mb.cfg, mb.sizerType, mb.sizer, mb.wp, mb.consumeFunc, mb.logger.With(zap.String("shard", key)))
 	newS.start(ctx, nil)
 	s, loaded := mb.shards.LoadOrStore(key, newS)
 	// If not loaded, there was a race condition in adding the new shard. Shutdown the newly created shard.

exporter/exporterhelper/internal/queuebatch/multi_batcher_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
@@ -33,6 +34,7 @@ func TestMultiBatcher_NoTimeout(t *testing.T) {
 		}),
 		next:       sink.Export,
 		maxWorkers: 1,
+		logger:     zap.NewNop(),
 	})
 
 	require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))
@@ -84,6 +86,7 @@ func TestMultiBatcher_Timeout(t *testing.T) {
 		}),
 		next:       sink.Export,
 		maxWorkers: 1,
+		logger:     zap.NewNop(),
 	})
 
 	require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))

exporter/exporterhelper/internal/queuebatch/queue_batch.go

@@ -8,6 +8,8 @@ import (
 	"errors"
 	"fmt"
 
+	"go.uber.org/zap"
+
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
@@ -72,6 +74,7 @@ func newQueueBatch(
 				partitioner: set.Partitioner,
 				next:        next,
 				maxWorkers:  cfg.NumConsumers,
+				logger:      set.Telemetry.Logger.With(zap.String("component", "multi_batcher")),
 			})
 		} else {
 			b = newMultiBatcher(*cfg.Batch, batcherSettings[request.Request]{
@@ -80,6 +83,7 @@ func newQueueBatch(
 				partitioner: set.Partitioner,
 				next:        next,
 				maxWorkers:  cfg.NumConsumers,
+				logger:      set.Telemetry.Logger.With(zap.String("component", "multi_batcher")),
 			})
 		}
 		// Keep the number of queue consumers to 1 if batching is enabled until we support sharding as described in

exporter/exporterhelper/internal/queuebatch/shard_batcher.go

@@ -5,12 +5,15 @@ package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhel
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"time"
 
 	"go.uber.org/multierr"
+	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/partialsuccess"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
 )
@@ -33,16 +36,18 @@ type shardBatcher struct {
 	currentBatch   *batch
 	timer          *time.Timer
 	shutdownCh     chan struct{}
+	logger         *zap.Logger
 }
 
-func newShard(cfg BatchConfig, sizerType request.SizerType, sizer request.Sizer[request.Request], workerPool chan struct{}, next sender.SendFunc[request.Request]) *shardBatcher {
+func newShard(cfg BatchConfig, sizerType request.SizerType, sizer request.Sizer[request.Request], workerPool chan struct{}, next sender.SendFunc[request.Request], logger *zap.Logger) *shardBatcher {
 	return &shardBatcher{
 		cfg:         cfg,
 		workerPool:  workerPool,
 		sizerType:   sizerType,
 		sizer:       sizer,
 		consumeFunc: next,
 		shutdownCh:  make(chan struct{}, 1),
+		logger:      logger,
 	}
 }
 
@@ -57,7 +62,24 @@ func (qb *shardBatcher) Consume(ctx context.Context, req request.Request, done D
 
 	if qb.currentBatch == nil {
 		reqList, mergeSplitErr := req.MergeSplit(ctx, int(qb.cfg.MaxSize), qb.sizerType, nil)
-		if mergeSplitErr != nil || len(reqList) == 0 {
+		if mergeSplitErr != nil {
+			var partialSuccessErr *partialsuccess.PartialSuccessError
+			if !errors.As(mergeSplitErr, &partialSuccessErr) {
+				done.OnDone(mergeSplitErr)
+				qb.currentBatchMu.Unlock()
+				return
+			}
+			qb.logger.Warn(
+				"failed to split request",
+				zap.Int("failure_count", partialSuccessErr.FailureCount),
+				zap.String("reason", partialSuccessErr.Reason),
+			)
+			done.OnDone(mergeSplitErr)
+			qb.currentBatchMu.Unlock()
+			return
+		}
+
+		if len(reqList) == 0 {
 			done.OnDone(mergeSplitErr)
 			qb.currentBatchMu.Unlock()
 			return
@@ -92,7 +114,24 @@ func (qb *shardBatcher) Consume(ctx context.Context, req request.Request, done D
 
 	reqList, mergeSplitErr := qb.currentBatch.req.MergeSplit(ctx, int(qb.cfg.MaxSize), qb.sizerType, req)
 	// If failed to merge signal all Done callbacks from the current batch as well as the current request and reset the current batch.
-	if mergeSplitErr != nil || len(reqList) == 0 {
+	if mergeSplitErr != nil {
+		var partialSuccessErr *partialsuccess.PartialSuccessError
+		if !errors.As(mergeSplitErr, &partialSuccessErr) {
+			done.OnDone(mergeSplitErr)
+			qb.currentBatchMu.Unlock()
+			return
+		}
+		qb.logger.Warn(
+			"failed to split request",
+			zap.Int("failure_count", partialSuccessErr.FailureCount),
+			zap.String("reason", partialSuccessErr.Reason),
+		)
+		done.OnDone(mergeSplitErr)
+		qb.currentBatchMu.Unlock()
+		return
+	}
+
+	if len(reqList) == 0 {
 		done.OnDone(mergeSplitErr)
 		qb.currentBatchMu.Unlock()
 		return