OBS-6820: add otelcol_exporter_retry_dropped_*, based from open-telemetry#13957

Signed-off-by: Jayson Cena <[email protected]>

Author: jaysoncena

exporter/exporterhelper/internal/experr/err.go

@@ -27,3 +27,24 @@ func IsShutdownErr(err error) bool {
 	var sdErr shutdownErr
 	return errors.As(err, &sdErr)
 }
+
+type retriesExhaustedErr struct {
+	err error
+}
+
+func NewRetriesExhaustedErr(err error) error {
+	return retriesExhaustedErr{err: err}
+}
+
+func (r retriesExhaustedErr) Error() string {
+	return "retries exhausted: " + r.err.Error()
+}
+
+func (r retriesExhaustedErr) Unwrap() error {
+	return r.err
+}
+
+func IsRetriesExhaustedErr(err error) bool {
+	var reErr retriesExhaustedErr
+	return errors.As(err, &reErr)
+}

exporter/exporterhelper/internal/experr/err_test.go

@@ -22,3 +22,15 @@ func TestIsShutdownErr(t *testing.T) {
 	err = NewShutdownErr(err)
 	require.True(t, IsShutdownErr(err))
 }
+
+func TestNewRetriesExhaustedErr(t *testing.T) {
+	err := NewRetriesExhaustedErr(errors.New("another error"))
+	assert.Equal(t, "retries exhausted: another error", err.Error())
+}
+
+func TestIsRetriesExhaustedErr(t *testing.T) {
+	err := errors.New("testError")
+	require.False(t, IsRetriesExhaustedErr(err))
+	err = NewRetriesExhaustedErr(err)
+	require.True(t, IsRetriesExhaustedErr(err))
+}

exporter/exporterhelper/internal/metadata/generated_telemetry.go

