Add messaging exception event extractor helpers

Author: trask

instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/messaging/MessagingExceptionEventExtractors.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.api.incubator.semconv.messaging;
+
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import io.opentelemetry.api.logs.Severity;
+import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
+import io.opentelemetry.instrumentation.api.internal.Experimental;
+
+/** Utilities for configuring messaging exception events. */
+public final class MessagingExceptionEventExtractors {
+
+  /** Configures the messaging create exception event name and severity. */
+  @CanIgnoreReturnValue
+  public static <REQUEST, RESPONSE>
+      InstrumenterBuilder<REQUEST, RESPONSE> setMessagingCreateExceptionEventExtractor(
+          InstrumenterBuilder<REQUEST, RESPONSE> builder) {
+    return setExceptionEventExtractor(builder, "messaging.create.exception", Severity.WARN);
+  }
+
+  /** Configures the messaging send exception event name and severity. */
+  @CanIgnoreReturnValue
+  public static <REQUEST, RESPONSE>
+      InstrumenterBuilder<REQUEST, RESPONSE> setMessagingSendExceptionEventExtractor(
+          InstrumenterBuilder<REQUEST, RESPONSE> builder) {
+    return setExceptionEventExtractor(builder, "messaging.send.exception", Severity.WARN);
+  }
+
+  /** Configures the messaging receive exception event name and severity. */
+  @CanIgnoreReturnValue
+  public static <REQUEST, RESPONSE>
+      InstrumenterBuilder<REQUEST, RESPONSE> setMessagingReceiveExceptionEventExtractor(
+          InstrumenterBuilder<REQUEST, RESPONSE> builder) {
+    return setExceptionEventExtractor(builder, "messaging.receive.exception", Severity.WARN);
+  }
+
+  /** Configures the messaging settle exception event name and severity. */
+  @CanIgnoreReturnValue
+  public static <REQUEST, RESPONSE>
+      InstrumenterBuilder<REQUEST, RESPONSE> setMessagingSettleExceptionEventExtractor(
+          InstrumenterBuilder<REQUEST, RESPONSE> builder) {
+    return setExceptionEventExtractor(builder, "messaging.settle.exception", Severity.WARN);
+  }
+
+  /** Configures the messaging process exception event name and severity. */
+  @CanIgnoreReturnValue
+  public static <REQUEST, RESPONSE>
+      InstrumenterBuilder<REQUEST, RESPONSE> setMessagingProcessExceptionEventExtractor(
+          InstrumenterBuilder<REQUEST, RESPONSE> builder) {
+    return setExceptionEventExtractor(builder, "messaging.process.exception", Severity.ERROR);
+  }
+
+  @CanIgnoreReturnValue
+  private static <REQUEST, RESPONSE>
+      InstrumenterBuilder<REQUEST, RESPONSE> setExceptionEventExtractor(
+          InstrumenterBuilder<REQUEST, RESPONSE> builder, String eventName, Severity severity) {
+    Experimental.setExceptionEventExtractor(
+        builder,
+        (logRecordBuilder, context, request) -> {
+          logRecordBuilder.setEventName(eventName);
+          logRecordBuilder.setSeverity(severity);
+        });
+    return builder;
+  }
+
+  private MessagingExceptionEventExtractors() {}
+}

instrumentation/jms/jms-common-1.1/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/jms/common/v1_1/JmsInstrumenterFactory.java

@@ -5,6 +5,9 @@
 
 package io.opentelemetry.javaagent.instrumentation.jms.common.v1_1;
 
+import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingExceptionEventExtractors.setMessagingProcessExceptionEventExtractor;
+import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingExceptionEventExtractors.setMessagingReceiveExceptionEventExtractor;
+import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingExceptionEventExtractors.setMessagingSendExceptionEventExtractor;
 import static java.util.Collections.emptyList;
 
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
@@ -50,11 +53,12 @@ public Instrumenter<MessageWithDestination, Void> createProducerInstrumenter() {
     JmsMessageAttributesGetter getter = new JmsMessageAttributesGetter();
     MessageOperation operation = MessageOperation.PUBLISH;
 
-    return Instrumenter.<MessageWithDestination, Void>builder(
-            openTelemetry,
-            instrumentationName,
-            MessagingSpanNameExtractor.create(getter, operation))
-        .addAttributesExtractor(createMessagingAttributesExtractor(operation))
+    return setMessagingSendExceptionEventExtractor(
+            Instrumenter.<MessageWithDestination, Void>builder(
+                    openTelemetry,
+                    instrumentationName,
+                    MessagingSpanNameExtractor.create(getter, operation))
+                .addAttributesExtractor(createMessagingAttributesExtractor(operation)))
         .buildProducerInstrumenter(new MessagePropertySetter());
   }
 
