otelgrpc: Fix data race in stats handlers when processing messages received and sent metrics

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

@@ -195,8 +195,8 @@ func (c *config) handleRPC(ctx context.Context, rs stats.RPCStats) {
 
 		metricAttrs = append(metricAttrs, rpcStatusAttr)
 		c.rpcDuration.Record(wctx, float64(rs.EndTime.Sub(rs.BeginTime)), metric.WithAttributes(metricAttrs...))
-		c.rpcRequestsPerRPC.Record(wctx, gctx.messagesReceived, metric.WithAttributes(metricAttrs...))
-		c.rpcResponsesPerRPC.Record(wctx, gctx.messagesSent, metric.WithAttributes(metricAttrs...))
+		c.rpcRequestsPerRPC.Record(wctx, atomic.LoadInt64(&gctx.messagesReceived), metric.WithAttributes(metricAttrs...))
+		c.rpcResponsesPerRPC.Record(wctx, atomic.LoadInt64(&gctx.messagesSent), metric.WithAttributes(metricAttrs...))
 
 	default:
 		return

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

@@ -17,6 +17,7 @@ package test
 import (
 	"context"
 	"net"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -1316,3 +1317,27 @@ func checkServerMetrics(t *testing.T, reader metric.Reader) {
 
 	metricdatatest.AssertEqual(t, expectedScopeMetric, rm.ScopeMetrics[0], metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreValue())
 }
+
+func TestStatsHandlerConcurrentSafe(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err, "failed to open port")
+	client := newGrpcTest(t, listener,
+		[]grpc.DialOption{
+			grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
+		},
+		[]grpc.ServerOption{
+			grpc.StatsHandler(otelgrpc.NewServerHandler()),
+		},
+	)
+
+	wg := &sync.WaitGroup{}
+	const n = 100
+	for i := 0; i < n; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			doCalls(client)
+		}()
+	}
+	wg.Wait()
+}