Fix cardinality issue for rpc.server.duration (#3536)

.gitignore

@@ -4,6 +4,7 @@ Thumbs.db
 .tools/
 .idea/
 .vscode/
+vendor/
 *.iml
 *.so
 coverage.*

CHANGELOG.md

@@ -20,6 +20,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ### Fixed
 
 - Prevent taking from reservoir in AWS XRay Remote Sampler when there is zero capacity in `go.opentelemetry.io/contrib/samplers/aws/xray`. (#3684)
+- Improved `go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc` to optionally, off by default, use peer information as an attribute for the meter rpc.server.duration (#3536)
 
 ## [1.16.0-rc.2/0.41.0-rc.2/0.10.0-rc.2] - 2023-03-23
 

instrumentation/google.golang.org/grpc/otelgrpc/config.go

@@ -43,8 +43,9 @@ type config struct {
 	TracerProvider trace.TracerProvider
 	MeterProvider  metric.MeterProvider
 
-	meter             metric.Meter
-	rpcServerDuration instrument.Int64Histogram
+	meter                           metric.Meter
+	rpcServerDuration               instrument.Int64Histogram
+	includeRPCServerDurationPeerCtx bool
 }
 
 // Option applies an option value for a config.
@@ -127,6 +128,20 @@ func (o meterProviderOption) apply(c *config) {
 	}
 }
 
+type includePeerAddrAsAttribute struct {
+	use bool
+}
+
+// WithPeerAddrAsAttribute returns an Option to use peer details when
+// recording with the meter rpc.server.duration.
+func WithPeerAddrAsAttribute(peerAddrAsAttribute bool) Option {
+	return includePeerAddrAsAttribute{use: peerAddrAsAttribute}
+}
+
+func (o includePeerAddrAsAttribute) apply(c *config) {
+	c.includeRPCServerDurationPeerCtx = o.use
+}
+
 // WithMeterProvider returns an Option to use the MeterProvider when
 // creating a Meter. If this option is not provide the global MeterProvider will be used.
 func WithMeterProvider(mp metric.MeterProvider) Option {

instrumentation/google.golang.org/grpc/otelgrpc/interceptor.go

@@ -321,12 +321,16 @@ func UnaryServerInterceptor(opts ...Option) grpc.UnaryServerInterceptor {
 
 		ctx = extract(ctx, cfg.Propagators)
 
-		name, attr := spanInfo(info.FullMethod, peerFromCtx(ctx))
+		name, nameAttr := internal.ParseFullMethod(info.FullMethod)
+		peerAttrs := peerAttr(peerFromCtx(ctx))
+		spanAttrs := append(nameAttr, peerAttrs...)
+		spanAttrs = append(spanAttrs, RPCSystemGRPC)
+
 		ctx, span := tracer.Start(
 			trace.ContextWithRemoteSpanContext(ctx, trace.SpanContextFromContext(ctx)),
 			name,
 			trace.WithSpanKind(trace.SpanKindServer),
-			trace.WithAttributes(attr...),
+			trace.WithAttributes(spanAttrs...),
 		)
 		defer span.End()
 
@@ -335,6 +339,10 @@ func UnaryServerInterceptor(opts ...Option) grpc.UnaryServerInterceptor {
 		var statusCode grpc_codes.Code
 		defer func(t time.Time) {
 			elapsedTime := time.Since(t) / time.Millisecond
+			attr := append(nameAttr, RPCSystemGRPC)
+			if cfg.includeRPCServerDurationPeerCtx {
+				attr = append(attr, peerAttrs...)
+			}
 			attr = append(attr, semconv.RPCGRPCStatusCodeKey.Int64(int64(statusCode)))
 			cfg.rpcServerDuration.Record(ctx, int64(elapsedTime), attr...)
 		}(time.Now())

instrumentation/google.golang.org/grpc/otelgrpc/test/bufconn_mock.go

@@ -0,0 +1,117 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package test // import "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc/test"
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"google.golang.org/grpc/test/bufconn"
+)
+
+const (
+	mockIP   = "1.1.1.1"
+	mockPort = 1234
+)
+
+var (
+	mockAddr, _ = net.ResolveTCPAddr("tcp", fmt.Sprintf("%s:%d", mockIP, mockPort))
+)
+
+// bufConnMock wraps a bufconn.Lister for the sake of returning a valid address. This is so we can properly test
+// returning net.sock.peer.addr and net.sock.peer.port attributes.
+type bufConnMock struct {
+	listener *bufconn.Listener
+}
+
+func newMockBufConn(size int) *bufConnMock {
+	return &bufConnMock{
+		listener: bufconn.Listen(size),
+	}
+}
+
+func (b *bufConnMock) Accept() (net.Conn, error) {
+	conn, err := b.listener.Accept()
+	if err != nil {
+		return nil, err
+	}
+
+	return &bufConn{
+		conn: conn,
+	}, nil
+}
+
+func (b *bufConnMock) Close() error {
+	return b.listener.Close()
+}
+
+func (b *bufConnMock) Addr() net.Addr {
+	return mockAddr
+}
+
+func (b *bufConnMock) Dial() (net.Conn, error) {
+	// bufConnect's listener Dial implementation just calls
+	// DialContext under the covers so don't wrap the connection in our mock here
+	return b.listener.DialContext(context.Background())
+}
+
+func (b *bufConnMock) DialContext(ctx context.Context) (net.Conn, error) {
+	conn, err := b.listener.DialContext(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &bufConn{
+		conn: conn,
+	}, nil
+}
+
+type bufConn struct {
+	conn net.Conn
+}
+
+func (b *bufConn) Read(bytes []byte) (n int, err error) {
+	return b.conn.Read(bytes)
+}
+
+func (b *bufConn) Write(bytes []byte) (n int, err error) {
+	return b.conn.Write(bytes)
+}
+
+func (b *bufConn) Close() error {
+	return b.conn.Close()
+}
+
+func (b *bufConn) LocalAddr() net.Addr {
+	return mockAddr
+}
+
+func (b *bufConn) RemoteAddr() net.Addr {
+	return mockAddr
+}
+
+func (b *bufConn) SetDeadline(t time.Time) error {
+	return b.conn.SetDeadline(t)
+}
+
+func (b *bufConn) SetReadDeadline(t time.Time) error {
+	return b.conn.SetReadDeadline(t)
+}
+
+func (b *bufConn) SetWriteDeadline(t time.Time) error {
+	return b.conn.SetWriteDeadline(t)
+}