Replace exporterhelper Request.Export with passing a ConsumeRequest func (#12637)

Fixes
#8950

The logic behind this, is that during merging and splitting we should
not care about the export func (see complicated logic removed in
FakeRequest of in the pdata requests implementations).

Also, this simplify what user should carry with the request.

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

Author: bogdandrutu

.chloggen/rm-request-export.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: Remove the Request.Export function in favor of an equivalent request consume func in the New[Traces|Metrics|Logs|Profiles]Request
+
+# One or more tracking issues or pull requests related to the change
+issues: [12637]
+
+# (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/constants.go

@@ -12,12 +12,14 @@ var (
 	errNilConfig = errors.New("nil config")
 	// errNilLogger is returned when a logger is nil
 	errNilLogger = errors.New("nil logger")
-	// errNilPushTraceData is returned when a nil PushTraces is given.
-	errNilPushTraceData = errors.New("nil PushTraces")
-	// errNilPushMetricsData is returned when a nil PushMetrics is given.
-	errNilPushMetricsData = errors.New("nil PushMetrics")
-	// errNilPushLogsData is returned when a nil PushLogs is given.
-	errNilPushLogsData = errors.New("nil PushLogs")
+	// errNilConsumeRequest is returned when a nil PushTraces is given.
+	errNilConsumeRequest = errors.New("nil RequestConsumeFunc")
+	// errNilPushTraces is returned when a nil PushTraces is given.
+	errNilPushTraces = errors.New("nil PushTraces")
+	// errNilPushMetrics is returned when a nil PushMetrics is given.
+	errNilPushMetrics = errors.New("nil PushMetrics")
+	// errNilPushLogs is returned when a nil PushLogs is given.
+	errNilPushLogs = errors.New("nil PushLogs")
 	// errNilTracesConverter is returned when a nil RequestFromTracesFunc is given.
 	errNilTracesConverter = errors.New("nil RequestFromTracesFunc")
 	// errNilMetricsConverter is returned when a nil RequestFromMetricsFunc is given.

exporter/exporterhelper/internal/base_exporter.go

@@ -50,7 +50,7 @@ type BaseExporter struct {
 	batcherCfg exporterbatcher.Config
 }
 
-func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, options ...Option) (*BaseExporter, error) {
+func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, pusher func(context.Context, request.Request) error, options ...Option) (*BaseExporter, error) {
 	be := &BaseExporter{
 		Set:        set,
 		timeoutCfg: NewDefaultTimeoutConfig(),
@@ -62,15 +62,13 @@ func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, options ...O
 		}
 	}
 
-	//nolint: staticcheck
+	//nolint:staticcheck
 	if be.batcherCfg.MinSizeItems != nil || be.batcherCfg.MaxSizeItems != nil {
 		set.Logger.Warn("Using of deprecated fields `min_size_items` and `max_size_items`")
 	}
 
 	// Consumer Sender is always initialized.
-	be.firstSender = newSender(func(ctx context.Context, req request.Request) error {
-		return req.Export(ctx)
-	})
+	be.firstSender = newSender(pusher)
 
 	// Next setup the timeout Sender since we want the timeout to control only the export functionality.
 	// Only initialize if not explicitly disabled.

exporter/exporterhelper/internal/base_exporter_test.go

@@ -36,19 +36,8 @@ var (
 	}()
 )
 
