[exporterhelper] New exporter helper with costom requests

Introduce a new exporter helper that operates over client provided requests instead of pdata. 

It opens a door for moving batching to the exporter where batches will be built from clients data format, instead of pdata. The batches can be properly sized by custom request size which can be different from OTLP. The same custom request sizing will be applied to the sending queue. It will also improve performance of the sending queue retries for non-OTLP exporters, they don't need to translate pdata on every retry. 

This is an experimental API, once stabilized it's intended to replace the existing helpers.

exporter/exporterhelper/common.go

@@ -56,14 +56,25 @@ func (req *baseRequest) OnProcessingFinished() {
 	}
 }
 
+type queueSettings struct {
+	config      QueueSettings
+	marshaler   internal.RequestMarshaler
+	unmarshaler internal.RequestUnmarshaler
+}
+
+func (qs *queueSettings) persistenceEnabled() bool {
+	return qs.config.StorageID != nil && qs.marshaler != nil && qs.unmarshaler != nil
+}
+
 // baseSettings represents all the options that users can configure.
 type baseSettings struct {
 	component.StartFunc
 	component.ShutdownFunc
 	consumerOptions []consumer.Option
 	TimeoutSettings
-	QueueSettings
+	queueSettings
 	RetrySettings
+	RequestSender
 }
 
 // fromOptions returns the internal options starting from the default and applying all configured options.
