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/stats_handler_test.go

@@ -0,0 +1,40 @@
+package otelgrpc
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"google.golang.org/grpc/stats"
+)
+
+func TestHandleRPC(t *testing.T) {
+	const iteration = 100
+	wg := &sync.WaitGroup{}
+	ctx := context.Background()
+	h := NewClientHandler()
+	ctx = h.TagRPC(ctx, &stats.RPCTagInfo{
+		FullMethodName: "test/method",
+	})
+	for i := 0; i < iteration; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			h.HandleRPC(ctx, &stats.Begin{})
+			wg.Add(3)
+			go func() {
+				defer wg.Done()
+				h.HandleRPC(ctx, &stats.InPayload{})
+			}()
+			go func() {
+				defer wg.Done()
+				h.HandleRPC(ctx, &stats.OutPayload{})
+			}()
+			go func() {
+				defer wg.Done()
+				h.HandleRPC(ctx, &stats.End{})
+			}()
+		}()
+	}
+	wg.Wait()
+}