Bump otel instrumentation to 2.5.0

smallrye · Jul 3, 2024 · 1f09b77 · 1f09b77
1 parent 198a76d
commit 1f09b77
Show file tree

Hide file tree

Showing 12 changed files with 164 additions and 53 deletions.
diff --git a/pom.xml b/pom.xml
@@ -90,8 +90,8 @@
 
     <kafka.version>3.7.1</kafka.version>
 
-    <opentelemetry.version>1.32.0-alpha</opentelemetry.version>
-    <opentelemetry-semconv.version>1.21.0-alpha</opentelemetry-semconv.version>
+    <opentelemetry.instrumentation.version>2.5.0-alpha</opentelemetry.instrumentation.version>
+    <opentelemetry-semconv.version>1.25.0-alpha</opentelemetry-semconv.version>
 
     <smallrye-vertx-mutiny-clients.version>3.13.0</smallrye-vertx-mutiny-clients.version>
     <smallrye-reactive-converters.version>3.0.0</smallrye-reactive-converters.version>
@@ -282,7 +282,7 @@
       <dependency>
         <groupId>io.opentelemetry.instrumentation</groupId>
         <artifactId>opentelemetry-instrumentation-bom-alpha</artifactId>
-        <version>${opentelemetry.version}</version>
+        <version>${opentelemetry.instrumentation.version}</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>

diff --git a/...qp/src/main/java/io/smallrye/reactive/messaging/amqp/tracing/AmqpAttributesExtractor.java b/...qp/src/main/java/io/smallrye/reactive/messaging/amqp/tracing/AmqpAttributesExtractor.java
@@ -1,9 +1,12 @@
 package io.smallrye.reactive.messaging.amqp.tracing;
 
