[chore] Move exporterhelper queue code to exporterqueue (#12033)

Signed-off-by: Bogdan Drutu <[email protected]>
open-telemetry · Jan 7, 2025 · d2c178b · d2c178b
1 parent ced38e8
commit d2c178b
Show file tree

Hide file tree

Showing 24 changed files with 282 additions and 254 deletions.
diff --git a/exporter/exporterhelper/exporterhelper.go b/exporter/exporterhelper/exporterhelper.go
@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 package exporterhelper // import "go.opentelemetry.io/collector/exporter/exporterhelper"
+
 import "go.opentelemetry.io/collector/exporter/internal"
 
 // Request represents a single request that can be sent to an external endpoint.

diff --git a/exporter/exporterhelper/internal/queue_sender_test.go b/exporter/exporterhelper/internal/queue_sender_test.go
@@ -23,7 +23,7 @@ import (
 	"go.opentelemetry.io/collector/exporter/exporterqueue"
 	"go.opentelemetry.io/collector/exporter/exportertest"
 	"go.opentelemetry.io/collector/exporter/internal"
-	"go.opentelemetry.io/collector/exporter/internal/queue"
+	"go.opentelemetry.io/collector/exporter/internal/storagetest"
 	"go.opentelemetry.io/collector/pipeline"
 )
 
@@ -435,7 +435,7 @@ func TestQueuedRetryPersistenceEnabled(t *testing.T) {
 			require.NoError(t, err)
 
 			extensions := map[component.ID]component.Component{
-				storageID: queue.NewMockStorageExtension(nil),
+				storageID: storagetest.NewMockStorageExtension(nil),
 			}
 			host := &MockHost{Ext: extensions}
 
@@ -468,7 +468,7 @@ func TestQueuedRetryPersistenceEnabledStorageError(t *testing.T) {
 			require.NoError(t, err)
 
 			extensions := map[component.ID]component.Component{
-				storageID: queue.NewMockStorageExtension(storageError),
+				storageID: storagetest.NewMockStorageExtension(storageError),
 			}
 			host := &MockHost{Ext: extensions}
 
@@ -500,7 +500,7 @@ func TestQueuedRetryPersistentEnabled_NoDataLossOnShutdown(t *testing.T) {
 			require.NoError(t, err)
 
 			extensions := map[component.ID]component.Component{
-				storageID: queue.NewMockStorageExtension(nil),
+				storageID: storagetest.NewMockStorageExtension(nil),
 			}
 			host := &MockHost{Ext: extensions}
 
@@ -540,11 +540,17 @@ func TestQueueSenderNoStartShutdown(t *testing.T) {
 	runTest := func(testName string, enableQueueBatcher bool) {
 		t.Run(testName, func(t *testing.T) {
 			defer setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)()
-			queue := queue.NewBoundedMemoryQueue[internal.Request](queue.MemoryQueueSettings[internal.Request]{})
 			set := exportertest.NewNopSettings()
+			queue := exporterqueue.NewMemoryQueueFactory[internal.Request]()(
+				context.Background(),
+				exporterqueue.Settings{
+					Signal:           pipeline.SignalTraces,
+					ExporterSettings: set,
+				},
+				exporterqueue.NewDefaultConfig())
 			obsrep, err := NewExporter(ObsReportSettings{
 				ExporterID:             exporterID,
-				ExporterCreateSettings: exportertest.NewNopSettings(),
+				ExporterCreateSettings: set,
 			})
 			require.NoError(t, err)
 			qs := NewQueueSender(queue, set, 1, "", obsrep, exporterbatcher.NewDefaultConfig())

diff --git a/exporter/exporterhelper/logs.go b/exporter/exporterhelper/logs.go
@@ -15,7 +15,6 @@ import (
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 	"go.opentelemetry.io/collector/exporter/exporterqueue"
-	"go.opentelemetry.io/collector/exporter/internal/queue"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pipeline"
 )
@@ -135,7 +134,7 @@ func NewLogsRequest(
 			return consumererror.NewPermanent(cErr)
 		}
 		sErr := be.Send(ctx, req)
-		if errors.Is(sErr, queue.ErrQueueIsFull) {
+		if errors.Is(sErr, exporterqueue.ErrQueueIsFull) {
 			be.Obsrep.RecordEnqueueFailure(ctx, pipeline.SignalLogs, int64(req.ItemsCount()))
 		}
 		return sErr

diff --git a/exporter/exporterhelper/logs_test.go b/exporter/exporterhelper/logs_test.go
@@ -27,7 +27,7 @@ import (
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 	"go.opentelemetry.io/collector/exporter/exportertest"
-	"go.opentelemetry.io/collector/exporter/internal/queue"
+	"go.opentelemetry.io/collector/exporter/internal/storagetest"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/testdata"
 )
@@ -169,7 +169,7 @@ func TestLogs_WithPersistentQueue(t *testing.T) {
 	require.NoError(t, err)
 
 	host := &internal.MockHost{Ext: map[component.ID]component.Component{
-		storageID: queue.NewMockStorageExtension(nil),
+		storageID: storagetest.NewMockStorageExtension(nil),
 	}}
 	require.NoError(t, te.Start(context.Background(), host))
 	t.Cleanup(func() { require.NoError(t, te.Shutdown(context.Background())) })

diff --git a/exporter/exporterhelper/metrics.go b/exporter/exporterhelper/metrics.go
@@ -15,7 +15,6 @@ import (
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 	"go.opentelemetry.io/collector/exporter/exporterqueue"
-	"go.opentelemetry.io/collector/exporter/internal/queue"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pipeline"
 )
@@ -135,7 +134,7 @@ func NewMetricsRequest(
 			return consumererror.NewPermanent(cErr)
 		}
 		sErr := be.Send(ctx, req)
-		if errors.Is(sErr, queue.ErrQueueIsFull) {
+		if errors.Is(sErr, exporterqueue.ErrQueueIsFull) {
 			be.Obsrep.RecordEnqueueFailure(ctx, pipeline.SignalMetrics, int64(req.ItemsCount()))
 		}
 		return sErr

diff --git a/exporter/exporterhelper/metrics_test.go b/exporter/exporterhelper/metrics_test.go
@@ -27,7 +27,7 @@ import (
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 	"go.opentelemetry.io/collector/exporter/exportertest"
-	"go.opentelemetry.io/collector/exporter/internal/queue"
+	"go.opentelemetry.io/collector/exporter/internal/storagetest"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/testdata"
 )
@@ -170,7 +170,7 @@ func TestMetrics_WithPersistentQueue(t *testing.T) {
 	require.NoError(t, err)
 
 	host := &internal.MockHost{Ext: map[component.ID]component.Component{
-		storageID: queue.NewMockStorageExtension(nil),
+		storageID: storagetest.NewMockStorageExtension(nil),
 	}}
 	require.NoError(t, te.Start(context.Background(), host))
 	t.Cleanup(func() { require.NoError(t, te.Shutdown(context.Background())) })

diff --git a/exporter/exporterhelper/traces.go b/exporter/exporterhelper/traces.go
@@ -15,7 +15,6 @@ import (
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 	"go.opentelemetry.io/collector/exporter/exporterqueue"
-	"go.opentelemetry.io/collector/exporter/internal/queue"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/pipeline"
 )
@@ -135,7 +134,7 @@ func NewTracesRequest(
 			return consumererror.NewPermanent(cErr)
 		}
 		sErr := be.Send(ctx, req)
-		if errors.Is(sErr, queue.ErrQueueIsFull) {
+		if errors.Is(sErr, exporterqueue.ErrQueueIsFull) {
 			be.Obsrep.RecordEnqueueFailure(ctx, pipeline.SignalTraces, int64(req.ItemsCount()))
 		}
 		return sErr

diff --git a/exporter/exporterhelper/traces_test.go b/exporter/exporterhelper/traces_test.go
@@ -27,7 +27,7 @@ import (
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 	"go.opentelemetry.io/collector/exporter/exportertest"
-	"go.opentelemetry.io/collector/exporter/internal/queue"
+	"go.opentelemetry.io/collector/exporter/internal/storagetest"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/pdata/testdata"
 )
@@ -167,7 +167,7 @@ func TestTraces_WithPersistentQueue(t *testing.T) {
 	require.NoError(t, err)
 
 	host := &internal.MockHost{Ext: map[component.ID]component.Component{
-		storageID: queue.NewMockStorageExtension(nil),
+		storageID: storagetest.NewMockStorageExtension(nil),
 	}}
 	require.NoError(t, te.Start(context.Background(), host))
 	t.Cleanup(func() { require.NoError(t, te.Shutdown(context.Background())) })

diff --git a/exporter/exporterhelper/xexporterhelper/profiles_test.go b/exporter/exporterhelper/xexporterhelper/profiles_test.go
@@ -30,7 +30,7 @@ import (
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 	"go.opentelemetry.io/collector/exporter/exportertest"
-	"go.opentelemetry.io/collector/exporter/internal/queue"
+	"go.opentelemetry.io/collector/exporter/internal/storagetest"
 	"go.opentelemetry.io/collector/exporter/xexporter"
 	"go.opentelemetry.io/collector/pdata/pprofile"
 	"go.opentelemetry.io/collector/pdata/testdata"
@@ -170,7 +170,7 @@ func TestProfilesExporter_WithPersistentQueue(t *testing.T) {
 	require.NoError(t, err)
 
 	host := &internal.MockHost{Ext: map[component.ID]component.Component{
-		storageID: queue.NewMockStorageExtension(nil),
+		storageID: storagetest.NewMockStorageExtension(nil),
 	}}
 	require.NoError(t, te.Start(context.Background(), host))
 	t.Cleanup(func() { require.NoError(t, te.Shutdown(context.Background())) })

diff --git a/...er/internal/queue/bounded_memory_queue.go → ...ter/exporterqueue/bounded_memory_queue.go b/...er/internal/queue/bounded_memory_queue.go → ...ter/exporterqueue/bounded_memory_queue.go
@@ -3,7 +3,7 @@
 // Copyright (c) 2017 Uber Technologies, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-package queue // import "go.opentelemetry.io/collector/exporter/internal/queue"
+package exporterqueue // import "go.opentelemetry.io/collector/exporter/exporterqueue"
 
 import (
 	"context"
@@ -17,21 +17,21 @@ import (
 type boundedMemoryQueue[T any] struct {
 	component.StartFunc
 	*sizedChannel[memQueueEl[T]]
-	sizer Sizer[T]
+	sizer sizer[T]
 }
 
-// MemoryQueueSettings defines internal parameters for boundedMemoryQueue creation.
-type MemoryQueueSettings[T any] struct {
-	Sizer    Sizer[T]
-	Capacity int64
+// memoryQueueSettings defines internal parameters for boundedMemoryQueue creation.
+type memoryQueueSettings[T any] struct {
+	sizer    sizer[T]
+	capacity int64
 }
 
-// NewBoundedMemoryQueue constructs the new queue of specified capacity, and with an optional
+// 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]) Queue[T] {
 	return &boundedMemoryQueue[T]{
-		sizedChannel: newSizedChannel[memQueueEl[T]](set.Capacity, nil, 0),
-		sizer:        set.Sizer,
+		sizedChannel: newSizedChannel[memQueueEl[T]](set.capacity, nil, 0),
+		sizer:        set.sizer,
 	}
 }
 

diff --git a/...ternal/queue/bounded_memory_queue_test.go → ...xporterqueue/bounded_memory_queue_test.go b/...ternal/queue/bounded_memory_queue_test.go → ...xporterqueue/bounded_memory_queue_test.go
@@ -3,12 +3,13 @@
 // Copyright (c) 2017 Uber Technologies, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-package queue
+package exporterqueue
 
 import (
 	"context"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -21,7 +22,7 @@ import (
 // We want to test the overflow behavior, so we block the consumer
 // by holding a startLock before submitting items to the queue.
 func TestBoundedQueue(t *testing.T) {
-	q := NewBoundedMemoryQueue[string](MemoryQueueSettings[string]{Sizer: &RequestSizer[string]{}, Capacity: 1})
+	q := newBoundedMemoryQueue[string](memoryQueueSettings[string]{sizer: &requestSizer[string]{}, capacity: 1})
 
 	require.NoError(t, q.Offer(context.Background(), "a"))
 
@@ -71,7 +72,7 @@ func TestBoundedQueue(t *testing.T) {
 // only after Stop will mean the consumers are still locked while
 // trying to perform the final consumptions.
 func TestShutdownWhileNotEmpty(t *testing.T) {
-	q := NewBoundedMemoryQueue[string](MemoryQueueSettings[string]{Sizer: &RequestSizer[string]{}, Capacity: 1000})
+	q := newBoundedMemoryQueue[string](memoryQueueSettings[string]{sizer: &requestSizer[string]{}, capacity: 1000})
 
 	assert.NoError(t, q.Start(context.Background(), componenttest.NewNopHost()))
 	for i := 0; i < 10; i++ {
@@ -97,11 +98,11 @@ func TestShutdownWhileNotEmpty(t *testing.T) {
 }
 
 func Benchmark_QueueUsage_1000_requests(b *testing.B) {
-	benchmarkQueueUsage(b, &RequestSizer[fakeReq]{}, 1000)
+	benchmarkQueueUsage(b, &requestSizer[fakeReq]{}, 1000)
 }
 
 func Benchmark_QueueUsage_100000_requests(b *testing.B) {
-	benchmarkQueueUsage(b, &RequestSizer[fakeReq]{}, 100000)
+	benchmarkQueueUsage(b, &requestSizer[fakeReq]{}, 100000)
 }
 
 func Benchmark_QueueUsage_10000_items(b *testing.B) {
@@ -116,40 +117,38 @@ func Benchmark_QueueUsage_1M_items(b *testing.B) {
 
 func TestQueueUsage(t *testing.T) {
 	t.Run("requests_based", func(t *testing.T) {
-		queueUsage(t, &RequestSizer[fakeReq]{}, 10)
+		queueUsage(t, &requestSizer[fakeReq]{}, 10)
 	})
 	t.Run("items_based", func(t *testing.T) {
 		queueUsage(t, &itemsSizer[fakeReq]{}, 10)
 	})
 }
 
-func benchmarkQueueUsage(b *testing.B, sizer Sizer[fakeReq], requestsCount int) {
+func benchmarkQueueUsage(b *testing.B, sizer sizer[fakeReq], requestsCount int) {
 	b.ReportAllocs()
 	for i := 0; i < b.N; i++ {
 		queueUsage(b, sizer, requestsCount)
 	}
 }
 
-func queueUsage(tb testing.TB, sizer Sizer[fakeReq], requestsCount int) {
-	var wg sync.WaitGroup
-	wg.Add(requestsCount)
-	q := NewBoundedMemoryQueue[fakeReq](MemoryQueueSettings[fakeReq]{Sizer: sizer, Capacity: int64(10 * requestsCount)})
-	consumers := NewQueueConsumers(q, 1, func(context.Context, fakeReq) error {
-		wg.Done()
+func queueUsage(tb testing.TB, sizer sizer[fakeReq], requestsCount int) {
+	q := newBoundedMemoryQueue[fakeReq](memoryQueueSettings[fakeReq]{sizer: sizer, capacity: int64(10 * requestsCount)})
+	consumed := &atomic.Int64{}
+	require.NoError(tb, q.Start(context.Background(), componenttest.NewNopHost()))
+	ac := newAsyncConsumer(q, 1, func(context.Context, fakeReq) error {
+		consumed.Add(1)
 		return nil
 	})
-	require.NoError(tb, q.Start(context.Background(), componenttest.NewNopHost()))
-	require.NoError(tb, consumers.Start(context.Background(), componenttest.NewNopHost()))
 	for j := 0; j < requestsCount; j++ {
 		require.NoError(tb, q.Offer(context.Background(), fakeReq{10}))
 	}
 	assert.NoError(tb, q.Shutdown(context.Background()))
-	assert.NoError(tb, consumers.Shutdown(context.Background()))
-	wg.Wait()
+	assert.NoError(tb, ac.Shutdown(context.Background()))
+	assert.Equal(tb, int64(requestsCount), consumed.Load())
 }
 
 func TestZeroSizeNoConsumers(t *testing.T) {
-	q := NewBoundedMemoryQueue[string](MemoryQueueSettings[string]{Sizer: &RequestSizer[string]{}, Capacity: 0})
+	q := newBoundedMemoryQueue[string](memoryQueueSettings[string]{sizer: &requestSizer[string]{}, capacity: 0})
 
 	err := q.Start(context.Background(), componenttest.NewNopHost())
 	require.NoError(t, err)
@@ -166,3 +165,43 @@ type fakeReq struct {
 func (r fakeReq) ItemsCount() int {
 	return r.itemsCount
 }
+
+func consume[T any](q Queue[T], consumeFunc func(context.Context, T) error) bool {
+	index, ctx, req, ok := q.Read(context.Background())
+	if !ok {
+		return false
+	}
+	consumeErr := consumeFunc(ctx, req)
+	q.OnProcessingFinished(index, consumeErr)
+	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 {
+				index, ctx, req, ok := q.Read(context.Background())
+				if !ok {
+					return
+				}
+				consumeErr := consumeFunc(ctx, req)
+				q.OnProcessingFinished(index, consumeErr)
+			}
+		}()
+	}
+	return ac
+}
+
+// Shutdown ensures that queue and all consumers are stopped.
+func (qc *asyncConsumer) Shutdown(_ context.Context) error {
+	qc.stopWG.Wait()
+	return nil
+}