Reduce OpenTelemetry events in gRPC interceptor

CHANGELOG.md

@@ -36,6 +36,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ### Changed
 
 - Use `strings.Cut()` instead of `string.SplitN()` for better readability and memory use. (#3822)
+- Change `otelgrpc` interceptors to disable `SENT`/`RECEIVED` events.  Use `WithMessageEvents()` to turn back on. (#3964)
 
 ## [1.17.0-rc.1/0.42.0-rc.1/0.11.0-rc.1] - 2023-05-17
 

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

@@ -42,6 +42,9 @@ type config struct {
 	TracerProvider trace.TracerProvider
 	MeterProvider  metric.MeterProvider
 
+	ReceivedEvent bool
+	SentEvent     bool
+
 	meter             metric.Meter
 	rpcServerDuration metric.Int64Histogram
 }
@@ -131,3 +134,38 @@ func (o meterProviderOption) apply(c *config) {
 func WithMeterProvider(mp metric.MeterProvider) Option {
 	return meterProviderOption{mp: mp}
 }
+
+// Event type that can be recorded, see WithMessageEvents.
+type Event int
+
+// Different types of events that can be recorded, see WithMessageEvents.
+const (
+	ReceivedEvents Event = iota
+	SentEvents
+)
+
+type messageEventsProviderOption struct {
+	events []Event
+}
+
+func (m messageEventsProviderOption) apply(c *config) {
+	for _, e := range m.events {
+		switch e {
+		case ReceivedEvents:
+			c.ReceivedEvent = true
+		case SentEvents:
+			c.SentEvent = true
+		}
+	}
+}
+
+// WithMessageEvents configures the Handler to record the specified events
+// (span.AddEvent) on spans. By default only summary attributes are added at the
+// end of the request.
+//
+// Valid events are:
+//   - ReceivedEvents: Record the number of bytes read after every gRPC read operation.
+//   - SentEvents: Record the number of bytes written after every gRPC write operation.
+func WithMessageEvents(events ...Event) Option {
+	return messageEventsProviderOption{events: events}
+}

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

@@ -95,11 +95,15 @@
 
 		ctx = inject(ctx, cfg.Propagators)
 
-		messageSent.Event(ctx, 1, req)
+		if cfg.SentEvent {
+			messageSent.Event(ctx, 1, req)
+		}
 
 		err := invoker(ctx, method, req, reply, cc, callOpts...)
 
-		messageReceived.Event(ctx, 1, reply)
+		if cfg.ReceivedEvent {
+			messageReceived.Event(ctx, 1, reply)
+		}
 
 		if err != nil {
 			s, _ := status.FromError(err)
@@ -135,6 +139,9 @@
 	eventsDone chan struct{}
 	finished   chan error
 
+	receivedEvent bool
+	sentEvent     bool
+
 	receivedMessageID int
 	sentMessageID     int
 }
@@ -152,7 +159,10 @@
 		w.sendStreamEvent(errorEvent, err)
 	} else {
 		w.receivedMessageID++
-		messageReceived.Event(w.Context(), w.receivedMessageID, m)
+
+		if w.receivedEvent {
+			messageReceived.Event(w.Context(), w.receivedMessageID, m)
+		}
 	}
 
 	return err
@@ -162,7 +172,10 @@
 	err := w.ClientStream.SendMsg(m)
 
 	w.sentMessageID++
-	messageSent.Event(w.Context(), w.sentMessageID, m)
+
+	if w.sentEvent {
+		messageSent.Event(w.Context(), w.sentMessageID, m)
+	}
 
 	if err != nil {
 		w.sendStreamEvent(errorEvent, err)
@@ -191,7 +204,7 @@
 	return err
 }
 
-func wrapClientStream(ctx context.Context, s grpc.ClientStream, desc *grpc.StreamDesc) *clientStream {
+func wrapClientStream(ctx context.Context, s grpc.ClientStream, desc *grpc.StreamDesc, cfg *config) *clientStream {
 	events := make(chan streamEvent)
 	eventsDone := make(chan struct{})
 	finished := make(chan error)
@@ -218,11 +231,13 @@
 	}()
 
 	return &clientStream{
-		ClientStream: s,
-		desc:         desc,
-		events:       events,
-		eventsDone:   eventsDone,
-		finished:     finished,
+		ClientStream:  s,
+		desc:          desc,
+		events:        events,
+		eventsDone:    eventsDone,
+		finished:      finished,
+		receivedEvent: cfg.ReceivedEvent,
+		sentEvent:     cfg.SentEvent,
 	}
 }
 
@@ -277,7 +292,7 @@
 			span.End()
 			return s, err
 		}
-		stream := wrapClientStream(ctx, s, desc)
+		stream := wrapClientStream(ctx, s, desc, cfg)
 
 		go func() {
 			err := <-stream.finished
@@ -331,7 +346,9 @@
 		)
 		defer span.End()
 
-		messageReceived.Event(ctx, 1, req)
+		if cfg.ReceivedEvent {
+			messageReceived.Event(ctx, 1, req)
+		}
 
 		var statusCode grpc_codes.Code
 		defer func(t time.Time) {
@@ -347,11 +364,15 @@
 			statusCode, msg := serverStatus(s)
 			span.SetStatus(statusCode, msg)
 			span.SetAttributes(statusCodeAttr(s.Code()))
-			messageSent.Event(ctx, 1, s.Proto())
+			if cfg.SentEvent {
+				messageSent.Event(ctx, 1, s.Proto())
+			}
 		} else {
 			statusCode = grpc_codes.OK
 			span.SetAttributes(statusCodeAttr(grpc_codes.OK))
-			messageSent.Event(ctx, 1, resp)
+			if cfg.SentEvent {
+				messageSent.Event(ctx, 1, resp)
+			}
 		}
 
 		return resp, err
@@ -366,6 +387,9 @@
 
 	receivedMessageID int
 	sentMessageID     int
+
+	receivedEvent bool
+	sentEvent     bool
 }
 
 func (w *serverStream) Context() context.Context {
@@ -377,7 +401,9 @@
 
 	if err == nil {
 		w.receivedMessageID++
-		messageReceived.Event(w.Context(), w.receivedMessageID, m)
+		if w.receivedEvent {
+			messageReceived.Event(w.Context(), w.receivedMessageID, m)
+		}
 	}
 
 	return err
@@ -387,15 +413,19 @@
 	err := w.ServerStream.SendMsg(m)
 
 	w.sentMessageID++
-	messageSent.Event(w.Context(), w.sentMessageID, m)
+	if w.sentEvent {
+		messageSent.Event(w.Context(), w.sentMessageID, m)
+	}
 
 	return err
 }
 