+import java.util.Collections;
+import java.util.List;
+
 import io.opentelemetry.api.common.AttributesBuilder;
 import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter;
 import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesGetter;
 import io.smallrye.reactive.messaging.amqp.AmqpMessage;
 
 public class AmqpAttributesExtractor implements AttributesExtractor<AmqpMessage<?>, Void> {
@@ -61,21 +64,44 @@ public String getConversationId(final AmqpMessage<?> amqpMessage) {
 
         // Recommended
         @Override
-        public Long getMessagePayloadSize(final AmqpMessage<?> amqpMessage) {
+        public String getMessageId(final AmqpMessage<?> amqpMessage, final Void unused) {
+            Object messageId = amqpMessage.getMessageId();
+            return messageId instanceof String ? (String) messageId : null;
+        }
+
+        @Override
+        public List<String> getMessageHeader(AmqpMessage<?> amqpMessage, String name) {
+            return Collections.emptyList();
+        }
+
+        @Override
+        public String getDestinationTemplate(AmqpMessage<?> amqpMessage) {
             return null;
         }
 
-        // Recommended
         @Override
-        public Long getMessagePayloadCompressedSize(final AmqpMessage<?> amqpMessage) {
+        public boolean isAnonymousDestination(AmqpMessage<?> amqpMessage) {
+            return false;
+        }
+
+        @Override
+        public Long getMessageBodySize(AmqpMessage<?> amqpMessage) {
             return null;
         }
 
-        // Recommended
         @Override
-        public String getMessageId(final AmqpMessage<?> amqpMessage, final Void unused) {
-            Object messageId = amqpMessage.getMessageId();
-            return messageId instanceof String ? (String) messageId : null;
+        public Long getMessageEnvelopeSize(AmqpMessage<?> amqpMessage) {
+            return null;
+        }
+
+        @Override
+        public String getClientId(AmqpMessage<?> amqpMessage) {
+            return null;
+        }
+
+        @Override
+        public Long getBatchMessageCount(AmqpMessage<?> amqpMessage, Void unused) {
+            return null;
         }
     }
 }
diff --git a/.../main/java/io/smallrye/reactive/messaging/amqp/tracing/AmqpOpenTelemetryInstrumenter.java b/.../main/java/io/smallrye/reactive/messaging/amqp/tracing/AmqpOpenTelemetryInstrumenter.java
@@ -5,12 +5,12 @@
 import org.eclipse.microprofile.reactive.messaging.Message;
 
 import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessageOperation;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesExtractor;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanNameExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessageOperation;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesExtractor;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesGetter;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingSpanNameExtractor;
 import io.smallrye.reactive.messaging.amqp.AmqpMessage;
 import io.smallrye.reactive.messaging.tracing.TracingUtils;
 

diff --git a/...ng-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/TracingAmqpToAppToAmqpTest.java b/...ng-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/TracingAmqpToAppToAmqpTest.java
@@ -9,7 +9,6 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNull;
 
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;

diff --git a/.../src/main/java/io/smallrye/reactive/messaging/kafka/tracing/KafkaAttributesExtractor.java b/.../src/main/java/io/smallrye/reactive/messaging/kafka/tracing/KafkaAttributesExtractor.java
@@ -1,15 +1,17 @@
 package io.smallrye.reactive.messaging.kafka.tracing;
 
 import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_CONSUMER_ID;
-import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_KAFKA_CLIENT_ID;
 import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_KAFKA_CONSUMER_GROUP;
 import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET;
 import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_KAFKA_PARTITION;
 
+import java.util.Collections;
+import java.util.List;
+
 import io.opentelemetry.api.common.AttributesBuilder;
 import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter;
 import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesGetter;
 
 public class KafkaAttributesExtractor implements AttributesExtractor<KafkaTrace, Void> {
     private final MessagingAttributesGetter<KafkaTrace, Void> messagingAttributesGetter;
@@ -39,9 +41,6 @@ public void onStart(final AttributesBuilder attributes, final Context parentCont
         if (groupId != null) {
             attributes.put(MESSAGING_KAFKA_CONSUMER_GROUP, groupId);
         }
-        if (clientId != null) {
-            attributes.put(MESSAGING_KAFKA_CLIENT_ID, clientId);
-        }
     }
 
     @Override
@@ -80,17 +79,45 @@ public String getConversationId(final KafkaTrace kafkaTrace) {
         }
 
         @Override
-        public Long getMessagePayloadSize(final KafkaTrace kafkaTrace) {
+        public String getMessageId(final KafkaTrace kafkaTrace, final Void unused) {
             return null;
         }
 
         @Override
-        public Long getMessagePayloadCompressedSize(final KafkaTrace kafkaTrace) {
+        public List<String> getMessageHeader(KafkaTrace kafkaTrace, String name) {
+            return Collections.emptyList();
+        }
+
+        @Override
+        public String getDestinationTemplate(KafkaTrace kafkaTrace) {
             return null;
         }
 
         @Override
-        public String getMessageId(final KafkaTrace kafkaTrace, final Void unused) {
+        public boolean isAnonymousDestination(KafkaTrace kafkaTrace) {
+            return false;
+        }
+
+        @Override
+        public Long getMessageBodySize(KafkaTrace kafkaTrace) {
+            return null;
+        }
+
+        @Override
+        public Long getMessageEnvelopeSize(KafkaTrace kafkaTrace) {
+            return null;
+        }
+
+        @Override
+        public String getClientId(KafkaTrace kafkaTrace) {
+            if (kafkaTrace.getClientId() == null) {
+                return null;
+            }
+            return kafkaTrace.getClientId();
+        }
+
+        @Override
+        public Long getBatchMessageCount(KafkaTrace kafkaTrace, Void unused) {
             return null;
         }
     }

diff --git a/...ain/java/io/smallrye/reactive/messaging/kafka/tracing/KafkaOpenTelemetryInstrumenter.java b/...ain/java/io/smallrye/reactive/messaging/kafka/tracing/KafkaOpenTelemetryInstrumenter.java
@@ -5,12 +5,12 @@
 import org.eclipse.microprofile.reactive.messaging.Message;
 
 import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessageOperation;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesExtractor;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanNameExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessageOperation;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesExtractor;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesGetter;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingSpanNameExtractor;
 import io.smallrye.reactive.messaging.tracing.TracingUtils;
 
 /**

diff --git a/...ka/src/test/java/io/smallrye/reactive/messaging/kafka/tracing/TracingPropagationTest.java b/...ka/src/test/java/io/smallrye/reactive/messaging/kafka/tracing/TracingPropagationTest.java
@@ -1,7 +1,7 @@
 package io.smallrye.reactive.messaging.kafka.tracing;
 
+import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_CLIENT_ID;
 import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_DESTINATION_NAME;
-import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_KAFKA_CLIENT_ID;
 import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET;
 import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_OPERATION;
 import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_SYSTEM;
@@ -117,7 +117,7 @@ public void testFromAppToKafka() {
             assertEquals("kafka", span.getAttributes().get(MESSAGING_SYSTEM));
             assertEquals("publish", span.getAttributes().get(MESSAGING_OPERATION));
             assertEquals(topic, span.getAttributes().get(MESSAGING_DESTINATION_NAME));
-            assertEquals("kafka-producer-kafka", span.getAttributes().get(MESSAGING_KAFKA_CLIENT_ID));
+            assertEquals("kafka-producer-kafka", span.getAttributes().get(MESSAGING_CLIENT_ID));
             assertEquals(0, span.getAttributes().get(MESSAGING_KAFKA_MESSAGE_OFFSET));
 
             assertEquals(topic + " publish", span.getName());
@@ -155,7 +155,8 @@ public void testFromAppToKafkaWithStructuredCloudEvents() {
             assertEquals("kafka", span.getAttributes().get(MESSAGING_SYSTEM));
             assertEquals("publish", span.getAttributes().get(MESSAGING_OPERATION));
             assertEquals(topic, span.getAttributes().get(MESSAGING_DESTINATION_NAME));
-            assertEquals("kafka-producer-kafka", span.getAttributes().get(MESSAGING_KAFKA_CLIENT_ID));
+            assertEquals("publish", span.getAttributes().get(MESSAGING_OPERATION));
+            assertEquals("kafka-producer-kafka", span.getAttributes().get(MESSAGING_CLIENT_ID));
             assertEquals(0, span.getAttributes().get(MESSAGING_KAFKA_MESSAGE_OFFSET));
             assertEquals(topic + " publish", span.getName());
         });
@@ -192,7 +193,8 @@ public void testFromAppToKafkaWithBinaryCloudEvents() {
             assertEquals("kafka", span.getAttributes().get(MESSAGING_SYSTEM));
             assertEquals("publish", span.getAttributes().get(MESSAGING_OPERATION));
             assertEquals(topic, span.getAttributes().get(MESSAGING_DESTINATION_NAME));
-            assertEquals("kafka-producer-kafka", span.getAttributes().get(MESSAGING_KAFKA_CLIENT_ID));
+            assertEquals("publish", span.getAttributes().get(MESSAGING_OPERATION));
+            assertEquals("kafka-producer-kafka", span.getAttributes().get(MESSAGING_CLIENT_ID));
             assertEquals(0, span.getAttributes().get(MESSAGING_KAFKA_MESSAGE_OFFSET));
             assertEquals(topic + " publish", span.getName());
         });
@@ -238,7 +240,7 @@ public void testFromKafkaToAppToKafka() {
             assertEquals("kafka", consumer.getAttributes().get(MESSAGING_SYSTEM));
             assertEquals("receive", consumer.getAttributes().get(MESSAGING_OPERATION));
             assertEquals(parentTopic, consumer.getAttributes().get(MESSAGING_DESTINATION_NAME));
-            assertEquals("kafka-consumer-source", consumer.getAttributes().get(MESSAGING_KAFKA_CLIENT_ID));
+            assertEquals("kafka-consumer-source", consumer.getAttributes().get(MESSAGING_CLIENT_ID));
             assertEquals(0, consumer.getAttributes().get(MESSAGING_KAFKA_MESSAGE_OFFSET));
             assertEquals(parentTopic + " receive", consumer.getName());
 
@@ -249,7 +251,8 @@ public void testFromKafkaToAppToKafka() {
             assertEquals("kafka", producer.getAttributes().get(MESSAGING_SYSTEM));
             assertEquals("publish", producer.getAttributes().get(MESSAGING_OPERATION));
             assertEquals(resultTopic, producer.getAttributes().get(MESSAGING_DESTINATION_NAME));
-            assertEquals("kafka-producer-kafka", producer.getAttributes().get(MESSAGING_KAFKA_CLIENT_ID));
+            assertEquals("publish", producer.getAttributes().get(MESSAGING_OPERATION));
+            assertEquals("kafka-producer-kafka", producer.getAttributes().get(MESSAGING_CLIENT_ID));
             assertEquals(0, producer.getAttributes().get(MESSAGING_KAFKA_MESSAGE_OFFSET));
             assertEquals(resultTopic + " publish", producer.getName());
         });
@@ -303,7 +306,7 @@ public void testFromKafkaToAppWithParentSpan() {
             assertEquals("kafka", consumer.getAttributes().get(MESSAGING_SYSTEM));
             assertEquals("receive", consumer.getAttributes().get(MESSAGING_OPERATION));
             assertEquals(parentTopic, consumer.getAttributes().get(MESSAGING_DESTINATION_NAME));
-            assertEquals("kafka-consumer-stuff", consumer.getAttributes().get(MESSAGING_KAFKA_CLIENT_ID));
+            assertEquals("kafka-consumer-stuff", consumer.getAttributes().get(MESSAGING_CLIENT_ID));
             assertEquals(0, consumer.getAttributes().get(MESSAGING_KAFKA_MESSAGE_OFFSET));
             assertEquals(parentTopic + " receive", consumer.getName());
         });

diff --git a/smallrye-reactive-messaging-otel/pom.xml b/smallrye-reactive-messaging-otel/pom.xml
@@ -37,7 +37,7 @@
     </dependency>
     <dependency>
       <groupId>io.opentelemetry.instrumentation</groupId>
-      <artifactId>opentelemetry-instrumentation-api-semconv</artifactId>
+      <artifactId>opentelemetry-instrumentation-api-incubator</artifactId>
       <exclusions>
         <exclusion>
           <groupId>io.opentelemetry.semconv</groupId>

diff --git a/...rc/main/java/io/smallrye/reactive/messaging/pulsar/tracing/PulsarAttributesExtractor.java b/...rc/main/java/io/smallrye/reactive/messaging/pulsar/tracing/PulsarAttributesExtractor.java
@@ -2,10 +2,13 @@
 
 import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_CONSUMER_ID;
 
+import java.util.Collections;
+import java.util.List;
+
 import io.opentelemetry.api.common.AttributesBuilder;
 import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter;
 import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesGetter;
 
 public class PulsarAttributesExtractor implements AttributesExtractor<PulsarTrace, Void> {
     private final MessagingAttributesGetter<PulsarTrace, Void> messagingAttributesGetter;
@@ -56,18 +59,43 @@ public String getConversationId(PulsarTrace pulsarTrace) {
         }
 
         @Override
-        public Long getMessagePayloadSize(PulsarTrace pulsarTrace) {
+        public String getMessageId(PulsarTrace pulsarTrace, Void unused) {
+            return pulsarTrace.getMessageId();
+        }
+
+        @Override
+        public List<String> getMessageHeader(PulsarTrace pulsarTrace, String name) {
+            return Collections.emptyList();
+        }
+
+        @Override
+        public String getDestinationTemplate(PulsarTrace pulsarTrace) {
+            return null;
+        }
+
+        @Override
+        public boolean isAnonymousDestination(PulsarTrace pulsarTrace) {
+            return false;
+        }
+
+        @Override
+        public Long getMessageBodySize(PulsarTrace pulsarTrace) {
             return pulsarTrace.getUncompressedPayloadSize();
         }
 
         @Override
-        public Long getMessagePayloadCompressedSize(PulsarTrace pulsarTrace) {
+        public Long getMessageEnvelopeSize(PulsarTrace pulsarTrace) {
             return null;
         }
 
         @Override
-        public String getMessageId(PulsarTrace pulsarTrace, Void unused) {
-            return pulsarTrace.getMessageId();
+        public String getClientId(PulsarTrace pulsarTrace) {
+            return null;
+        }
+
+        @Override
+        public Long getBatchMessageCount(PulsarTrace pulsarTrace, Void unused) {
+            return null;
         }
     }
 }
diff --git a/...n/java/io/smallrye/reactive/messaging/pulsar/tracing/PulsarOpenTelemetryInstrumenter.java b/...n/java/io/smallrye/reactive/messaging/pulsar/tracing/PulsarOpenTelemetryInstrumenter.java
@@ -5,12 +5,12 @@
 import org.eclipse.microprofile.reactive.messaging.Message;
 
 import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessageOperation;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesExtractor;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanNameExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessageOperation;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesExtractor;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesGetter;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingSpanNameExtractor;
 import io.smallrye.reactive.messaging.tracing.TracingUtils;
 
 public class PulsarOpenTelemetryInstrumenter {

diff --git a/...va/io/smallrye/reactive/messaging/rabbitmq/tracing/RabbitMQOpenTelemetryInstrumenter.java b/...va/io/smallrye/reactive/messaging/rabbitmq/tracing/RabbitMQOpenTelemetryInstrumenter.java
@@ -5,12 +5,12 @@
 import org.eclipse.microprofile.reactive.messaging.Message;
 
 import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessageOperation;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesExtractor;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanNameExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessageOperation;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesExtractor;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesGetter;
-import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingSpanNameExtractor;
 import io.smallrye.reactive.messaging.tracing.TracingUtils;
 
 public class RabbitMQOpenTelemetryInstrumenter {