[chore] Move code from exporter/internal to exporterhelper/internal (open-telemetry#12342)

This PR is a preparation to simplify dependency between "exporter"
module and the new "exporterhelper" module that will be created next.

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

Author: bogdandrutu

exporter/exporterhelper/exporterhelper.go

@@ -3,17 +3,19 @@
 
 package exporterhelper // import "go.opentelemetry.io/collector/exporter/exporterhelper"
 
-import "go.opentelemetry.io/collector/exporter/internal"
+import (
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
+)
 
 // Request represents a single request that can be sent to an external endpoint.
 // 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.
-type Request = internal.Request
+type Request = request.Request
 
 // RequestErrorHandler is an optional interface that can be implemented by Request to provide a way handle partial
 // temporary failures. For example, if some items failed to process and can be retried, this interface allows to
 // return a new Request that contains the items left to be sent. Otherwise, the original Request should be returned.
 // If not implemented, the original Request will be returned assuming the error is applied to the whole Request.
 // 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.
-type RequestErrorHandler = internal.RequestErrorHandler
+type RequestErrorHandler = request.RequestErrorHandler

exporter/exporterhelper/internal/base_exporter.go

@@ -19,8 +19,8 @@ import (
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterqueue" // BaseExporter contains common fields between different exporter types.
-	"go.opentelemetry.io/collector/exporter/internal"
 	"go.opentelemetry.io/collector/pipeline"
 )
 
@@ -31,8 +31,8 @@ type BaseExporter struct {
 	component.StartFunc
 	component.ShutdownFunc
 
-	Marshaler   exporterqueue.Marshaler[internal.Request]
-	Unmarshaler exporterqueue.Unmarshaler[internal.Request]
+	Marshaler   exporterqueue.Marshaler[request.Request]
+	Unmarshaler exporterqueue.Unmarshaler[request.Request]
 
 	Set exporter.Settings
 
@@ -42,17 +42,17 @@ type BaseExporter struct {
 	// Chain of senders that the exporter helper applies before passing the data to the actual exporter.
 	// The data is handled by each sender in the respective order starting from the queueSender.
 	// Most of the senders are optional, and initialized with a no-op path-through sender.
-	QueueSender  Sender[internal.Request]
-	ObsrepSender Sender[internal.Request]
-	RetrySender  Sender[internal.Request]
+	QueueSender  Sender[request.Request]
+	ObsrepSender Sender[request.Request]
+	RetrySender  Sender[request.Request]
 
-	firstSender Sender[internal.Request]
+	firstSender Sender[request.Request]
 
 	ConsumerOptions []consumer.Option
 
 	timeoutCfg   TimeoutConfig
 	retryCfg     configretry.BackOffConfig
-	queueFactory exporterqueue.Factory[internal.Request]
+	queueFactory exporterqueue.Factory[request.Request]
 	queueCfg     exporterqueue.Config
 	batcherCfg   exporterbatcher.Config
 }
@@ -61,7 +61,7 @@ func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, options ...O
 	be := &BaseExporter{
 		Set:          set,
 		timeoutCfg:   NewDefaultTimeoutConfig(),
-		queueFactory: exporterqueue.NewMemoryQueueFactory[internal.Request](),
+		queueFactory: exporterqueue.NewMemoryQueueFactory[request.Request](),
 	}
 
 	for _, op := range options {
@@ -105,7 +105,7 @@ func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, options ...O
 }
 
 // Send sends the request using the first sender in the chain.
-func (be *BaseExporter) Send(ctx context.Context, req internal.Request) error {
+func (be *BaseExporter) Send(ctx context.Context, req request.Request) error {
 	// Have to read the number of items before sending the request since the request can
 	// be modified by the downstream components like the batcher.
 	itemsCount := req.ItemsCount()
@@ -206,7 +206,7 @@ func WithQueue(config QueueConfig) Option {
 			QueueSize:    config.QueueSize,
 			Blocking:     config.Blocking,
 		}
-		o.queueFactory = exporterqueue.NewPersistentQueueFactory[internal.Request](config.StorageID, exporterqueue.PersistentQueueSettings[internal.Request]{
+		o.queueFactory = exporterqueue.NewPersistentQueueFactory[request.Request](config.StorageID, exporterqueue.PersistentQueueSettings[request.Request]{
 			Marshaler:   o.Marshaler,
 			Unmarshaler: o.Unmarshaler,
 		})
@@ -218,7 +218,7 @@ func WithQueue(config QueueConfig) 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 WithRequestQueue(cfg exporterqueue.Config, queueFactory exporterqueue.Factory[internal.Request]) Option {
+func WithRequestQueue(cfg exporterqueue.Config, queueFactory exporterqueue.Factory[request.Request]) Option {
 	return func(o *BaseExporter) error {
 		if o.Marshaler != nil || o.Unmarshaler != nil {
 			return errors.New("WithRequestQueue option must be used with the new request exporters only, use WithQueue instead")
@@ -257,7 +257,7 @@ func WithBatcher(cfg exporterbatcher.Config) Option {
 
 // WithMarshaler is used to set the request marshaler for the new exporter helper.
 // It must be provided as the first option when creating a new exporter helper.
-func WithMarshaler(marshaler exporterqueue.Marshaler[internal.Request]) Option {
+func WithMarshaler(marshaler exporterqueue.Marshaler[request.Request]) Option {
 	return func(o *BaseExporter) error {
 		o.Marshaler = marshaler
 		return nil
@@ -266,7 +266,7 @@ func WithMarshaler(marshaler exporterqueue.Marshaler[internal.Request]) Option {
 
 // WithUnmarshaler is used to set the request unmarshaler for the new exporter helper.
 // It must be provided as the first option when creating a new exporter helper.
-func WithUnmarshaler(unmarshaler exporterqueue.Unmarshaler[internal.Request]) Option {
+func WithUnmarshaler(unmarshaler exporterqueue.Unmarshaler[request.Request]) Option {
 	return func(o *BaseExporter) error {
 		o.Unmarshaler = unmarshaler
 		return nil

exporter/exporterhelper/internal/base_exporter_test.go

@@ -18,10 +18,10 @@ import (
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/exporter"
 	"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/exporter/internal"
-	"go.opentelemetry.io/collector/exporter/internal/requesttest"
 	"go.opentelemetry.io/collector/pipeline"
 )
 
@@ -39,11 +39,11 @@ var (
 type noopSender struct {
 	component.StartFunc
 	component.ShutdownFunc
-	SendFunc[internal.Request]
+	SendFunc[request.Request]
 }
 
-func newNoopExportSender() Sender[internal.Request] {
-	return &noopSender{SendFunc: func(ctx context.Context, req internal.Request) error {
+func newNoopExportSender() Sender[request.Request] {
+	return &noopSender{SendFunc: func(ctx context.Context, req request.Request) error {
 		select {
 		case <-ctx.Done():
 			return ctx.Err() // Returns the cancellation error
@@ -103,7 +103,7 @@ func TestQueueOptionsWithRequestExporter(t *testing.T) {
 			_, err = NewBaseExporter(exportertest.NewNopSettings(), defaultSignal,
 				WithMarshaler(mockRequestMarshaler), WithUnmarshaler(mockRequestUnmarshaler(&requesttest.FakeRequest{Items: 1})),
 				WithRetry(configretry.NewDefaultBackOffConfig()),
-				WithRequestQueue(exporterqueue.NewDefaultConfig(), exporterqueue.NewMemoryQueueFactory[internal.Request]()))
+				WithRequestQueue(exporterqueue.NewDefaultConfig(), exporterqueue.NewMemoryQueueFactory[request.Request]()))
 			require.Error(t, err)
 		})
 	}
@@ -123,7 +123,7 @@ func TestBaseExporterLogging(t *testing.T) {
 			qCfg := exporterqueue.NewDefaultConfig()
 			qCfg.Enabled = false
 			bs, err := NewBaseExporter(set, defaultSignal,
-				WithRequestQueue(qCfg, exporterqueue.NewMemoryQueueFactory[internal.Request]()),
+				WithRequestQueue(qCfg, exporterqueue.NewMemoryQueueFactory[request.Request]()),
 				WithBatcher(exporterbatcher.NewDefaultConfig()),
 				WithRetry(rCfg))
 			require.NoError(t, err)
@@ -172,7 +172,7 @@ func TestQueueRetryWithDisabledQueue(t *testing.T) {
 				func() Option {
 					qs := exporterqueue.NewDefaultConfig()
 					qs.Enabled = false
-					return WithRequestQueue(qs, exporterqueue.NewMemoryQueueFactory[internal.Request]())
+					return WithRequestQueue(qs, exporterqueue.NewMemoryQueueFactory[request.Request]())
 				}(),
 				func() Option {
 					bs := exporterbatcher.NewDefaultConfig()

exporter/exporterhelper/internal/batch_sender.go

@@ -14,7 +14,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
-	"go.opentelemetry.io/collector/exporter/internal"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 )
 
 // BatchSender is a component that places requests into batches before passing them to the downstream senders.
@@ -24,7 +24,7 @@ import (
 // - concurrencyLimit is reached.
 type BatchSender struct {
 	cfg  exporterbatcher.Config
-	next Sender[internal.Request]
+	next Sender[request.Request]
 
 	// concurrencyLimit is the maximum number of goroutines that can be blocked by the batcher.
 	// If this number is reached and all the goroutines are busy, the batch will be sent right away.
@@ -44,7 +44,7 @@ type BatchSender struct {
 }
 
 // NewBatchSender returns a new batch consumer component.
-func NewBatchSender(cfg exporterbatcher.Config, set exporter.Settings, concurrencyLimit int64, next Sender[internal.Request]) *BatchSender {
+func NewBatchSender(cfg exporterbatcher.Config, set exporter.Settings, concurrencyLimit int64, next Sender[request.Request]) *BatchSender {
 	bs := &BatchSender{
 		cfg:                cfg,
 		next:               next,
@@ -101,7 +101,7 @@ func (bs *BatchSender) Start(_ context.Context, _ component.Host) error {
 
 type batch struct {
 	ctx     context.Context
-	request internal.Request
+	request request.Request
 	done    chan struct{}
 	err     error
 
@@ -137,7 +137,7 @@ func (bs *BatchSender) isActiveBatchReady() bool {
 		(bs.concurrencyLimit > 0 && bs.activeRequests.Load() >= bs.concurrencyLimit)
 }
 
-func (bs *BatchSender) Send(ctx context.Context, req internal.Request) error {
+func (bs *BatchSender) Send(ctx context.Context, req request.Request) error {
 	// Stopped batch sender should act as pass-through to allow the queue to be drained.
 	if bs.stopped.Load() {
 		return bs.next.Send(ctx, req)
@@ -150,10 +150,10 @@ func (bs *BatchSender) Send(ctx context.Context, req internal.Request) error {
 }
 
 // sendMergeSplitBatch sends the request to the batch which may be split into multiple requests.
-func (bs *BatchSender) sendMergeSplitBatch(ctx context.Context, req internal.Request) error {
+func (bs *BatchSender) sendMergeSplitBatch(ctx context.Context, req request.Request) error {
 	bs.mu.Lock()
 
-	var reqs []internal.Request
+	var reqs []request.Request
 	var mergeSplitErr error
 	if bs.activeBatch.request == nil {
 		reqs, mergeSplitErr = req.MergeSplit(ctx, bs.cfg.MaxSizeConfig, nil)
@@ -200,7 +200,7 @@ func (bs *BatchSender) sendMergeSplitBatch(ctx context.Context, req internal.Req
 }
 
 // sendMergeBatch sends the request to the batch and waits for the batch to be exported.
-func (bs *BatchSender) sendMergeBatch(ctx context.Context, req internal.Request) error {
+func (bs *BatchSender) sendMergeBatch(ctx context.Context, req request.Request) error {
 	bs.mu.Lock()
 
 	if bs.activeBatch.request != nil {
@@ -228,7 +228,7 @@ func (bs *BatchSender) sendMergeBatch(ctx context.Context, req internal.Request)
 // The context is only set once and is not updated after the first call.
 // Merging the context would be complex and require an additional goroutine to handle the context cancellation.
 // We take the approach of using the context from the first request since it's likely to have the shortest timeout.
-func (bs *BatchSender) updateActiveBatch(ctx context.Context, req internal.Request) {
+func (bs *BatchSender) updateActiveBatch(ctx context.Context, req request.Request) {
 	if bs.activeBatch.request == nil {
 		bs.activeBatch.ctx = ctx
 	}

exporter/exporterhelper/internal/batch_sender_test.go

@@ -16,8 +16,8 @@ import (
 
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
 	"go.opentelemetry.io/collector/exporter/exporterqueue"
-	"go.opentelemetry.io/collector/exporter/internal/requesttest"
 )
 
 func TestBatchSender_Merge(t *testing.T) {

exporter/internal/queue/batcher.go renamed to exporter/exporterhelper/internal/batcher/batcher.go

@@ -1,25 +1,25 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package queue // import "go.opentelemetry.io/collector/exporter/internal/queue"
+package batcher // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/batcher"
 
 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/exporterqueue"
-	"go.opentelemetry.io/collector/exporter/internal"
 )
 
 // Batcher is in charge of reading items from the queue and send them out asynchronously.
 type Batcher interface {
 	component.Component
-	Consume(context.Context, internal.Request, exporterqueue.Done)
+	Consume(context.Context, request.Request, exporterqueue.Done)
 }
 
 func NewBatcher(batchCfg exporterbatcher.Config,
-	exportFunc func(ctx context.Context, req internal.Request) error,
+	exportFunc func(ctx context.Context, req request.Request) error,
 	maxWorkers int,
 ) (Batcher, error) {
 	if !batchCfg.Enabled {

exporter/internal/queue/default_batcher.go renamed to exporter/exporterhelper/internal/batcher/default_batcher.go

@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package queue // import "go.opentelemetry.io/collector/exporter/internal/queue"
+package batcher // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/batcher"
 
 import (
 	"context"
@@ -12,21 +12,21 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterqueue"
-	"go.opentelemetry.io/collector/exporter/internal"
 )
 
 type batch struct {
 	ctx  context.Context
-	req  internal.Request
+	req  request.Request
 	done multiDone
 }
 
 // defaultBatcher continuously batch incoming requests and flushes asynchronously if minimum size limit is met or on timeout.
 type defaultBatcher struct {
 	batchCfg       exporterbatcher.Config
 	workerPool     chan struct{}
-	exportFunc     func(ctx context.Context, req internal.Request) error
+	exportFunc     func(ctx context.Context, req request.Request) error
 	stopWG         sync.WaitGroup
 	currentBatchMu sync.Mutex
 	currentBatch   *batch
@@ -35,7 +35,7 @@ type defaultBatcher struct {
 }
 
 func newDefaultBatcher(batchCfg exporterbatcher.Config,
-	exportFunc func(ctx context.Context, req internal.Request) error,
+	exportFunc func(ctx context.Context, req request.Request) error,
 	maxWorkers int,
 ) *defaultBatcher {
 	// TODO: Determine what is the right behavior for this in combination with async queue.
@@ -61,10 +61,10 @@ func (qb *defaultBatcher) resetTimer() {
 	}
 }
 
-func (qb *defaultBatcher) Consume(ctx context.Context, req internal.Request, done exporterqueue.Done) {
+func (qb *defaultBatcher) Consume(ctx context.Context, req request.Request, done exporterqueue.Done) {
 	qb.currentBatchMu.Lock()
 
-	var reqList []internal.Request
+	var reqList []request.Request
 	var mergeSplitErr error
 	if qb.currentBatch == nil {
 		reqList, mergeSplitErr = req.MergeSplit(ctx, qb.batchCfg.MaxSizeConfig, nil)
@@ -201,7 +201,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 internal.Request, done exporterqueue.Done) {
+func (qb *defaultBatcher) flush(ctx context.Context, req request.Request, done exporterqueue.Done) {
 	qb.stopWG.Add(1)
 	if qb.workerPool != nil {
 		<-qb.workerPool