-func newNoopExportSender() Sender[request.Request] {
-	return newSender(func(ctx context.Context, req request.Request) error {
-		select {
-		case <-ctx.Done():
-			return ctx.Err() // Returns the cancellation error
-		default:
-			return req.Export(ctx)
-		}
-	})
-}
-
 func TestBaseExporter(t *testing.T) {
-	be, err := NewBaseExporter(defaultSettings, defaultSignal)
+	be, err := NewBaseExporter(defaultSettings, defaultSignal, noopExport)
 	require.NoError(t, err)
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
 	require.NoError(t, be.Shutdown(context.Background()))
@@ -57,7 +46,7 @@ func TestBaseExporter(t *testing.T) {
 func TestBaseExporterWithOptions(t *testing.T) {
 	want := errors.New("my error")
 	be, err := NewBaseExporter(
-		defaultSettings, defaultSignal,
+		defaultSettings, defaultSignal, noopExport,
 		WithStart(func(context.Context, component.Host) error { return want }),
 		WithShutdown(func(context.Context) error { return want }),
 		WithTimeout(NewDefaultTimeoutConfig()),
@@ -68,18 +57,18 @@ func TestBaseExporterWithOptions(t *testing.T) {
 }
 
 func TestQueueOptionsWithRequestExporter(t *testing.T) {
-	bs, err := NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), defaultSignal,
+	bs, err := NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), defaultSignal, noopExport,
 		WithRetry(configretry.NewDefaultBackOffConfig()))
 	require.NoError(t, err)
 	require.Nil(t, bs.encoding)
-	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), defaultSignal,
+	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), defaultSignal, noopExport,
 		WithRetry(configretry.NewDefaultBackOffConfig()), WithQueue(exporterqueue.NewDefaultConfig()))
 	require.Error(t, err)
 
 	qCfg := exporterqueue.NewDefaultConfig()
 	storageID := component.NewID(component.MustNewType("test"))
 	qCfg.StorageID = &storageID
-	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), defaultSignal,
+	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), defaultSignal, noopExport,
 		WithEncoding(newFakeEncoding(&requesttest.FakeRequest{Items: 1})),
 		WithRetry(configretry.NewDefaultBackOffConfig()),
 		WithRequestQueue(qCfg, nil))
@@ -94,14 +83,13 @@ func TestBaseExporterLogging(t *testing.T) {
 	rCfg.Enabled = false
 	qCfg := exporterqueue.NewDefaultConfig()
 	qCfg.Enabled = false
-	bs, err := NewBaseExporter(set, defaultSignal,
+	bs, err := NewBaseExporter(set, defaultSignal, errExport,
 		WithRequestQueue(qCfg, newFakeEncoding(&requesttest.FakeRequest{})),
 		WithBatcher(exporterbatcher.NewDefaultConfig()),
 		WithRetry(rCfg))
 	require.NoError(t, err)
 	require.NoError(t, bs.Start(context.Background(), componenttest.NewNopHost()))
-	sink := requesttest.NewSink()
-	sendErr := bs.Send(context.Background(), &requesttest.FakeRequest{Items: 2, Sink: sink, ExportErr: errors.New("my error")})
+	sendErr := bs.Send(context.Background(), &requesttest.FakeRequest{Items: 2})
 	require.Error(t, sendErr)
 
 	require.Len(t, observed.FilterLevelExact(zap.ErrorLevel).All(), 2)
@@ -155,16 +143,22 @@ func TestQueueRetryWithDisabledQueue(t *testing.T) {
 			set := exportertest.NewNopSettings(exportertest.NopType)
 			logger, observed := observer.New(zap.ErrorLevel)
 			set.Logger = zap.New(logger)
-			be, err := NewBaseExporter(set, pipeline.SignalLogs, tt.queueOptions...)
+			be, err := NewBaseExporter(set, pipeline.SignalLogs, errExport, tt.queueOptions...)
 			require.NoError(t, err)
 			require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
-			sink := requesttest.NewSink()
-			mockR := &requesttest.FakeRequest{Items: 2, Sink: sink, ExportErr: errors.New("some error")}
+			mockR := &requesttest.FakeRequest{Items: 2}
 			require.Error(t, be.Send(context.Background(), mockR))
 			assert.Len(t, observed.All(), 1)
 			assert.Equal(t, "Exporting failed. Rejecting data. Try enabling sending_queue to survive temporary failures.", observed.All()[0].Message)
 			require.NoError(t, be.Shutdown(context.Background()))
-			assert.Empty(t, 0, sink.RequestsCount())
 		})
 	}
 }
+
+func errExport(context.Context, request.Request) error {
+	return errors.New("my error")
+}
+
+func noopExport(context.Context, request.Request) error {
+	return nil
+}