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/default_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"
 )
@@ -40,9 +43,10 @@ type defaultBatcher struct {
 	currentBatch   *batch
 	timer          *time.Timer
 	shutdownCh     chan struct{}
+	logger         *zap.Logger
 }
 
-func newDefaultBatcher(bCfg BatchConfig, bSet batcherSettings[request.Request]) *defaultBatcher {
+func newDefaultBatcher(bCfg BatchConfig, bSet batcherSettings[request.Request], set Settings[request.Request]) *defaultBatcher {
 	// TODO: Determine what is the right behavior for this in combination with async queue.
 	var workerPool chan struct{}
 	if bSet.maxWorkers != 0 {
@@ -59,6 +63,7 @@ func newDefaultBatcher(bCfg BatchConfig, bSet batcherSettings[request.Request])
 		consumeFunc: bSet.next,
 		stopWG:      sync.WaitGroup{},
 		shutdownCh:  make(chan struct{}, 1),
+		logger:      set.Telemetry.Logger,
 	}
 }
 
@@ -73,8 +78,21 @@ func (qb *defaultBatcher) Consume(ctx context.Context, req request.Request, done
 
 	if qb.currentBatch == nil {
 		reqList, mergeSplitErr := req.MergeSplit(ctx, int(qb.cfg.MaxSize), qb.sizerType, nil)
-		if mergeSplitErr != nil || len(reqList) == 0 {
-			done.OnDone(mergeSplitErr)
+		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),
+			)
+		}
+
+		if len(reqList) == 0 {
 			qb.currentBatchMu.Unlock()
 			return
 		}
@@ -108,8 +126,20 @@ func (qb *defaultBatcher) Consume(ctx context.Context, req request.Request, done
 
 	reqList, mergeSplitErr := qb.currentBatch.req.MergeSplit(ctx, int(qb.cfg.MaxSize), qb.sizerType, req)
 	// If failed to merge signal all Done callbacks from current batch as well as the current request and reset the current batch.
-	if mergeSplitErr != nil || len(reqList) == 0 {
-		done.OnDone(mergeSplitErr)
+	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),
+		)
+	}
+	if len(reqList) == 0 {
 		qb.currentBatchMu.Unlock()
 		return
 	}

exporter/exporterhelper/internal/queuebatch/default_batcher_test.go

@@ -13,7 +13,9 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
 
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
@@ -64,6 +66,11 @@ func TestDefaultBatcher_NoSplit_MinThresholdZero_TimeoutDisabled(t *testing.T) {
 				sizer:      tt.sizer,
 				next:       sink.Export,
 				maxWorkers: tt.maxWorkers,
+			}, Settings[request.Request]{
+				ID: exporterID,
+				Telemetry: component.TelemetrySettings{
+					Logger: zap.NewNop(),
+				},
 			})
 			require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))
 			t.Cleanup(func() {
@@ -134,7 +141,13 @@ func TestDefaultBatcher_NoSplit_TimeoutDisabled(t *testing.T) {
 				sizer:      tt.sizer,
 				next:       sink.Export,
 				maxWorkers: tt.maxWorkers,
-			})
+			},
+				Settings[request.Request]{
+					ID: exporterID,
+					Telemetry: component.TelemetrySettings{
+						Logger: zap.NewNop(),
+					},
+				})
 			require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))
 
 			done := newFakeDone()
@@ -219,6 +232,11 @@ func TestDefaultBatcher_NoSplit_WithTimeout(t *testing.T) {
 				sizer:      tt.sizer,
 				next:       sink.Export,
 				maxWorkers: tt.maxWorkers,
+			}, Settings[request.Request]{
+				ID: exporterID,
+				Telemetry: component.TelemetrySettings{
+					Logger: zap.NewNop(),
+				},
 			})
 			require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))
 			t.Cleanup(func() {
@@ -295,6 +313,11 @@ func TestDefaultBatcher_Split_TimeoutDisabled(t *testing.T) {
 				sizer:      tt.sizer,
 				next:       sink.Export,
 				maxWorkers: tt.maxWorkers,
+			}, Settings[request.Request]{
+				ID: exporterID,
+				Telemetry: component.TelemetrySettings{
+					Logger: zap.NewNop(),
+				},
 			})
 			require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))
 
@@ -347,6 +370,11 @@ func TestDefaultBatcher_Shutdown(t *testing.T) {
 		sizer:      request.NewItemsSizer(),
 		next:       sink.Export,
 		maxWorkers: 2,
+	}, Settings[request.Request]{
+		ID: exporterID,
+		Telemetry: component.TelemetrySettings{
+			Logger: zap.NewNop(),
+		},
 	})
 	require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))
 
@@ -380,6 +408,11 @@ func TestDefaultBatcher_MergeError(t *testing.T) {
 		sizer:      request.NewItemsSizer(),
 		next:       sink.Export,
 		maxWorkers: 2,
+	}, Settings[request.Request]{
+		ID: exporterID,
+		Telemetry: component.TelemetrySettings{
+			Logger: zap.NewNop(),
+		},
 	})
 
 	require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))