-func wrapServerStream(ctx context.Context, ss grpc.ServerStream) *serverStream {
+func wrapServerStream(ctx context.Context, ss grpc.ServerStream, cfg *config) *serverStream {
 	return &serverStream{
-		ServerStream: ss,
-		ctx:          ctx,
+		ServerStream:  ss,
+		ctx:           ctx,
+		receivedEvent: cfg.ReceivedEvent,
+		sentEvent:     cfg.SentEvent,
 	}
 }
 
@@ -420,7 +450,7 @@
 			Type:             StreamServer,
 		}
 		if cfg.Filter != nil && !cfg.Filter(i) {
-			return handler(srv, wrapServerStream(ctx, ss))
+			return handler(srv, wrapServerStream(ctx, ss, cfg))
 		}
 
 		ctx = extract(ctx, cfg.Propagators)
@@ -434,7 +464,7 @@
 		)
 		defer span.End()
 
-		err := handler(srv, wrapServerStream(ctx, ss))
+		err := handler(srv, wrapServerStream(ctx, ss, cfg))
 		if err != nil {
 			s, _ := status.FromError(err)
 			statusCode, msg := serverStatus(s)

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

@@ -94,12 +94,25 @@ func TestInterceptors(t *testing.T) {
 
 	assert.NoError(t, doCalls(
 		[]grpc.DialOption{
-			grpc.WithUnaryInterceptor(otelgrpc.UnaryClientInterceptor(otelgrpc.WithTracerProvider(clientUnaryTP))),
-			grpc.WithStreamInterceptor(otelgrpc.StreamClientInterceptor(otelgrpc.WithTracerProvider(clientStreamTP))),
+			grpc.WithUnaryInterceptor(otelgrpc.UnaryClientInterceptor(
+				otelgrpc.WithTracerProvider(clientUnaryTP),
+				otelgrpc.WithMessageEvents(otelgrpc.ReceivedEvents, otelgrpc.SentEvents),
+			)),
+			grpc.WithStreamInterceptor(otelgrpc.StreamClientInterceptor(
+				otelgrpc.WithTracerProvider(clientStreamTP),
+				otelgrpc.WithMessageEvents(otelgrpc.ReceivedEvents, otelgrpc.SentEvents),
+			)),
 		},
 		[]grpc.ServerOption{
-			grpc.UnaryInterceptor(otelgrpc.UnaryServerInterceptor(otelgrpc.WithTracerProvider(serverUnaryTP), otelgrpc.WithMeterProvider(serverUnaryMP))),
-			grpc.StreamInterceptor(otelgrpc.StreamServerInterceptor(otelgrpc.WithTracerProvider(serverStreamTP))),
+			grpc.UnaryInterceptor(otelgrpc.UnaryServerInterceptor(
+				otelgrpc.WithTracerProvider(serverUnaryTP),
+				otelgrpc.WithMeterProvider(serverUnaryMP),
+				otelgrpc.WithMessageEvents(otelgrpc.ReceivedEvents, otelgrpc.SentEvents),
+			)),
+			grpc.StreamInterceptor(otelgrpc.StreamServerInterceptor(
+				otelgrpc.WithTracerProvider(serverStreamTP),
+				otelgrpc.WithMessageEvents(otelgrpc.ReceivedEvents, otelgrpc.SentEvents),
+			)),
 		},
 	))