Migrate RabbitMQ messaging telemetry to v1.43

Author: trask

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

@@ -44,7 +44,14 @@ tasks {
     systemProperty("metadataConfig", "otel.instrumentation.rabbitmq.experimental-span-attributes=true")
   }
 
+  val testMessagingPreview = register<Test>("testMessagingPreview") {
+    testClassesDirs = sourceSets.test.get().output.classesDirs
+    classpath = sourceSets.test.get().runtimeClasspath
+    jvmArgs("-Dotel.semconv-stability.preview=messaging")
+    systemProperty("metadataConfig", "otel.semconv-stability.preview=messaging")
+  }
+
   check {
-    dependsOn(testExperimental)
+    dependsOn(testExperimental, testMessagingPreview)
   }
 }

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

@@ -6,12 +6,25 @@
 package io.opentelemetry.javaagent.instrumentation.rabbitmq.v2_7;
 
 import io.opentelemetry.instrumentation.api.semconv.network.NetworkAttributesGetter;
+import io.opentelemetry.instrumentation.api.semconv.network.ServerAttributesGetter;
 import java.net.Inet4Address;
 import java.net.Inet6Address;
 import java.net.InetAddress;
 import javax.annotation.Nullable;
 
-class RabbitChannelNetAttributesGetter implements NetworkAttributesGetter<ChannelAndMethod, Void> {
+class RabbitChannelNetAttributesGetter
+    implements NetworkAttributesGetter<ChannelAndMethod, Void>,
+        ServerAttributesGetter<ChannelAndMethod> {
+
+  @Override
+  public String getServerAddress(ChannelAndMethod channelAndMethod) {
+    return channelAndMethod.getChannel().getConnection().getAddress().getHostAddress();
+  }
+
+  @Override
+  public Integer getServerPort(ChannelAndMethod channelAndMethod) {
+    return channelAndMethod.getChannel().getConnection().getPort();
+  }
 
   @Nullable
   @Override

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

@@ -5,6 +5,9 @@
 
 package io.opentelemetry.javaagent.instrumentation.rabbitmq.v2_7;
 
+import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitStableMessagingSemconv;
+import static io.opentelemetry.javaagent.instrumentation.rabbitmq.v2_7.RabbitInstrumenterHelper.consumerDestinationName;
+import static io.opentelemetry.javaagent.instrumentation.rabbitmq.v2_7.RabbitInstrumenterHelper.isGeneratedQueueName;
 import static java.util.Collections.emptyList;
 import static java.util.Collections.singletonList;
 
@@ -24,11 +27,13 @@ public String getSystem(DeliveryRequest request) {
   @Nullable
   @Override
   public String getDestination(DeliveryRequest request) {
-    if (request.getEnvelope() != null) {
-      return normalizeExchangeName(request.getEnvelope().getExchange());
-    } else {
-      return null;
+    if (emitStableMessagingSemconv()) {
+      return consumerDestinationName(
+          request.getEnvelope().getExchange(),
+          request.getEnvelope().getRoutingKey(),
+          request.getQueue());
     }
+    return normalizeExchangeName(request.getEnvelope().getExchange());
   }
 
   @Nullable
@@ -48,7 +53,7 @@ public boolean isTemporaryDestination(DeliveryRequest request) {
 
   @Override
   public boolean isAnonymousDestination(DeliveryRequest request) {
-    return false;
+    return isGeneratedQueueName(request.getQueue());
   }
 
   @Nullable

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

@@ -5,16 +5,22 @@
 
 package io.opentelemetry.javaagent.instrumentation.rabbitmq.v2_7;
 
+import static io.opentelemetry.api.common.AttributeKey.longKey;
+import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitStableMessagingSemconv;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY;
 
 import com.rabbitmq.client.Envelope;
+import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.common.AttributesBuilder;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
 import javax.annotation.Nullable;
 
 class RabbitDeliveryExtraAttributesExtractor implements AttributesExtractor<DeliveryRequest, Void> {
 
+  private static final AttributeKey<Long> MESSAGING_RABBITMQ_MESSAGE_DELIVERY_TAG =
+      longKey("messaging.rabbitmq.message.delivery_tag");
+
   @Override
   public void onStart(
       AttributesBuilder attributes, Context parentContext, DeliveryRequest request) {
@@ -23,6 +29,9 @@ public void onStart(
     if (routingKey != null && !routingKey.isEmpty()) {
       attributes.put(MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY, routingKey);
     }
+    if (emitStableMessagingSemconv()) {
+      attributes.put(MESSAGING_RABBITMQ_MESSAGE_DELIVERY_TAG, envelope.getDeliveryTag());
+    }
   }
 
   @Override

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

@@ -6,12 +6,25 @@
 package io.opentelemetry.javaagent.instrumentation.rabbitmq.v2_7;
 
 import io.opentelemetry.instrumentation.api.semconv.network.NetworkAttributesGetter;
+import io.opentelemetry.instrumentation.api.semconv.network.ServerAttributesGetter;
 import java.net.Inet4Address;
 import java.net.Inet6Address;
 import java.net.InetAddress;
 import javax.annotation.Nullable;
 
-class RabbitDeliveryNetAttributesGetter implements NetworkAttributesGetter<DeliveryRequest, Void> {
+class RabbitDeliveryNetAttributesGetter
+    implements NetworkAttributesGetter<DeliveryRequest, Void>,
+        ServerAttributesGetter<DeliveryRequest> {
+
+  @Override
+  public String getServerAddress(DeliveryRequest request) {
+    return request.getConnection().getAddress().getHostAddress();
+  }
+
+  @Override
+  public Integer getServerPort(DeliveryRequest request) {
+    return request.getConnection().getPort();
+  }
 
   @Nullable
   @Override

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

@@ -5,7 +5,9 @@
 
 package io.opentelemetry.javaagent.instrumentation.rabbitmq.v2_7;
 
+import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitStableMessagingSemconv;
 import static io.opentelemetry.javaagent.instrumentation.rabbitmq.v2_7.RabbitSingletons.CHANNEL_AND_METHOD_CONTEXT_KEY;
+import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_ANONYMOUS;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY;
 
@@ -17,6 +19,7 @@
 import io.opentelemetry.context.Context;
 import io.opentelemetry.instrumentation.api.incubator.config.internal.DeclarativeConfigUtil;
 import java.util.Map;
+import javax.annotation.Nullable;
 
 public class RabbitInstrumenterHelper {
   static final AttributeKey<String> RABBITMQ_COMMAND = AttributeKey.stringKey("rabbitmq.command");
@@ -33,8 +36,19 @@ public static RabbitInstrumenterHelper helper() {
 
   public void onPublish(Span span, String exchange, String routingKey) {
     String exchangeName = normalizeExchangeName(exchange);
-    span.setAttribute(MESSAGING_DESTINATION_NAME, exchangeName);
-    span.updateName(exchangeName + " publish");
+    if (emitStableMessagingSemconv()) {
+      String destinationName = producerDestinationName(exchange, routingKey);
+      span.setAttribute(MESSAGING_DESTINATION_NAME, destinationName);
+      boolean anonymousDestination =
+          isDefaultExchange(exchange) && isGeneratedQueueName(routingKey);
+      if (anonymousDestination) {
+        span.setAttribute(MESSAGING_DESTINATION_ANONYMOUS, true);
+      }
+      span.updateName(anonymousDestination ? "publish" : "publish " + destinationName);
+    } else {
+      span.setAttribute(MESSAGING_DESTINATION_NAME, exchangeName);
+      span.updateName(exchangeName + " publish");
+    }
     if (routingKey != null && !routingKey.isEmpty()) {
       span.setAttribute(MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY, routingKey);
     }
@@ -58,7 +72,66 @@ public void onProps(Context context, Span span, AMQP.BasicProperties props) {
   }
 
   private static String normalizeExchangeName(String exchange) {
-    return exchange == null || exchange.isEmpty() ? "<default>" : exchange;
+    return isDefaultExchange(exchange) ? "<default>" : exchange;
+  }
+
+  private static boolean isDefaultExchange(@Nullable String exchange) {
+    return exchange == null || exchange.isEmpty();
+  }
+
+  static boolean isGeneratedQueueName(@Nullable String queue) {
+    if (queue == null) {
+      return false;
+    }
+    if (queue.startsWith("amq.gen-") || queue.startsWith("spring.gen-")) {
+      return true;
+    }
+    return isCanonicalUuid(queue);
+  }
+
+  private static boolean isCanonicalUuid(String value) {
+    if (value.length() != 36) {
+      return false;
+    }
+    for (int i = 0; i < value.length(); i++) {
+      char ch = value.charAt(i);
+      if (i == 8 || i == 13 || i == 18 || i == 23) {
+        if (ch != '-') {
+          return false;
+        }
+      } else if (!((ch >= '0' && ch <= '9') || (ch >= 'a' && ch <= 'f'))) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  static String producerDestinationName(String exchange, String routingKey) {
+    StringBuilder destination = new StringBuilder();
+    appendDestinationPart(destination, exchange);
+    appendDestinationPart(destination, routingKey);
+    return destination.length() == 0 ? "amq.default" : destination.toString();
+  }
+
+  @Nullable
+  static String consumerDestinationName(String exchange, String routingKey, String queue) {
+    StringBuilder destination = new StringBuilder();
+    appendDestinationPart(destination, exchange);
+    appendDestinationPart(destination, routingKey);
+    if (queue != null && !queue.equals(routingKey)) {
+      appendDestinationPart(destination, queue);
+    }
+    return destination.length() == 0 ? null : destination.toString();
+  }
+
+  private static void appendDestinationPart(StringBuilder destination, String part) {
+    if (part == null || part.isEmpty()) {
+      return;
+    }
+    if (destination.length() != 0) {
+      destination.append(':');
+    }
+    destination.append(part);
   }
 
   public static void onCommand(Span span, Command command) {

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

@@ -5,6 +5,9 @@
 
 package io.opentelemetry.javaagent.instrumentation.rabbitmq.v2_7;
 
+import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitStableMessagingSemconv;
+import static io.opentelemetry.javaagent.instrumentation.rabbitmq.v2_7.RabbitInstrumenterHelper.consumerDestinationName;
+import static io.opentelemetry.javaagent.instrumentation.rabbitmq.v2_7.RabbitInstrumenterHelper.isGeneratedQueueName;
 import static java.util.Collections.emptyList;
 import static java.util.Collections.singletonList;
 
@@ -25,11 +28,17 @@ public String getSystem(ReceiveRequest request) {
   @Nullable
   @Override
   public String getDestination(ReceiveRequest request) {
-    if (request.getResponse() != null) {
-      return normalizeExchangeName(request.getResponse().getEnvelope().getExchange());
-    } else {
+    GetResponse response = request.getResponse();
+    if (emitStableMessagingSemconv()) {
+      return consumerDestinationName(
+          response == null ? null : response.getEnvelope().getExchange(),
+          response == null ? null : response.getEnvelope().getRoutingKey(),
+          request.getQueue());
+    }
+    if (response == null) {
       return null;
     }
+    return normalizeExchangeName(response.getEnvelope().getExchange());
   }
 
   @Nullable
@@ -49,7 +58,7 @@ public boolean isTemporaryDestination(ReceiveRequest request) {
 
   @Override
   public boolean isAnonymousDestination(ReceiveRequest request) {
-    return false;
+    return isGeneratedQueueName(request.getQueue());
   }
 
   @Nullable

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

@@ -0,0 +1,51 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rabbitmq.v2_7;
+
+import static io.opentelemetry.api.common.AttributeKey.longKey;
+import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitStableMessagingSemconv;
+import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY;
+
+import com.rabbitmq.client.GetResponse;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
+import javax.annotation.Nullable;
+
+class RabbitReceiveExtraAttributesExtractor
+    implements AttributesExtractor<ReceiveRequest, GetResponse> {
+
+  private static final AttributeKey<Long> MESSAGING_RABBITMQ_MESSAGE_DELIVERY_TAG =
+      longKey("messaging.rabbitmq.message.delivery_tag");
+
+  @Override
+  public void onStart(
+      AttributesBuilder attributes, Context parentContext, ReceiveRequest request) {}
+
+  @Override
+  public void onEnd(
+      AttributesBuilder attributes,
+      Context context,
+      ReceiveRequest request,
+      @Nullable GetResponse response,
+      @Nullable Throwable error) {
+    if (response == null) {
+      response = request.getResponse();
+      if (response == null) {
+        return;
+      }
+    }
+    String routingKey = response.getEnvelope().getRoutingKey();
+    if (routingKey != null && !routingKey.isEmpty()) {
+      attributes.put(MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY, routingKey);
+    }
+    if (emitStableMessagingSemconv()) {
+      attributes.put(
+          MESSAGING_RABBITMQ_MESSAGE_DELIVERY_TAG, response.getEnvelope().getDeliveryTag());
+    }
+  }
+}

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

@@ -0,0 +1,32 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rabbitmq.v2_7;
+
+import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_BATCH_MESSAGE_COUNT;
+
+import com.rabbitmq.client.GetResponse;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
+import javax.annotation.Nullable;
+
+final class RabbitReceiveMessageCountAttributesExtractor
+    implements AttributesExtractor<ReceiveRequest, GetResponse> {
+
+  @Override
+  public void onStart(
+      AttributesBuilder attributes, Context parentContext, ReceiveRequest request) {}
+
+  @Override
+  public void onEnd(
+      AttributesBuilder attributes,
+      Context context,
+      ReceiveRequest request,
+      @Nullable GetResponse response,
+      @Nullable Throwable error) {
+    attributes.put(MESSAGING_BATCH_MESSAGE_COUNT, request.getResponse() == null ? 0 : 1);
+  }
+}

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

@@ -7,13 +7,25 @@
 
 import com.rabbitmq.client.GetResponse;
 import io.opentelemetry.instrumentation.api.semconv.network.NetworkAttributesGetter;
+import io.opentelemetry.instrumentation.api.semconv.network.ServerAttributesGetter;
 import java.net.Inet4Address;
 import java.net.Inet6Address;
 import java.net.InetAddress;
 import javax.annotation.Nullable;
 
 class RabbitReceiveNetAttributesGetter
-    implements NetworkAttributesGetter<ReceiveRequest, GetResponse> {
+    implements NetworkAttributesGetter<ReceiveRequest, GetResponse>,
+        ServerAttributesGetter<ReceiveRequest> {
+
+  @Override
+  public String getServerAddress(ReceiveRequest request) {
+    return request.getConnection().getAddress().getHostAddress();
+  }
+
+  @Override
+  public Integer getServerPort(ReceiveRequest request) {
+    return request.getConnection().getPort();
+  }
 
   @Nullable
   @Override