otelgrpc: Add peer attributes to spans generated by stats handlers (#4873)

Author: pellared

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 peer attributes to spans recorded by `NewClientHandler`, `NewServerHandler` in `go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc`. (#4873)
 
 ## [1.22.0/0.47.0/0.16.0/0.2.0] - 2024-01-18
 

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

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	grpc_codes "google.golang.org/grpc/codes"
+	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/stats"
 	"google.golang.org/grpc/status"
 
@@ -179,6 +180,10 @@ func (c *config) handleRPC(ctx context.Context, rs stats.RPCStats, isServer bool
 			)
 		}
 	case *stats.OutTrailer:
+	case *stats.OutHeader:
+		if p, ok := peer.FromContext(ctx); ok {
+			span.SetAttributes(peerAttr(p.Addr.String())...)
+		}
 	case *stats.End:
 		var rpcStatusAttr attribute.KeyValue
 

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

@@ -18,6 +18,7 @@ import (
 	"context"
 	"io"
 	"net"
+	"strconv"
 	"sync"
 	"testing"
 
@@ -74,7 +75,7 @@ func TestStatsHandler(t *testing.T) {
 	doCalls(client)
 
 	t.Run("ClientSpans", func(t *testing.T) {
-		checkClientSpans(t, clientSR.Ended())
+		checkClientSpans(t, clientSR.Ended(), listener.Addr().String())
 	})
 
 	t.Run("ClientMetrics", func(t *testing.T) {
@@ -90,9 +91,14 @@ func TestStatsHandler(t *testing.T) {
 	})
 }
 
-func checkClientSpans(t *testing.T, spans []trace.ReadOnlySpan) {
+func checkClientSpans(t *testing.T, spans []trace.ReadOnlySpan, addr string) {
 	require.Len(t, spans, 5)
 
+	host, p, err := net.SplitHostPort(addr)
+	require.NoError(t, err)
+	port, err := strconv.Atoi(p)
+	require.NoError(t, err)
+
 	emptySpan := spans[0]
 	assert.False(t, emptySpan.EndTime().IsZero())
 	assert.Equal(t, "grpc.testing.TestService/EmptyCall", emptySpan.Name())
@@ -121,6 +127,8 @@ func checkClientSpans(t *testing.T, spans []trace.ReadOnlySpan) {
 		semconv.RPCServiceKey.String("grpc.testing.TestService"),
 		otelgrpc.RPCSystemGRPC,
 		otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
+		semconv.NetSockPeerAddr(host),
+		semconv.NetSockPeerPort(port),
 	}, emptySpan.Attributes())
 
 	largeSpan := spans[1]
@@ -151,6 +159,8 @@ func checkClientSpans(t *testing.T, spans []trace.ReadOnlySpan) {
 		semconv.RPCServiceKey.String("grpc.testing.TestService"),
 		otelgrpc.RPCSystemGRPC,
 		otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
+		semconv.NetSockPeerAddr(host),
+		semconv.NetSockPeerPort(port),
 	}, largeSpan.Attributes())
 
 	streamInput := spans[2]
@@ -209,6 +219,8 @@ func checkClientSpans(t *testing.T, spans []trace.ReadOnlySpan) {
 		semconv.RPCServiceKey.String("grpc.testing.TestService"),
 		otelgrpc.RPCSystemGRPC,
 		otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
+		semconv.NetSockPeerAddr(host),
+		semconv.NetSockPeerPort(port),
 	}, streamInput.Attributes())
 
 	streamOutput := spans[3]
@@ -266,6 +278,8 @@ func checkClientSpans(t *testing.T, spans []trace.ReadOnlySpan) {
 		semconv.RPCServiceKey.String("grpc.testing.TestService"),
 		otelgrpc.RPCSystemGRPC,
 		otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
+		semconv.NetSockPeerAddr(host),
+		semconv.NetSockPeerPort(port),
 	}, streamOutput.Attributes())
 
 	pingPong := spans[4]
@@ -350,6 +364,8 @@ func checkClientSpans(t *testing.T, spans []trace.ReadOnlySpan) {
 		semconv.RPCServiceKey.String("grpc.testing.TestService"),
 		otelgrpc.RPCSystemGRPC,
 		otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
+		semconv.NetSockPeerAddr(host),
+		semconv.NetSockPeerPort(port),
 	}, pingPong.Attributes())
 }
 
