otelgrpc: Fix stats handlers to honor WithMessageEvents option (#4536)

Author: fatsheep9146

CHANGELOG.md

@@ -32,6 +32,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 - The `go.opentelemetry.io/contrib/samplers/jaegerremote` sampler does not panic when the default HTTP round-tripper (`http.DefaultTransport`) is not `*http.Transport`. (#4045)
 - The `UnaryServerInterceptor` in `go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc` now sets gRPC status code correctly for the `rpc.server.duration` metric. (#4481)
+- The `NewClientHandler`, `NewServerHandler` in `go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc` now honor `otelgrpc.WithMessageEvents` options. (#4536) 
 
 ## [1.20.0/0.45.0/0.14.0] - 2023-09-28
 

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

@@ -153,28 +153,32 @@ func (c *config) handleRPC(ctx context.Context, rs stats.RPCStats) {
 			c.rpcRequestSize.Record(wctx, int64(rs.Length), metric.WithAttributes(metricAttrs...))
 		}
 
-		span.AddEvent("message",
-			trace.WithAttributes(
-				semconv.MessageTypeReceived,
-				semconv.MessageIDKey.Int64(messageId),
-				semconv.MessageCompressedSizeKey.Int(rs.CompressedLength),
-				semconv.MessageUncompressedSizeKey.Int(rs.Length),
-			),
-		)
+		if c.ReceivedEvent {
+			span.AddEvent("message",
+				trace.WithAttributes(
+					semconv.MessageTypeReceived,
+					semconv.MessageIDKey.Int64(messageId),
+					semconv.MessageCompressedSizeKey.Int(rs.CompressedLength),
+					semconv.MessageUncompressedSizeKey.Int(rs.Length),
+				),
+			)
+		}
 	case *stats.OutPayload:
 		if gctx != nil {
 			messageId = atomic.AddInt64(&gctx.messagesSent, 1)
 			c.rpcResponseSize.Record(wctx, int64(rs.Length), metric.WithAttributes(metricAttrs...))
 		}
 
-		span.AddEvent("message",
-			trace.WithAttributes(
-				semconv.MessageTypeSent,
-				semconv.MessageIDKey.Int64(messageId),
-				semconv.MessageCompressedSizeKey.Int(rs.CompressedLength),
-				semconv.MessageUncompressedSizeKey.Int(rs.Length),
-			),
-		)
+		if c.SentEvent {
+			span.AddEvent("message",
+				trace.WithAttributes(
+					semconv.MessageTypeSent,
+					semconv.MessageIDKey.Int64(messageId),
+					semconv.MessageCompressedSizeKey.Int(rs.CompressedLength),
+					semconv.MessageUncompressedSizeKey.Int(rs.Length),
+				),
+			)
+		}
 	case *stats.OutTrailer:
 	case *stats.End:
 		var rpcStatusAttr attribute.KeyValue

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

@@ -52,10 +52,18 @@ func TestStatsHandler(t *testing.T) {
 	err := newGrpcTest(
 		listener,
 		[]grpc.DialOption{
-			grpc.WithStatsHandler(otelgrpc.NewClientHandler(otelgrpc.WithTracerProvider(clientTP), otelgrpc.WithMeterProvider(clientMP))),
+			grpc.WithStatsHandler(otelgrpc.NewClientHandler(
+				otelgrpc.WithTracerProvider(clientTP),
+				otelgrpc.WithMeterProvider(clientMP),
+				otelgrpc.WithMessageEvents(otelgrpc.ReceivedEvents, otelgrpc.SentEvents)),
+			),
 		},
 		[]grpc.ServerOption{
-			grpc.StatsHandler(otelgrpc.NewServerHandler(otelgrpc.WithTracerProvider(serverTP), otelgrpc.WithMeterProvider(serverMP))),
+			grpc.StatsHandler(otelgrpc.NewServerHandler(
+				otelgrpc.WithTracerProvider(serverTP),
+				otelgrpc.WithMeterProvider(serverMP),
+				otelgrpc.WithMessageEvents(otelgrpc.ReceivedEvents, otelgrpc.SentEvents)),
+			),
 		},
 	)
 	require.NoError(t, err)