Move Queue to internal, don't expect other implementations (open-telemetry#12680)

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

Author: bogdandrutu

.chloggen/move-queue-to-internal.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: exporterqueue
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Move Queue interface to internal, disallow alternative implementations
+
+# One or more tracking issues or pull requests related to the change
+issues: [12680]
+
+# (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/base_exporter.go

@@ -94,7 +94,7 @@ func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, pusher func(
 	}
 
 	if be.queueCfg.Enabled || be.batcherCfg.Enabled {
-		qSet := exporterqueue.Settings[request.Request]{
+		qSet := queuebatch.QueueSettings[request.Request]{
 			Signal:           signal,
 			ExporterSettings: set,
 			Encoding:         be.queueBatchSettings.Encoding,

exporter/exporterhelper/internal/batcher/batcher.go

@@ -8,14 +8,14 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
-	"go.opentelemetry.io/collector/exporter/exporterqueue"
 )
 
 // Batcher is in charge of reading items from the queue and send them out asynchronously.
 type Batcher interface {
 	component.Component
-	Consume(context.Context, request.Request, exporterqueue.Done)
+	Consume(context.Context, request.Request, queuebatch.Done)
 }
 
 func NewBatcher(batchCfg exporterbatcher.Config,

exporter/exporterhelper/internal/batcher/default_batcher.go

@@ -12,8 +12,8 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
-	"go.opentelemetry.io/collector/exporter/exporterqueue"
 )
 
 type batch struct {
@@ -61,7 +61,7 @@ func (qb *defaultBatcher) resetTimer() {
 	}
 }
 
-func (qb *defaultBatcher) Consume(ctx context.Context, req request.Request, done exporterqueue.Done) {
+func (qb *defaultBatcher) Consume(ctx context.Context, req request.Request, done queuebatch.Done) {
 	qb.currentBatchMu.Lock()
 
 	if qb.currentBatch == nil {
@@ -199,7 +199,7 @@ func (qb *defaultBatcher) flushCurrentBatchIfNecessary() {
 }
 
 // flush starts a goroutine that calls exportFunc. It blocks until a worker is available if necessary.
-func (qb *defaultBatcher) flush(ctx context.Context, req request.Request, done exporterqueue.Done) {
+func (qb *defaultBatcher) flush(ctx context.Context, req request.Request, done queuebatch.Done) {
 	qb.stopWG.Add(1)
 	if qb.workerPool != nil {
 		<-qb.workerPool
@@ -222,7 +222,7 @@ func (qb *defaultBatcher) Shutdown(_ context.Context) error {
 	return nil
 }
 
-type multiDone []exporterqueue.Done
+type multiDone []queuebatch.Done
 
 func (mdc multiDone) OnDone(err error) {
 	for _, d := range mdc {
@@ -231,13 +231,13 @@ func (mdc multiDone) OnDone(err error) {
 }
 
 type refCountDone struct {
-	done     exporterqueue.Done
+	done     queuebatch.Done
 	mu       sync.Mutex
 	refCount int64
 	err      error
 }
 
-func newRefCountDone(done exporterqueue.Done, refCount int64) exporterqueue.Done {
+func newRefCountDone(done queuebatch.Done, refCount int64) queuebatch.Done {
 	return &refCountDone{
 		done:     done,
 		refCount: refCount,

exporter/exporterhelper/internal/batcher/disabled_batcher.go

@@ -7,8 +7,8 @@ import (
 	"context"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
-	"go.opentelemetry.io/collector/exporter/exporterqueue"
 )
 
 // disabledBatcher is a special-case of Batcher that has no size limit for sending. Any items read from the queue will
@@ -19,7 +19,7 @@ type disabledBatcher[T any] struct {
 	exportFunc func(context.Context, T) error
 }
 
-func (db *disabledBatcher[T]) Consume(ctx context.Context, req T, done exporterqueue.Done) {
+func (db *disabledBatcher[T]) Consume(ctx context.Context, req T, done queuebatch.Done) {
 	done.OnDone(db.exportFunc(ctx, req))
 }
 

exporter/exporterhelper/internal/batcher/disabled_batcher_test.go

@@ -14,6 +14,7 @@ import (
 
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
 	"go.opentelemetry.io/collector/exporter/exporterqueue"
@@ -44,9 +45,9 @@ func TestDisabledBatcher_Basic(t *testing.T) {
 			ba, err := NewBatcher(cfg, sink.Export, tt.maxWorkers)
 			require.NoError(t, err)
 
-			q := exporterqueue.NewQueue[request.Request](
+			q := queuebatch.NewQueue[request.Request](
 				context.Background(),
-				exporterqueue.Settings[request.Request]{
+				queuebatch.QueueSettings[request.Request]{
 					Signal:           pipeline.SignalTraces,
 					ExporterSettings: exportertest.NewNopSettings(exportertest.NopType),
 				},

exporter/exporterhelper/internal/obs_queue.go

@@ -10,20 +10,20 @@ import (
 	"go.opentelemetry.io/otel/metric"
 
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadata"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
-	"go.opentelemetry.io/collector/exporter/exporterqueue"
 	"go.opentelemetry.io/collector/pipeline"
 )
 
 // obsQueue is a helper to add observability to a queue.
 type obsQueue[T request.Request] struct {
-	exporterqueue.Queue[T]
+	queuebatch.Queue[T]
 	tb                *metadata.TelemetryBuilder
 	metricAttr        metric.MeasurementOption
 	enqueueFailedInst metric.Int64Counter
 }
 
-func newObsQueue[T request.Request](set exporterqueue.Settings[T], delegate exporterqueue.Queue[T]) (exporterqueue.Queue[T], error) {
+func newObsQueue[T request.Request](set queuebatch.QueueSettings[T], delegate queuebatch.Queue[T]) (queuebatch.Queue[T], error) {
 	tb, err := metadata.NewTelemetryBuilder(set.ExporterSettings.TelemetrySettings)
 	if err != nil {
 		return nil, err

exporter/exporterhelper/internal/obs_queue_test.go

@@ -17,14 +17,14 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadatatest"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"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/pipeline"
 )
 
 type fakeQueue[T any] struct {
-	exporterqueue.Queue[T]
+	queuebatch.Queue[T]
 	offerErr error
 	size     int64
 	capacity int64
@@ -42,15 +42,15 @@ func (fq *fakeQueue[T]) Offer(context.Context, T) error {
 	return fq.offerErr
 }
 
-func newFakeQueue[T request.Request](offerErr error, size, capacity int64) exporterqueue.Queue[T] {
+func newFakeQueue[T request.Request](offerErr error, size, capacity int64) queuebatch.Queue[T] {
 	return &fakeQueue[T]{offerErr: offerErr, size: size, capacity: capacity}
 }
 
 func TestObsQueueLogsSizeCapacity(t *testing.T) {
 	tt := componenttest.NewTelemetry()
 	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
 
-	te, err := newObsQueue[request.Request](exporterqueue.Settings[request.Request]{
+	te, err := newObsQueue[request.Request](queuebatch.QueueSettings[request.Request]{
 		Signal:           pipeline.SignalLogs,
 		ExporterSettings: exporter.Settings{ID: exporterID, TelemetrySettings: tt.NewTelemetrySettings(), BuildInfo: component.NewDefaultBuildInfo()},
 	}, newFakeQueue[request.Request](nil, 7, 9))
@@ -80,7 +80,7 @@ func TestObsQueueLogsFailure(t *testing.T) {
 	tt := componenttest.NewTelemetry()
 	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
 
-	te, err := newObsQueue[request.Request](exporterqueue.Settings[request.Request]{
+	te, err := newObsQueue[request.Request](queuebatch.QueueSettings[request.Request]{
 		Signal:           pipeline.SignalLogs,
 		ExporterSettings: exporter.Settings{ID: exporterID, TelemetrySettings: tt.NewTelemetrySettings(), BuildInfo: component.NewDefaultBuildInfo()},
 	}, newFakeQueue[request.Request](errors.New("my error"), 7, 9))
@@ -100,7 +100,7 @@ func TestObsQueueTracesSizeCapacity(t *testing.T) {
 	tt := componenttest.NewTelemetry()
 	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
 
-	te, err := newObsQueue[request.Request](exporterqueue.Settings[request.Request]{
+	te, err := newObsQueue[request.Request](queuebatch.QueueSettings[request.Request]{
 		Signal:           pipeline.SignalTraces,
 		ExporterSettings: exporter.Settings{ID: exporterID, TelemetrySettings: tt.NewTelemetrySettings(), BuildInfo: component.NewDefaultBuildInfo()},
 	}, newFakeQueue[request.Request](nil, 17, 19))
@@ -130,7 +130,7 @@ func TestObsQueueTracesFailure(t *testing.T) {
 	tt := componenttest.NewTelemetry()
 	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
 
-	te, err := newObsQueue[request.Request](exporterqueue.Settings[request.Request]{
+	te, err := newObsQueue[request.Request](queuebatch.QueueSettings[request.Request]{
 		Signal:           pipeline.SignalTraces,
 		ExporterSettings: exporter.Settings{ID: exporterID, TelemetrySettings: tt.NewTelemetrySettings(), BuildInfo: component.NewDefaultBuildInfo()},
 	}, newFakeQueue[request.Request](errors.New("my error"), 0, 0))
@@ -150,7 +150,7 @@ func TestObsQueueMetrics(t *testing.T) {
 	tt := componenttest.NewTelemetry()
 	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
 
-	te, err := newObsQueue[request.Request](exporterqueue.Settings[request.Request]{
+	te, err := newObsQueue[request.Request](queuebatch.QueueSettings[request.Request]{
 		Signal:           pipeline.SignalMetrics,
 		ExporterSettings: exporter.Settings{ID: exporterID, TelemetrySettings: tt.NewTelemetrySettings(), BuildInfo: component.NewDefaultBuildInfo()},
 	}, newFakeQueue[request.Request](nil, 27, 29))
@@ -180,7 +180,7 @@ func TestObsQueueMetricsFailure(t *testing.T) {
 	tt := componenttest.NewTelemetry()
 	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
 
-	te, err := newObsQueue[request.Request](exporterqueue.Settings[request.Request]{
+	te, err := newObsQueue[request.Request](queuebatch.QueueSettings[request.Request]{
 		Signal:           pipeline.SignalMetrics,
 		ExporterSettings: exporter.Settings{ID: exporterID, TelemetrySettings: tt.NewTelemetrySettings(), BuildInfo: component.NewDefaultBuildInfo()},
 	}, newFakeQueue[request.Request](errors.New("my error"), 0, 0))

exporter/exporterhelper/internal/queue_sender.go

@@ -12,6 +12,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/batcher"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterqueue"
 	"go.opentelemetry.io/collector/featuregate"
@@ -26,12 +27,12 @@ var _ = featuregate.GlobalRegistry().MustRegister(
 )
 
 type QueueSender struct {
-	queue   exporterqueue.Queue[request.Request]
+	queue   queuebatch.Queue[request.Request]
 	batcher component.Component
 }
 
 func NewQueueSender(
-	qSet exporterqueue.Settings[request.Request],
+	qSet queuebatch.QueueSettings[request.Request],
 	qCfg exporterqueue.Config,
 	bCfg exporterbatcher.Config,
 	exportFailureMessage string,
@@ -57,7 +58,7 @@ func NewQueueSender(
 	if bCfg.Enabled {
 		qCfg.NumConsumers = 1
 	}
-	q, err := newObsQueue(qSet, exporterqueue.NewQueue(context.Background(), qSet, qCfg, b.Consume))
+	q, err := newObsQueue(qSet, queuebatch.NewQueue(context.Background(), qSet, qCfg, b.Consume))
 	if err != nil {
 		return nil, err
 	}

exporter/exporterhelper/internal/queue_sender_test.go

@@ -20,6 +20,7 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
 	"go.opentelemetry.io/collector/exporter/exporterqueue"
@@ -42,7 +43,7 @@ func newFakeEncoding(mr request.Request) exporterqueue.Encoding[request.Request]
 	return &fakeEncoding{mr: mr}
 }
 
-var defaultQueueSettings = exporterqueue.Settings[request.Request]{
+var defaultQueueSettings = queuebatch.QueueSettings[request.Request]{
 	Signal:           defaultSignal,
 	ExporterSettings: defaultSettings,
 }
@@ -126,7 +127,7 @@ func TestQueueFailedRequestDropped(t *testing.T) {
 }
 
 func TestQueueBatcherPersistenceEnabled(t *testing.T) {
-	qSet := exporterqueue.Settings[request.Request]{
+	qSet := queuebatch.QueueSettings[request.Request]{
 		Signal:           defaultSignal,
 		ExporterSettings: defaultSettings,
 		Encoding:         newFakeEncoding(&requesttest.FakeRequest{}),
@@ -150,7 +151,7 @@ func TestQueueBatcherPersistenceEnabled(t *testing.T) {
 func TestQueueBatcherPersistenceEnabledStorageError(t *testing.T) {
 	storageError := errors.New("could not get storage client")
 
-	qSet := exporterqueue.Settings[request.Request]{
+	qSet := queuebatch.QueueSettings[request.Request]{
 		Signal:           defaultSignal,
 		ExporterSettings: defaultSettings,
 		Encoding:         newFakeEncoding(&requesttest.FakeRequest{}),
@@ -183,7 +184,7 @@ func TestQueueBatcherPersistentEnabled_NoDataLossOnShutdown(t *testing.T) {
 	require.NoError(t, rs.Start(context.Background(), componenttest.NewNopHost()))
 
 	mockReq := &requesttest.FakeRequest{Items: 2}
-	qSet := exporterqueue.Settings[request.Request]{
+	qSet := queuebatch.QueueSettings[request.Request]{
 		Signal:           defaultSignal,
 		ExporterSettings: defaultSettings,
 		Encoding:         newFakeEncoding(mockReq),
@@ -338,7 +339,7 @@ func TestQueueBatcher_BatchExportError(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			sink := requesttest.NewSink()
-			qSet := exporterqueue.Settings[request.Request]{
+			qSet := queuebatch.QueueSettings[request.Request]{
 				Signal:           defaultSignal,
 				ExporterSettings: defaultSettings,
 			}