Add messaging v1.43 shared contract

Author: trask

instrumentation-api-incubator/build.gradle.kts

@@ -92,13 +92,15 @@ tasks {
     testClassesDirs = sourceSets.test.get().output.classesDirs
     classpath = sourceSets.test.get().runtimeClasspath
     jvmArgs("-Dotel.semconv-stability.opt-in=database,code,service.peer,rpc")
+    jvmArgs("-Dotel.semconv-stability.preview=messaging")
     inputs.dir(jflexOutputDir)
   }
 
   val testBothSemconv = register<Test>("testBothSemconv") {
     testClassesDirs = sourceSets.test.get().output.classesDirs
     classpath = sourceSets.test.get().runtimeClasspath
     jvmArgs("-Dotel.semconv-stability.opt-in=database/dup,code/dup,service.peer/dup,rpc/dup")
+    jvmArgs("-Dotel.semconv-stability.preview=messaging/dup")
     inputs.dir(jflexOutputDir)
   }
 
@@ -117,6 +119,11 @@ tasks {
   }
 
   check {
-    dependsOn(testStableSemconv, testBothSemconv, testExceptionSignalLogs, testExceptionSignalLogsDup)
+    dependsOn(
+      testStableSemconv,
+      testBothSemconv,
+      testExceptionSignalLogs,
+      testExceptionSignalLogsDup,
+    )
   }
 }

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

@@ -5,24 +5,19 @@
 
 package io.opentelemetry.instrumentation.api.incubator.semconv.messaging;
 
