Merge branch 'main' into fix/processconv-observable-metrics

Author: abhay1999

CHANGELOG.md

@@ -48,6 +48,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Fix gzipped request body replay on redirect in `go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp`. (#8152)
 - `go.opentelemetry.io/otel/exporters/prometheus` now uses `Value.String` formatting for label values following the [OpenTelemetry AnyValue representation for non-OTLP protocols](https://opentelemetry.io/docs/specs/otel/common/#anyvalue). (#8170)
 - Propagate errors from the exporter when calling `Shutdown` on `BatchSpanProcessor` in `go.opentelemetry.io/otel/sdk/trace`. (#8197)
+- 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)
 
 <!-- Released section -->

exporters/otlp/otlplog/otlploghttp/client.go

@@ -177,6 +177,7 @@ func (c *httpClient) uploadLogs(ctx context.Context, data []*logpb.ResourceLogs)
 		default:
 		}
 
+		statusCode = 0
 		request.reset(iCtx)
 		// nolint:gosec // URL is constructed from validated OTLP endpoint configuration
 		resp, err := c.client.Do(request.Request)

exporters/otlp/otlplog/otlploghttp/client_test.go

@@ -1183,3 +1183,76 @@ func BenchmarkExporterExportLogs(b *testing.B) {
 
 	b.Run("NoObservability", run)
 }
+
+func TestClientInstrumentationStaleStatusCode(t *testing.T) {
+	t.Setenv("OTEL_GO_X_OBSERVABILITY", "true")
+	const id = 0
+	SetExporterID(id)
+
+	orig := otel.GetMeterProvider()
+	t.Cleanup(func() { otel.SetMeterProvider(orig) })
+
+	reader := metric.NewManualReader()
+	mp := metric.NewMeterProvider(metric.WithReader(reader))
+	otel.SetMeterProvider(mp)
+
+	// Use a client that returns a 503 error once and then a network error.
+	var calls int
+	client := &http.Client{
+		Transport: roundTripperFunc(func(_ *http.Request) (*http.Response, error) {
+			calls++
+			if calls == 1 {
+				return &http.Response{
+					StatusCode: http.StatusServiceUnavailable,
+					Status:     "503 " + http.StatusText(http.StatusServiceUnavailable),
+					Header:     make(http.Header),
+					Body:       io.NopCloser(strings.NewReader("")),
+				}, nil
+			}
+			return nil, errors.New("network error")
+		}),
+	}
+
+	opts := []Option{
+		WithHTTPClient(client),
+		WithInsecure(),
+		WithRetry(RetryConfig{
+			Enabled:         true,
+			InitialInterval: time.Nanosecond,
+			MaxInterval:     time.Nanosecond,
+			MaxElapsedTime:  time.Second,
+		}),
+	}
+	cfg := newConfig(opts)
+	c, err := newHTTPClient(t.Context(), cfg)
+	require.NoError(t, err)
+
+	err = c.UploadLogs(t.Context(), resourceLogs)
+	assert.Error(t, err)
+
+	// Validate that the status code is 0 and not the stale 503 on self-observability metrics.
+	var got metricdata.ResourceMetrics
+	require.NoError(t, reader.Collect(t.Context(), &got))
+
+	require.Len(t, got.ScopeMetrics, 1)
+	metrics := got.ScopeMetrics[0].Metrics
+	var found bool
+	for _, m := range metrics {
+		if m.Name != (otelconv.SDKExporterOperationDuration{}).Name() {
+			continue
+		}
+		found = true
+		data := m.Data.(metricdata.Histogram[float64])
+		require.NotEmpty(t, data.DataPoints)
+		dp := data.DataPoints[0]
+		_, ok := dp.Attributes.Value(otelconv.SDKExporterOperationDuration{}.AttrHTTPResponseStatusCode(0).Key)
+		assert.False(t, ok, "should not report status code when the request fails before getting a response.")
+	}
+	assert.True(t, found, "expected to find operation duration metric")
+}
+
+type roundTripperFunc func(*http.Request) (*http.Response, error)
+
+func (f roundTripperFunc) RoundTrip(r *http.Request) (*http.Response, error) {
+	return f(r)
+}

exporters/otlp/otlplog/otlploghttp/internal/observ/instrumentation.go

@@ -268,11 +268,10 @@ func (e ExportOp) recordOption(err error, code int) metric.RecordOption {
 	defer put(attrsPool, attrs)
 
 	*attrs = append(*attrs, e.inst.presetAttrs...)
-	*attrs = append(
-		*attrs,
-		semconv.HTTPResponseStatusCode(code),
-		semconv.ErrorType(err),
-	)
+	if code != 0 {
+		*attrs = append(*attrs, semconv.HTTPResponseStatusCode(code))
+	}
+	*attrs = append(*attrs, semconv.ErrorType(err))
 	return metric.WithAttributeSet(attribute.NewSet(*attrs...))
 }
 

exporters/otlp/otlptrace/otlptracehttp/client.go

@@ -186,6 +186,7 @@ func (d *client) UploadTraces(ctx context.Context, protoSpans []*tracepb.Resourc
 		default:
 		}
 
+		statusCode = 0
 		request.reset(ctx)
 		// nolint:gosec // URL is constructed from validated OTLP endpoint configuration
 		resp, err := d.client.Do(request.Request)

exporters/otlp/otlptrace/otlptracehttp/client_test.go

@@ -8,6 +8,7 @@ import (
 	"compress/gzip"
 	"context"
 	"crypto/tls"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -818,3 +819,79 @@ func BenchmarkExporterExportSpans(b *testing.B) {
 		run(b)
 	})
 }
