Add support for queued retry in the exporter helper.

exporter/exporterhelper/common.go

@@ -16,72 +16,217 @@ package exporterhelper
 
 import (
 	"context"
+	"sync"
+	"time"
 
 	"go.opencensus.io/trace"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/component/componenterror"
+	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/consumer/consumererror"
 )
 
 var (
 	okStatus = trace.Status{Code: trace.StatusCodeOK}
 )
 
+// Settings for timeout. The timeout applies to individual attempts to send data to the backend.
+type TimeoutSettings struct {
+	// Timeout is the timeout for every attempt to send data to the backend.
+	Timeout time.Duration `mapstructure:"timeout"`
+}
+
+// CreateDefaultTimeoutSettings returns the default settings for TimeoutSettings.
+func CreateDefaultTimeoutSettings() TimeoutSettings {
+	return TimeoutSettings{
+		Timeout: 5 * time.Second,
+	}
+}
+
+// request is an abstraction of an individual request (batch of data) independent of the type of the data (traces, metrics, logs).
+type request interface {
+	// context returns the Context of the requests.
+	context() context.Context
+	// setContext updates the Context of the requests.
+	setContext(context.Context)
+	export(ctx context.Context) (int, error)
+	// Returns a new request that contains the items left to be sent.
+	onPartialError(consumererror.PartialError) request
+	// Returns the count of spans/metric points or log records.
+	count() int
+}
+
+// requestSender is an abstraction of a sender for a request independent of the type of the data (traces, metrics, logs).
+type requestSender interface {
+	send(req request) (int, error)
+}
+
+// baseRequest is a base implementation for the request.
+type baseRequest struct {
+	ctx context.Context
+}
+
+func (req *baseRequest) context() context.Context {
+	return req.ctx
+}
+
+func (req *baseRequest) setContext(ctx context.Context) {
+	req.ctx = ctx
+}
+
 // Start specifies the function invoked when the exporter is being started.
 type Start func(context.Context, component.Host) error
 
 // Shutdown specifies the function invoked when the exporter is being shutdown.
 type Shutdown func(context.Context) error
 
+// internalOptions represents all the options that users can configure.
+type internalOptions struct {
+	TimeoutSettings
+	QueueSettings
+	RetrySettings
+	Start
+	Shutdown
+}
+
+// fromConfiguredOptions returns the internal options starting from the default and applying all configured options.
+func fromConfiguredOptions(options ...ExporterOption) *internalOptions {
+	// Start from the default options:
+	opts := &internalOptions{
+		TimeoutSettings: CreateDefaultTimeoutSettings(),
+		// TODO: Enable queuing by default (call CreateDefaultQueueSettings)
+		QueueSettings: QueueSettings{Disabled: true},
+		// TODO: Enable retry by default (call CreateDefaultRetrySettings)
+		RetrySettings: RetrySettings{Disabled: true},
+		Start:         func(ctx context.Context, host component.Host) error { return nil },
+		Shutdown:      func(ctx context.Context) error { return nil },
+	}
+
+	for _, op := range options {
+		op(opts)
+	}
+
+	return opts
+}
+
 // ExporterOption apply changes to internalOptions.
