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.

exporter/exporterhelper/common.go

@@ -16,16 +16,65 @@ 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}
 )
 
+type TimeoutSettings struct {
+	// Timeout is the timeout for each operation.
+	Timeout time.Duration `mapstructure:"timeout"`
+}
+
+func CreateDefaultTimeoutSettings() TimeoutSettings {
+	return TimeoutSettings{
+		Timeout: 5 * time.Second,
+	}
+}
+
+type settings struct {
+	configmodels.Exporter
+	TimeoutSettings
+	QueuedSettings
+	RetrySettings
+}
+
+type request interface {
+	context() context.Context
+	setContext(context.Context)
+	export(ctx context.Context) (int, error)
+	// Returns a new queue request that contains the items left to be exported.
+	onPartialError(consumererror.PartialError) request
+	// Returns the cnt of spans/metric points or log records.
+	count() int
+}
+
+type requestSender interface {
+	send(req request) (int, error)
+}
+
+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
 
@@ -51,37 +100,121 @@ func WithStart(start Start) ExporterOption {
 	}
 }
 
+// WithShutdown overrides the default TimeoutSettings for an exporter.
+// The default TimeoutSettings is 5 seconds.
+func WithTimeout(timeout TimeoutSettings) ExporterOption {
+	return func(o *baseExporter) {
+		o.cfg.TimeoutSettings = timeout
+	}
+}
+
+// WithRetry overrides the default RetrySettings for an exporter.
+// The default RetrySettings is to disable retries.
+func WithRetry(retry RetrySettings) ExporterOption {
+	return func(o *baseExporter) {
+		o.cfg.RetrySettings = retry
+	}
+}
+
+// WithQueued overrides the default QueuedSettings for an exporter.
+// The default QueuedSettings is to disable queueing.
+func WithQueued(queued QueuedSettings) ExporterOption {
+	return func(o *baseExporter) {
+		o.cfg.QueuedSettings = queued
+	}
+}
+
 // internalOptions contains internalOptions concerning how an Exporter is configured.
 type baseExporter struct {
-	exporterFullName string
-	start            Start
-	shutdown         Shutdown
+	cfg          *settings
+	sender       requestSender
+	rSender      *retrySender
+	qSender      *queuedSender
+	start        Start
+	shutdown     Shutdown
+	startOnce    sync.Once
+	shutdownOnce sync.Once
 }
 
 // Construct the internalOptions from multiple ExporterOption.
-func newBaseExporter(exporterFullName string, options ...ExporterOption) baseExporter {
-	be := baseExporter{
-		exporterFullName: exporterFullName,
+func newBaseExporter(cfg configmodels.Exporter, options ...ExporterOption) *baseExporter {
+	be := &baseExporter{
+		cfg: &settings{
+			Exporter:        cfg,
+			TimeoutSettings: CreateDefaultTimeoutSettings(),
+			// TODO: Enable queuing by default (call CreateDefaultQueuedSettings
+			QueuedSettings: QueuedSettings{Disabled: true},
+			// TODO: Enable retry by default (call CreateDefaultRetrySettings)
+			RetrySettings: RetrySettings{Disabled: true},
+		},
 	}
 
 	for _, op := range options {
-		op(&be)
+		op(be)
+	}
+
+	if be.start == nil {
+		be.start = func(ctx context.Context, host component.Host) error { return nil }
 	}
 
+	if be.shutdown == nil {
+		be.shutdown = func(ctx context.Context) error { return nil }
+	}
+
+	be.sender = &timeoutSender{cfg: &be.cfg.TimeoutSettings}
+
+	be.rSender = newRetrySender(&be.cfg.RetrySettings, be.sender)
+	be.sender = be.rSender
+
+	be.qSender = newQueuedSender(&be.cfg.QueuedSettings, be.sender)
+	be.sender = be.qSender
+
 	return be
 }
 
 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 nextSender
+		err = be.start(ctx, host)
+		if err != nil {
+			return
+		}
+
+		// If no error then start the queuedSender
+		be.qSender.start()
+	})
+	return err
 }
 
