Use AtomicLongUpdater for gRPC message ID (#3137)

open-telemetry · May 31, 2021 · f755b3f · f755b3f
1 parent c3ed9a3
commit f755b3f
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 6 deletions.
diff --git a/...ry/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/TracingClientInterceptor.java b/...ry/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/TracingClientInterceptor.java
@@ -25,10 +25,14 @@
 import io.opentelemetry.context.propagation.ContextPropagators;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import java.net.SocketAddress;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 
 final class TracingClientInterceptor implements ClientInterceptor {
 
+  @SuppressWarnings("rawtypes")
+  private static final AtomicLongFieldUpdater<TracingClientCallListener> MESSAGE_ID_UPDATER =
+      AtomicLongFieldUpdater.newUpdater(TracingClientCallListener.class, "messageId");
+
   private final Instrumenter<GrpcRequest, Status> instrumenter;
   private final ContextPropagators propagators;
 
@@ -97,10 +101,13 @@ public void sendMessage(REQUEST message) {
 
   final class TracingClientCallListener<RESPONSE>
       extends ForwardingClientCallListener.SimpleForwardingClientCallListener<RESPONSE> {
+
     private final Context context;
     private final GrpcRequest request;
 
-    private final AtomicLong messageId = new AtomicLong();
+    // Used by MESSAGE_ID_UPDATER
+    @SuppressWarnings("UnusedVariable")
+    volatile long messageId;
 
     TracingClientCallListener(Listener<RESPONSE> delegate, Context context, GrpcRequest request) {
       super(delegate);
@@ -113,7 +120,10 @@ public void onMessage(RESPONSE message) {
       Span span = Span.fromContext(context);
       Attributes attributes =
           Attributes.of(
-              GrpcHelper.MESSAGE_TYPE, "SENT", GrpcHelper.MESSAGE_ID, messageId.incrementAndGet());
+              GrpcHelper.MESSAGE_TYPE,
+              "SENT",
+              GrpcHelper.MESSAGE_ID,
+              MESSAGE_ID_UPDATER.incrementAndGet(this));
       span.addEvent("message", attributes);
       try (Scope ignored = context.makeCurrent()) {
         delegate().onMessage(message);

diff --git a/...ry/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/TracingServerInterceptor.java b/...ry/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/TracingServerInterceptor.java
@@ -20,10 +20,14 @@
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.Scope;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 
 final class TracingServerInterceptor implements ServerInterceptor {
 
+  @SuppressWarnings("rawtypes")
+  private static final AtomicLongFieldUpdater<TracingServerCallListener> MESSAGE_ID_UPDATER =
+      AtomicLongFieldUpdater.newUpdater(TracingServerCallListener.class, "messageId");
+
   private final Instrumenter<GrpcRequest, Status> instrumenter;
   private final boolean captureExperimentalSpanAttributes;
 
@@ -89,7 +93,9 @@ final class TracingServerCallListener<REQUEST>
     private final Context context;
     private final GrpcRequest request;
 
-    private final AtomicLong messageId = new AtomicLong();
+    // Used by MESSAGE_ID_UPDATER
+    @SuppressWarnings("UnusedVariable")
+    volatile long messageId;
 
     TracingServerCallListener(Listener<REQUEST> delegate, Context context, GrpcRequest request) {
       super(delegate);
@@ -105,7 +111,7 @@ public void onMessage(REQUEST message) {
               GrpcHelper.MESSAGE_TYPE,
               "RECEIVED",
               GrpcHelper.MESSAGE_ID,
-              messageId.incrementAndGet());
+              MESSAGE_ID_UPDATER.incrementAndGet(this));
       Span.fromContext(context).addEvent("message", attributes);
       delegate().onMessage(message);
     }