@@ -72,7 +83,9 @@ func fromOptions(options ...Option) *baseSettings {
 	opts := &baseSettings{
 		TimeoutSettings: NewDefaultTimeoutSettings(),
 		// TODO: Enable queuing by default (call DefaultQueueSettings)
-		QueueSettings: QueueSettings{Enabled: false},
+		queueSettings: queueSettings{
+			config: QueueSettings{Enabled: false},
+		},
 		// TODO: Enable retry by default (call DefaultRetrySettings)
 		RetrySettings: RetrySettings{Enabled: false},
 	}
@@ -121,9 +134,35 @@ func WithRetry(retrySettings RetrySettings) Option {
 
 // WithQueue overrides the default QueueSettings for an exporter.
 // The default QueueSettings is to disable queueing.
-func WithQueue(queueSettings QueueSettings) Option {
+func WithQueue(config QueueSettings) Option {
 	return func(o *baseSettings) {
-		o.QueueSettings = queueSettings
+		o.queueSettings.config = config
+	}
+}
+
+// WithPersistentQueue overrides the default QueueSettings for an exporter with persistent queue.
+// Takes effect only for the new style helpers: NewTracesExporterV2, NewMetricsExporterV2, NewLogsExporterV2.
+func WithPersistentQueue(config QueueSettings, marshaler RequestMarshaler, unmarshaler RequestUnmarshaler) Option {
+	return func(o *baseSettings) {
+		o.queueSettings = queueSettings{
+			config: config,
+			marshaler: func(req internal.Request) ([]byte, error) {
+				return marshaler(req.(*request).Request)
+			},
+			unmarshaler: func(buf []byte) (internal.Request, error) {
+				req, err := unmarshaler(buf)
+				if err != nil {
+					return nil, err
+				}
+				return &request{
+					Request: req,
+					baseRequest: baseRequest{
+						ctx: context.Background(),
+					},
+					sender: o.RequestSender,
+				}, nil
+			},
+		}
 	}
 }
 
@@ -145,7 +184,7 @@ type baseExporter struct {
 	qrSender *queuedRetrySender
 }
 
-func newBaseExporter(set exporter.CreateSettings, bs *baseSettings, signal component.DataType, reqUnmarshaler internal.RequestUnmarshaler) (*baseExporter, error) {
+func newBaseExporter(set exporter.CreateSettings, bs *baseSettings, signal component.DataType) (*baseExporter, error) {
 	be := &baseExporter{}
 
 	var err error
@@ -154,7 +193,7 @@ func newBaseExporter(set exporter.CreateSettings, bs *baseSettings, signal compo
 		return nil, err
 	}
 
-	be.qrSender = newQueuedRetrySender(set.ID, signal, bs.QueueSettings, bs.RetrySettings, reqUnmarshaler, &timeoutSender{cfg: bs.TimeoutSettings}, set.Logger)
+	be.qrSender = newQueuedRetrySender(set.ID, signal, bs.queueSettings, bs.RetrySettings, &timeoutSender{cfg: bs.TimeoutSettings}, set.Logger)
 	be.sender = be.qrSender
 	be.StartFunc = func(ctx context.Context, host component.Host) error {
 		// First start the wrapped exporter.

exporter/exporterhelper/common_test.go

@@ -16,7 +16,6 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/exporter"
-	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 	"go.opentelemetry.io/collector/exporter/exportertest"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 )
@@ -35,7 +34,7 @@ var (
 )
 
 func TestBaseExporter(t *testing.T) {
-	be, err := newBaseExporter(defaultSettings, fromOptions(), "", nopRequestUnmarshaler())
+	be, err := newBaseExporter(defaultSettings, fromOptions(), "")
 	require.NoError(t, err)
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
 	require.NoError(t, be.Shutdown(context.Background()))
@@ -50,7 +49,6 @@ func TestBaseExporterWithOptions(t *testing.T) {
 			WithShutdown(func(ctx context.Context) error { return want }),
 			WithTimeout(NewDefaultTimeoutSettings())),
 		"",
-		nopRequestUnmarshaler(),
 	)
 	require.NoError(t, err)
 	require.Equal(t, want, be.Start(context.Background(), componenttest.NewNopHost()))
@@ -71,7 +69,3 @@ func nopTracePusher() consumer.ConsumeTracesFunc {
 		return nil
 	}
 }
-
-func nopRequestUnmarshaler() internal.RequestUnmarshaler {
-	return newTraceRequestUnmarshalerFunc(nopTracePusher())
-}

exporter/exporterhelper/constants.go

@@ -18,4 +18,12 @@ var (
 	errNilPushMetricsData = errors.New("nil PushMetrics")
 	// errNilPushLogsData is returned when a nil PushLogs is given.
 	errNilPushLogsData = errors.New("nil PushLogs")
+	// errNilRequestFromTracesConverter is returned when a nil RequestFromTracesConverter is given.
+	errNilRequestFromTracesConverter = errors.New("nil RequestFromTracesConverter")
+	// errNilRequestFromMetricsConverter is returned when a nil RequestFromMetricsConverter is given.
+	errNilRequestFromMetricsConverter = errors.New("nil RequestFromMetricsConverter")
+	// errNilRequestFromLogsConverter is returned when a nil RequestFromLogsConverter is given.
+	errNilRequestFromLogsConverter = errors.New("nil RequestFromLogsConverter")
+	// errNilRequestSender is returned when a nil RequestSender is given.
+	errNilRequestSender = errors.New("nil RequestSender")
 )

exporter/exporterhelper/internal/persistent_queue.go

@@ -35,11 +35,28 @@ func buildPersistentStorageName(name string, signal component.DataType) string {
 	return fmt.Sprintf("%s-%s", name, signal)
 }
 
+type PersistentQueueSettings struct {
+	// Name of the queue
+	Name string
+	// Signal type of the queue
+	Signal component.DataType
+	// Capacity of the queue
+	Capacity uint64
+	// Logger to be used by the queue
+	Logger *zap.Logger
+	// Client to be used by the queue
+	Client storage.Client
+	// Unmarshaler to be used by the queue
+	Unmarshaler RequestUnmarshaler
+	// Marshaler to be used by the queue
+	Marshaler RequestMarshaler
+}
+
 // NewPersistentQueue creates a new queue backed by file storage; name and signal must be a unique combination that identifies the queue storage
-func NewPersistentQueue(ctx context.Context, name string, signal component.DataType, capacity int, logger *zap.Logger, client storage.Client, unmarshaler RequestUnmarshaler) ProducerConsumerQueue {
+func NewPersistentQueue(ctx context.Context, params PersistentQueueSettings) ProducerConsumerQueue {
 	return &persistentQueue{
 		stopChan: make(chan struct{}),
-		storage:  newPersistentContiguousStorage(ctx, buildPersistentStorageName(name, signal), uint64(capacity), logger, client, unmarshaler),
+		storage:  newPersistentContiguousStorage(ctx, buildPersistentStorageName(params.Name, params.Signal), params),
 	}
 }
 

exporter/exporterhelper/internal/persistent_queue_test.go

@@ -28,7 +28,15 @@ func createTestQueue(extension storage.Extension, capacity int) *persistentQueue
 		panic(err)
 	}
 
-	wq := NewPersistentQueue(context.Background(), "foo", component.DataTypeTraces, capacity, logger, client, newFakeTracesRequestUnmarshalerFunc())
+	wq := NewPersistentQueue(context.Background(), PersistentQueueSettings{
+		Name:        "foo",
+		Signal:      component.DataTypeTraces,
+		Capacity:    uint64(capacity),
+		Logger:      logger,
+		Client:      client,
+		Unmarshaler: newFakeTracesRequestUnmarshalerFunc(),
+		Marshaler:   newFakeTracesRequestMarshalerFunc(),
+	})
 	return wq.(*persistentQueue)
 }
 

exporter/exporterhelper/internal/persistent_storage.go

@@ -12,7 +12,7 @@ import (
 	"sync/atomic"
 
 	"go.uber.org/zap"
-
+	
 	"go.opentelemetry.io/collector/extension/experimental/storage"
 )
 
@@ -43,6 +43,7 @@ type persistentContiguousStorage struct {
 	queueName   string
 	client      storage.Client
 	unmarshaler RequestUnmarshaler
+	marshaler   RequestMarshaler
 
 	putChan  chan struct{}
 	stopChan chan struct{}
@@ -80,14 +81,15 @@ var (
 
 // newPersistentContiguousStorage creates a new file-storage extension backed queue;
 // queueName parameter must be a unique value that identifies the queue.
-func newPersistentContiguousStorage(ctx context.Context, queueName string, capacity uint64, logger *zap.Logger, client storage.Client, unmarshaler RequestUnmarshaler) *persistentContiguousStorage {
+func newPersistentContiguousStorage(ctx context.Context, queueName string, set PersistentQueueSettings) *persistentContiguousStorage {
 	pcs := &persistentContiguousStorage{
-		logger:      logger,
-		client:      client,
+		logger:      set.Logger,
+		client:      set.Client,
 		queueName:   queueName,
-		unmarshaler: unmarshaler,
-		capacity:    capacity,
-		putChan:     make(chan struct{}, capacity),
+		unmarshaler: set.Unmarshaler,
+		marshaler:   set.Marshaler,
+		capacity:    set.Capacity,
+		putChan:     make(chan struct{}, set.Capacity),
 		reqChan:     make(chan Request),
 		stopChan:    make(chan struct{}),
 		itemsCount:  &atomic.Uint64{},

exporter/exporterhelper/internal/persistent_storage_batch.go

@@ -137,7 +137,7 @@ func (bof *batchStruct) getItemIndexArrayResult(key string) ([]itemIndex, error)
 
 // setRequest adds Set operation over a given request to the batch
 func (bof *batchStruct) setRequest(key string, value Request) *batchStruct {
-	return bof.set(key, value, requestToBytes)
+	return bof.set(key, value, bof.requestToBytes)
 }
 
 // setItemIndex adds Set operation over a given itemIndex to the batch
@@ -206,8 +206,8 @@ func bytesToItemIndexArray(b []byte) (any, error) {
 	return val, err
 }
 
-func requestToBytes(req any) ([]byte, error) {
-	return req.(Request).Marshal()
+func (bof *batchStruct) requestToBytes(req any) ([]byte, error) {
+	return bof.pcs.marshaler(req.(Request))
 }
 
 func (bof *batchStruct) bytesToRequest(b []byte) (any, error) {

exporter/exporterhelper/internal/persistent_storage_test.go

@@ -36,7 +36,13 @@ func createTestClient(extension storage.Extension) storage.Client {
 }
 
 func createTestPersistentStorageWithLoggingAndCapacity(client storage.Client, logger *zap.Logger, capacity uint64) *persistentContiguousStorage {
-	return newPersistentContiguousStorage(context.Background(), "foo", capacity, logger, client, newFakeTracesRequestUnmarshalerFunc())
+	return newPersistentContiguousStorage(context.Background(), "foo", PersistentQueueSettings{
+		Capacity:    capacity,
+		Logger:      logger,
+		Client:      client,
+		Unmarshaler: newFakeTracesRequestUnmarshalerFunc(),
+		Marshaler:   newFakeTracesRequestMarshalerFunc(),
+	})
 }
 
 func createTestPersistentStorage(client storage.Client) *persistentContiguousStorage {
@@ -82,6 +88,13 @@ func newFakeTracesRequestUnmarshalerFunc() RequestUnmarshaler {
 	}
 }
 
+func newFakeTracesRequestMarshalerFunc() RequestMarshaler {
+	return func(req Request) ([]byte, error) {
+		marshaler := ptrace.ProtoMarshaler{}
+		return marshaler.MarshalTraces(req.(*fakeTracesRequest).td)
+	}
+}
+
 func TestPersistentStorage_CorruptedData(t *testing.T) {
 	path := t.TempDir()
 

exporter/exporterhelper/internal/request.go

@@ -19,11 +19,8 @@ type Request interface {
 	// Otherwise, it should return the original Request.
 	OnError(error) Request
 
-	// Count returns the count of spans/metric points or log records.
-	Count() int
-
-	// Marshal serializes the current request into a byte stream
-	Marshal() ([]byte, error)
+	// ItemsCount returns the number of basic items in the request (spans, date points or log records for OTLP)
+	ItemsCount() int
 
 	// OnProcessingFinished calls the optional callback function to handle cleanup after all processing is finished
 	OnProcessingFinished()
@@ -34,3 +31,6 @@ type Request interface {
 
 // RequestUnmarshaler defines a function which takes a byte slice and unmarshals it into a relevant request
 type RequestUnmarshaler func([]byte) (Request, error)
+
+// RequestMarshaler defines a function which takes a request and marshals it into a byte slice
+type RequestMarshaler func(Request) ([]byte, error)