Replace exporterhelper Request.Export with passing a ConsumeRequest func

.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
+}