Add support for queued retry in the exporter helper. (#1386)

* Add support for queued retry in the exporter helper.

Changed only the OTLP exporter for the moment to use the new settings.

Timeout is enabled for all the exporters. Fixes #1193

There are some missing features that will be added in a followup PR:
1. Enforcing errors. For the moment added the Throttle error as a hack to keep backwards compatibility with OTLP.
2. Enable queued and retry for all exporters.
3. Fix observability metrics for the case when requests are dropped because the queue is full.

* First round of comments addressed

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
 }