[otelhttp] transport metrics

CHANGELOG.md

@@ -14,6 +14,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Add the new `go.opentelemetry.io/contrib/instrgen` package to provide auto-generated source code instrumentation. (#3068, #3108)
 - The `go.opentelemetry.io/contrib/exporters/autoexport` package to provide configuration of trace exporters with useful defaults and envar support. (#2753, #4100, #4129, #4132, #4134)
 - `WithRouteTag` in `go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp` adds HTTP route attribute to metrics. (#615)
+- Add metrics to `NewTransport` in `go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp`. (#3769)
 
 ### Fixed
 

instrumentation/net/http/otelhttp/common.go

@@ -37,6 +37,13 @@ const (
 	ServerLatency         = "http.server.duration"                // Incoming end to end duration, microseconds
 )
 
+// Client HTTP metrics.
+const (
+	ClientRequestContentLength  = "http.client.request_content_length"  // Outgoing request bytes total
+	ClientResponseContentLength = "http.client.response_content_length" // Outgoing response bytes total
+	ClientLatency               = "http.client.duration"                // Outgoing end to end duration, microseconds
+)
+
 // Filter is a predicate used to determine whether a given http.request should
 // be traced. A Filter must return true if the request should be traced.
 type Filter func(*http.Request) bool

instrumentation/net/http/otelhttp/config.go

@@ -20,6 +20,7 @@ import (
 	"net/http/httptrace"
 
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/propagation"
 	"go.opentelemetry.io/otel/trace"
@@ -32,18 +33,20 @@ const (
 // config represents the configuration options available for the http.Handler
 // and http.Transport types.
 type config struct {
-	ServerName        string
-	Tracer            trace.Tracer
-	Meter             metric.Meter
-	Propagators       propagation.TextMapPropagator
-	SpanStartOptions  []trace.SpanStartOption
-	PublicEndpoint    bool
-	PublicEndpointFn  func(*http.Request) bool
-	ReadEvent         bool
-	WriteEvent        bool
-	Filters           []Filter
-	SpanNameFormatter func(string, *http.Request) string
-	ClientTrace       func(context.Context) *httptrace.ClientTrace
+	ServerName            string
+	Tracer                trace.Tracer
+	Meter                 metric.Meter
+	Propagators           propagation.TextMapPropagator
+	SpanStartOptions      []trace.SpanStartOption
+	PublicEndpoint        bool
+	PublicEndpointFn      func(*http.Request) bool
+	ReadEvent             bool
+	WriteEvent            bool
+	Filters               []Filter
+	SpanNameFormatter     func(string, *http.Request) string
+	ClientTrace           func(context.Context) *httptrace.ClientTrace
+	GetRequestAttributes  func(*http.Request) []attribute.KeyValue
+	GetResponseAttributes func(*http.Response) []attribute.KeyValue
 
 	TracerProvider trace.TracerProvider
 	MeterProvider  metric.MeterProvider
@@ -206,3 +209,17 @@ func WithServerName(server string) Option {
 		c.ServerName = server
 	})
 }
+
+// WithRequestAttributeGetter extracts additional attributes from the request.
+func WithRequestAttributeGetter(fn func(req *http.Request) []attribute.KeyValue) Option {
+	return optionFunc(func(o *config) {
+		o.GetRequestAttributes = fn
+	})
+}
+
+// WithResponseAttributeGetter extracts additional attributes from the response.
+func WithResponseAttributeGetter(fn func(req *http.Response) []attribute.KeyValue) Option {
+	return optionFunc(func(o *config) {
+		o.GetResponseAttributes = fn
+	})
+}

instrumentation/net/http/otelhttp/handler.go

@@ -216,15 +216,15 @@ func (h *middleware) serveHTTP(w http.ResponseWriter, r *http.Request, next http
 
 	next.ServeHTTP(w, r.WithContext(ctx))
 
-	setAfterServeAttributes(span, bw.read, rww.written, rww.statusCode, bw.err, rww.err)
+	setAfterServeAttributes(span, bw.read.Load(), rww.written, rww.statusCode, bw.err, rww.err)
 
 	// Add metrics
 	attributes := append(labeler.Get(), semconvutil.HTTPServerRequest(h.server, r)...)
 	if rww.statusCode > 0 {
 		attributes = append(attributes, semconv.HTTPStatusCode(rww.statusCode))
 	}
 	o := metric.WithAttributes(attributes...)
-	h.counters[RequestContentLength].Add(ctx, bw.read, o)
+	h.counters[RequestContentLength].Add(ctx, bw.read.Load(), o)
 	h.counters[ResponseContentLength].Add(ctx, rww.written, o)
 
 	// Use floating point division here for higher precision (instead of Millisecond method).

instrumentation/net/http/otelhttp/test/transport_test.go

@@ -16,25 +16,75 @@ package test
 
 import (
 	"context"
+	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/http/httptrace"
 	"runtime"
+	"strconv"
 	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/sdk/instrumentation"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
+	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
 	"go.opentelemetry.io/otel/trace"
 )
 
+func assertClientScopeMetrics(t *testing.T, sm metricdata.ScopeMetrics, attrs attribute.Set) {
+	assert.Equal(t, instrumentation.Scope{
+		Name:    "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp",
+		Version: otelhttp.Version(),
+	}, sm.Scope)
+
+	require.Len(t, sm.Metrics, 3)
+
+	want := metricdata.Metrics{
+		Name: "http.client.request_content_length",
+		Data: metricdata.Sum[int64]{
+			DataPoints:  []metricdata.DataPoint[int64]{{Attributes: attrs, Value: 0}},
+			Temporality: metricdata.CumulativeTemporality,
+			IsMonotonic: true,
+		},
+	}
+	metricdatatest.AssertEqual(t, want, sm.Metrics[0], metricdatatest.IgnoreTimestamp())
+
+	want = metricdata.Metrics{
+		Name: "http.client.response_content_length",
+		Data: metricdata.Sum[int64]{
+			DataPoints:  []metricdata.DataPoint[int64]{{Attributes: attrs, Value: 13}},
+			Temporality: metricdata.CumulativeTemporality,
+			IsMonotonic: true,
+		},
+	}
+	metricdatatest.AssertEqual(t, want, sm.Metrics[1], metricdatatest.IgnoreTimestamp())
+
+	// Duration value is not predictable.
+	dur := sm.Metrics[2]
+	assert.Equal(t, "http.client.duration", dur.Name)
+	require.IsType(t, dur.Data, metricdata.Histogram[float64]{})
+	hist := dur.Data.(metricdata.Histogram[float64])
+	assert.Equal(t, metricdata.CumulativeTemporality, hist.Temporality)
+	require.Len(t, hist.DataPoints, 1)
+	dPt := hist.DataPoints[0]
+	assert.Equal(t, attrs, dPt.Attributes, "attributes")
+	assert.Equal(t, uint64(1), dPt.Count, "count")
+	assert.Equal(t, []float64{0, 5, 10, 25, 50, 75, 100, 250, 500, 750, 1000, 2500, 5000, 7500, 10000}, dPt.Bounds, "bounds")
+}
+
 func TestTransportUsesFormatter(t *testing.T) {
 	prop := propagation.TraceContext{}
 	spanRecorder := tracetest.NewSpanRecorder()
@@ -84,6 +134,9 @@ func TestTransportErrorStatus(t *testing.T) {
 	spanRecorder := tracetest.NewSpanRecorder()
 	provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(spanRecorder))
 
+	reader := metric.NewManualReader()
+	meterProvider := metric.NewMeterProvider(metric.WithReader(reader))
+
 	// Run a server and stop to make sure nothing is listening and force the error.
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
 	server.Close()
@@ -92,6 +145,7 @@ func TestTransportErrorStatus(t *testing.T) {
 	tr := otelhttp.NewTransport(
 		http.DefaultTransport,
 		otelhttp.WithTracerProvider(provider),
+		otelhttp.WithMeterProvider(meterProvider),
 	)
 	c := http.Client{Transport: tr}
 	r, err := http.NewRequest(http.MethodGet, server.URL, nil)
@@ -238,3 +292,67 @@ func TestWithHTTPTrace(t *testing.T) {
 	assert.Equal(t, spans[2].SpanContext().SpanID(), spans[0].Parent().SpanID())
 	assert.Equal(t, spans[1].SpanContext().SpanID(), spans[2].Parent().SpanID())
 }
+
+func TestTransportMetrics(t *testing.T) {
+	reader := metric.NewManualReader()
+	meterProvider := metric.NewMeterProvider(metric.WithReader(reader))
+
+	content := []byte("Hello, world!")
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		if _, err := w.Write(content); err != nil {
+			t.Fatal(err)
+		}
+	}))
+	defer ts.Close()
+
+	r, err := http.NewRequest(http.MethodGet, ts.URL, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tr := otelhttp.NewTransport(
+		http.DefaultTransport,
+		otelhttp.WithMeterProvider(meterProvider),
+		otelhttp.WithRequestAttributeGetter(func(req *http.Request) []attribute.KeyValue {
+			return []attribute.KeyValue{attribute.String("test_req", "attribute_req")}
+		}),
+		otelhttp.WithResponseAttributeGetter(func(req *http.Response) []attribute.KeyValue {
+			return []attribute.KeyValue{attribute.String("test_resp", "attribute_resp")}
+		}),
+	)
+
+	c := http.Client{Transport: tr}
+	res, err := c.Do(r)
+	if err != nil {
+		t.Fatal(err)
+	}
+	require.NoError(t, res.Body.Close())
+
+	host, portStr, _ := net.SplitHostPort(r.Host)
+	if host == "" {
+		host = "127.0.0.1"
+	}
+	port, err := strconv.Atoi(portStr)
+	if err != nil {
+		port = 0
+	}
+
+	rm := metricdata.ResourceMetrics{}
+	err = reader.Collect(context.Background(), &rm)
+	require.NoError(t, err)
+	require.Len(t, rm.ScopeMetrics, 1)
+	attrs := attribute.NewSet(
+		semconv.NetPeerName(host),
+		semconv.NetPeerPort(port),
+		semconv.HTTPURL(ts.URL),
+		semconv.HTTPFlavorKey.String(fmt.Sprintf("1.%d", r.ProtoMinor)),
+		semconv.HTTPMethod("GET"),
+		semconv.HTTPResponseContentLength(13),
+		attribute.String("test_req", "attribute_req"),
+		attribute.String("test_resp", "attribute_resp"),
+		semconv.HTTPStatusCode(200),
+	)
+	assertClientScopeMetrics(t, rm.ScopeMetrics[0], attrs)
+}