Report metric about current size of the exporter retry queue

This commit adds observability to queue_retry exporter helper. It adds the first metric "queue_length" that indicates current size of the queue per exporter. The metrics is updated every second.

This is the first commit to address the issue open-telemetry#2434

exporter/exporterhelper/common.go

@@ -23,6 +23,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenthelper"
 	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/obsreport"
 )
 
 // TimeoutSettings for timeout. The timeout applies to individual attempts to send data to the backend.
@@ -158,15 +159,15 @@ type baseExporter struct {
 	convertResourceToTelemetry bool
 }
 
-func newBaseExporter(cfg config.Exporter, logger *zap.Logger, options ...Option) *baseExporter {
+func newBaseExporter(cfg config.Exporter, logger *zap.Logger, obsrep *obsreport.Exporter, options ...Option) *baseExporter {
 	bs := fromOptions(options)
 	be := &baseExporter{
 		Component:                  componenthelper.New(bs.componentOptions...),
 		cfg:                        cfg,
 		convertResourceToTelemetry: bs.ResourceToTelemetrySettings.Enabled,
 	}
 
-	be.qrSender = newQueuedRetrySender(cfg.Name(), bs.QueueSettings, bs.RetrySettings, &timeoutSender{cfg: bs.TimeoutSettings}, logger)
+	be.qrSender = newQueuedRetrySender(cfg.Name(), bs.QueueSettings, bs.RetrySettings, &timeoutSender{cfg: bs.TimeoutSettings}, obsrep, logger)
 	be.sender = be.qrSender
 
 	return be

exporter/exporterhelper/common_test.go

@@ -25,6 +25,8 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/config/configtelemetry"
+	"go.opentelemetry.io/collector/obsreport"
 )
 
 var okStatus = trace.Status{Code: trace.StatusCodeOK}
@@ -40,7 +42,7 @@ func TestErrorToStatus(t *testing.T) {
 }
 
 func TestBaseExporter(t *testing.T) {
-	be := newBaseExporter(defaultExporterCfg, zap.NewNop())
+	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), newTestObsExporter())
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
 	require.NoError(t, be.Shutdown(context.Background()))
 }