@@ -379,11 +395,15 @@ func checkServerSpans(t *testing.T, spans []trace.ReadOnlySpan) {
 			},
 		},
 	}, emptySpan.Events())
+	port, ok := findAttribute(emptySpan.Attributes(), semconv.NetSockPeerPortKey)
+	assert.True(t, ok)
 	assert.ElementsMatch(t, []attribute.KeyValue{
 		semconv.RPCMethodKey.String("EmptyCall"),
 		semconv.RPCServiceKey.String("grpc.testing.TestService"),
 		otelgrpc.RPCSystemGRPC,
 		otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
+		semconv.NetSockPeerAddr("127.0.0.1"),
+		port,
 	}, emptySpan.Attributes())
 
 	largeSpan := spans[1]
@@ -409,11 +429,15 @@ func checkServerSpans(t *testing.T, spans []trace.ReadOnlySpan) {
 			},
 		},
 	}, largeSpan.Events())
+	port, ok = findAttribute(largeSpan.Attributes(), semconv.NetSockPeerPortKey)
+	assert.True(t, ok)
 	assert.ElementsMatch(t, []attribute.KeyValue{
 		semconv.RPCMethodKey.String("UnaryCall"),
 		semconv.RPCServiceKey.String("grpc.testing.TestService"),
 		otelgrpc.RPCSystemGRPC,
 		otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
+		semconv.NetSockPeerAddr("127.0.0.1"),
+		port,
 	}, largeSpan.Attributes())
 
 	streamInput := spans[2]
@@ -467,11 +491,15 @@ func checkServerSpans(t *testing.T, spans []trace.ReadOnlySpan) {
 		},
 		// client does not record an event for the server response.
 	}, streamInput.Events())
+	port, ok = findAttribute(streamInput.Attributes(), semconv.NetSockPeerPortKey)
+	assert.True(t, ok)
 	assert.ElementsMatch(t, []attribute.KeyValue{
 		semconv.RPCMethodKey.String("StreamingInputCall"),
 		semconv.RPCServiceKey.String("grpc.testing.TestService"),
 		otelgrpc.RPCSystemGRPC,
 		otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
+		semconv.NetSockPeerAddr("127.0.0.1"),
+		port,
 	}, streamInput.Attributes())
 
 	streamOutput := spans[3]
@@ -524,11 +552,15 @@ func checkServerSpans(t *testing.T, spans []trace.ReadOnlySpan) {
 			},
 		},
 	}, streamOutput.Events())
+	port, ok = findAttribute(streamOutput.Attributes(), semconv.NetSockPeerPortKey)
+	assert.True(t, ok)
 	assert.ElementsMatch(t, []attribute.KeyValue{
 		semconv.RPCMethodKey.String("StreamingOutputCall"),
 		semconv.RPCServiceKey.String("grpc.testing.TestService"),
 		otelgrpc.RPCSystemGRPC,
 		otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
+		semconv.NetSockPeerAddr("127.0.0.1"),
+		port,
 	}, streamOutput.Attributes())
 
 	pingPong := spans[4]
@@ -608,11 +640,15 @@ func checkServerSpans(t *testing.T, spans []trace.ReadOnlySpan) {
 			},
 		},
 	}, pingPong.Events())
+	port, ok = findAttribute(pingPong.Attributes(), semconv.NetSockPeerPortKey)
+	assert.True(t, ok)
 	assert.ElementsMatch(t, []attribute.KeyValue{
 		semconv.RPCMethodKey.String("FullDuplexCall"),
 		semconv.RPCServiceKey.String("grpc.testing.TestService"),
 		otelgrpc.RPCSystemGRPC,
 		otelgrpc.GRPCStatusCodeKey.Int64(int64(codes.OK)),
+		semconv.NetSockPeerAddr("127.0.0.1"),
+		port,
 	}, pingPong.Attributes())
 }