fix counting of spans/logs in self-observability metrics in otlp trace and log exporters (#8254)

Fixes https://github.com/open-telemetry/opentelemetry-go/issues/8253

CHANGELOG.md

@@ -75,6 +75,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Fix stale status code reporting on self-observability metrics in `go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp` and `go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp`. (#8226)
 - Fix a concurrent `Collect` data race and potential panic in `go.opentelemetry.io/otel/exporters/prometheus` when `WithResourceAsConstantLabels` option is used. (#8227)
 - Fix race condition in `FixedSizeReservoir` in `go.opentelemetry.io/otel/sdk/metric/exemplar` by reverting #7447. (#8249)
+- Fix counting of spans and logs in self-observability metrics in `go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc`, `go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp`, `go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc`, and `go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp`. (#8254)
 
 <!-- Released section -->
 <!-- Don't change this section unless doing release -->

exporters/otlp/otlplog/otlploggrpc/client.go

@@ -151,8 +151,13 @@ func (c *client) UploadLogs(ctx context.Context, rl []*logpb.ResourceLogs) (uplo
 	defer cancel()
 
 	pbRequest := &collogpb.ExportLogsServiceRequest{ResourceLogs: rl}
-	count := int64(len(rl))
 	if c.instrumentation != nil {
+		var count int64
+		for _, resLogs := range rl {
+			for _, scopeLogs := range resLogs.ScopeLogs {
+				count += int64(len(scopeLogs.LogRecords))
+			}
+		}
 		eo := c.instrumentation.ExportLogs(ctx, count)
 		defer func() {
 			eo.End(uploadErr)

exporters/otlp/otlplog/otlploggrpc/client_test.go

@@ -705,7 +705,7 @@ func TestClientObservability(t *testing.T) {
 											serverAddrAttrs[0],
 											serverAddrAttrs[1],
 										),
-										Value: int64(len(resourceLogs)),
+										Value: int64(len(logRecords)),
 									},
 								},
 							},
@@ -831,7 +831,7 @@ func TestClientObservability(t *testing.T) {
 											serverAddrAttrs[0],
 											serverAddrAttrs[1],
 										),
-										Value: 0,
+										Value: 2,
 									},
 									{
 										Attributes: attribute.NewSet(
@@ -843,7 +843,7 @@ func TestClientObservability(t *testing.T) {
 											serverAddrAttrs[1],
 											semconv.ErrorType(wantErr),
 										),
-										Value: 1,
+										Value: 2,
 									},
 								},
 							},
@@ -976,7 +976,7 @@ func TestClientObservability(t *testing.T) {
 											serverAddrAttrs[1],
 											wantErrTypeAttr,
 										),
-										Value: 1,
+										Value: int64(len(logRecords)),
 									},
 								},
 							},
@@ -1144,7 +1144,7 @@ func TestClientObservabilityWithRetry(t *testing.T) {
 								serverAddrAttrs[0],
 								serverAddrAttrs[1],
 							),