@@ -13,6 +13,7 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/experr"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadata"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
@@ -40,13 +41,14 @@ type obsReportSender[K request.Request] struct {
 	component.StartFunc
 	component.ShutdownFunc
 
-	spanName        string
-	tracer          trace.Tracer
-	spanAttrs       trace.SpanStartEventOption
-	metricAttr      metric.MeasurementOption
-	itemsSentInst   metric.Int64Counter
-	itemsFailedInst metric.Int64Counter
-	next            sender.Sender[K]
+	spanName              string
+	tracer                trace.Tracer
+	spanAttrs             trace.SpanStartEventOption
+	metricAttr            metric.MeasurementOption
+	itemsSentInst         metric.Int64Counter
+	itemsFailedInst       metric.Int64Counter
+	itemsRetryDroppedInst metric.Int64Counter
+	next                  sender.Sender[K]
 }
 
 func newObsReportSender[K request.Request](set exporter.Settings, signal pipeline.Signal, next sender.Sender[K]) (sender.Sender[K], error) {
@@ -70,14 +72,17 @@ func newObsReportSender[K request.Request](set exporter.Settings, signal pipelin
 	case pipeline.SignalTraces:
 		or.itemsSentInst = telemetryBuilder.ExporterSentSpans
 		or.itemsFailedInst = telemetryBuilder.ExporterSendFailedSpans
+		or.itemsRetryDroppedInst = telemetryBuilder.ExporterRetryDroppedSpans
 
 	case pipeline.SignalMetrics:
 		or.itemsSentInst = telemetryBuilder.ExporterSentMetricPoints
 		or.itemsFailedInst = telemetryBuilder.ExporterSendFailedMetricPoints
+		or.itemsRetryDroppedInst = telemetryBuilder.ExporterRetryDroppedMetricPoints
 
 	case pipeline.SignalLogs:
 		or.itemsSentInst = telemetryBuilder.ExporterSentLogRecords
 		or.itemsFailedInst = telemetryBuilder.ExporterSendFailedLogRecords
+		or.itemsRetryDroppedInst = telemetryBuilder.ExporterRetryDroppedLogRecords
 	}
 
 	return or, nil
@@ -116,6 +121,10 @@ func (ors *obsReportSender[K]) endOp(ctx context.Context, numLogRecords int, err
 	if ors.itemsFailedInst != nil {
 		ors.itemsFailedInst.Add(ctx, numFailedToSend, ors.metricAttr)
 	}
+	// Count drops after retries were exhausted.
+	if err != nil && ors.itemsRetryDroppedInst != nil && experr.IsRetriesExhaustedErr(err) {
+		ors.itemsRetryDroppedInst.Add(ctx, numFailedToSend, ors.metricAttr)
+	}
 
 	span := trace.SpanFromContext(ctx)
 	defer span.End()

exporter/exporterhelper/internal/metadatatest/generated_telemetrytest.go

@@ -18,6 +18,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/experr"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadatatest"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
@@ -31,6 +32,43 @@ var (
 	errFake = errors.New("errFake")
 )
 
+func TestExportTraceRetryDroppedMetric(t *testing.T) {
+	tt := componenttest.NewTelemetry()
+	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
+
+	obsrep, err := newObsReportSender(
+		exporter.Settings{ID: exporterID, TelemetrySettings: tt.NewTelemetrySettings(), BuildInfo: component.NewDefaultBuildInfo()},
+		pipeline.SignalTraces,
+		sender.NewSender(func(context.Context, request.Request) error {
+			return experr.NewRetriesExhaustedErr(errFake)
+		}),
+	)
+	require.NoError(t, err)
+
+	req := &requesttest.FakeRequest{Items: 7}
+	sendErr := obsrep.Send(context.Background(), req)
+	require.Error(t, sendErr)
+	require.True(t, experr.IsRetriesExhaustedErr(sendErr))
+
+	wantAttrs := attribute.NewSet(attribute.String("exporter", exporterID.String()))
+
+	metadatatest.AssertEqualExporterSendFailedSpans(t, tt,
+		[]metricdata.DataPoint[int64]{
+			{
+				Attributes: wantAttrs,
+				Value:      int64(req.Items),
+			},
+		}, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+
+	metadatatest.AssertEqualExporterRetryDroppedSpans(t, tt,
+		[]metricdata.DataPoint[int64]{
+			{
+				Attributes: wantAttrs,
+				Value:      int64(req.Items),
+			},
+		}, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+}
+
 func TestExportTraceDataOp(t *testing.T) {
 	tt := componenttest.NewTelemetry()
 	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })

exporter/exporterhelper/internal/metadatatest/generated_telemetrytest_test.go

@@ -79,10 +79,17 @@ func (rs *retrySender) Send(ctx context.Context, req request.Request) error {
 	}
 	span := trace.SpanFromContext(ctx)
 	retryNum := int64(0)
+	retried := false
 	var maxElapsedTime time.Time
 	if rs.cfg.MaxElapsedTime > 0 {
 		maxElapsedTime = time.Now().Add(rs.cfg.MaxElapsedTime)
 	}
+	wrapRetryErr := func(e error) error {
+		if retried {
+			return experr.NewRetriesExhaustedErr(e)
+		}
+		return e
+	}
 	for {
 		span.AddEvent(
 			"Sending request.",
@@ -104,7 +111,7 @@ func (rs *retrySender) Send(ctx context.Context, req request.Request) error {
 
 		backoffDelay := expBackoff.NextBackOff()
 		if backoffDelay == backoff.Stop {
-			return fmt.Errorf("no more retries left: %w", err)
+			return wrapRetryErr(fmt.Errorf("no more retries left: %w", err))
 		}
 
 		throttleErr := throttleRetry{}
@@ -115,13 +122,13 @@ func (rs *retrySender) Send(ctx context.Context, req request.Request) error {
 		nextRetryTime := time.Now().Add(backoffDelay)
 		if !maxElapsedTime.IsZero() && maxElapsedTime.Before(nextRetryTime) {
 			// The delay is longer than the maxElapsedTime.
-			return fmt.Errorf("no more retries left: %w", err)
+			return wrapRetryErr(fmt.Errorf("no more retries left: %w", err))
 		}
 
 		if deadline, has := ctx.Deadline(); has && deadline.Before(nextRetryTime) {
 			// The delay is longer than the deadline.  There is no point in
 			// waiting for cancelation.
-			return fmt.Errorf("request will be cancelled before next retry: %w", err)
+			return wrapRetryErr(fmt.Errorf("request will be cancelled before next retry: %w", err))
 		}
 
 		backoffDelayStr := backoffDelay.String()
@@ -140,7 +147,7 @@ func (rs *retrySender) Send(ctx context.Context, req request.Request) error {
 		// back-off, but get interrupted when shutting down or request is cancelled or timed out.
 		select {
 		case <-ctx.Done():
-			return fmt.Errorf("request is cancelled or timed out: %w", err)
+			return wrapRetryErr(fmt.Errorf("request is cancelled or timed out: %w", err))
 		case <-rs.stopCh:
 			return experr.NewShutdownErr(err)
 		case <-time.After(backoffDelay):

exporter/exporterhelper/internal/obs_report_sender.go

@@ -18,6 +18,7 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/consumer/consumererror"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/experr"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
@@ -76,6 +77,26 @@ func TestRetrySenderMaxElapsedTime(t *testing.T) {
 	require.NoError(t, rs.Shutdown(context.Background()))
 }
 
+func TestRetrySenderRetriesExhaustedErrorWrapped(t *testing.T) {
+	rCfg := configretry.NewDefaultBackOffConfig()
+	rCfg.InitialInterval = time.Millisecond
+	rCfg.RandomizationFactor = 0
+	rCfg.Multiplier = 1
+	rCfg.MaxInterval = time.Millisecond
+	rCfg.MaxElapsedTime = 4 * time.Millisecond
+	var attempts int
+	rs := newRetrySender(rCfg, exportertest.NewNopSettings(exportertest.NopType), sender.NewSender(func(context.Context, request.Request) error {
+		attempts++
+		return errors.New("transient error")
+	}))
+	require.NoError(t, rs.Start(context.Background(), componenttest.NewNopHost()))
+	err := rs.Send(context.Background(), &requesttest.FakeRequest{Items: 2})
+	require.Error(t, err)
+	require.True(t, experr.IsRetriesExhaustedErr(err))
+	require.GreaterOrEqual(t, attempts, 2)
+	require.NoError(t, rs.Shutdown(context.Background()))
+}
+
 func TestRetrySenderThrottleError(t *testing.T) {
 	rCfg := configretry.NewDefaultBackOffConfig()
 	rCfg.InitialInterval = 10 * time.Millisecond

exporter/exporterhelper/internal/obs_report_sender_test.go

@@ -121,3 +121,33 @@ telemetry:
       gauge:
         value_type: int
         async: true
+
+    exporter_retry_dropped_log_records:
+      enabled: true
+      stability:
+        level: alpha
+      description: Number of log records dropped after exhausting configured retries.
+      unit: "{records}"
+      sum:
+        value_type: int
+        monotonic: true
+
+    exporter_retry_dropped_metric_points:
+      enabled: true
+      stability:
+        level: alpha
+      description: Number of metric points dropped after exhausting configured retries.
+      unit: "{datapoints}"
+      sum:
+        value_type: int
+        monotonic: true
+
+    exporter_retry_dropped_spans:
+      enabled: true
+      stability:
+        level: alpha
+      description: Number of spans dropped after exhausting configured retries.
+      unit: "{spans}"
+      sum:
+        value_type: int
+        monotonic: true