@@ -68,6 +72,7 @@ public Instrumenter<MessageWithDestination, Void> createConsumerReceiveInstrumen
                 instrumentationName,
                 MessagingSpanNameExtractor.create(getter, operation))
             .addAttributesExtractor(createMessagingAttributesExtractor(operation));
+    setMessagingReceiveExceptionEventExtractor(builder);
     if (messagingReceiveInstrumentationEnabled) {
       builder.addSpanLinksExtractor(
           new PropagatorBasedSpanLinksExtractor<>(
@@ -88,6 +93,7 @@ public Instrumenter<MessageWithDestination, Void> createConsumerProcessInstrumen
                 instrumentationName,
                 MessagingSpanNameExtractor.create(getter, operation))
             .addAttributesExtractor(createMessagingAttributesExtractor(operation));
+    setMessagingProcessExceptionEventExtractor(builder);
     if (canHaveReceiveInstrumentation && messagingReceiveInstrumentationEnabled) {
       builder.addSpanLinksExtractor(
           new PropagatorBasedSpanLinksExtractor<>(

instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/KafkaInstrumenterFactory.java

@@ -5,6 +5,9 @@
 
 package io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal;
 
+import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingExceptionEventExtractors.setMessagingProcessExceptionEventExtractor;
+import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingExceptionEventExtractors.setMessagingReceiveExceptionEventExtractor;
+import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingExceptionEventExtractors.setMessagingSendExceptionEventExtractor;
 import static java.util.Collections.emptyList;
 
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
@@ -91,6 +94,7 @@ public Instrumenter<KafkaProducerRequest, RecordMetadata> createProducerInstrume
     if (captureExperimentalSpanAttributes) {
       builder.addAttributesExtractor(new KafkaProducerExperimentalAttributesExtractor());
     }
+    setMessagingSendExceptionEventExtractor(builder);
     return builder.buildInstrumenter(SpanKindExtractor.alwaysProducer());
   }
 
@@ -103,16 +107,17 @@ public Instrumenter<KafkaReceiveRequest, Void> createConsumerReceiveInstrumenter
     KafkaReceiveAttributesGetter getter = new KafkaReceiveAttributesGetter();
     MessageOperation operation = MessageOperation.RECEIVE;
 
-    return Instrumenter.<KafkaReceiveRequest, Void>builder(
-            openTelemetry,
-            instrumentationName,
-            MessagingSpanNameExtractor.create(getter, operation))
-        .addAttributesExtractor(
-            buildMessagingAttributesExtractor(getter, operation, capturedHeaders))
-        .addAttributesExtractor(new KafkaReceiveAttributesExtractor())
-        .addAttributesExtractors(extractors)
-        .setErrorCauseExtractor(errorCauseExtractor)
-        .setEnabled(messagingReceiveInstrumentationEnabled)
+    return setMessagingReceiveExceptionEventExtractor(
+            Instrumenter.<KafkaReceiveRequest, Void>builder(
+                    openTelemetry,
+                    instrumentationName,
+                    MessagingSpanNameExtractor.create(getter, operation))
+                .addAttributesExtractor(
+                    buildMessagingAttributesExtractor(getter, operation, capturedHeaders))
+                .addAttributesExtractor(new KafkaReceiveAttributesExtractor())
+                .addAttributesExtractors(extractors)
+                .setErrorCauseExtractor(errorCauseExtractor)
+                .setEnabled(messagingReceiveInstrumentationEnabled))
         .buildInstrumenter(SpanKindExtractor.alwaysConsumer());
   }
 
@@ -138,6 +143,7 @@ public Instrumenter<KafkaProcessRequest, Void> createConsumerProcessInstrumenter
     if (captureExperimentalSpanAttributes) {
       builder.addAttributesExtractor(new KafkaConsumerExperimentalAttributesExtractor());
     }
+    setMessagingProcessExceptionEventExtractor(builder);
 
     if (messagingReceiveInstrumentationEnabled) {
       builder.addSpanLinksExtractor(
@@ -154,17 +160,18 @@ public Instrumenter<KafkaReceiveRequest, Void> createBatchProcessInstrumenter()
     KafkaReceiveAttributesGetter getter = new KafkaReceiveAttributesGetter();
     MessageOperation operation = MessageOperation.PROCESS;
 
-    return Instrumenter.<KafkaReceiveRequest, Void>builder(
-            openTelemetry,
-            instrumentationName,
-            MessagingSpanNameExtractor.create(getter, operation))
-        .addAttributesExtractor(
-            buildMessagingAttributesExtractor(getter, operation, capturedHeaders))
-        .addAttributesExtractor(new KafkaReceiveAttributesExtractor())
-        .addSpanLinksExtractor(
-            new KafkaBatchProcessSpanLinksExtractor(
-                openTelemetry.getPropagators().getTextMapPropagator()))
-        .setErrorCauseExtractor(errorCauseExtractor)
+    return setMessagingProcessExceptionEventExtractor(
+            Instrumenter.<KafkaReceiveRequest, Void>builder(
+                    openTelemetry,
+                    instrumentationName,
+                    MessagingSpanNameExtractor.create(getter, operation))
+                .addAttributesExtractor(
+                    buildMessagingAttributesExtractor(getter, operation, capturedHeaders))
+                .addAttributesExtractor(new KafkaReceiveAttributesExtractor())
+                .addSpanLinksExtractor(
+                    new KafkaBatchProcessSpanLinksExtractor(
+                        openTelemetry.getPropagators().getTextMapPropagator()))
+                .setErrorCauseExtractor(errorCauseExtractor))
         .buildInstrumenter(SpanKindExtractor.alwaysConsumer());
   }
 

instrumentation/kafka/kafka-connect-2.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaconnect/v2_6/KafkaConnectSingletons.java

@@ -5,6 +5,8 @@
 
 package io.opentelemetry.javaagent.instrumentation.kafkaconnect.v2_6;
 
+import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingExceptionEventExtractors.setMessagingProcessExceptionEventExtractor;
+
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.context.propagation.TextMapPropagator;
 import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessageOperation;
@@ -26,16 +28,17 @@ public class KafkaConnectSingletons {
         new KafkaConnectBatchProcessSpanLinksExtractor(propagator);
 
     instrumenter =
-        Instrumenter.<KafkaConnectTask, Void>builder(
-                GlobalOpenTelemetry.get(),
-                INSTRUMENTATION_NAME,
-                MessagingSpanNameExtractor.create(
-                    new KafkaConnectAttributesGetter(), MessageOperation.PROCESS))
-            .addAttributesExtractor(
-                MessagingAttributesExtractor.builder(
-                        new KafkaConnectAttributesGetter(), MessageOperation.PROCESS)
-                    .build())
-            .addSpanLinksExtractor(spanLinksExtractor)
+        setMessagingProcessExceptionEventExtractor(
+                Instrumenter.<KafkaConnectTask, Void>builder(
+                        GlobalOpenTelemetry.get(),
+                        INSTRUMENTATION_NAME,
+                        MessagingSpanNameExtractor.create(
+                            new KafkaConnectAttributesGetter(), MessageOperation.PROCESS))
+                    .addAttributesExtractor(
+                        MessagingAttributesExtractor.builder(
+                                new KafkaConnectAttributesGetter(), MessageOperation.PROCESS)
+                            .build())
+                    .addSpanLinksExtractor(spanLinksExtractor))
             .buildInstrumenter(SpanKindExtractor.alwaysConsumer());
   }
 

instrumentation/nats/nats-2.17/library/src/main/java/io/opentelemetry/instrumentation/nats/v2_17/internal/NatsInstrumenterFactory.java

@@ -5,6 +5,9 @@
 
 package io.opentelemetry.instrumentation.nats.v2_17.internal;
 
+import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingExceptionEventExtractors.setMessagingProcessExceptionEventExtractor;
+import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingExceptionEventExtractors.setMessagingSendExceptionEventExtractor;
+
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessageOperation;
 import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesExtractor;
@@ -22,16 +25,17 @@ public final class NatsInstrumenterFactory {
 
   public static Instrumenter<NatsRequest, NatsRequest> createProducerInstrumenter(
       OpenTelemetry openTelemetry, List<String> capturedHeaders) {
-    return Instrumenter.<NatsRequest, NatsRequest>builder(
-            openTelemetry,
-            INSTRUMENTATION_NAME,
-            MessagingSpanNameExtractor.create(
-                new NatsRequestMessagingAttributesGetter(), MessageOperation.PUBLISH))
-        .addAttributesExtractor(
-            MessagingAttributesExtractor.builder(
-                    new NatsRequestMessagingAttributesGetter(), MessageOperation.PUBLISH)
-                .setCapturedHeaders(capturedHeaders)
-                .build())
+    return setMessagingSendExceptionEventExtractor(
+            Instrumenter.<NatsRequest, NatsRequest>builder(
+                    openTelemetry,
+                    INSTRUMENTATION_NAME,
+                    MessagingSpanNameExtractor.create(
+                        new NatsRequestMessagingAttributesGetter(), MessageOperation.PUBLISH))
+                .addAttributesExtractor(
+                    MessagingAttributesExtractor.builder(
+                            new NatsRequestMessagingAttributesGetter(), MessageOperation.PUBLISH)
+                        .setCapturedHeaders(capturedHeaders)
+                        .build()))
         .buildProducerInstrumenter(new NatsRequestTextMapSetter());
   }
 
@@ -48,6 +52,7 @@ public static Instrumenter<NatsRequest, Void> createConsumerProcessInstrumenter(
                         new NatsRequestMessagingAttributesGetter(), MessageOperation.PROCESS)
                     .setCapturedHeaders(capturedHeaders)
                     .build());
+    setMessagingProcessExceptionEventExtractor(builder);
 
     return builder.buildConsumerInstrumenter(new NatsRequestTextMapGetter());
   }

instrumentation/pulsar/pulsar-2.8/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/telemetry/PulsarSingletons.java

@@ -5,6 +5,10 @@
 
 package io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.telemetry;
 
+import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingExceptionEventExtractors.setMessagingProcessExceptionEventExtractor;
+import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingExceptionEventExtractors.setMessagingReceiveExceptionEventExtractor;
+import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingExceptionEventExtractors.setMessagingSendExceptionEventExtractor;
+
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.context.Context;
@@ -77,6 +81,7 @@ private static Instrumenter<PulsarRequest, Void> createConsumerReceiveInstrument
             .addOperationMetrics(MessagingConsumerMetrics.get())
             .addAttributesExtractor(
                 ServerAttributesExtractor.create(new PulsarNetClientAttributesGetter()));
+    setMessagingReceiveExceptionEventExtractor(instrumenterBuilder);
 
     if (receiveInstrumentationEnabled) {
       return instrumenterBuilder
@@ -91,16 +96,17 @@ private static Instrumenter<PulsarBatchRequest, Void> createConsumerBatchReceive
     MessagingAttributesGetter<PulsarBatchRequest, Void> getter =
         new PulsarBatchMessagingAttributesGetter();
 
-    return Instrumenter.<PulsarBatchRequest, Void>builder(
-            telemetry,
-            INSTRUMENTATION_NAME,
-            MessagingSpanNameExtractor.create(getter, MessageOperation.RECEIVE))
-        .addAttributesExtractor(
-            createMessagingAttributesExtractor(getter, MessageOperation.RECEIVE))
-        .addAttributesExtractor(
-            ServerAttributesExtractor.create(new PulsarNetClientAttributesGetter()))
-        .addSpanLinksExtractor(new PulsarBatchRequestSpanLinksExtractor(propagator))
-        .addOperationMetrics(MessagingConsumerMetrics.get())
+    return setMessagingReceiveExceptionEventExtractor(
+            Instrumenter.<PulsarBatchRequest, Void>builder(
+                    telemetry,
+                    INSTRUMENTATION_NAME,
+                    MessagingSpanNameExtractor.create(getter, MessageOperation.RECEIVE))
+                .addAttributesExtractor(
+                    createMessagingAttributesExtractor(getter, MessageOperation.RECEIVE))
+                .addAttributesExtractor(
+                    ServerAttributesExtractor.create(new PulsarNetClientAttributesGetter()))
+                .addSpanLinksExtractor(new PulsarBatchRequestSpanLinksExtractor(propagator))
+                .addOperationMetrics(MessagingConsumerMetrics.get()))
         .buildInstrumenter(SpanKindExtractor.alwaysConsumer());
   }
 
@@ -114,6 +120,7 @@ private static Instrumenter<PulsarRequest, Void> createConsumerProcessInstrument
                 MessagingSpanNameExtractor.create(getter, MessageOperation.PROCESS))
             .addAttributesExtractor(
                 createMessagingAttributesExtractor(getter, MessageOperation.PROCESS));
+    setMessagingProcessExceptionEventExtractor(instrumenterBuilder);
 
     if (receiveInstrumentationEnabled) {
       SpanLinksExtractor<PulsarRequest> spanLinksExtractor =
@@ -142,6 +149,7 @@ private static Instrumenter<PulsarRequest, Void> createProducerInstrumenter() {
         .getBoolean("experimental_span_attributes/development", false)) {
       builder.addAttributesExtractor(new ExperimentalProducerAttributesExtractor());
     }
+    setMessagingSendExceptionEventExtractor(builder);
 
     return builder.buildProducerInstrumenter(new MessageTextMapSetter());
   }