-import java.util.Locale;
-
-/**
- * Represents type of <a
- * href="https://github.com/open-telemetry/semantic-conventions/blob/main/docs/messaging/messaging-spans.md#operation-names">operations</a>
- * that may be used in a messaging system.
- */
+/** Represents an operation that may be used in a messaging system. */
 public enum MessageOperation {
-  PUBLISH,
-  RECEIVE,
-  PROCESS;
+  PUBLISH(MessagingOperationType.SEND),
+  RECEIVE(MessagingOperationType.RECEIVE),
+  PROCESS(MessagingOperationType.PROCESS);
+
+  private final MessagingOperationType operationType;
+
+  MessageOperation(MessagingOperationType operationType) {
+    this.operationType = operationType;
+  }
 
-  /**
-   * Returns the operation name as defined in <a
-   * href="https://github.com/open-telemetry/semantic-conventions/blob/main/docs/messaging/messaging-spans.md#operation-names">the
-   * specification</a>.
-   */
-  String operationName() {
-    return name().toLowerCase(Locale.ROOT);
+  MessagingOperationType type() {
+    return operationType;
   }
 }

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

@@ -5,6 +5,10 @@
 
 package io.opentelemetry.instrumentation.api.incubator.semconv.messaging;
 
+import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitOldMessagingSemconv;
+import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitStableMessagingSemconv;
+import static io.opentelemetry.semconv.ErrorAttributes.ERROR_TYPE;
+
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.common.AttributesBuilder;
 import io.opentelemetry.context.Context;
@@ -17,7 +21,7 @@
 
 /**
  * Extractor of <a
- * href="https://github.com/open-telemetry/semantic-conventions/blob/main/docs/messaging/messaging-spans.md">messaging
+ * href="https://github.com/open-telemetry/semantic-conventions/blob/v1.43.0/docs/messaging/messaging-spans.md">messaging
  * attributes</a>.
  *
  * <p>This class delegates to a type-specific {@link MessagingAttributesGetter} for individual
@@ -29,8 +33,10 @@ public final class MessagingAttributesExtractor<REQUEST, RESPONSE>
   // copied from MessagingIncubatingAttributes
   private static final AttributeKey<Long> MESSAGING_BATCH_MESSAGE_COUNT =
       AttributeKey.longKey("messaging.batch.message_count");
-  private static final AttributeKey<String> MESSAGING_CLIENT_ID =
+  private static final AttributeKey<String> MESSAGING_CLIENT_ID_OLD =
       AttributeKey.stringKey("messaging.client_id");
+  private static final AttributeKey<String> MESSAGING_CLIENT_ID =
+      AttributeKey.stringKey("messaging.client.id");
   private static final AttributeKey<Boolean> MESSAGING_DESTINATION_ANONYMOUS =
       AttributeKey.booleanKey("messaging.destination.anonymous");
   private static final AttributeKey<String> MESSAGING_DESTINATION_NAME =
@@ -51,64 +57,105 @@ public final class MessagingAttributesExtractor<REQUEST, RESPONSE>
       AttributeKey.stringKey("messaging.message.id");
   private static final AttributeKey<String> MESSAGING_OPERATION =
       AttributeKey.stringKey("messaging.operation");
+  private static final AttributeKey<String> MESSAGING_OPERATION_NAME =
+      AttributeKey.stringKey("messaging.operation.name");
+  private static final AttributeKey<String> MESSAGING_OPERATION_TYPE =
+      AttributeKey.stringKey("messaging.operation.type");
   private static final AttributeKey<String> MESSAGING_SYSTEM =
       AttributeKey.stringKey("messaging.system");
 
   static final String TEMP_DESTINATION_NAME = "(temporary)";
 
-  /**
-   * Creates the messaging attributes extractor for the given {@link MessageOperation operation}
-   * with default configuration.
-   */
+  /** Creates the messaging attributes extractor for the given operation type. */
+  public static <REQUEST, RESPONSE> AttributesExtractor<REQUEST, RESPONSE> createForOperationType(
+      MessagingAttributesGetter<REQUEST, RESPONSE> getter,
+      @Nullable MessagingOperationType operationType) {
+    return builderForOperationType(getter, operationType).build();
+  }
+
+  /** Creates the messaging attributes extractor for the given operation. */
   public static <REQUEST, RESPONSE> AttributesExtractor<REQUEST, RESPONSE> create(
-      MessagingAttributesGetter<REQUEST, RESPONSE> getter, MessageOperation operation) {
+      MessagingAttributesGetter<REQUEST, RESPONSE> getter, @Nullable MessageOperation operation) {
     return builder(getter, operation).build();
   }
 
   /**
-   * Returns a new {@link MessagingAttributesExtractorBuilder} for the given {@link MessageOperation
-   * operation} that can be used to configure the messaging attributes extractor.
+   * Returns a new {@link MessagingAttributesExtractorBuilder} configured for the given operation
+   * type.
    */
+  public static <REQUEST, RESPONSE>
+      MessagingAttributesExtractorBuilder<REQUEST, RESPONSE> builderForOperationType(
+          MessagingAttributesGetter<REQUEST, RESPONSE> getter,
+          @Nullable MessagingOperationType operationType) {
+    return new MessagingAttributesExtractorBuilder<>(getter, operationType, true);
+  }
+
+  /** Returns a new messaging attributes extractor builder for the given operation. */
   public static <REQUEST, RESPONSE> MessagingAttributesExtractorBuilder<REQUEST, RESPONSE> builder(
-      MessagingAttributesGetter<REQUEST, RESPONSE> getter, MessageOperation operation) {
-    return new MessagingAttributesExtractorBuilder<>(getter, operation);
+      MessagingAttributesGetter<REQUEST, RESPONSE> getter, @Nullable MessageOperation operation) {
+    return new MessagingAttributesExtractorBuilder<>(
+        getter, operation == null ? null : operation.type(), false);
   }
 
   private final MessagingAttributesGetter<REQUEST, RESPONSE> getter;
-  private final MessageOperation operation;
+  @Nullable private final MessagingOperationType operationType;
+  @Nullable private final String operationName;
+  private final boolean supportsStableSemconv;
   private final List<String> capturedHeaders;
 
   MessagingAttributesExtractor(
       MessagingAttributesGetter<REQUEST, RESPONSE> getter,
-      MessageOperation operation,
+      @Nullable MessagingOperationType operationType,
+      @Nullable String operationName,
+      boolean supportsStableSemconv,
       List<String> capturedHeaders) {
     this.getter = getter;
-    this.operation = operation;
+    this.operationType = operationType;
+    this.operationName = operationName;
+    this.supportsStableSemconv = supportsStableSemconv;
     this.capturedHeaders = new ArrayList<>(capturedHeaders);
   }
 
   @Override
   public void onStart(AttributesBuilder attributes, Context parentContext, REQUEST request) {
+    boolean emitOldSemconv = !supportsStableSemconv || emitOldMessagingSemconv();
+    boolean emitStableSemconv = supportsStableSemconv && emitStableMessagingSemconv();
     attributes.put(MESSAGING_SYSTEM, getter.getSystem(request));
     boolean isTemporaryDestination = getter.isTemporaryDestination(request);
     if (isTemporaryDestination) {
       attributes.put(MESSAGING_DESTINATION_TEMPORARY, true);
-      attributes.put(MESSAGING_DESTINATION_NAME, TEMP_DESTINATION_NAME);
+      if (emitStableSemconv) {
+        attributes.put(MESSAGING_DESTINATION_NAME, getter.getDestination(request));
+        attributes.put(MESSAGING_DESTINATION_TEMPLATE, getter.getDestinationTemplate(request));
+      } else {
+        attributes.put(MESSAGING_DESTINATION_NAME, TEMP_DESTINATION_NAME);
+      }
     } else {
       attributes.put(MESSAGING_DESTINATION_NAME, getter.getDestination(request));
       attributes.put(MESSAGING_DESTINATION_TEMPLATE, getter.getDestinationTemplate(request));
     }
     attributes.put(MESSAGING_DESTINATION_PARTITION_ID, getter.getDestinationPartitionId(request));
     boolean isAnonymousDestination = getter.isAnonymousDestination(request);
-    if (isAnonymousDestination) {
+    if (emitStableSemconv && isAnonymousDestination) {
       attributes.put(MESSAGING_DESTINATION_ANONYMOUS, true);
     }
     attributes.put(MESSAGING_MESSAGE_CONVERSATION_ID, getter.getConversationId(request));
     attributes.put(MESSAGING_MESSAGE_BODY_SIZE, getter.getMessageBodySize(request));
     attributes.put(MESSAGING_MESSAGE_ENVELOPE_SIZE, getter.getMessageEnvelopeSize(request));
-    attributes.put(MESSAGING_CLIENT_ID, getter.getClientId(request));
-    if (operation != null) {
-      attributes.put(MESSAGING_OPERATION, operation.operationName());
+    if (emitOldSemconv) {
+      attributes.put(MESSAGING_CLIENT_ID_OLD, getter.getClientId(request));
+    }
+    if (emitStableSemconv) {
+      attributes.put(MESSAGING_CLIENT_ID, getter.getClientId(request));
+    }
+    if (emitOldSemconv && operationType != null) {
+      attributes.put(MESSAGING_OPERATION, operationType.defaultOperationName());
+    }
+    if (emitStableSemconv) {
+      attributes.put(MESSAGING_OPERATION_NAME, operationName);
+      if (operationType != null) {
+        attributes.put(MESSAGING_OPERATION_TYPE, operationType.value());
+      }
     }
   }
 
@@ -121,6 +168,13 @@ public void onEnd(
       @Nullable Throwable error) {
     attributes.put(MESSAGING_MESSAGE_ID, getter.getMessageId(request, response));
     attributes.put(MESSAGING_BATCH_MESSAGE_COUNT, getter.getBatchMessageCount(request, response));
+    if (supportsStableSemconv && emitStableMessagingSemconv()) {
+      String errorType = getter.getErrorType(request, response, error);
+      if (errorType == null && error != null) {
+        errorType = error.getClass().getName();
+      }
+      attributes.put(ERROR_TYPE, errorType);
+    }
 
     for (String name : capturedHeaders) {
       List<String> values = getter.getMessageHeader(request, name);
@@ -137,17 +191,21 @@ public void onEnd(
   @Override
   @Nullable
   public SpanKey internalGetSpanKey() {
-    if (operation == null) {
+    if (operationType == null) {
       return null;
     }
 
-    switch (operation) {
-      case PUBLISH:
+    switch (operationType) {
+      case CREATE:
+        return SpanKey.PRODUCER_CREATE;
+      case SEND:
         return SpanKey.PRODUCER;
       case RECEIVE:
         return SpanKey.CONSUMER_RECEIVE;
       case PROCESS:
         return SpanKey.CONSUMER_PROCESS;
+      case SETTLE:
+        return SpanKey.CONSUMER_SETTLE;
     }
     throw new IllegalStateException("Can't possibly happen");
   }

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

@@ -6,24 +6,40 @@
 package io.opentelemetry.instrumentation.api.incubator.semconv.messaging;
 
 import static java.util.Collections.emptyList;
+import static java.util.Objects.requireNonNull;
 
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import javax.annotation.Nullable;
 
 /** A builder of {@link MessagingAttributesExtractor}. */
 public final class MessagingAttributesExtractorBuilder<REQUEST, RESPONSE> {
 
   final MessagingAttributesGetter<REQUEST, RESPONSE> getter;
-  final MessageOperation operation;
+  @Nullable MessagingOperationType operationType;
+  @Nullable String operationName;
+  private final boolean supportsStableSemconv;
   List<String> capturedHeaders = emptyList();
 
   MessagingAttributesExtractorBuilder(
-      MessagingAttributesGetter<REQUEST, RESPONSE> getter, MessageOperation operation) {
+      MessagingAttributesGetter<REQUEST, RESPONSE> getter,
+      @Nullable MessagingOperationType operationType,
+      boolean supportsStableSemconv) {
     this.getter = getter;
-    this.operation = operation;
+    this.operationType = operationType;
+    this.operationName = operationType == null ? null : operationType.defaultOperationName();
+    this.supportsStableSemconv = supportsStableSemconv;
+  }
+
+  /** Configures the system-specific operation name emitted as {@code messaging.operation.name}. */
+  @CanIgnoreReturnValue
+  public MessagingAttributesExtractorBuilder<REQUEST, RESPONSE> setOperationName(
+      String operationName) {
+    this.operationName = requireNonNull(operationName, "operationName");
+    return this;
   }
 
   /**
@@ -47,6 +63,7 @@ public MessagingAttributesExtractorBuilder<REQUEST, RESPONSE> setCapturedHeaders
    * MessagingAttributesExtractorBuilder}.
    */
   public AttributesExtractor<REQUEST, RESPONSE> build() {
-    return new MessagingAttributesExtractor<>(getter, operation, capturedHeaders);
+    return new MessagingAttributesExtractor<>(
+        getter, operationType, operationName, supportsStableSemconv, capturedHeaders);
   }
 }

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

@@ -55,6 +55,21 @@ default String getDestinationPartitionId(REQUEST request) {
     return null;
   }
 
+  /**
+   * Returns a description of a class of error the operation ended with.
+   *
+   * <p>If this method returns {@code null}, the exception class name (if any) will be used as error
+   * type.
+   *
+   * <p>The cardinality of the error type should be low. The instrumentations implementing this
+   * method are recommended to document the custom values they support.
+   */
+  @Nullable
+  default String getErrorType(
+      REQUEST request, @Nullable RESPONSE response, @Nullable Throwable error) {
+    return null;
+  }
+
   /**
    * Extracts all values of header named {@code name} from the request, or an empty list if there
    * were none.

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.api.incubator.semconv.messaging;
+
+/**
+ * Represents a <a
+ * href="https://github.com/open-telemetry/semantic-conventions/blob/v1.43.0/docs/messaging/messaging-spans.md#operation-types">messaging
+ * operation type</a>.
+ */
+public enum MessagingOperationType {
+  CREATE("create", "create"),
+  SEND("send", "publish"),
+  RECEIVE("receive", "receive"),
+  PROCESS("process", "process"),
+  SETTLE("settle", "settle");
+
+  private final String value;
+  private final String defaultOperationName;
+
+  MessagingOperationType(String value, String defaultOperationName) {
+    this.value = value;
+    this.defaultOperationName = defaultOperationName;
+  }
+
+  String value() {
+    return value;
+  }
+
+  String defaultOperationName() {
+    return defaultOperationName;
+  }
+}