-// Shutdown stops the exporter and is invoked during shutdown.
+// Shutdown stops the nextSender and is invoked during 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 stop the retry goroutines
+		be.rSender.shutdown()
+
+		// All operations will try to export once but will not retry because retrying was disabled when be.rSender stopped.
+		be.qSender.shutdown()
+
+		// Last shutdown the nextSender itself.
+		err = be.shutdown(ctx)
+	})
+	return err
+}
+
+type timeoutSender struct {
+	cfg *TimeoutSettings
+}
+
+func (te *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 te.cfg.Timeout > 0 {
+		var cancelFunc func()
+		ctx, cancelFunc = context.WithTimeout(req.context(), te.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 {
+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...)
+
+	// Record metrics on the consumer.
+	be.qSender.nextSender = &logsExporterWithObservability{
+		exporterName: cfg.Name(),
+		sender:       be.qSender.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
+	sender       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.sender.send(req)
+	obsreport.EndLogsExportOp(req.context(), req.count(), numDroppedLogs, err)
+	return numDroppedLogs, err
 }

exporter/exporterhelper/logshelper_test.go

@@ -24,6 +24,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/internal/data/testdata"
 	"go.opentelemetry.io/collector/obsreport"
@@ -43,6 +44,13 @@ var (
 	}
 )
 
+func TestLogsRequest(t *testing.T) {
+	mr := newLogsRequest(context.Background(), testdata.GenerateLogDataEmpty(), nil)
+
+	partialErr := consumererror.PartialTracesError(errors.New("some error"), testdata.GenerateTraceDataOneSpan())
+	assert.Same(t, mr, mr.onPartialError(partialErr.(consumererror.PartialError)))
+}
+
 func TestLogsExporter_InvalidName(t *testing.T) {
 	me, err := NewLogsExporter(nil, newPushLogsData(0, nil))
 	require.Nil(t, me)
@@ -178,7 +186,7 @@ func generateLogsTraffic(t *testing.T, me component.LogExporter, numRequests int
 	}
 }
 
-func checkWrapSpanForLogsExporter(t *testing.T, me component.LogExporter, wantError error, numMetricPoints int64) {
+func checkWrapSpanForLogsExporter(t *testing.T, me component.LogExporter, wantError error, numLogRecords int64) {
 	ocSpansSaver := new(testOCTraceExporter)
 	trace.RegisterExporter(ocSpansSaver)
 	defer trace.UnregisterExporter(ocSpansSaver)
@@ -201,13 +209,13 @@ func checkWrapSpanForLogsExporter(t *testing.T, me component.LogExporter, wantEr
 		require.Equalf(t, parentSpan.SpanContext.SpanID, sd.ParentSpanID, "Exporter span not a child\nSpanData %v", sd)
 		require.Equalf(t, errToStatus(wantError), sd.Status, "SpanData %v", sd)
 
-		sentMetricPoints := numMetricPoints
-		var failedToSendMetricPoints int64
+		sentLogRecords := numLogRecords
+		var failedToSendLogRecords int64
 		if wantError != nil {
-			sentMetricPoints = 0
-			failedToSendMetricPoints = numMetricPoints
+			sentLogRecords = 0
+			failedToSendLogRecords = numLogRecords
 		}
-		require.Equalf(t, sentMetricPoints, sd.Attributes[obsreport.SentLogRecordsKey], "SpanData %v", sd)
-		require.Equalf(t, failedToSendMetricPoints, sd.Attributes[obsreport.FailedToSendLogRecordsKey], "SpanData %v", sd)
+		require.Equalf(t, sentLogRecords, sd.Attributes[obsreport.SentLogRecordsKey], "SpanData %v", sd)
+		require.Equalf(t, failedToSendLogRecords, sd.Attributes[obsreport.FailedToSendLogRecordsKey], "SpanData %v", sd)
 	}
 }