@@ -50,6 +52,7 @@ func TestBaseExporterWithOptions(t *testing.T) {
 	be := newBaseExporter(
 		defaultExporterCfg,
 		zap.NewNop(),
+		newTestObsExporter(),
 		WithStart(func(ctx context.Context, host component.Host) error { return want }),
 		WithShutdown(func(ctx context.Context) error { return want }),
 		WithResourceToTelemetryConversion(defaultResourceToTelemetrySettings()),
@@ -65,3 +68,10 @@ func errToStatus(err error) trace.Status {
 	}
 	return okStatus
 }
+
+func newTestObsExporter() *obsreport.Exporter {
+	return obsreport.NewExporter(obsreport.ExporterSettings{
+		Level:        configtelemetry.LevelNormal,
+		ExporterName: defaultExporterCfg.Name(),
+	})
+}

exporter/exporterhelper/logshelper.go

@@ -89,13 +89,14 @@ func NewLogsExporter(
 		return nil, errNilPushLogsData
 	}
 
-	be := newBaseExporter(cfg, logger, options...)
+	obsrep := obsreport.NewExporter(obsreport.ExporterSettings{
+		Level:        configtelemetry.GetMetricsLevelFlagValue(),
+		ExporterName: cfg.Name(),
+	})
+	be := newBaseExporter(cfg, logger, obsrep, options...)
 	be.wrapConsumerSender(func(nextSender requestSender) requestSender {
 		return &logsExporterWithObservability{
-			obsrep: obsreport.NewExporter(obsreport.ExporterSettings{
-				Level:        configtelemetry.GetMetricsLevelFlagValue(),
-				ExporterName: cfg.Name(),
-			}),
+			obsrep:     obsrep,
 			nextSender: nextSender,
 		}
 	})

exporter/exporterhelper/metricshelper.go

@@ -93,13 +93,14 @@ func NewMetricsExporter(
 		return nil, errNilPushMetricsData
 	}
 
-	be := newBaseExporter(cfg, logger, options...)
+	obsrep := obsreport.NewExporter(obsreport.ExporterSettings{
+		Level:        configtelemetry.GetMetricsLevelFlagValue(),
+		ExporterName: cfg.Name(),
+	})
+	be := newBaseExporter(cfg, logger, obsrep, options...)
 	be.wrapConsumerSender(func(nextSender requestSender) requestSender {
 		return &metricsSenderWithObservability{
-			obsrep: obsreport.NewExporter(obsreport.ExporterSettings{
-				Level:        configtelemetry.GetMetricsLevelFlagValue(),
-				ExporterName: cfg.Name(),
-			}),
+			obsrep:     obsrep,
 			nextSender: nextSender,
 		}
 	})

exporter/exporterhelper/queued_retry.go

@@ -30,6 +30,11 @@ import (
 	"go.opentelemetry.io/collector/obsreport"
 )
 
+var (
+	// Time interval to report queue length metric
+	queueLengthReportInterval = 1 * time.Second
+)
+
 // QueueSettings defines configuration for queueing batches before sending to the consumerSender.
 type QueueSettings struct {
 	// Enabled indicates whether to not enqueue batches before sending to the consumerSender.
@@ -85,6 +90,7 @@ type queuedRetrySender struct {
 	retryStopCh     chan struct{}
 	traceAttributes []trace.Attribute
 	logger          *zap.Logger
+	obsrep          *obsreport.Exporter
 }
 
 func createSampledLogger(logger *zap.Logger) *zap.Logger {
@@ -106,7 +112,14 @@ func createSampledLogger(logger *zap.Logger) *zap.Logger {
 	return logger.WithOptions(opts)
 }
 
-func newQueuedRetrySender(fullName string, qCfg QueueSettings, rCfg RetrySettings, nextSender requestSender, logger *zap.Logger) *queuedRetrySender {
+func newQueuedRetrySender(
+	fullName string,
+	qCfg QueueSettings,
+	rCfg RetrySettings,
+	nextSender requestSender,
+	obsrep *obsreport.Exporter,
+	logger *zap.Logger,
+) *queuedRetrySender {
 	retryStopCh := make(chan struct{})
 	sampledLogger := createSampledLogger(logger)
 	traceAttr := trace.StringAttribute(obsreport.ExporterKey, fullName)
@@ -122,6 +135,7 @@ func newQueuedRetrySender(fullName string, qCfg QueueSettings, rCfg RetrySetting
 		queue:           queue.NewBoundedQueue(qCfg.QueueSize, func(item interface{}) {}),
 		retryStopCh:     retryStopCh,
 		traceAttributes: []trace.Attribute{traceAttr},
+		obsrep:          obsrep,
 		logger:          sampledLogger,
 	}
 }
@@ -132,6 +146,21 @@ func (qrs *queuedRetrySender) start() {
 		req := item.(request)
 		_ = qrs.consumerSender.send(req)
 	})
+
+	// Start a timer to report the queue length every second.
+	ticker := time.NewTicker(queueLengthReportInterval)
+	go func() {
+		defer ticker.Stop()
+		for {
+			select {
+			case <-qrs.retryStopCh:
+				return
+			case <-ticker.C:
+				qrs.obsrep.ReportQueueLength(context.Background(), qrs.queue.Size())
+			}
+		}
+	}()
+
 }
 
 // send implements the requestSender interface

exporter/exporterhelper/queued_retry_test.go

@@ -35,7 +35,7 @@ import (
 func TestQueuedRetry_DropOnPermanentError(t *testing.T) {
 	qCfg := DefaultQueueSettings()
 	rCfg := DefaultRetrySettings()
-	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), WithRetry(rCfg), WithQueue(qCfg))
+	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), newTestObsExporter(), WithRetry(rCfg), WithQueue(qCfg))
 	ocs := newObservabilityConsumerSender(be.qrSender.consumerSender)
 	be.qrSender.consumerSender = ocs
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
@@ -59,7 +59,7 @@ func TestQueuedRetry_DropOnNoRetry(t *testing.T) {
 	qCfg := DefaultQueueSettings()
 	rCfg := DefaultRetrySettings()
 	rCfg.Enabled = false
-	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), WithRetry(rCfg), WithQueue(qCfg))
+	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), newTestObsExporter(), WithRetry(rCfg), WithQueue(qCfg))
 	ocs := newObservabilityConsumerSender(be.qrSender.consumerSender)
 	be.qrSender.consumerSender = ocs
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
@@ -84,7 +84,7 @@ func TestQueuedRetry_OnError(t *testing.T) {
 	qCfg.NumConsumers = 1
 	rCfg := DefaultRetrySettings()
 	rCfg.InitialInterval = 0
-	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), WithRetry(rCfg), WithQueue(qCfg))
+	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), newTestObsExporter(), WithRetry(rCfg), WithQueue(qCfg))
 	ocs := newObservabilityConsumerSender(be.qrSender.consumerSender)
 	be.qrSender.consumerSender = ocs
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
@@ -110,7 +110,7 @@ func TestQueuedRetry_StopWhileWaiting(t *testing.T) {
 	qCfg := DefaultQueueSettings()
 	qCfg.NumConsumers = 1
 	rCfg := DefaultRetrySettings()
-	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), WithRetry(rCfg), WithQueue(qCfg))
+	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), newTestObsExporter(), WithRetry(rCfg), WithQueue(qCfg))
 	ocs := newObservabilityConsumerSender(be.qrSender.consumerSender)
 	be.qrSender.consumerSender = ocs
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
@@ -143,7 +143,7 @@ func TestQueuedRetry_DoNotPreserveCancellation(t *testing.T) {
 	qCfg := DefaultQueueSettings()
 	qCfg.NumConsumers = 1
 	rCfg := DefaultRetrySettings()
-	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), WithRetry(rCfg), WithQueue(qCfg))
+	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), newTestObsExporter(), WithRetry(rCfg), WithQueue(qCfg))
 	ocs := newObservabilityConsumerSender(be.qrSender.consumerSender)
 	be.qrSender.consumerSender = ocs
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
@@ -172,7 +172,7 @@ func TestQueuedRetry_MaxElapsedTime(t *testing.T) {
 	rCfg := DefaultRetrySettings()
 	rCfg.InitialInterval = time.Millisecond
 	rCfg.MaxElapsedTime = 100 * time.Millisecond
-	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), WithRetry(rCfg), WithQueue(qCfg))
+	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), newTestObsExporter(), WithRetry(rCfg), WithQueue(qCfg))
 	ocs := newObservabilityConsumerSender(be.qrSender.consumerSender)
 	be.qrSender.consumerSender = ocs
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
@@ -210,7 +210,7 @@ func TestQueuedRetry_ThrottleError(t *testing.T) {
 	qCfg.NumConsumers = 1
 	rCfg := DefaultRetrySettings()
 	rCfg.InitialInterval = 10 * time.Millisecond
-	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), WithRetry(rCfg), WithQueue(qCfg))
+	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), newTestObsExporter(), WithRetry(rCfg), WithQueue(qCfg))
 	ocs := newObservabilityConsumerSender(be.qrSender.consumerSender)
 	be.qrSender.consumerSender = ocs
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
@@ -241,7 +241,7 @@ func TestQueuedRetry_RetryOnError(t *testing.T) {
 	qCfg.QueueSize = 1
 	rCfg := DefaultRetrySettings()
 	rCfg.InitialInterval = 0
-	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), WithRetry(rCfg), WithQueue(qCfg))
+	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), newTestObsExporter(), WithRetry(rCfg), WithQueue(qCfg))
 	ocs := newObservabilityConsumerSender(be.qrSender.consumerSender)
 	be.qrSender.consumerSender = ocs
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
@@ -267,7 +267,7 @@ func TestQueuedRetry_DropOnFull(t *testing.T) {
 	qCfg := DefaultQueueSettings()
 	qCfg.QueueSize = 0
 	rCfg := DefaultRetrySettings()
-	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), WithRetry(rCfg), WithQueue(qCfg))
+	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), newTestObsExporter(), WithRetry(rCfg), WithQueue(qCfg))
 	ocs := newObservabilityConsumerSender(be.qrSender.consumerSender)
 	be.qrSender.consumerSender = ocs
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
@@ -285,7 +285,7 @@ func TestQueuedRetryHappyPath(t *testing.T) {
 
 	qCfg := DefaultQueueSettings()
 	rCfg := DefaultRetrySettings()
-	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), WithRetry(rCfg), WithQueue(qCfg))
+	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), newTestObsExporter(), WithRetry(rCfg), WithQueue(qCfg))
 	ocs := newObservabilityConsumerSender(be.qrSender.consumerSender)
 	be.qrSender.consumerSender = ocs
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
@@ -315,6 +315,32 @@ func TestQueuedRetryHappyPath(t *testing.T) {
 	ocs.checkDroppedItemsCount(t, 0)
 }
 