-type ExporterOption func(*baseExporter)
+type ExporterOption func(*internalOptions)
 
 // WithShutdown overrides the default Shutdown function for an exporter.
 // The default shutdown function does nothing and always returns nil.
 func WithShutdown(shutdown Shutdown) ExporterOption {
-	return func(o *baseExporter) {
-		o.shutdown = shutdown
+	return func(o *internalOptions) {
+		o.Shutdown = shutdown
 	}
 }
 
 // WithStart overrides the default Start function for an exporter.
 // The default shutdown function does nothing and always returns nil.
 func WithStart(start Start) ExporterOption {
-	return func(o *baseExporter) {
-		o.start = start
+	return func(o *internalOptions) {
+		o.Start = start
 	}
 }
 
-// internalOptions contains internalOptions concerning how an Exporter is configured.
-type baseExporter struct {
-	exporterFullName string
-	start            Start
-	shutdown         Shutdown
+// WithTimeout overrides the default TimeoutSettings for an exporter.
+// The default TimeoutSettings is 5 seconds.
+func WithTimeout(timeoutSettings TimeoutSettings) ExporterOption {
+	return func(o *internalOptions) {
+		o.TimeoutSettings = timeoutSettings
+	}
 }
 
-// Construct the internalOptions from multiple ExporterOption.
-func newBaseExporter(exporterFullName string, options ...ExporterOption) baseExporter {
-	be := baseExporter{
-		exporterFullName: exporterFullName,
+// WithRetry overrides the default RetrySettings for an exporter.
+// The default RetrySettings is to disable retries.
+func WithRetry(retrySettings RetrySettings) ExporterOption {
+	return func(o *internalOptions) {
+		o.RetrySettings = retrySettings
 	}
+}
 
-	for _, op := range options {
-		op(&be)
+// WithQueue overrides the default QueueSettings for an exporter.
+// The default QueueSettings is to disable queueing.
+func WithQueue(queueSettings QueueSettings) ExporterOption {
+	return func(o *internalOptions) {
+		o.QueueSettings = queueSettings
 	}
+}
+
+// baseExporter contains common fields between different exporter types.
+type baseExporter struct {
+	cfg          configmodels.Exporter
+	sender       requestSender
+	qrSender     *queuedRetrySender
+	start        Start
+	shutdown     Shutdown
+	startOnce    sync.Once
+	shutdownOnce sync.Once
+}
+
+func newBaseExporter(cfg configmodels.Exporter, options ...ExporterOption) *baseExporter {
+	opts := fromConfiguredOptions(options...)
+	be := &baseExporter{
+		cfg:      cfg,
+		start:    opts.Start,
+		shutdown: opts.Shutdown,
+	}
+
+	be.qrSender = newQueuedRetrySender(opts.QueueSettings, opts.RetrySettings, &timeoutSender{cfg: opts.TimeoutSettings})
+	be.sender = be.qrSender
 
 	return be
 }
 
+// wrapConsumerSender wraps the consumer sender (the sender that uses retries and timeout) with the given wrapper.
+// This can be used to wrap with observability (create spans, record metrics) the consumer sender.
+func (be *baseExporter) wrapConsumerSender(f func(consumer requestSender) requestSender) {
+	be.qrSender.consumerSender = f(be.qrSender.consumerSender)
+}
+
+// Start all senders and exporter and is invoked during service start.
 func (be *baseExporter) Start(ctx context.Context, host component.Host) error {
-	if be.start != nil {
-		return be.start(ctx, host)
-	}
-	return nil
+	err := componenterror.ErrAlreadyStarted
+	be.startOnce.Do(func() {
+		// First start the wrapped exporter.
+		err = be.start(ctx, host)
+		if err != nil {
+			// TODO: Log errors, or check if it is recorded by the caller.
+			return
+		}
+
+		// If no error then start the queuedRetrySender.
+		be.qrSender.start()
+	})
+	return err
 }
 
-// Shutdown stops the exporter and is invoked during shutdown.
+// Shutdown all senders and exporter and is invoked during service shutdown.
 func (be *baseExporter) Shutdown(ctx context.Context) error {
-	if be.shutdown != nil {
-		return be.shutdown(ctx)
+	err := componenterror.ErrAlreadyStopped
+	be.shutdownOnce.Do(func() {
+		// First shutdown the queued retry sender
+		be.qrSender.shutdown()
+		// Last shutdown the wrapped exporter itself.
+		err = be.shutdown(ctx)
+	})
+	return err
+}
+
+// timeoutSender is a request sender that adds a `timeout` to every request that passes this sender.
+type timeoutSender struct {
+	cfg TimeoutSettings
+}
+
+// send implements the requestSender interface
+func (ts *timeoutSender) send(req request) (int, error) {
+	// Intentionally don't overwrite the context inside the request, because in case of retries deadline will not be
+	// updated because this deadline most likely is before the next one.
+	ctx := req.context()
+	if ts.cfg.Timeout > 0 {
+		var cancelFunc func()
+		ctx, cancelFunc = context.WithTimeout(req.context(), ts.cfg.Timeout)
+		defer cancelFunc()
 	}
-	return nil
+	return req.export(ctx)
 }

exporter/exporterhelper/common_test.go

@@ -23,22 +23,28 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config/configmodels"
 )
 
+var defaultExporterCfg = &configmodels.ExporterSettings{
+	TypeVal: "test",
+	NameVal: "test",
+}
+
 func TestErrorToStatus(t *testing.T) {
 	require.Equal(t, okStatus, errToStatus(nil))
 	require.Equal(t, trace.Status{Code: trace.StatusCodeUnknown, Message: "my_error"}, errToStatus(errors.New("my_error")))
 }
 
 func TestBaseExporter(t *testing.T) {
-	be := newBaseExporter("test")
+	be := newBaseExporter(defaultExporterCfg)
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
 	require.NoError(t, be.Shutdown(context.Background()))
 }
 
 func TestBaseExporterWithOptions(t *testing.T) {
 	be := newBaseExporter(
-		"test",
+		defaultExporterCfg,
 		WithStart(func(ctx context.Context, host component.Host) error { return errors.New("my error") }),
 		WithShutdown(func(ctx context.Context) error { return errors.New("my error") }))
 	require.Error(t, be.Start(context.Background(), componenttest.NewNopHost()))

exporter/exporterhelper/logshelper.go

@@ -19,6 +19,7 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/internal/data"
 	"go.opentelemetry.io/collector/obsreport"
 )
@@ -27,44 +28,76 @@ import (
 // the number of dropped logs.
 type PushLogsData func(ctx context.Context, md data.Logs) (droppedTimeSeries int, err error)
 
+type logsRequest struct {
+	baseRequest
+	ld     data.Logs
+	pusher PushLogsData
+}
+
+func newLogsRequest(ctx context.Context, ld data.Logs, pusher PushLogsData) request {
+	return &logsRequest{
+		baseRequest: baseRequest{ctx: ctx},
+		ld:          ld,
+		pusher:      pusher,
+	}
+}
+
+func (req *logsRequest) onPartialError(partialErr consumererror.PartialError) request {
+	// TODO: Implement this
+	return req
+}
+
+func (req *logsRequest) export(ctx context.Context) (int, error) {
+	return req.pusher(ctx, req.ld)
+}
+
+func (req *logsRequest) count() int {
+	return req.ld.LogRecordCount()
+}
+
 type logsExporter struct {
-	baseExporter
+	*baseExporter
 	pushLogsData PushLogsData
 }
 
-func (me *logsExporter) ConsumeLogs(ctx context.Context, md data.Logs) error {
-	exporterCtx := obsreport.ExporterContext(ctx, me.exporterFullName)
-	_, err := me.pushLogsData(exporterCtx, md)
+func (lexp *logsExporter) ConsumeLogs(ctx context.Context, ld data.Logs) error {
+	exporterCtx := obsreport.ExporterContext(ctx, lexp.cfg.Name())
+	_, err := lexp.sender.send(newLogsRequest(exporterCtx, ld, lexp.pushLogsData))
 	return err
 }
 
-// NewLogsExporter creates an LogsExporter that can record logs and can wrap every request with a Span.
-// TODO: Add support for retries.
-func NewLogsExporter(config configmodels.Exporter, pushLogsData PushLogsData, options ...ExporterOption) (component.LogExporter, error) {
-	if config == nil {
+// NewLogsExporter creates an LogsExporter that records observability metrics and wraps every request with a Span.
+func NewLogsExporter(cfg configmodels.Exporter, pushLogsData PushLogsData, options ...ExporterOption) (component.LogExporter, error) {
+	if cfg == nil {
 		return nil, errNilConfig
 	}
 
 	if pushLogsData == nil {
 		return nil, errNilPushLogsData
 	}
 
-	pushLogsData = pushLogsWithObservability(pushLogsData, config.Name())
+	be := newBaseExporter(cfg, options...)
+	be.wrapConsumerSender(func(nextSender requestSender) requestSender {
+		return &logsExporterWithObservability{
+			exporterName: cfg.Name(),
+			nextSender:   nextSender,
+		}
+	})
 
 	return &logsExporter{
-		baseExporter: newBaseExporter(config.Name(), options...),
+		baseExporter: be,
 		pushLogsData: pushLogsData,
 	}, nil
 }
 
-func pushLogsWithObservability(next PushLogsData, exporterName string) PushLogsData {
-	return func(ctx context.Context, ld data.Logs) (int, error) {
-		ctx = obsreport.StartLogsExportOp(ctx, exporterName)
-		numDroppedLogs, err := next(ctx, ld)
-
-		numLogs := ld.LogRecordCount()
+type logsExporterWithObservability struct {
+	exporterName string
+	nextSender   requestSender
+}
 
-		obsreport.EndLogsExportOp(ctx, numLogs, numDroppedLogs, err)
-		return numLogs, err
-	}
+func (lewo *logsExporterWithObservability) send(req request) (int, error) {
+	req.setContext(obsreport.StartLogsExportOp(req.context(), lewo.exporterName))
+	numDroppedLogs, err := lewo.nextSender.send(req)
+	obsreport.EndLogsExportOp(req.context(), req.count(), numDroppedLogs, err)
+	return numDroppedLogs, err
 }