[chore] Start merging queue and batcher

Signed-off-by: Bogdan Drutu <[email protected]>

Author: bogdandrutu

exporter/exporterhelper/internal/base_exporter.go

@@ -48,7 +48,7 @@ type BaseExporter struct {
 	timeoutCfg TimeoutConfig
 	retryCfg   configretry.BackOffConfig
 
-	queueBatchSettings queuebatch.Settings[request.Request]
+	queueBatchSettings QueueBatchSettings[request.Request]
 	queueCfg           exporterqueue.Config
 	batcherCfg         exporterbatcher.Config
 }
@@ -91,11 +91,12 @@ func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, pusher sende
 	}
 
 	if be.queueCfg.Enabled || be.batcherCfg.Enabled {
-		qSet := queuebatch.QueueSettings[request.Request]{
+		qSet := queuebatch.Settings[request.Request]{
 			Signal:    signal,
 			ID:        set.ID,
 			Telemetry: set.TelemetrySettings,
-			Settings:  be.queueBatchSettings,
+			Encoding:  be.queueBatchSettings.Encoding,
+			Sizers:    be.queueBatchSettings.Sizers,
 		}
 		be.QueueSender, err = NewQueueSender(qSet, be.queueCfg, be.batcherCfg, be.ExportFailureMessage, be.firstSender)
 		if err != nil {
@@ -207,7 +208,7 @@ func WithQueue(cfg exporterqueue.Config) Option {
 // This option should be used with the new exporter helpers New[Traces|Metrics|Logs]RequestExporter.
 // Experimental: This API is at the early stage of development and may change without backward compatibility
 // until https://github.com/open-telemetry/opentelemetry-collector/issues/8122 is resolved.
-func WithQueueBatch(cfg exporterqueue.Config, set queuebatch.Settings[request.Request]) Option {
+func WithQueueBatch(cfg exporterqueue.Config, set QueueBatchSettings[request.Request]) Option {
 	return func(o *BaseExporter) error {
 		if !cfg.Enabled {
 			o.ExportFailureMessage += " Try enabling sending_queue to survive temporary failures."
@@ -244,9 +245,9 @@ func WithBatcher(cfg exporterbatcher.Config) Option {
 	}
 }
 
-// WithQueueBatchSettings is used to set the queuebatch.Settings for the new request based exporter helper.
+// WithQueueBatchSettings is used to set the QueueBatchSettings for the new request based exporter helper.
 // It must be provided as the first option when creating a new exporter helper.
-func WithQueueBatchSettings(set queuebatch.Settings[request.Request]) Option {
+func WithQueueBatchSettings(set QueueBatchSettings[request.Request]) Option {
 	return func(o *BaseExporter) error {
 		o.queueBatchSettings = set
 		return nil

exporter/exporterhelper/internal/base_exporter_test.go

@@ -58,9 +58,9 @@ func TestQueueOptionsWithRequestExporter(t *testing.T) {
 	storageID := component.NewID(component.MustNewType("test"))
 	qCfg.StorageID = &storageID
 	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport,
-		WithQueueBatchSettings(newFakeQueueBatch(&requesttest.FakeRequest{Items: 1})),
+		WithQueueBatchSettings(newFakeQueueBatch()),
 		WithRetry(configretry.NewDefaultBackOffConfig()),
-		WithQueueBatch(qCfg, queuebatch.Settings[request.Request]{}))
+		WithQueueBatch(qCfg, QueueBatchSettings[request.Request]{}))
 	require.Error(t, err)
 }
 
@@ -73,7 +73,8 @@ func TestBaseExporterLogging(t *testing.T) {
 	qCfg := exporterqueue.NewDefaultConfig()
 	qCfg.Enabled = false
 	bs, err := NewBaseExporter(set, pipeline.SignalMetrics, errExport,
-		WithQueueBatch(qCfg, newFakeQueueBatch(&requesttest.FakeRequest{})),
+		WithQueueBatchSettings(newFakeQueueBatch()),
+		WithQueue(qCfg),
 		WithBatcher(exporterbatcher.NewDefaultConfig()),
 		WithRetry(rCfg))
 	require.NoError(t, err)
@@ -97,7 +98,7 @@ func TestQueueRetryWithDisabledQueue(t *testing.T) {
 		{
 			name: "WithQueue",
 			queueOptions: []Option{
-				WithQueueBatchSettings(newFakeQueueBatch(&requesttest.FakeRequest{Items: 1})),
+				WithQueueBatchSettings(newFakeQueueBatch()),
 				func() Option {
 					qs := exporterqueue.NewDefaultConfig()
 					qs.Enabled = false
@@ -116,7 +117,7 @@ func TestQueueRetryWithDisabledQueue(t *testing.T) {
 				func() Option {
 					qs := exporterqueue.NewDefaultConfig()
 					qs.Enabled = false
-					return WithQueueBatch(qs, newFakeQueueBatch(&requesttest.FakeRequest{Items: 1}))
+					return WithQueueBatch(qs, newFakeQueueBatch())
 				}(),
 				func() Option {
 					bs := exporterbatcher.NewDefaultConfig()
@@ -152,18 +153,21 @@ func noopExport(context.Context, request.Request) error {
 	return nil
 }
 
-func newFakeQueueBatch(mr request.Request) queuebatch.Settings[request.Request] {
-	return queuebatch.Settings[request.Request]{Encoding: &fakeEncoding{mr: mr}}
+func newFakeQueueBatch() QueueBatchSettings[request.Request] {
+	return QueueBatchSettings[request.Request]{
+		Encoding: fakeEncoding{},
+		Sizers: map[exporterbatcher.SizerType]queuebatch.Sizer[request.Request]{
+			exporterbatcher.SizerTypeRequests: queuebatch.RequestsSizer[request.Request]{},
+		},
+	}
 }
 
-type fakeEncoding struct {
-	mr request.Request
-}
+type fakeEncoding struct{}
 
 func (f fakeEncoding) Marshal(request.Request) ([]byte, error) {
 	return []byte("mockRequest"), nil
 }
 
 func (f fakeEncoding) Unmarshal([]byte) (request.Request, error) {
-	return f.mr, nil
+	return &requesttest.FakeRequest{}, nil
 }

exporter/exporterhelper/internal/queue_sender.go

@@ -15,8 +15,14 @@ import (
 	"go.opentelemetry.io/collector/exporter/exporterqueue"
 )
 
+// QueueBatchSettings is a subset of the queuebatch.Settings that are needed when used within an Exporter.
+type QueueBatchSettings[K any] struct {
+	Encoding exporterqueue.Encoding[K]
+	Sizers   map[exporterbatcher.SizerType]queuebatch.Sizer[K]
+}
+
 func NewQueueSender(
-	qSet queuebatch.QueueSettings[request.Request],
+	qSet queuebatch.Settings[request.Request],
 	qCfg exporterqueue.Config,
 	bCfg exporterbatcher.Config,
 	exportFailureMessage string,

exporter/exporterhelper/internal/queue_sender_test.go

@@ -26,14 +26,12 @@ import (
 )
 
 func TestNewQueueSenderFailedRequestDropped(t *testing.T) {
-	qSet := queuebatch.QueueSettings[request.Request]{
+	qSet := queuebatch.Settings[request.Request]{
 		Signal:    pipeline.SignalMetrics,
 		ID:        component.NewID(exportertest.NopType),
 		Telemetry: componenttest.NewNopTelemetrySettings(),
-		Settings: queuebatch.Settings[request.Request]{
-			Sizers: map[exporterbatcher.SizerType]queuebatch.Sizer[request.Request]{
-				exporterbatcher.SizerTypeRequests: queuebatch.RequestsSizer[request.Request]{},
-			},
+		Sizers: map[exporterbatcher.SizerType]queuebatch.Sizer[request.Request]{
+			exporterbatcher.SizerTypeRequests: queuebatch.RequestsSizer[request.Request]{},
 		},
 	}
 	logger, observed := observer.New(zap.ErrorLevel)

exporter/exporterhelper/internal/queuebatch/batcher.go

@@ -7,20 +7,10 @@ import (
 	"context"
 
 	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/exporter/exporterbatcher"
-	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
-	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
 )
 
 // Batcher is in charge of reading items from the queue and send them out asynchronously.
 type Batcher[K any] interface {
 	component.Component
 	Consume(context.Context, K, Done)
 }
-
-func NewBatcher(batchCfg exporterbatcher.Config, exportFunc sender.SendFunc[request.Request], maxWorkers int) (Batcher[request.Request], error) {
-	if !batchCfg.Enabled {
-		return newDisabledBatcher[request.Request](exportFunc), nil
-	}
-	return newDefaultBatcher(batchCfg, exportFunc, maxWorkers), nil
-}

exporter/exporterhelper/internal/queuebatch/default_batcher_test.go

@@ -44,8 +44,7 @@ func TestDefaultBatcher_NoSplit_MinThresholdZero_TimeoutDisabled(t *testing.T) {
 			}
 
 			sink := requesttest.NewSink()
-			ba, err := NewBatcher(cfg, sink.Export, tt.maxWorkers)
-			require.NoError(t, err)
+			ba := newDefaultBatcher(cfg, sink.Export, tt.maxWorkers)
 			require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))
 			t.Cleanup(func() {
 				require.NoError(t, ba.Shutdown(context.Background()))
@@ -95,8 +94,7 @@ func TestDefaultBatcher_NoSplit_TimeoutDisabled(t *testing.T) {
 			}
 
 			sink := requesttest.NewSink()
-			ba, err := NewBatcher(cfg, sink.Export, tt.maxWorkers)
-			require.NoError(t, err)
+			ba := newDefaultBatcher(cfg, sink.Export, tt.maxWorkers)
 			require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))
 
 			done := newFakeDone()
@@ -161,8 +159,7 @@ func TestDefaultBatcher_NoSplit_WithTimeout(t *testing.T) {
 			}
 
 			sink := requesttest.NewSink()
-			ba, err := NewBatcher(cfg, sink.Export, tt.maxWorkers)
-			require.NoError(t, err)
+			ba := newDefaultBatcher(cfg, sink.Export, tt.maxWorkers)
 			require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))
 			t.Cleanup(func() {
 				require.NoError(t, ba.Shutdown(context.Background()))
@@ -218,8 +215,7 @@ func TestDefaultBatcher_Split_TimeoutDisabled(t *testing.T) {
 			}
 
 			sink := requesttest.NewSink()
-			ba, err := NewBatcher(cfg, sink.Export, tt.maxWorkers)
-			require.NoError(t, err)
+			ba := newDefaultBatcher(cfg, sink.Export, tt.maxWorkers)
 			require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))
 
 			done := newFakeDone()
@@ -265,8 +261,7 @@ func TestDefaultBatcher_Shutdown(t *testing.T) {
 	batchCfg.FlushTimeout = 100 * time.Second
 
 	sink := requesttest.NewSink()
-	ba, err := NewBatcher(batchCfg, sink.Export, 2)
-	require.NoError(t, err)
+	ba := newDefaultBatcher(batchCfg, sink.Export, 2)
 	require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))
 
 	done := newFakeDone()
@@ -292,8 +287,7 @@ func TestDefaultBatcher_MergeError(t *testing.T) {
 	batchCfg.MaxSize = 7
 
 	sink := requesttest.NewSink()
-	ba, err := NewBatcher(batchCfg, sink.Export, 2)
-	require.NoError(t, err)
+	ba := newDefaultBatcher(batchCfg, sink.Export, 2)
 
 	require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))
 	t.Cleanup(func() {

exporter/exporterhelper/internal/queuebatch/disabled_batcher_test.go

@@ -12,17 +12,12 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
-	"go.opentelemetry.io/collector/exporter/exporterbatcher"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
-	"go.opentelemetry.io/collector/exporter/exporterqueue"
-	"go.opentelemetry.io/collector/exporter/exportertest"
-	"go.opentelemetry.io/collector/pipeline"
 )
 
-func TestDisabledBatcher_Basic(t *testing.T) {
+func TestDisabledBatcher(t *testing.T) {
 	tests := []struct {
 		name       string
 		maxWorkers int
@@ -38,28 +33,15 @@ func TestDisabledBatcher_Basic(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			cfg := exporterbatcher.NewDefaultConfig()
-			cfg.Enabled = false
-
 			sink := requesttest.NewSink()
-			ba, err := NewBatcher(cfg, sink.Export, tt.maxWorkers)
-			require.NoError(t, err)
+			ba := newDisabledBatcher(sink.Export)
 
-			q, err := NewQueue[request.Request](
-				context.Background(),
-				QueueSettings[request.Request]{
-					Signal:    pipeline.SignalTraces,
-					ID:        component.NewID(exportertest.NopType),
-					Telemetry: componenttest.NewNopTelemetrySettings(),
-					Settings: Settings[request.Request]{
-						Sizers: map[exporterbatcher.SizerType]Sizer[request.Request]{
-							exporterbatcher.SizerTypeRequests: RequestsSizer[request.Request]{},
-						},
-					},
-				},
-				exporterqueue.NewDefaultConfig(),
-				ba.Consume)
-			require.NoError(t, err)
+			mq := newMemoryQueue[request.Request](memoryQueueSettings[request.Request]{
+				sizer:    RequestsSizer[request.Request]{},
+				capacity: 1000,
+				blocking: true,
+			})
+			q := newAsyncQueue(mq, tt.maxWorkers, ba.Consume)
 
 			require.NoError(t, q.Start(context.Background(), componenttest.NewNopHost()))
 			require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))

exporter/exporterhelper/internal/queuebatch/obs_queue.go

@@ -30,7 +30,7 @@ type obsQueue[T request.Request] struct {
 	enqueueFailedInst metric.Int64Counter
 }
 
-func newObsQueue[T request.Request](set QueueSettings[T], delegate Queue[T]) (Queue[T], error) {
+func newObsQueue[T request.Request](set Settings[T], delegate Queue[T]) (Queue[T], error) {
 	tb, err := metadata.NewTelemetryBuilder(set.Telemetry)
 	if err != nil {
 		return nil, err

exporter/exporterhelper/internal/queuebatch/obs_queue_test.go

@@ -51,7 +51,7 @@ func TestObsQueueLogsSizeCapacity(t *testing.T) {
 	tt := componenttest.NewTelemetry()
 	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
 
-	te, err := newObsQueue[request.Request](QueueSettings[request.Request]{
+	te, err := newObsQueue[request.Request](Settings[request.Request]{
 		Signal:    pipeline.SignalLogs,
 		ID:        exporterID,
 		Telemetry: tt.NewTelemetrySettings(),
@@ -82,7 +82,7 @@ func TestObsQueueLogsFailure(t *testing.T) {
 	tt := componenttest.NewTelemetry()
 	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
 
-	te, err := newObsQueue[request.Request](QueueSettings[request.Request]{
+	te, err := newObsQueue[request.Request](Settings[request.Request]{
 		Signal:    pipeline.SignalLogs,
 		ID:        exporterID,
 		Telemetry: tt.NewTelemetrySettings(),
@@ -103,7 +103,7 @@ func TestObsQueueTracesSizeCapacity(t *testing.T) {
 	tt := componenttest.NewTelemetry()
 	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
 
-	te, err := newObsQueue[request.Request](QueueSettings[request.Request]{
+	te, err := newObsQueue[request.Request](Settings[request.Request]{
 		Signal:    pipeline.SignalTraces,
 		ID:        exporterID,
 		Telemetry: tt.NewTelemetrySettings(),
@@ -134,7 +134,7 @@ func TestObsQueueTracesFailure(t *testing.T) {
 	tt := componenttest.NewTelemetry()
 	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
 
-	te, err := newObsQueue[request.Request](QueueSettings[request.Request]{
+	te, err := newObsQueue[request.Request](Settings[request.Request]{
 		Signal:    pipeline.SignalTraces,
 		ID:        exporterID,
 		Telemetry: tt.NewTelemetrySettings(),
@@ -155,7 +155,7 @@ func TestObsQueueMetrics(t *testing.T) {
 	tt := componenttest.NewTelemetry()
 	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
 
-	te, err := newObsQueue[request.Request](QueueSettings[request.Request]{
+	te, err := newObsQueue[request.Request](Settings[request.Request]{
 		Signal:    pipeline.SignalMetrics,
 		ID:        exporterID,
 		Telemetry: tt.NewTelemetrySettings(),
@@ -186,7 +186,7 @@ func TestObsQueueMetricsFailure(t *testing.T) {
 	tt := componenttest.NewTelemetry()
 	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
 
-	te, err := newObsQueue[request.Request](QueueSettings[request.Request]{
+	te, err := newObsQueue[request.Request](Settings[request.Request]{
 		Signal:    pipeline.SignalMetrics,
 		ID:        exporterID,
 		Telemetry: tt.NewTelemetrySettings(),