Convert RabbitMQ to Instrumenter

instrumentation/rabbitmq-2.7/javaagent/build.gradle.kts

@@ -14,6 +14,9 @@ muzzle {
 dependencies {
   library("com.rabbitmq:amqp-client:2.7.0")
 
+  compileOnly("com.google.auto.value:auto-value-annotations")
+  annotationProcessor("com.google.auto.value:auto-value")
+
   testLibrary("org.springframework.amqp:spring-rabbit:1.1.0.RELEASE") {
     exclude("com.rabbitmq", "amqp-client")
   }

instrumentation/rabbitmq-2.7/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rabbitmq/ChannelAndMethod.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rabbitmq;
+
+import com.google.auto.value.AutoValue;
+import com.rabbitmq.client.Channel;
+
+@AutoValue
+public abstract class ChannelAndMethod {
+
+  public static ChannelAndMethod create(Channel channel, String method) {
+    return new AutoValue_ChannelAndMethod(channel, method);
+  }
+
+  abstract Channel getChannel();
+
+  abstract String getMethod();
+}

instrumentation/rabbitmq-2.7/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rabbitmq/DeliveryRequest.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rabbitmq;
+
+import com.google.auto.value.AutoValue;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Envelope;
+
+@AutoValue
+abstract class DeliveryRequest {
+
+  static DeliveryRequest create(
+      String queue, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
+    return new AutoValue_DeliveryRequest(queue, envelope, properties, body);
+  }
+
+  abstract String getQueue();
+
+  abstract Envelope getEnvelope();
+
+  abstract AMQP.BasicProperties getProperties();
+
+  @SuppressWarnings("mutable")
+  abstract byte[] getBody();
+
+  String spanName() {
+    String queue = getQueue();
+    if (queue == null || queue.isEmpty()) {
+      return "<default> process";
+    } else if (queue.startsWith("amq.gen-")) {
+      return "<generated> process";
+    } else {
+      return queue + " process";
+    }
+  }
+}

instrumentation/rabbitmq-2.7/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rabbitmq/RabbitChannelAttributesExtractor.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rabbitmq;
+
+import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessageOperation;
+import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesExtractor;
+import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
+import javax.annotation.Nullable;
+
+public class RabbitChannelAttributesExtractor
+    extends MessagingAttributesExtractor<ChannelAndMethod, Void> {
+  @Override
+  public MessageOperation operation() {
+    return MessageOperation.SEND;
+  }
+
+  @Override
+  protected String system(ChannelAndMethod channelAndMethod) {
+    return "rabbitmq";
+  }
+
+  @Override
+  protected String destinationKind(ChannelAndMethod channelAndMethod) {
+    return SemanticAttributes.MessagingDestinationKindValues.QUEUE;
+  }
+
+  @Nullable
+  @Override
+  protected String destination(ChannelAndMethod channelAndMethod) {
+    return null;
+  }
+
+  @Override
+  protected boolean temporaryDestination(ChannelAndMethod channelAndMethod) {
+    return false;
+  }
+
+  @Nullable
+  @Override
+  protected String protocol(ChannelAndMethod channelAndMethod) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  protected String protocolVersion(ChannelAndMethod channelAndMethod) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  protected String url(ChannelAndMethod channelAndMethod) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  protected String conversationId(ChannelAndMethod channelAndMethod) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  protected Long messagePayloadSize(ChannelAndMethod channelAndMethod) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  protected Long messagePayloadCompressedSize(ChannelAndMethod channelAndMethod) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  protected String messageId(ChannelAndMethod channelAndMethod, @Nullable Void unused) {
+    return null;
+  }
+}

instrumentation/rabbitmq-2.7/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rabbitmq/RabbitChannelInstrumentation.java

@@ -8,7 +8,9 @@
 import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.hasClassesNamed;
 import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.implementsInterface;
 import static io.opentelemetry.javaagent.instrumentation.rabbitmq.RabbitCommandInstrumentation.SpanHolder.CURRENT_RABBIT_CONTEXT;
-import static io.opentelemetry.javaagent.instrumentation.rabbitmq.RabbitTracer.tracer;
+import static io.opentelemetry.javaagent.instrumentation.rabbitmq.RabbitInstrumenterHelper.helper;
+import static io.opentelemetry.javaagent.instrumentation.rabbitmq.RabbitSingletons.channelInstrumenter;
+import static io.opentelemetry.javaagent.instrumentation.rabbitmq.RabbitSingletons.receiveInstrumenter;
 import static net.bytebuddy.matcher.ElementMatchers.canThrow;
 import static net.bytebuddy.matcher.ElementMatchers.isGetter;
 import static net.bytebuddy.matcher.ElementMatchers.isMethod;
@@ -94,13 +96,21 @@ public static void onEnter(
         @Advice.Origin("Channel.#m") String method,
         @Advice.Local("otelCallDepth") CallDepth callDepth,
         @Advice.Local("otelContext") Context context,
-        @Advice.Local("otelScope") Scope scope) {
+        @Advice.Local("otelScope") Scope scope,
+        @Advice.Local("otelRequest") ChannelAndMethod request) {
       callDepth = CallDepth.forClass(Channel.class);
       if (callDepth.getAndIncrement() > 0) {
         return;
       }
 
-      context = tracer().startSpan(method, channel.getConnection());
+      Context parentContext = Java8BytecodeBridge.currentContext();
+      request = ChannelAndMethod.create(channel, method);
+
+      if (!channelInstrumenter().shouldStart(parentContext, request)) {
+        return;
+      }
+
+      context = channelInstrumenter().start(parentContext, request);
       CURRENT_RABBIT_CONTEXT.set(context);
       scope = context.makeCurrent();
     }
@@ -110,19 +120,16 @@ public static void stopSpan(
         @Advice.Thrown Throwable throwable,
         @Advice.Local("otelCallDepth") CallDepth callDepth,
         @Advice.Local("otelContext") Context context,
-        @Advice.Local("otelScope") Scope scope) {
+        @Advice.Local("otelScope") Scope scope,
+        @Advice.Local("otelRequest") ChannelAndMethod request) {
       if (callDepth.decrementAndGet() > 0) {
         return;
       }
 
       scope.close();
 
       CURRENT_RABBIT_CONTEXT.remove();
-      if (throwable != null) {
-        tracer().endExceptionally(context, throwable);
-      } else {
-        tracer().end(context);
-      }
+      channelInstrumenter().end(context, request, null, throwable);
     }
   }
 
@@ -139,7 +146,7 @@ public static void setSpanNameAddHeaders(
       Span span = Java8BytecodeBridge.spanFromContext(context);
 
       if (span.getSpanContext().isValid()) {
-        tracer().onPublish(span, exchange, routingKey);
+        helper().onPublish(span, exchange, routingKey);
         if (body != null) {
           span.setAttribute(
               SemanticAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES, (long) body.length);
@@ -149,13 +156,13 @@ public static void setSpanNameAddHeaders(
         if (props == null) {
           props = MessageProperties.MINIMAL_BASIC;
         }
-        tracer().onProps(span, props);
+        helper().onProps(span, props);
 
         // We need to copy the BasicProperties and provide a header map we can modify
         Map<String, Object> headers = props.getHeaders();
         headers = (headers == null) ? new HashMap<>() : new HashMap<>(headers);
 
-        tracer().inject(context, headers, MapSetter.INSTANCE);
+        helper().inject(context, headers, MapSetter.INSTANCE);
 
         props =
             new AMQP.BasicProperties(
@@ -181,32 +188,37 @@ public static void setSpanNameAddHeaders(
   public static class ChannelGetAdvice {
 
     @Advice.OnMethodEnter
-    public static long takeTimestamp(@Advice.Local("otelCallDepth") CallDepth callDepth) {
+    public static void takeTimestamp(
+        @Advice.Local("otelCallDepth") CallDepth callDepth,
+        @Advice.Local("otelTimer") Timer timer) {
       callDepth = CallDepth.forClass(Channel.class);
       callDepth.getAndIncrement();
-      return System.currentTimeMillis();
+      timer = Timer.start();
     }
 
     @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
     public static void extractAndStartSpan(
         @Advice.This Channel channel,
         @Advice.Argument(0) String queue,
-        @Advice.Enter long startTime,
         @Advice.Return GetResponse response,
         @Advice.Thrown Throwable throwable,
-        @Advice.Local("otelCallDepth") CallDepth callDepth) {
+        @Advice.Local("otelCallDepth") CallDepth callDepth,
+        @Advice.Local("otelTimer") Timer timer) {
       if (callDepth.decrementAndGet() > 0) {
         return;
       }
 
+      Context parentContext = Java8BytecodeBridge.currentContext();
+      ReceiveRequest request =
+          ReceiveRequest.create(queue, timer, response, channel.getConnection());
+      if (!receiveInstrumenter().shouldStart(parentContext, request)) {
+        return;
+      }
+
       // can't create span and put into scope in method enter above, because can't add parent after
       // span creation
-      Context context = tracer().startGetSpan(queue, startTime, response, channel.getConnection());
-      if (throwable != null) {
-        tracer().endExceptionally(context, throwable);
-      } else {
-        tracer().end(context);
-      }
+      Context context = receiveInstrumenter().start(parentContext, request);
+      receiveInstrumenter().end(context, request, null, throwable);
     }
   }
 

instrumentation/rabbitmq-2.7/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rabbitmq/RabbitChannelNetAttributesExtractor.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rabbitmq;
+
+import io.opentelemetry.instrumentation.api.instrumenter.net.NetClientAttributesExtractor;
+import javax.annotation.Nullable;
+
+public class RabbitChannelNetAttributesExtractor
+    extends NetClientAttributesExtractor<ChannelAndMethod, Void> {
+  @Nullable
+  @Override
+  public String transport(ChannelAndMethod channelAndMethod, @Nullable Void unused) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public String peerName(ChannelAndMethod channelAndMethod, @Nullable Void unused) {
+    return channelAndMethod.getChannel().getConnection().getAddress().getHostName();
+  }
+
+  @Nullable
+  @Override
+  public Integer peerPort(ChannelAndMethod channelAndMethod, @Nullable Void unused) {
+    return channelAndMethod.getChannel().getConnection().getPort();
+  }
+
+  @Nullable
+  @Override
+  public String peerIp(ChannelAndMethod channelAndMethod, @Nullable Void unused) {
+    return channelAndMethod.getChannel().getConnection().getAddress().getHostAddress();
+  }
+}

instrumentation/rabbitmq-2.7/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rabbitmq/RabbitCommandInstrumentation.java

@@ -51,7 +51,7 @@ public static void setSpanNameAddHeaders(@Advice.This Command command) {
 
       Context context = CURRENT_RABBIT_CONTEXT.get();
       if (context != null && command.getMethod() != null) {
-        RabbitTracer.onCommand(Java8BytecodeBridge.spanFromContext(context), command);
+        RabbitInstrumenterHelper.onCommand(Java8BytecodeBridge.spanFromContext(context), command);
       }
     }
   }