[net/http] add http.server.active_requests metric

CHANGELOG.md

@@ -11,6 +11,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ### Added
 
 - Add the new `go.opentelemetry.io/contrib/instrgen` package to provide auto-generated source code instrumentation. (#3068, #3108)
+- Add the `http.server.active_requests` metric to `go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp`. (#4543)
 
 ## [1.21.0/0.46.0/0.15.0/0.1.0] - 2023-11-10
 

instrumentation/net/http/otelhttp/common.go

@@ -35,6 +35,7 @@ const (
 	RequestContentLength  = "http.server.request_content_length"  // Incoming request bytes total
 	ResponseContentLength = "http.server.response_content_length" // Incoming response bytes total
 	ServerLatency         = "http.server.duration"                // Incoming end to end duration, milliseconds
+	ActiveRequests        = "http.server.active_requests"         // Number of active requests
 )
 
 // Filter is a predicate used to determine whether a given http.request should

instrumentation/net/http/otelhttp/go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/stretchr/testify v1.8.4
 	go.opentelemetry.io/otel v1.20.0
 	go.opentelemetry.io/otel/metric v1.20.0
+	go.opentelemetry.io/otel/sdk/metric v1.20.0
 	go.opentelemetry.io/otel/trace v1.20.0
 )
 
@@ -15,5 +16,7 @@ require (
 	github.com/go-logr/logr v1.3.0 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.20.0 // indirect
+	golang.org/x/sys v0.14.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

instrumentation/net/http/otelhttp/go.sum

@@ -16,8 +16,14 @@ go.opentelemetry.io/otel v1.20.0 h1:vsb/ggIY+hUjD/zCAQHpzTmndPqv/ml2ArbsbfBYTAc=
 go.opentelemetry.io/otel v1.20.0/go.mod h1:oUIGj3D77RwJdM6PPZImDpSZGDvkD9fhesHny69JFrs=
 go.opentelemetry.io/otel/metric v1.20.0 h1:ZlrO8Hu9+GAhnepmRGhSU7/VkpjrNowxRN9GyKR4wzA=
 go.opentelemetry.io/otel/metric v1.20.0/go.mod h1:90DRw3nfK4D7Sm/75yQ00gTJxtkBxX+wu6YaNymbpVM=
+go.opentelemetry.io/otel/sdk v1.20.0 h1:5Jf6imeFZlZtKv9Qbo6qt2ZkmWtdWx/wzcCbNUlAWGM=
+go.opentelemetry.io/otel/sdk v1.20.0/go.mod h1:rmkSx1cZCm/tn16iWDn1GQbLtsW/LvsdEEFzCSRM6V0=
+go.opentelemetry.io/otel/sdk/metric v1.20.0 h1:5eD40l/H2CqdKmbSV7iht2KMK0faAIL2pVYzJOWobGk=
+go.opentelemetry.io/otel/sdk/metric v1.20.0/go.mod h1:AGvpC+YF/jblITiafMTYgvRBUiwi9hZf0EYE2E5XlS8=
 go.opentelemetry.io/otel/trace v1.20.0 h1:+yxVAPZPbQhbC3OfAkeIVTky6iTFpcr4SiY9om7mXSQ=
 go.opentelemetry.io/otel/trace v1.20.0/go.mod h1:HJSK7F/hA5RlzpZ0zKDCHCDHm556LCDtKaAo6JmBFUU=
+golang.org/x/sys v0.14.0 h1:Vz7Qs629MkJkGyHxUlRHizWJRG2j8fbQKjELVSNhy7Q=
+golang.org/x/sys v0.14.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

instrumentation/net/http/otelhttp/handler.go

@@ -44,6 +44,7 @@ type middleware struct {
 	filters           []Filter
 	spanNameFormatter func(string, *http.Request) string
 	counters          map[string]metric.Int64Counter
+	upDownCounters    map[string]metric.Int64UpDownCounter
 	valueRecorders    map[string]metric.Float64Histogram
 	publicEndpoint    bool
 	publicEndpointFn  func(*http.Request) bool
@@ -106,6 +107,7 @@ func handleErr(err error) {
 func (h *middleware) createMeasures() {
 	h.counters = make(map[string]metric.Int64Counter)
 	h.valueRecorders = make(map[string]metric.Float64Histogram)
+	h.upDownCounters = make(map[string]metric.Int64UpDownCounter)
 
 	requestBytesCounter, err := h.meter.Int64Counter(
 		RequestContentLength,
@@ -128,11 +130,27 @@ func (h *middleware) createMeasures() {
 	)
 	handleErr(err)
 
+	activeRequestsCounter, err := h.meter.Int64UpDownCounter(
+		ActiveRequests,
+		metric.WithUnit("{request}"),
+		metric.WithDescription("Number of active HTTP server requests."),
+	)
+	handleErr(err)
+
+	h.upDownCounters[ActiveRequests] = activeRequestsCounter
 	h.counters[RequestContentLength] = requestBytesCounter
 	h.counters[ResponseContentLength] = responseBytesCounter
 	h.valueRecorders[ServerLatency] = serverLatencyMeasure
 }
 
+func httpSchemeFromRequest(r *http.Request) attribute.KeyValue {
+	if r.TLS != nil {
+		return semconv.HTTPSchemeHTTPS
+	} else {
+		return semconv.HTTPSchemeHTTP
+	}
+}
+
 // serveHTTP sets up tracing and calls the given next http.Handler with the span
 // context injected into the request context.
 func (h *middleware) serveHTTP(w http.ResponseWriter, r *http.Request, next http.Handler) {
@@ -226,6 +244,10 @@ func (h *middleware) serveHTTP(w http.ResponseWriter, r *http.Request, next http
 	labeler := &Labeler{}
 	ctx = injectLabeler(ctx, labeler)
 
+	attrs := metric.WithAttributes(httpSchemeFromRequest(r), semconv.HTTPMethod(r.Method))
+	h.upDownCounters[ActiveRequests].Add(ctx, 1, attrs)
+	defer h.upDownCounters[ActiveRequests].Add(ctx, -1, attrs)
+
 	next.ServeHTTP(w, r.WithContext(ctx))
 
 	setAfterServeAttributes(span, bw.read, rww.written, rww.statusCode, bw.err, rww.err)

instrumentation/net/http/otelhttp/handler_test.go

@@ -15,10 +15,18 @@
 package otelhttp_test
 
 import (
+	"context"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
+	semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"slices"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -27,6 +35,90 @@ import (
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 )
 
+func assertActiveMetric(t *testing.T, ctx context.Context, reader *metric.ManualReader, actual metricdata.Aggregation, opts ...metricdatatest.Option) bool {
+	var rm metricdata.ResourceMetrics
+	err := reader.Collect(ctx, &rm)
+
+	if !assert.NoError(t, err, "failed to read the metric reader") {
+		return false
+	}
+
+	if !assert.Len(t, rm.ScopeMetrics, 1, "too many metrics") {
+		return false
+	}
+
+	metrics := rm.ScopeMetrics[0].Metrics
+	m := metrics[slices.IndexFunc(metrics, func(m metricdata.Metrics) bool {
+		return m.Name == otelhttp.ActiveRequests
+	})]
+	return metricdatatest.AssertAggregationsEqual(t, actual, m.Data, opts...)
+}
+
+func activeMetric(val int64) metricdata.Sum[int64] {
+	return metricdata.Sum[int64]{
+		Temporality: metricdata.CumulativeTemporality,
+		IsMonotonic: false,
+		DataPoints: []metricdata.DataPoint[int64]{
+			{
+				Value: val,
+				Attributes: attribute.NewSet(
+					semconv.HTTPMethod("GET"),
+					semconv.HTTPScheme("http"),
+				),
+			},
+		},
+	}
+}
+
+func TestActiveMetrics(t *testing.T) {
+	ctx, done := context.WithCancel(context.Background())
+	defer done()
+
+	reader := metric.NewManualReader()
+	var g sync.WaitGroup
+	var ag sync.WaitGroup
+	var l sync.Mutex
+
+	handler := otelhttp.NewHandler(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		g.Done()
+		l.Lock()
+		defer l.Unlock()
+		ag.Done()
+	}), "test", otelhttp.WithMeterProvider(metric.NewMeterProvider(metric.WithReader(reader))))
+
+	g.Add(5)
+	ag.Add(5)
+	l.Lock()
+
+	for i := 0; i < 5; i++ {
+		go handler.ServeHTTP(
+			httptest.NewRecorder(),
+			httptest.NewRequest("GET", "/foo/bar", nil),
+		)
+	}
+
+	g.Wait()
+
+	assertActiveMetric(t, ctx, reader, activeMetric(5), metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+
+	g.Add(5)
+	ag.Add(5)
+	for i := 0; i < 5; i++ {
+		go handler.ServeHTTP(
+			httptest.NewRecorder(),
+			httptest.NewRequest("GET", "/foo/bar", nil),
+		)
+	}
+	g.Wait()
+
+	assertActiveMetric(t, ctx, reader, activeMetric(10), metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+
+	l.Unlock()
+	ag.Wait()
+
+	assertActiveMetric(t, ctx, reader, activeMetric(0), metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+}
+
 func TestHandler(t *testing.T) {
 	testCases := []struct {
 		name               string