support mongo-driver's db metrics (#7983)

Co-authored-by: Preston Vasquez <[email protected]>
Co-authored-by: Damien Mathieu <[email protected]>

Author: 3 people

CHANGELOG.md

@@ -21,6 +21,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Add a `WithSpanNameFormatter` option to `go.opentelemetry.io/contrib/instrumentation/go.mongodb.org/mongo-driver/v2/mongo/otelmongo`. (#7986)
 - Add unmarshaling and validation for `AttributeType`, `AttributeNameValue`, `SimpleSpanProcessor`, `SimpleLogRecordProcessor`, `ZipkinSpanExporter`, `NameStringValuePair`, `InstrumentType`, `ExperimentalPeerInstrumentationServiceMappingElem`, `ExporterDefaultHistogramAggregation`, `PullMetricReader` to v1.0.0 model in `go.opentelemetry.io/contrib/otelconf`. (#8127)
 - Updated `go.opentelemetry.io/contrib/otelconf` to include the [v1.0.0-rc2](https://github.com/open-telemetry/opentelemetry-configuration/releases/tag/v1.0.0-rc.2) release candidate of schema which includes backwards incompatible changes. (#8026)
+- Support `db.client.operation.duration` metric for `go.opentelemetry.io/contrib/instrumentation/go.mongodb.org/mongo-driver/v2/mongo/otelmongo`. (#7983)
 
 ### Changed
 

instrumentation/go.mongodb.org/mongo-driver/v2/mongo/otelmongo/config.go

@@ -6,6 +6,7 @@ package otelmongo // import "go.opentelemetry.io/contrib/instrumentation/go.mong
 import (
 	"go.mongodb.org/mongo-driver/v2/event"
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -14,8 +15,10 @@ const ScopeName = "go.opentelemetry.io/contrib/instrumentation/go.mongodb.org/mo
 
 // config is used to configure the mongo tracer.
 type config struct {
+	MeterProvider  metric.MeterProvider
 	TracerProvider trace.TracerProvider
 
+	Meter  metric.Meter
 	Tracer trace.Tracer
 
 	CommandAttributeDisabled bool
@@ -26,6 +29,7 @@ type config struct {
 // newConfig returns a config with all Options set.
 func newConfig(opts ...Option) config {
 	cfg := config{
+		MeterProvider:            otel.GetMeterProvider(),
 		TracerProvider:           otel.GetTracerProvider(),
 		CommandAttributeDisabled: true,
 	}
@@ -43,6 +47,11 @@ func newConfig(opts ...Option) config {
 		opt.apply(&cfg)
 	}
 
+	cfg.Meter = cfg.MeterProvider.Meter(
+		ScopeName,
+		metric.WithInstrumentationVersion(Version()),
+	)
+
 	cfg.Tracer = cfg.TracerProvider.Tracer(
 		ScopeName,
 		trace.WithInstrumentationVersion(Version()),
@@ -61,6 +70,16 @@ func (o optionFunc) apply(c *config) {
 	o(c)
 }
 
+// WithMeterProvider specifies a [metric.MeterProvider] to use for creating a Meter.
+// If none is specified, the global MeterProvider is used.
+func WithMeterProvider(provider metric.MeterProvider) Option {
+	return optionFunc(func(cfg *config) {
+		if provider != nil {
+			cfg.MeterProvider = provider
+		}
+	})
+}
+
 // SpanNameFormatterFunc is a function that resolves the span name given an
 // *event.CommandStartedEvent.
 type SpanNameFormatterFunc func(e *event.CommandStartedEvent) string

instrumentation/go.mongodb.org/mongo-driver/v2/mongo/otelmongo/doc.go

@@ -4,7 +4,7 @@
 // Package otelmongo instruments go.mongodb.org/mongo-driver/v2/mongo.
 //
 // `NewMonitor` will return an event.CommandMonitor which is used to trace
-// requests.
+// requests and collect its metrics.
 //
 // This code was originally based on the following:
 // - https://github.com/open-telemetry/opentelemetry-go-contrib/tree/323e373a6c15ae310bdd0617e3ed52d8cb8e4e6f/instrumentation/go.mongodb.org/mongo-driver/mongo/otelmongo

instrumentation/go.mongodb.org/mongo-driver/v2/mongo/otelmongo/go.mod

@@ -6,7 +6,9 @@ require (
 	github.com/stretchr/testify v1.11.1
 	go.mongodb.org/mongo-driver/v2 v2.4.0
 	go.opentelemetry.io/otel v1.38.0
+	go.opentelemetry.io/otel/metric v1.38.0
 	go.opentelemetry.io/otel/sdk v1.38.0
+	go.opentelemetry.io/otel/sdk/metric v1.38.0
 	go.opentelemetry.io/otel/trace v1.38.0
 )
 
@@ -23,7 +25,6 @@ require (
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
-	go.opentelemetry.io/otel/metric v1.38.0 // indirect
 	golang.org/x/crypto v0.45.0 // indirect
 	golang.org/x/sync v0.18.0 // indirect
 	golang.org/x/sys v0.38.0 // indirect

instrumentation/go.mongodb.org/mongo-driver/v2/mongo/otelmongo/metrics_test.go

@@ -0,0 +1,145 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package otelmongo
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/drivertest"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+)
+
+const (
+	testAddr = "mongodb://localhost:27017/?connect=direct"
+)
+
+func TestMetricsOperationDuration(t *testing.T) {
+	reader := metric.NewManualReader()
+	provider := metric.NewMeterProvider(metric.WithReader(reader))
+
+	md := drivertest.NewMockDeployment()
+
+	ctx, cancel := context.WithTimeout(t.Context(), time.Second*3)
+	defer cancel()
+
+	opts := options.Client()
+	opts.Deployment = md //nolint:staticcheck // This method is the current documented way to set the mongodb mock. See https://github.com/mongodb/mongo-go-driver/blob/v2.0.0/x/mongo/driver/drivertest/opmsg_deployment_test.go#L24
+	opts.Monitor = NewMonitor(
+		WithMeterProvider(provider),
+		WithCommandAttributeDisabled(false),
+	)
+	opts.ApplyURI(testAddr)
+
+	md.AddResponses([]bson.D{{{Key: "ok", Value: 1}}}...)
+	client, err := mongo.Connect(opts)
+	require.NoError(t, err)
+	defer func() {
+		err := client.Disconnect(t.Context())
+		require.NoError(t, err)
+	}()
+
+	// Perform an insert operation
+	_, err = client.Database("test-database").Collection("test-collection").InsertOne(ctx, bson.D{{Key: "test-item", Value: "test-value"}})
+	require.NoError(t, err)
+
+	// Collect metrics
+	var rm metricdata.ResourceMetrics
+	err = reader.Collect(ctx, &rm)
+	require.NoError(t, err)
+
+	// Verify metrics were recorded
+	require.Len(t, rm.ScopeMetrics, 1)
+	scopeMetrics := rm.ScopeMetrics[0]
+	assert.Equal(t, ScopeName, scopeMetrics.Scope.Name)
+
+	// Find the operation duration metric
+	var foundDuration bool
+	for _, m := range scopeMetrics.Metrics {
+		if m.Name != "db.client.operation.duration" {
+			continue
+		}
+		foundDuration = true
+		histogram, ok := m.Data.(metricdata.Histogram[float64])
+		assert.True(t, ok, "expected histogram data type")
+		assert.NotEmpty(t, histogram.DataPoints)
+
+		// Check that attributes are present
+		dp := histogram.DataPoints[0]
+		attrs := dp.Attributes.ToSlice()
+		hasDBSystem := false
+		hasOperation := false
+		for _, attr := range attrs {
+			if attr.Key == "db.system.name" && attr.Value.AsString() == "mongodb" {
+				hasDBSystem = true
+			}
+			if attr.Key == "db.operation.name" && attr.Value.AsString() == "insert" {
+				hasOperation = true
+			}
+		}
+		assert.True(t, hasDBSystem, "expected db.system.name attribute")
+		assert.True(t, hasOperation, "expected db.operation.name attribute")
+	}
+	assert.True(t, foundDuration, "expected db.client.operation.duration metric")
+}
+
+func TestMetricsOperationFailure(t *testing.T) {
+	reader := metric.NewManualReader()
+	provider := metric.NewMeterProvider(metric.WithReader(reader))
+
+	md := drivertest.NewMockDeployment()
+
+	ctx, cancel := context.WithTimeout(t.Context(), time.Second*3)
+	defer cancel()
+
+	opts := options.Client()
+	opts.Deployment = md //nolint:staticcheck // This method is the current documented way to set the mongodb mock. See https://github.com/mongodb/mongo-go-driver/blob/v2.0.0/x/mongo/driver/drivertest/opmsg_deployment_test.go#L24
+	opts.Monitor = NewMonitor(
+		WithMeterProvider(provider),
+		WithCommandAttributeDisabled(true),
+	)
+	opts.ApplyURI(testAddr)
+
+	// Simulate an error response
+	md.AddResponses([]bson.D{{{Key: "ok", Value: 0}, {Key: "errmsg", Value: "test error"}}}...)
+	client, err := mongo.Connect(opts)
+	require.NoError(t, err)
+	defer func() {
+		err := client.Disconnect(t.Context())
+		require.NoError(t, err)
+	}()
+
+	_, err = client.Database("test-database").Collection("test-collection").InsertOne(ctx, bson.D{{Key: "test-item", Value: "test-value"}})
+	require.Error(t, err)
+
+	// Collect metrics
+	var rm metricdata.ResourceMetrics
+	err = reader.Collect(ctx, &rm)
+	require.NoError(t, err)
+
+	// Verify metrics were recorded even for failed operations
+	require.Len(t, rm.ScopeMetrics, 1)
+	scopeMetrics := rm.ScopeMetrics[0]
+	assert.NotEmpty(t, scopeMetrics.Metrics)
+}
+
+func TestNewMonitorWithInvalidMeterProvider(t *testing.T) {
+	// This test verifies that NewMonitor handles errors gracefully
+	// even if metric creation fails. The function should not panic
+	// and should return a valid monitor that can be used.
+
+	// Using a nil meter provider will use the global one, which should work
+	monitor := NewMonitor()
+	assert.NotNil(t, monitor)
+	assert.NotNil(t, monitor.Started)
+	assert.NotNil(t, monitor.Succeeded)
+	assert.NotNil(t, monitor.Failed)
+}

instrumentation/go.mongodb.org/mongo-driver/v2/mongo/otelmongo/mongo.go

@@ -12,9 +12,12 @@ import (
 
 	"go.mongodb.org/mongo-driver/v2/bson"
 	"go.mongodb.org/mongo-driver/v2/event"
+	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/metric"
 	semconv "go.opentelemetry.io/otel/semconv/v1.37.0"
+	"go.opentelemetry.io/otel/semconv/v1.37.0/dbconv"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -24,13 +27,15 @@ type spanKey struct {
 }
 
 type monitor struct {
+	ClientOperationDuration *dbconv.ClientOperationDuration
+
 	sync.Mutex
 	spans map[spanKey]trace.Span
 	cfg   config
 }
 
 func (m *monitor) Started(ctx context.Context, evt *event.CommandStartedEvent) {
-	hostname, port := peerInfo(evt)
+	hostname, port := peerInfo(evt.ConnectionID)
 
 	attrs := []attribute.KeyValue{
 		semconv.DBSystemNameMongoDB,
@@ -65,12 +70,66 @@ func (m *monitor) Started(ctx context.Context, evt *event.CommandStartedEvent) {
 	m.Unlock()
 }
 
-func (m *monitor) Succeeded(_ context.Context, evt *event.CommandSucceededEvent) {
+func (m *monitor) Succeeded(ctx context.Context, evt *event.CommandSucceededEvent) {
 	m.Finished(&evt.CommandFinishedEvent, nil)
+	if m.ClientOperationDuration == nil {
+		return
+	}
+
+	hostname, port := peerInfo(evt.ConnectionID)
+	attrs := attribute.NewSet(
+		semconv.DBSystemNameMongoDB,
+		// No need to add semconv.DBSystemMongoDB, it will be added by metrics recorder.
+		semconv.DBOperationName(evt.CommandName),
+		semconv.DBNamespace(evt.DatabaseName),
+		semconv.NetworkPeerAddress(hostname),
+		semconv.NetworkPeerPort(port),
+		semconv.NetworkTransportTCP,
+		// `db.response.status_code` is excluded for succeeded events.
+		// Succeeded processes an [go.mongodb.org/mongo-driver/v2/event.CommandSucceededEvent] for OTel,
+		// including collecting metrics. The status code metric is excluded since MongoDB server indicates
+		// a successful operation with {ok: 1}, which doesn't map to a traditional status code.
+	)
+	// TODO: db.query.text attribute is currently disabled by default.
+	// Because event does not provide the query text directly.
+	// command := m.extractCommand(evt)
+	// attrs = append(attrs, semconv.DBQueryText(sanitizeCommand(evt.Command)))
+
+	m.ClientOperationDuration.RecordSet(
+		ctx,
+		evt.Duration.Seconds(),
+		attrs,
+	)
 }
 
-func (m *monitor) Failed(_ context.Context, evt *event.CommandFailedEvent) {
+func (m *monitor) Failed(ctx context.Context, evt *event.CommandFailedEvent) {
 	m.Finished(&evt.CommandFinishedEvent, evt.Failure)
+	if m.ClientOperationDuration == nil {
+		return
+	}
+
+	hostname, port := peerInfo(evt.ConnectionID)
+	attrs := attribute.NewSet(
+		semconv.DBSystemNameMongoDB,
+		semconv.DBOperationName(evt.CommandName),
+		semconv.NetworkPeerAddress(hostname),
+		semconv.NetworkPeerPort(port),
+		semconv.NetworkTransportTCP,
+		// TODO: The status code should not be static, but reflect server behavior.
+		// Assert the error as [go.mongodb.org/mongo-driver/v2/x/mongo/driver.Error] and pull the code from there.
+		// ref. https://jira.mongodb.org/browse/GODRIVER-3690
+		semconv.ErrorType(evt.Failure),
+	)
+	// TODO: db.query.text attribute is currently disabled by default.
+	// Because event does not provide the query text directly.
+	// command := m.extractCommand(evt)
+	// attrs = append(attrs, semconv.DBQueryText(sanitizeCommand(evt.Command)))
+
+	m.ClientOperationDuration.RecordSet(
+		ctx,
+		evt.Duration.Seconds(),
+		attrs,
+	)
 }
 
 func (m *monitor) Finished(evt *event.CommandFinishedEvent, err error) {
@@ -125,9 +184,23 @@ func extractCollection(evt *event.CommandStartedEvent) (string, error) {
 // NewMonitor creates a new mongodb event CommandMonitor.
 func NewMonitor(opts ...Option) *event.CommandMonitor {
 	cfg := newConfig(opts...)
+	var clientOperationDuration *dbconv.ClientOperationDuration
+	operationDuration, err := dbconv.NewClientOperationDuration(
+		cfg.Meter,
+		metric.WithExplicitBucketBoundaries(0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5, 10),
+	)
+	if err != nil {
+		clientOperationDuration = nil
+		otel.Handle(err)
+	} else {
+		clientOperationDuration = &operationDuration
+	}
+
 	m := &monitor{
 		spans: make(map[spanKey]trace.Span),
 		cfg:   cfg,
+
+		ClientOperationDuration: clientOperationDuration,
 	}
 	return &event.CommandMonitor{
 		Started:   m.Started,
@@ -137,12 +210,12 @@ func NewMonitor(opts ...Option) *event.CommandMonitor {
 }
 
 // peerInfo will parse the hostname and port from the mongo connection ID.
-func peerInfo(evt *event.CommandStartedEvent) (hostname string, port int) {
+func peerInfo(connectionID string) (hostname string, port int) {
 	defaultMongoPort := 27017
-	hostname, portStr, err := net.SplitHostPort(evt.ConnectionID)
+	hostname, portStr, err := net.SplitHostPort(connectionID)
 	if err != nil {
 		// If parsing fails, assume default MongoDB port and return the entire ConnectionID as hostname
-		hostname = evt.ConnectionID
+		hostname = connectionID
 		port = defaultMongoPort
 		return hostname, port
 	}

instrumentation/go.mongodb.org/mongo-driver/v2/mongo/otelmongo/mongo_test.go

@@ -376,10 +376,7 @@ func TestPeerInfo(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			evt := &event.CommandStartedEvent{
-				ConnectionID: tc.connectionID,
-			}
-			host, port := peerInfo(evt)
+			host, port := peerInfo(tc.connectionID)
 			assert.Equal(t, tc.expectedHost, host)
 			assert.Equal(t, tc.expectedPort, port)
 		})