+
+func TestClientInstrumentationStaleStatusCode(t *testing.T) {
+	t.Setenv("OTEL_GO_X_OBSERVABILITY", "true")
+	const id = 0
+	counter.SetExporterID(id)
+
+	orig := otel.GetMeterProvider()
+	t.Cleanup(func() { otel.SetMeterProvider(orig) })
+
+	reader := metric.NewManualReader()
+	mp := metric.NewMeterProvider(metric.WithReader(reader))
+	otel.SetMeterProvider(mp)
+
+	// Use a client that returns a 503 error once and then a network error.
+	var calls int
+	client := &http.Client{
+		Transport: roundTripperFunc(func(_ *http.Request) (*http.Response, error) {
+			calls++
+			if calls == 1 {
+				return &http.Response{
+					StatusCode: http.StatusServiceUnavailable,
+					Status: fmt.Sprintf("%d %s",
+						http.StatusServiceUnavailable,
+						http.StatusText(http.StatusServiceUnavailable)),
+					Header: make(http.Header),
+					Body:   io.NopCloser(strings.NewReader("")),
+				}, nil
+			}
+			return nil, errors.New("network error")
+		}),
+	}
+
+	driver := otlptracehttp.NewClient(
+		otlptracehttp.WithHTTPClient(client),
+		otlptracehttp.WithInsecure(),
+		otlptracehttp.WithRetry(otlptracehttp.RetryConfig{
+			Enabled:         true,
+			InitialInterval: time.Nanosecond,
+			MaxInterval:     time.Nanosecond,
+			MaxElapsedTime:  time.Second,
+		}),
+	)
+	exporter, err := otlptrace.New(t.Context(), driver)
+	require.NoError(t, err)
+
+	err = exporter.ExportSpans(t.Context(), otlptracetest.SingleReadOnlySpan())
+	assert.Error(t, err)
+
+	require.NoError(t, exporter.Shutdown(t.Context()))
+
+	// Validate that the status code is 0 and not the stale 503 on self-observability metrics.
+	var got metricdata.ResourceMetrics
+	require.NoError(t, reader.Collect(t.Context(), &got))
+
+	require.Len(t, got.ScopeMetrics, 1)
+	metrics := got.ScopeMetrics[0].Metrics
+	var found bool
+	for _, m := range metrics {
+		if m.Name != (otelconv.SDKExporterOperationDuration{}).Name() {
+			continue
+		}
+		found = true
+		data := m.Data.(metricdata.Histogram[float64])
+		require.NotEmpty(t, data.DataPoints)
+		dp := data.DataPoints[0]
+		_, ok := dp.Attributes.Value(otelconv.SDKExporterOperationDuration{}.AttrHTTPResponseStatusCode(0).Key)
+		assert.False(t, ok, "should not report status code when the request fails before getting a response.")
+	}
+	assert.True(t, found, "expected to find operation duration metric")
+}
+
+type roundTripperFunc func(*http.Request) (*http.Response, error)
+
+func (f roundTripperFunc) RoundTrip(r *http.Request) (*http.Response, error) {
+	return f(r)
+}

exporters/otlp/otlptrace/otlptracehttp/internal/observ/instrumentation.go

@@ -345,7 +345,7 @@ func (e ExportOp) End(err error, status int) {
 //
 // Otherwise, a new RecordOption is returned with the base attributes of the
 // Instrumentation plus the http.response.status_code attribute set to the
-// provided status, and if err is not nil, the error.type attribute set
+// provided status (if non-zero), and if err is not nil, the error.type attribute set
 // to the type of the error.
 func (i *Instrumentation) recordOption(err error, status int) metric.RecordOption {
 	if err == nil && status == http.StatusOK {
@@ -356,7 +356,9 @@ func (i *Instrumentation) recordOption(err error, status int) metric.RecordOptio
 	defer put(measureAttrsPool, attrs)
 	*attrs = append(*attrs, i.attrs...)
 
-	*attrs = append(*attrs, semconv.HTTPResponseStatusCode(status))
+	if status != 0 {
+		*attrs = append(*attrs, semconv.HTTPResponseStatusCode(status))
+	}
 	if err != nil {
 		*attrs = append(*attrs, semconv.ErrorType(err))
 	}