Change queue to embed the consumers (#12242)

This PR will facilitate:
1. The implementation of the disabled queue, can be a simpler call into
the "batcher" with the done callback setup and block until that is
called.
2. The implementation of the disabled batcher becomes trivial, and does
not suffer from the limitation that only one reader of the queue is
setup. Even though the request was sent async by multiple go-routines,
still one reader which may have been a bottleneck.

In a separate PR will try to get rid of readableQueue. Initial version
of the PR had "read" as a private func on Queue, but the "unused" linter
complain that `boundedMemoryQueue.read` was unused.

Technically this is a breaking change, but Queue is an experimental API
that we allow changes.

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

Author: bogdandrutu

.chloggen/consumer-queue.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: breaking
+
+# 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: Change queue to embed the async consumers.
+
+# One or more tracking issues or pull requests related to the change
+issues: [12242]
+
+# (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: [api]

exporter/exporterhelper/internal/obs_queue_test.go

@@ -24,8 +24,7 @@ import (
 )
 
 type fakeQueue[T any] struct {
-	component.StartFunc
-	component.ShutdownFunc
+	exporterqueue.Queue[T]
 	offerErr error
 	size     int64
 	capacity int64
@@ -39,10 +38,6 @@ func (fq *fakeQueue[T]) Capacity() int64 {
 	return fq.capacity
 }
 
-func (fq *fakeQueue[T]) Read(context.Context) (context.Context, T, exporterqueue.DoneCallback, bool) {
-	panic("implement me")
-}
-
 func (fq *fakeQueue[T]) Offer(context.Context, T) error {
 	return fq.offerErr
 }

exporter/exporterhelper/internal/queue_sender.go

@@ -66,9 +66,8 @@ func (qCfg *QueueConfig) Validate() error {
 }
 
 type QueueSender struct {
-	queue     exporterqueue.Queue[internal.Request]
-	batcher   queue.Batcher
-	consumers *queue.Consumers[internal.Request]
+	queue   exporterqueue.Queue[internal.Request]
+	batcher component.Component
 }
 
 func NewQueueSender(
@@ -79,15 +78,6 @@ func NewQueueSender(
 	exportFailureMessage string,
 	next Sender[internal.Request],
 ) (*QueueSender, error) {
-	q, err := newObsQueue(qSet, qf(context.Background(), qSet, qCfg))
-	if err != nil {
-		return nil, err
-	}
-
-	qs := &QueueSender{
-		queue: q,
-	}
-
 	exportFunc := func(ctx context.Context, req internal.Request) error {
 		err := next.Send(ctx, req)
 		if err != nil {
@@ -96,12 +86,30 @@ func NewQueueSender(
 		}
 		return err
 	}
-	if usePullingBasedExporterQueueBatcher.IsEnabled() {
-		qs.batcher, _ = queue.NewBatcher(bCfg, q, exportFunc, qCfg.NumConsumers)
-	} else {
-		qs.consumers = queue.NewQueueConsumers[internal.Request](q, qCfg.NumConsumers, exportFunc)
+	if !usePullingBasedExporterQueueBatcher.IsEnabled() {
+		q, err := newObsQueue(qSet, qf(context.Background(), qSet, qCfg, func(ctx context.Context, req internal.Request, done exporterqueue.DoneCallback) {
+			done(exportFunc(ctx, req))
+		}))
+		if err != nil {
+			return nil, err
+		}
+		return &QueueSender{queue: q}, nil
 	}
-	return qs, nil
+
+	b, err := queue.NewBatcher(bCfg, exportFunc, qCfg.NumConsumers)
+	if err != nil {
+		return nil, err
+	}
+	// TODO: https://github.com/open-telemetry/opentelemetry-collector/issues/12244
+	if bCfg.Enabled {
+		qCfg.NumConsumers = 1
+	}
+	q, err := newObsQueue(qSet, qf(context.Background(), qSet, qCfg, b.Consume))
+	if err != nil {
+		return nil, err
+	}
+
+	return &QueueSender{queue: q, batcher: b}, nil
 }
 
 // Start is invoked during service startup.
@@ -113,25 +121,22 @@ func (qs *QueueSender) Start(ctx context.Context, host component.Host) error {
 	if usePullingBasedExporterQueueBatcher.IsEnabled() {
 		return qs.batcher.Start(ctx, host)
 	}
-	return qs.consumers.Start(ctx, host)
+
+	return nil
 }
 
 // Shutdown is invoked during service shutdown.
 func (qs *QueueSender) Shutdown(ctx context.Context) error {
-	// Stop the queue and consumers, this will drain the queue and will call the retry (which is stopped) that will only
+	// Stop the queue and batcher, this will drain the queue and will call the retry (which is stopped) that will only
 	// try once every request.
-
-	if err := qs.queue.Shutdown(ctx); err != nil {
-		return err
-	}
+	err := qs.queue.Shutdown(ctx)
 	if usePullingBasedExporterQueueBatcher.IsEnabled() {
-		return qs.batcher.Shutdown(ctx)
+		return errors.Join(err, qs.batcher.Shutdown(ctx))
 	}
-	err := qs.consumers.Shutdown(ctx)
 	return err
 }
 
-// send implements the requestSender interface. It puts the request in the queue.
+// Send implements the requestSender interface. It puts the request in the queue.
 func (qs *QueueSender) Send(ctx context.Context, req internal.Request) error {
 	// Prevent cancellation and deadline to propagate to the context stored in the queue.
 	// The grpc/http based receivers will cancel the request context after this function returns.

exporter/exporterqueue/bounded_memory_queue.go

@@ -30,7 +30,7 @@ type memoryQueueSettings[T any] struct {
 
 // newBoundedMemoryQueue constructs the new queue of specified capacity, and with an optional
 // callback for dropped items (e.g. useful to emit metrics).
-func newBoundedMemoryQueue[T any](set memoryQueueSettings[T]) Queue[T] {
+func newBoundedMemoryQueue[T any](set memoryQueueSettings[T]) readableQueue[T] {
 	return &boundedMemoryQueue[T]{
 		sizedQueue: newSizedQueue[T](set.capacity, set.sizer, set.blocking),
 	}

exporter/exporterqueue/bounded_memory_queue_test.go

@@ -74,11 +74,11 @@ func TestBoundedQueue(t *testing.T) {
 func TestShutdownWhileNotEmpty(t *testing.T) {
 	q := newBoundedMemoryQueue[string](memoryQueueSettings[string]{sizer: &requestSizer[string]{}, capacity: 1000})
 
-	assert.NoError(t, q.Start(context.Background(), componenttest.NewNopHost()))
+	require.NoError(t, q.Start(context.Background(), componenttest.NewNopHost()))
 	for i := 0; i < 10; i++ {
 		require.NoError(t, q.Offer(context.Background(), strconv.FormatInt(int64(i), 10)))
 	}
-	assert.NoError(t, q.Shutdown(context.Background()))
+	require.NoError(t, q.Shutdown(context.Background()))
 
 	assert.Equal(t, int64(10), q.Size())
 	numConsumed := 0
@@ -115,16 +115,15 @@ func TestQueueUsage(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			q := newBoundedMemoryQueue[uint64](memoryQueueSettings[uint64]{sizer: tt.sizer, capacity: int64(100)})
 			consumed := &atomic.Int64{}
-			require.NoError(t, q.Start(context.Background(), componenttest.NewNopHost()))
-			ac := newAsyncConsumer(q, 1, func(context.Context, uint64) error {
+			ac := newConsumerQueue(q, 1, func(_ context.Context, _ uint64, done DoneCallback) {
 				consumed.Add(1)
-				return nil
+				done(nil)
 			})
+			require.NoError(t, ac.Start(context.Background(), componenttest.NewNopHost()))
 			for j := 0; j < 10; j++ {
 				require.NoError(t, q.Offer(context.Background(), uint64(10)))
 			}
-			assert.NoError(t, q.Shutdown(context.Background()))
-			assert.NoError(t, ac.Shutdown(context.Background()))
+			require.NoError(t, ac.Shutdown(context.Background()))
 			assert.Equal(t, int64(10), consumed.Load())
 		})
 	}
@@ -148,11 +147,11 @@ func TestBlockingQueueUsage(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			q := newBoundedMemoryQueue[uint64](memoryQueueSettings[uint64]{sizer: tt.sizer, capacity: int64(100), blocking: true})
 			consumed := &atomic.Int64{}
-			require.NoError(t, q.Start(context.Background(), componenttest.NewNopHost()))
-			ac := newAsyncConsumer(q, 10, func(context.Context, uint64) error {
+			ac := newConsumerQueue(q, 10, func(_ context.Context, _ uint64, done DoneCallback) {
 				consumed.Add(1)
-				return nil
+				done(nil)
 			})
+			require.NoError(t, ac.Start(context.Background(), componenttest.NewNopHost()))
 			wg := &sync.WaitGroup{}
 			for i := 0; i < 10; i++ {
 				wg.Add(1)
@@ -164,25 +163,21 @@ func TestBlockingQueueUsage(t *testing.T) {
 				}()
 			}
 			wg.Wait()
-			assert.NoError(t, q.Shutdown(context.Background()))
-			assert.NoError(t, ac.Shutdown(context.Background()))
+			require.NoError(t, ac.Shutdown(context.Background()))
 			assert.Equal(t, int64(1_000_000), consumed.Load())
 		})
 	}
 }
 
 func TestZeroSizeNoConsumers(t *testing.T) {
 	q := newBoundedMemoryQueue[string](memoryQueueSettings[string]{sizer: &requestSizer[string]{}, capacity: 0})
-
 	err := q.Start(context.Background(), componenttest.NewNopHost())
 	require.NoError(t, err)
-
 	require.ErrorIs(t, q.Offer(context.Background(), "a"), ErrQueueIsFull) // in process
-
 	assert.NoError(t, q.Shutdown(context.Background()))
 }
 
-func consume[T any](q Queue[T], consumeFunc func(context.Context, T) error) bool {
+func consume[T any](q readableQueue[T], consumeFunc func(context.Context, T) error) bool {
 	ctx, req, done, ok := q.Read(context.Background())
 	if !ok {
 		return false
@@ -191,35 +186,6 @@ func consume[T any](q Queue[T], consumeFunc func(context.Context, T) error) bool
 	return true
 }
 
-type asyncConsumer struct {
-	stopWG sync.WaitGroup
-}
-
-func newAsyncConsumer[T any](q Queue[T], numConsumers int, consumeFunc func(context.Context, T) error) *asyncConsumer {
-	ac := &asyncConsumer{}
-
-	ac.stopWG.Add(numConsumers)
-	for i := 0; i < numConsumers; i++ {
-		go func() {
-			defer ac.stopWG.Done()
-			for {
-				ctx, req, done, ok := q.Read(context.Background())
-				if !ok {
-					return
-				}
-				done(consumeFunc(ctx, req))
-			}
-		}()
-	}
-	return ac
-}
-
-// Shutdown ensures that queue and all consumers are stopped.
-func (qc *asyncConsumer) Shutdown(_ context.Context) error {
-	qc.stopWG.Wait()
-	return nil
-}
-
 func BenchmarkOffer(b *testing.B) {
 	tests := []struct {
 		name  string
@@ -236,20 +202,20 @@ func BenchmarkOffer(b *testing.B) {
 	}
 	for _, tt := range tests {
 		b.Run(tt.name, func(b *testing.B) {
-			q := newBoundedMemoryQueue[uint64](memoryQueueSettings[uint64]{sizer: &requestSizer[uint64]{}, capacity: int64(10 * b.N)})
+			q := newBoundedMemoryQueue[uint64](memoryQueueSettings[uint64]{sizer: tt.sizer, capacity: int64(10 * b.N)})
 			consumed := &atomic.Int64{}
 			require.NoError(b, q.Start(context.Background(), componenttest.NewNopHost()))
-			ac := newAsyncConsumer(q, 1, func(context.Context, uint64) error {
+			ac := newConsumerQueue(q, 1, func(_ context.Context, _ uint64, done DoneCallback) {
 				consumed.Add(1)
-				return nil
+				done(nil)
 			})
+			require.NoError(b, ac.Start(context.Background(), componenttest.NewNopHost()))
 			b.ResetTimer()
 			b.ReportAllocs()
 			for j := 0; j < b.N; j++ {
 				require.NoError(b, q.Offer(context.Background(), uint64(10)))
 			}
-			assert.NoError(b, q.Shutdown(context.Background()))
-			assert.NoError(b, ac.Shutdown(context.Background()))
+			require.NoError(b, ac.Shutdown(context.Background()))
 			assert.Equal(b, int64(b.N), consumed.Load())
 		})
 	}

exporter/exporterqueue/consumers.go

@@ -0,0 +1,59 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package exporterqueue // import "go.opentelemetry.io/collector/exporter/exporterqueue"
+
+import (
+	"context"
+	"sync"
+
+	"go.opentelemetry.io/collector/component"
+)
+
+type consumerQueue[T any] struct {
+	readableQueue[T]
+	numConsumers int
+	consumeFunc  ConsumeFunc[T]
+	stopWG       sync.WaitGroup
+}
+
+func newConsumerQueue[T any](q readableQueue[T], numConsumers int, consumeFunc ConsumeFunc[T]) *consumerQueue[T] {
+	return &consumerQueue[T]{
+		readableQueue: q,
+		numConsumers:  numConsumers,
+		consumeFunc:   consumeFunc,
+	}
+}
+
+// Start ensures that queue and all consumers are started.
+func (qc *consumerQueue[T]) Start(ctx context.Context, host component.Host) error {
+	if err := qc.readableQueue.Start(ctx, host); err != nil {
+		return err
+	}
+	var startWG sync.WaitGroup
+	for i := 0; i < qc.numConsumers; i++ {
+		qc.stopWG.Add(1)
+		startWG.Add(1)
+		go func() {
+			startWG.Done()
+			defer qc.stopWG.Done()
+			for {
+				ctx, req, done, ok := qc.readableQueue.Read(context.Background())
+				if !ok {
+					return
+				}
+				qc.consumeFunc(ctx, req, done)
+			}
+		}()
+	}
+	startWG.Wait()
+
+	return nil
+}
+
+// Shutdown ensures that queue and all consumers are stopped.
+func (qc *consumerQueue[T]) Shutdown(ctx context.Context) error {
+	err := qc.readableQueue.Shutdown(ctx)
+	qc.stopWG.Wait()
+	return err
+}

exporter/exporterqueue/persistent_queue.go

@@ -92,7 +92,7 @@ type persistentQueue[T any] struct {
 }
 
 // newPersistentQueue creates a new queue backed by file storage; name and signal must be a unique combination that identifies the queue storage
-func newPersistentQueue[T any](set persistentQueueSettings[T]) Queue[T] {
+func newPersistentQueue[T any](set persistentQueueSettings[T]) readableQueue[T] {
 	_, isRequestSized := set.sizer.(*requestSizer[T])
 	pq := &persistentQueue[T]{
 		set:            set,