exporter/exporterhelper/internal/queuebatch/queue_batch.go

@@ -70,14 +70,14 @@ func newQueueBatch(
 				sizer:      request.NewItemsSizer(),
 				next:       next,
 				maxWorkers: cfg.NumConsumers,
-			})
+			}, set)
 		} else {
 			b = newDefaultBatcher(*cfg.Batch, batcherSettings[request.Request]{
 				sizerType:  cfg.Sizer,
 				sizer:      sizer,
 				next:       next,
 				maxWorkers: cfg.NumConsumers,
-			})
+			}, set)
 		}
 		// Keep the number of queue consumers to 1 if batching is enabled until we support sharding as described in
 		// https://github.com/open-telemetry/opentelemetry-collector/issues/12473

exporter/exporterhelper/logs_batch.go

@@ -6,7 +6,9 @@ package exporterhelper // import "go.opentelemetry.io/collector/exporter/exporte
 import (
 	"context"
 	"errors"
+	"strconv"
 
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/partialsuccess"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sizer"
 	"go.opentelemetry.io/collector/pdata/plog"
 )
@@ -36,7 +38,7 @@ func (req *logsRequest) MergeSplit(_ context.Context, maxSize int, szt RequestSi
 		return []Request{req}, nil
 	}
 
-	return req.split(maxSize, sz), nil
+	return req.split(maxSize, sz)
 }
 
 func (req *logsRequest) mergeTo(dst *logsRequest, sz sizer.LogsSizer) {
@@ -47,15 +49,34 @@ func (req *logsRequest) mergeTo(dst *logsRequest, sz sizer.LogsSizer) {
 	req.ld.ResourceLogs().MoveAndAppendTo(dst.ld.ResourceLogs())
 }
 
-func (req *logsRequest) split(maxSize int, sz sizer.LogsSizer) []Request {
+func (req *logsRequest) split(maxSize int, sz sizer.LogsSizer) ([]Request, error) {
 	var res []Request
-	for req.size(sz) > maxSize {
-		ld, rmSize := extractLogs(req.ld, maxSize, sz)
-		req.setCachedSize(req.size(sz) - rmSize)
-		res = append(res, newLogsRequest(ld))
+	var ld plog.Logs
+	rmSize := -1
+
+	previousSize := req.size(sz)
+
+	for req.size(sz) > maxSize && rmSize != 0 {
+		ld, rmSize = extractLogs(req.ld, maxSize, sz)
+		if ld.LogRecordCount() > 0 {
+			req.setCachedSize(req.size(sz) - rmSize)
+			res = append(res, newLogsRequest(ld))
+		}
 	}
+
+	if req.size(sz) == previousSize && req.size(sz) > maxSize {
+		// This means we cannot split the log and is not possible
+		// to fit it into the max size.
+		err := &partialsuccess.PartialSuccessError{
+			FailureCount: req.ld.LogRecordCount(),
+			Reason: "failed to split logs request: size is greater than max size. size: " +
+				strconv.Itoa(req.size(sz)) + ", max_size: " + strconv.Itoa(maxSize),
+		}
+		return res, err
+	}
+
 	res = append(res, req)
-	return res
+	return res, nil
 }
 
 // extractLogs extracts logs from the input logs and returns a new logs with the specified number of log records.

exporter/exporterhelper/logs_batch_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/partialsuccess"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sizer"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/testdata"
@@ -128,12 +129,13 @@ func TestMergeSplitLogs(t *testing.T) {
 
 func TestMergeSplitLogsBasedOnByteSize(t *testing.T) {
 	tests := []struct {
-		name     string
-		szt      RequestSizerType
-		maxSize  int
-		lr1      Request
-		lr2      Request
-		expected []Request
+		name               string
+		szt                RequestSizerType
+		maxSize            int
+		lr1                Request
+		lr2                Request
+		expected           []Request
+		expectPartialError bool
 	}{
 		{
 			name:     "both_requests_empty",
@@ -219,10 +221,30 @@ func TestMergeSplitLogsBasedOnByteSize(t *testing.T) {
 				}()),
 			},
 		},
+		{
+			name:    "unsplittable_large_log",
+			szt:     RequestSizerTypeBytes,
+			maxSize: 10,
+			lr1: newLogsRequest(func() plog.Logs {
+				ld := testdata.GenerateLogs(1)
+				ld.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0).Body().SetStr(string(make([]byte, 100)))
+				return ld
+			}()),
+			lr2:                nil,
+			expected:           []Request{},
+			expectPartialError: true,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			res, err := tt.lr1.MergeSplit(context.Background(), tt.maxSize, tt.szt, tt.lr2)
+			if tt.expectPartialError {
+				require.Error(t, err)
+				var partialErr *partialsuccess.PartialSuccessError
+				require.ErrorAs(t, err, &partialErr)
+				assert.Contains(t, err.Error(), "failed to split logs request: size is greater than max size")
+				return
+			}
 			require.NoError(t, err)
 			assert.Len(t, res, len(tt.expected))
 			for i := range res {