-							Value: int64(len(resourceLogs)) - n,
+							Value: int64(len(logRecords)) - n,
 						},
 						{
 							Attributes: attribute.NewSet(

exporters/otlp/otlplog/otlploghttp/client.go

@@ -164,7 +164,13 @@ func (c *httpClient) uploadLogs(ctx context.Context, data []*logpb.ResourceLogs)
 
 	var statusCode int
 	if c.inst != nil {
-		op := c.inst.ExportLogs(ctx, int64(len(data)))
+		var count int64
+		for _, resLogs := range data {
+			for _, scopeLogs := range resLogs.ScopeLogs {
+				count += int64(len(scopeLogs.LogRecords))
+			}
+		}
+		op := c.inst.ExportLogs(ctx, count)
 		defer func() { op.End(uploadErr, statusCode) }()
 	}
 

exporters/otlp/otlplog/otlploghttp/client_test.go

@@ -1038,7 +1038,7 @@ func TestClientInstrumentation(t *testing.T) {
 				Unit:        otelconv.SDKExporterLogInflight{}.Unit(),
 				Data: metricdata.Sum[int64]{
 					DataPoints: []metricdata.DataPoint[int64]{
-						{Attributes: attribute.NewSet(baseAttrs...)},
+						{Attributes: attribute.NewSet(baseAttrs...), Value: 0},
 					},
 					Temporality: metricdata.CumulativeTemporality,
 				},
@@ -1049,11 +1049,11 @@ func TestClientInstrumentation(t *testing.T) {
 				Unit:        otelconv.SDKExporterLogExported{}.Unit(),
 				Data: metricdata.Sum[int64]{
 					DataPoints: []metricdata.DataPoint[int64]{
-						{Attributes: attribute.NewSet(baseAttrs...)},
+						{Attributes: attribute.NewSet(baseAttrs...), Value: 2},
 						{Attributes: attribute.NewSet(append(
 							baseAttrs,
 							otelconv.SDKExporterLogExported{}.AttrErrorType("*errors.joinError"),
-						)...)},
+						)...), Value: 2},
 					},
 					Temporality: 0x1,
 					IsMonotonic: true,
@@ -1078,12 +1078,25 @@ func TestClientInstrumentation(t *testing.T) {
 	}
 
 	require.Len(t, got.ScopeMetrics, 1)
-	opt := []metricdatatest.Option{
+
+	gotMetrics := got.ScopeMetrics[0].Metrics
+	require.Len(t, gotMetrics, 3, "expected 3 metrics")
+
+	// Assert counters without IgnoreValue
+	optCounters := []metricdatatest.Option{
+		metricdatatest.IgnoreTimestamp(),
+		metricdatatest.IgnoreExemplars(),
+	}
+	metricdatatest.AssertEqual(t, want.Metrics[0], gotMetrics[0], optCounters...)
+	metricdatatest.AssertEqual(t, want.Metrics[1], gotMetrics[1], optCounters...)
+
+	// Assert duration with IgnoreValue
+	optDuration := []metricdatatest.Option{
 		metricdatatest.IgnoreTimestamp(),
 		metricdatatest.IgnoreExemplars(),
 		metricdatatest.IgnoreValue(),
 	}
-	metricdatatest.AssertEqual(t, want, got.ScopeMetrics[0], opt...)
+	metricdatatest.AssertEqual(t, want.Metrics[2], gotMetrics[2], optDuration...)
 }
 
 func TestResponseBodySizeLimit(t *testing.T) {

exporters/otlp/otlptrace/otlptracegrpc/client.go

@@ -215,7 +215,13 @@ func (c *client) UploadTraces(ctx context.Context, protoSpans []*tracepb.Resourc
 
 	code := codes.Unknown
 	if c.inst != nil {
-		op := c.inst.ExportSpans(ctx, len(protoSpans))
+		var spanCount int
+		for _, rs := range protoSpans {
+			for _, ss := range rs.ScopeSpans {
+				spanCount += len(ss.Spans)
+			}
+		}
+		op := c.inst.ExportSpans(ctx, spanCount)
 		defer func() { op.End(uploadErr, code) }()
 	}
 

exporters/otlp/otlptrace/otlptracegrpc/client_test.go

@@ -491,7 +491,8 @@ func TestClientInstrumentation(t *testing.T) {
 	t.Cleanup(func() { require.NoError(t, mc.stop()) })
 
 	exp := newGRPCExporter(t, t.Context(), mc.endpoint)
-	err := exp.ExportSpans(t.Context(), roSpans)
+	localSpans := tracetest.SpanStubs{{Name: "Span 0"}, {Name: "Span 1"}}.Snapshots()
+	err := exp.ExportSpans(t.Context(), localSpans)
 	assert.ErrorIs(t, err, internal.TracePartialSuccessError(n, msg))
 	require.NoError(t, exp.Shutdown(t.Context()))
 
@@ -512,7 +513,7 @@ func TestClientInstrumentation(t *testing.T) {
 				Unit:        otelconv.SDKExporterSpanInflight{}.Unit(),
 				Data: metricdata.Sum[int64]{
 					DataPoints: []metricdata.DataPoint[int64]{
-						{Attributes: attribute.NewSet(attrs...)},
+						{Attributes: attribute.NewSet(attrs...), Value: 0},
 					},
 					Temporality: metricdata.CumulativeTemporality,
 				},
@@ -523,11 +524,11 @@ func TestClientInstrumentation(t *testing.T) {
 				Unit:        otelconv.SDKExporterSpanExported{}.Unit(),
 				Data: metricdata.Sum[int64]{
 					DataPoints: []metricdata.DataPoint[int64]{
-						{Attributes: attribute.NewSet(attrs...)},
+						{Attributes: attribute.NewSet(attrs...), Value: 0},
 						{Attributes: attribute.NewSet(append(
 							attrs,
 							otelconv.SDKExporterSpanExported{}.AttrErrorType("*errors.joinError"),
-						)...)},
+						)...), Value: 2},
 					},
 					Temporality: 0x1,
 					IsMonotonic: true,
@@ -553,12 +554,18 @@ func TestClientInstrumentation(t *testing.T) {
 		},
 	}
 	require.Len(t, got.ScopeMetrics, 1)
-	opt := []metricdatatest.Option{
+	gotMetrics := got.ScopeMetrics[0].Metrics
+	require.Len(t, gotMetrics, 3)
+
+	metricdatatest.AssertEqual(t, want.Metrics[0], gotMetrics[0], metricdatatest.IgnoreTimestamp())
+	metricdatatest.AssertEqual(t, want.Metrics[1], gotMetrics[1], metricdatatest.IgnoreTimestamp())
+	metricdatatest.AssertEqual(
+		t,
+		want.Metrics[2],
+		gotMetrics[2],
 		metricdatatest.IgnoreTimestamp(),
-		metricdatatest.IgnoreExemplars(),
 		metricdatatest.IgnoreValue(),
-	}
+	)
-	metricdatatest.AssertEqual(t, want, got.ScopeMetrics[0], opt...)
 }
 
 func canonical(t *testing.T, endpoint string) string {

exporters/otlp/otlptrace/otlptracehttp/client.go

@@ -179,7 +179,13 @@ func (d *client) UploadTraces(ctx context.Context, protoSpans []*tracepb.Resourc
 
 	var statusCode int
 	if d.inst != nil {
-		op := d.inst.ExportSpans(ctx, len(protoSpans))
+		var spanCount int
+		for _, rs := range protoSpans {
+			for _, ss := range rs.ScopeSpans {
+				spanCount += len(ss.Spans)
+			}
+		}
+		op := d.inst.ExportSpans(ctx, spanCount)
 		defer func() { op.End(uploadErr, statusCode) }()
 	}
 

exporters/otlp/otlptrace/otlptracehttp/client_test.go

@@ -531,7 +531,8 @@ func TestClientInstrumentation(t *testing.T) {
 	exporter, err := otlptrace.New(t.Context(), driver)
 	require.NoError(t, err)
 
-	err = exporter.ExportSpans(t.Context(), otlptracetest.SingleReadOnlySpan())
+	localSpans := tracetest.SpanStubs{{Name: "Span 0"}, {Name: "Span 1"}}.Snapshots()
+	err = exporter.ExportSpans(t.Context(), localSpans)
 	assert.Error(t, err)
 
 	require.NoError(t, exporter.Shutdown(t.Context()))
@@ -553,7 +554,7 @@ func TestClientInstrumentation(t *testing.T) {
 				Unit:        otelconv.SDKExporterSpanInflight{}.Unit(),
 				Data: metricdata.Sum[int64]{
 					DataPoints: []metricdata.DataPoint[int64]{
-						{Attributes: attribute.NewSet(attrs...)},
+						{Attributes: attribute.NewSet(attrs...), Value: 0},
 					},
 					Temporality: metricdata.CumulativeTemporality,
 				},
@@ -564,11 +565,11 @@ func TestClientInstrumentation(t *testing.T) {
 				Unit:        otelconv.SDKExporterSpanExported{}.Unit(),
 				Data: metricdata.Sum[int64]{
 					DataPoints: []metricdata.DataPoint[int64]{
-						{Attributes: attribute.NewSet(attrs...)},
+						{Attributes: attribute.NewSet(attrs...), Value: 0},
 						{Attributes: attribute.NewSet(append(
 							attrs,
 							otelconv.SDKExporterSpanExported{}.AttrErrorType("*errors.joinError"),
-						)...)},
+						)...), Value: 2},
 					},
 					Temporality: 0x1,
 					IsMonotonic: true,
@@ -592,12 +593,18 @@ func TestClientInstrumentation(t *testing.T) {
 		},
 	}
 	require.Len(t, got.ScopeMetrics, 1)
-	opt := []metricdatatest.Option{
+	gotMetrics := got.ScopeMetrics[0].Metrics
+	require.Len(t, gotMetrics, 3)
+
+	metricdatatest.AssertEqual(t, want.Metrics[0], gotMetrics[0], metricdatatest.IgnoreTimestamp())
+	metricdatatest.AssertEqual(t, want.Metrics[1], gotMetrics[1], metricdatatest.IgnoreTimestamp())
+	metricdatatest.AssertEqual(
+		t,
+		want.Metrics[2],
+		gotMetrics[2],
 		metricdatatest.IgnoreTimestamp(),
-		metricdatatest.IgnoreExemplars(),
 		metricdatatest.IgnoreValue(),
-	}
+	)
-	metricdatatest.AssertEqual(t, want, got.ScopeMetrics[0], opt...)
 }
 
 func TestResponseBodySizeLimit(t *testing.T) {