Migrate Kafka messaging telemetry to semconv v1.43 preview

CHANGELOG.md

@@ -4,6 +4,8 @@
 
 ### 🚫 Deprecations
 
+- Deprecate `MessageOperation` in favor of `MessagingOperationType`.
+  ([#19233](https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19233))
 - Deprecate `DeclarativeConfigPropertiesBridgeBuilder`. Read declarative component configuration
   through `DeclarativeConfigProperties` directly. To expose `ConfigProperties` through the
   declarative configuration API, use `ConfigPropertiesBackedConfigProvider`.

instrumentation-api-incubator/build.gradle.kts

@@ -92,13 +92,23 @@ 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)
+  }
+
+  val testV3Preview = register<Test>("testV3Preview") {
+    testClassesDirs = sourceSets.test.get().output.classesDirs
+    classpath = sourceSets.test.get().runtimeClasspath
+    jvmArgs("-Dotel.instrumentation.common.v3-preview=true")
+    jvmArgs("-Dotel.semconv-stability.preview=messaging")
     inputs.dir(jflexOutputDir)
   }
 
@@ -117,6 +127,12 @@ tasks {
   }
 
   check {
-    dependsOn(testStableSemconv, testBothSemconv, testExceptionSignalLogs, testExceptionSignalLogsDup)
+    dependsOn(
+      testStableSemconv,
+      testBothSemconv,
+      testV3Preview,
+      testExceptionSignalLogs,
+      testExceptionSignalLogsDup,
+    )
   }
 }

instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/db/SqlQueryAnalyzer.java

@@ -78,7 +78,9 @@ private static SqlQuery analyzeWithSummaryImpl(String query, SqlDialect dialect)
 
   // visible for tests
   static boolean isCached(String query, SqlDialect dialect) {
-    return sqlToQueryCache.get(CacheKey.create(query, dialect)) != null;
+    Cache<CacheKey, SqlQuery> cache =
+        SemconvStability.v3Preview() ? sqlToQueryCacheWithSummary : sqlToQueryCache;
+    return cache.get(CacheKey.create(query, dialect)) != null;
   }
 
   @AutoValue

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

@@ -5,24 +5,24 @@
 
 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.
+ *
+ * @deprecated Use {@link MessagingOperationType}. Will be removed in 3.0.
  */
+@Deprecated // to be removed in 3.0
 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,39 +57,65 @@ 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 operation type. */
+  public static <REQUEST, RESPONSE> AttributesExtractor<REQUEST, RESPONSE> create(
+      MessagingAttributesGetter<REQUEST, RESPONSE> getter,
+      @Nullable MessagingOperationType operationType) {
+    return builder(getter, operationType).build();
+  }
+
   /**
-   * Creates the messaging attributes extractor for the given {@link MessageOperation operation}
-   * with default configuration.
+   * @deprecated Use {@link #create(MessagingAttributesGetter, MessagingOperationType)}. Will be
+   *     removed in 3.0.
    */
+  @Deprecated // to be removed in 3.0
   public static <REQUEST, RESPONSE> AttributesExtractor<REQUEST, RESPONSE> create(
-      MessagingAttributesGetter<REQUEST, RESPONSE> getter, MessageOperation operation) {
-    return builder(getter, operation).build();
+      MessagingAttributesGetter<REQUEST, RESPONSE> getter, @Nullable MessageOperation operation) {
+    return create(getter, operation == null ? null : operation.type());
+  }
+
+  /**
+   * Returns a new {@link MessagingAttributesExtractorBuilder} configured for the given operation
+   * type.
+   */
+  public static <REQUEST, RESPONSE> MessagingAttributesExtractorBuilder<REQUEST, RESPONSE> builder(
+      MessagingAttributesGetter<REQUEST, RESPONSE> getter,
+      @Nullable MessagingOperationType operationType) {
+    return new MessagingAttributesExtractorBuilder<>(getter, operationType);
   }
 
   /**
-   * Returns a new {@link MessagingAttributesExtractorBuilder} for the given {@link MessageOperation
-   * operation} that can be used to configure the messaging attributes extractor.
+   * @deprecated Use {@link #builder(MessagingAttributesGetter, MessagingOperationType)}. Will be
+   *     removed in 3.0.
    */
+  @Deprecated // to be removed in 3.0
   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 builder(getter, operation == null ? null : operation.type());
   }
 
   private final MessagingAttributesGetter<REQUEST, RESPONSE> getter;
-  private final MessageOperation operation;
+  @Nullable private final MessagingOperationType operationType;
+  @Nullable private final String operationName;
   private final List<String> capturedHeaders;
 
   MessagingAttributesExtractor(
       MessagingAttributesGetter<REQUEST, RESPONSE> getter,
-      MessageOperation operation,
+      @Nullable MessagingOperationType operationType,
+      @Nullable String operationName,
       List<String> capturedHeaders) {
     this.getter = getter;
-    this.operation = operation;
+    this.operationType = operationType;
+    this.operationName = operationName;
     this.capturedHeaders = new ArrayList<>(capturedHeaders);
   }
 
@@ -93,7 +125,12 @@ public void onStart(AttributesBuilder attributes, Context parentContext, REQUEST
     boolean isTemporaryDestination = getter.isTemporaryDestination(request);
     if (isTemporaryDestination) {
       attributes.put(MESSAGING_DESTINATION_TEMPORARY, true);
-      attributes.put(MESSAGING_DESTINATION_NAME, TEMP_DESTINATION_NAME);
+      if (emitStableMessagingSemconv()) {
+        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));
@@ -106,9 +143,20 @@ public void onStart(AttributesBuilder attributes, Context parentContext, REQUEST
     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 (emitOldMessagingSemconv()) {
+      attributes.put(MESSAGING_CLIENT_ID_OLD, getter.getClientId(request));
+    }
+    if (emitStableMessagingSemconv()) {
+      attributes.put(MESSAGING_CLIENT_ID, getter.getClientId(request));
+    }
+    if (emitOldMessagingSemconv() && operationType != null) {
+      attributes.put(MESSAGING_OPERATION, operationType.defaultOperationName());
+    }
+    if (emitStableMessagingSemconv()) {
+      attributes.put(MESSAGING_OPERATION_NAME, operationName);
+      if (operationType != null) {
+        attributes.put(MESSAGING_OPERATION_TYPE, operationType.value());
+      }
     }
   }
 
@@ -121,6 +169,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 (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 +192,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,37 @@
 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;
   List<String> capturedHeaders = emptyList();
 
   MessagingAttributesExtractorBuilder(
-      MessagingAttributesGetter<REQUEST, RESPONSE> getter, MessageOperation operation) {
+      MessagingAttributesGetter<REQUEST, RESPONSE> getter,
+      @Nullable MessagingOperationType operationType) {
     this.getter = getter;
-    this.operation = operation;
+    this.operationType = operationType;
+    this.operationName = operationType == null ? null : operationType.defaultOperationName();
+  }
+
+  /** 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 +60,7 @@ public MessagingAttributesExtractorBuilder<REQUEST, RESPONSE> setCapturedHeaders
    * MessagingAttributesExtractorBuilder}.
    */
   public AttributesExtractor<REQUEST, RESPONSE> build() {
-    return new MessagingAttributesExtractor<>(getter, operation, capturedHeaders);
+    return new MessagingAttributesExtractor<>(
+        getter, operationType, operationName, 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.