[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

@@ -60,7 +60,7 @@ func TestQueueOptionsWithRequestExporter(t *testing.T) {
 	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport,
 		WithQueueBatchSettings(newFakeQueueBatch(&requesttest.FakeRequest{Items: 1})),
 		WithRetry(configretry.NewDefaultBackOffConfig()),
-		WithQueueBatch(qCfg, queuebatch.Settings[request.Request]{}))
+		WithQueueBatch(qCfg, QueueBatchSettings[request.Request]{}))
 	require.Error(t, err)
 }
 
@@ -152,8 +152,13 @@ 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(mr request.Request) QueueBatchSettings[request.Request] {
+	return QueueBatchSettings[request.Request]{
+		Encoding: &fakeEncoding{mr: mr},
+		Sizers: map[exporterbatcher.SizerType]queuebatch.Sizer[request.Request]{
+			exporterbatcher.SizerTypeRequests: queuebatch.RequestsSizer[request.Request]{},
+		},
+	}
 }
 
 type fakeEncoding struct {

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,13 @@ 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
@@ -42,24 +38,14 @@ func TestDisabledBatcher_Basic(t *testing.T) {
 			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(),

exporter/exporterhelper/internal/queuebatch/queue.go

@@ -8,9 +8,6 @@ import (
 	"errors"
 
 	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/exporter/exporterbatcher"
-	"go.opentelemetry.io/collector/exporter/exporterqueue"
-	"go.opentelemetry.io/collector/pipeline"
 )
 
 // ErrQueueIsFull is the error returned when an item is offered to the Queue and the queue is full and setup to
@@ -55,47 +52,3 @@ type readableQueue[T any] interface {
 	// If the queue is stopped returns false, otherwise true.
 	Read(context.Context) (context.Context, T, Done, bool)
 }
-
-// QueueSettings defines settings for creating a queue.
-type QueueSettings[K any] struct {
-	Signal    pipeline.Signal
-	ID        component.ID
-	Telemetry component.TelemetrySettings
-	Settings[K]
-}
-
-// NewQueue returns a queue
-// 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 NewQueue[T any](_ context.Context, set QueueSettings[T], cfg exporterqueue.Config, consume ConsumeFunc[T]) (Queue[T], error) {
-	if !cfg.Enabled {
-		return newDisabledQueue(consume), nil
-	}
-
-	sizer, ok := set.Sizers[exporterbatcher.SizerTypeRequests]
-	if !ok {
-		return nil, errors.New("unsupported queue_batch sizer")
-	}
-
-	if cfg.StorageID != nil {
-		q := newPersistentQueue[T](persistentQueueSettings[T]{
-			sizer:     sizer,
-			capacity:  int64(cfg.QueueSize),
-			blocking:  cfg.Blocking,
-			signal:    set.Signal,
-			storageID: *cfg.StorageID,
-			encoding:  set.Encoding,
-			id:        set.ID,
-			telemetry: set.Telemetry,
-		})
-		return newAsyncQueue(q, cfg.NumConsumers, consume), nil
-	}
-
-	q := newMemoryQueue[T](memoryQueueSettings[T]{
-		sizer:    sizer,
-		capacity: int64(cfg.QueueSize),
-		blocking: cfg.Blocking,
-	})
-
-	return newAsyncQueue(q, cfg.NumConsumers, consume), nil
-}