+func TestQueuedRetry_QueueMetricsReported(t *testing.T) {
+	doneFn, err := obsreporttest.SetupRecordedMetricsTest()
+	require.NoError(t, err)
+	defer doneFn()
+
+	queueLengthReportInterval = 10 * time.Millisecond
+
+	qCfg := DefaultQueueSettings()
+	qCfg.NumConsumers = 1
+	rCfg := DefaultRetrySettings()
+	be := newBaseExporter(defaultExporterCfg, zap.NewNop(), newTestObsExporter(), WithRetry(rCfg), WithQueue(qCfg))
+	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
+	t.Cleanup(func() {
+		assert.NoError(t, be.Shutdown(context.Background()))
+	})
+
+	for i := 0; i < 10; i++ {
+		be.sender.send(newErrorRequest(context.Background()))
+	}
+
+	<-time.After(20 * time.Millisecond)
+
+	// 1 request is in-flight, 9 requests are in the queue
+	obsreporttest.CheckExporterQueueRetryViews(t, defaultExporterCfg.Name(), int64(9))
+}
+
 func TestNoCancellationContext(t *testing.T) {
 	deadline := time.Now().Add(1 * time.Second)
 	ctx, cancelFunc := context.WithDeadline(context.Background(), deadline)

exporter/exporterhelper/tracehelper.go

@@ -90,14 +90,15 @@ func NewTraceExporter(
 		return nil, errNilPushTraceData
 	}
 
-	be := newBaseExporter(cfg, logger, options...)
+	obsrep := obsreport.NewExporter(
+		obsreport.ExporterSettings{
+			Level:        configtelemetry.GetMetricsLevelFlagValue(),
+			ExporterName: cfg.Name(),
+		})
+	be := newBaseExporter(cfg, logger, obsrep, options...)
 	be.wrapConsumerSender(func(nextSender requestSender) requestSender {
 		return &tracesExporterWithObservability{
-			obsrep: obsreport.NewExporter(
-				obsreport.ExporterSettings{
-					Level:        configtelemetry.GetMetricsLevelFlagValue(),
-					ExporterName: cfg.Name(),
-				}),
+			obsrep:     obsrep,
 			nextSender: nextSender,
 		}
 	})

obsreport/obsreport.go

@@ -119,6 +119,13 @@ func AllViews() (views []*view.View) {
 	tagKeys = []tag.Key{tagKeyExporter}
 	views = append(views, genViews(measures, tagKeys, view.Sum())...)
 
+	// Queued retry views.
+	measures = []*stats.Int64Measure{
+		mQueueLength,
+	}
+	tagKeys = []tag.Key{tagKeyExporter}
+	views = append(views, genViews(measures, tagKeys, view.LastValue())...)
+
 	// Processor views.
 	measures = []*stats.Int64Measure{
 		mProcessorAcceptedSpans,

obsreport/obsreport_exporter.go

@@ -42,6 +42,9 @@ const (
 	SentLogRecordsKey = "sent_log_records"
 	// Key used to track logs that failed to be sent by exporters.
 	FailedToSendLogRecordsKey = "send_failed_log_records"
+
+	// Queue retry metrics
+	QueueLengthKey = "queue_length"
 )
 
 var (
@@ -81,6 +84,12 @@ var (
 		exporterPrefix+FailedToSendLogRecordsKey,
 		"Number of log records in failed attempts to send to destination.",
 		stats.UnitDimensionless)
+
+	// Queued retry related metrics
+	mQueueLength = stats.Int64(
+		exporterPrefix+QueueLengthKey,
+		"Current length of the queue (in batches)",
+		stats.UnitDimensionless)
 )
 
 type Exporter struct {
@@ -146,6 +155,15 @@ func (eor *Exporter) EndLogsExportOp(ctx context.Context, numLogRecords int, err
 	endSpan(ctx, err, numSent, numFailedToSend, SentLogRecordsKey, FailedToSendLogRecordsKey)
 }
 
+// ReportQueueLength reports the current queue length metric
+func (eor *Exporter) ReportQueueLength(ctx context.Context, queueLength int) {
+	if gLevel == configtelemetry.LevelNone {
+		return
+	}
+	// Ignore the error for now. This should not happen.
+	_ = stats.RecordWithTags(ctx, eor.mutators, mQueueLength.M(int64(queueLength)))
+}
+
 // startSpan creates the span used to trace the operation. Returning
 // the updated context and the created span.
 func (eor *Exporter) startSpan(ctx context.Context, operationSuffix string) context.Context {