Pin messaging semantic conventions to v1.43 behind preview

Author: trask

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

@@ -9,7 +9,7 @@
 
 /**
  * Represents type of <a
- * href="https://github.com/open-telemetry/semantic-conventions/blob/main/docs/messaging/messaging-spans.md#operation-names">operations</a>
+ * href="https://github.com/open-telemetry/semantic-conventions/blob/v1.43.0/docs/messaging/messaging-spans.md#operation-types">operations</a>
  * that may be used in a messaging system.
  */
 public enum MessageOperation {
@@ -18,9 +18,8 @@ public enum MessageOperation {
   PROCESS;
 
   /**
-   * 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>.
+   * Returns the legacy operation name. The v1.43 operation name defaults to this value unless an
+   * instrumentation supplies a system-specific override.
    */
   String operationName() {
     return name().toLowerCase(Locale.ROOT);

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,6 +57,10 @@ 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");
 
@@ -61,26 +71,51 @@ public final class MessagingAttributesExtractor<REQUEST, RESPONSE>
    * with default configuration.
    */
   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();
   }
 
+  /**
+   * Creates the messaging attributes extractor with a system-specific v1.43 operation name.
+   *
+   * <p>The {@code operationName} is emitted as {@code messaging.operation.name}. The legacy {@code
+   * messaging.operation} value remains derived from {@code operation}.
+   */
+  public static <REQUEST, RESPONSE> AttributesExtractor<REQUEST, RESPONSE> create(
+      MessagingAttributesGetter<REQUEST, RESPONSE> getter,
+      MessageOperation operation,
+      String operationName) {
+    return builder(getter, operation, operationName).build();
+  }
+
   /**
    * Returns a new {@link MessagingAttributesExtractorBuilder} for the given {@link MessageOperation
    * operation} that can be used to configure the messaging attributes extractor.
    */
   public static <REQUEST, RESPONSE> MessagingAttributesExtractorBuilder<REQUEST, RESPONSE> builder(
-      MessagingAttributesGetter<REQUEST, RESPONSE> getter, MessageOperation operation) {
+      MessagingAttributesGetter<REQUEST, RESPONSE> getter, @Nullable MessageOperation operation) {
     return new MessagingAttributesExtractorBuilder<>(getter, operation);
   }
 
+  /**
+   * Returns a new {@link MessagingAttributesExtractorBuilder} with a system-specific v1.43
+   * operation name.
+   */
+  public static <REQUEST, RESPONSE> MessagingAttributesExtractorBuilder<REQUEST, RESPONSE> builder(
+      MessagingAttributesGetter<REQUEST, RESPONSE> getter,
+      MessageOperation operation,
+      String operationName) {
+    return new MessagingAttributesExtractorBuilder<>(
+        getter, MessagingOperation.create(operation, operationName));
+  }
+
   private final MessagingAttributesGetter<REQUEST, RESPONSE> getter;
-  private final MessageOperation operation;
+  @Nullable private final MessagingOperation operation;
   private final List<String> capturedHeaders;
 
   MessagingAttributesExtractor(
       MessagingAttributesGetter<REQUEST, RESPONSE> getter,
-      MessageOperation operation,
+      @Nullable MessagingOperation operation,
       List<String> capturedHeaders) {
     this.getter = getter;
     this.operation = operation;
@@ -93,7 +128,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 +146,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 (emitOldMessagingSemconv()) {
+      attributes.put(MESSAGING_CLIENT_ID_OLD, getter.getClientId(request));
+    }
+    if (emitStableMessagingSemconv()) {
+      attributes.put(MESSAGING_CLIENT_ID, getter.getClientId(request));
+    }
     if (operation != null) {
-      attributes.put(MESSAGING_OPERATION, operation.operationName());
+      if (emitOldMessagingSemconv()) {
+        attributes.put(MESSAGING_OPERATION, operation.operation().operationName());
+      }
+      if (emitStableMessagingSemconv()) {
+        attributes.put(MESSAGING_OPERATION_NAME, operation.name());
+        attributes.put(MESSAGING_OPERATION_TYPE, operation.type());
+      }
     }
   }
 
@@ -121,6 +172,9 @@ 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() && error != null) {
+      attributes.put(ERROR_TYPE, error.getClass().getName());
+    }
 
     for (String name : capturedHeaders) {
       List<String> values = getter.getMessageHeader(request, name);
@@ -141,7 +195,7 @@ public SpanKey internalGetSpanKey() {
       return null;
     }
 
-    switch (operation) {
+    switch (operation.operation()) {
       case PUBLISH:
         return SpanKey.PRODUCER;
       case RECEIVE:

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

@@ -12,16 +12,22 @@
 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 final MessagingOperation operation;
   List<String> capturedHeaders = emptyList();
 
   MessagingAttributesExtractorBuilder(
-      MessagingAttributesGetter<REQUEST, RESPONSE> getter, MessageOperation operation) {
+      MessagingAttributesGetter<REQUEST, RESPONSE> getter, @Nullable MessageOperation operation) {
+    this(getter, MessagingOperation.createNullable(operation));
+  }
+
+  MessagingAttributesExtractorBuilder(
+      MessagingAttributesGetter<REQUEST, RESPONSE> getter, @Nullable MessagingOperation operation) {
     this.getter = getter;
     this.operation = operation;
   }

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

@@ -5,6 +5,8 @@
 
 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 java.util.concurrent.TimeUnit.SECONDS;
 import static java.util.logging.Level.FINE;
 
@@ -23,10 +25,11 @@
 import io.opentelemetry.instrumentation.api.instrumenter.OperationMetrics;
 import io.opentelemetry.instrumentation.api.internal.OperationMetricsUtil;
 import java.util.logging.Logger;
+import javax.annotation.Nullable;
 
 /**
  * {@link OperationListener} which keeps track of <a
- * href="https://github.com/open-telemetry/semantic-conventions/blob/v1.26.0/docs/messaging/messaging-metrics.md#consumer-metrics">Consumer
+ * href="https://github.com/open-telemetry/semantic-conventions/blob/v1.43.0/docs/messaging/messaging-metrics.md#consumer-metrics">consumer
  * metrics</a>.
  */
 public final class MessagingConsumerMetrics implements OperationListener {
@@ -39,26 +42,17 @@ public final class MessagingConsumerMetrics implements OperationListener {
       ContextKey.named("messaging-consumer-metrics-state");
   private static final Logger logger = Logger.getLogger(MessagingConsumerMetrics.class.getName());
 
-  private final DoubleHistogram receiveDurationHistogram;
-  private final LongCounter receiveMessageCount;
+  @Nullable private final DoubleHistogram receiveDurationHistogram;
+  @Nullable private final LongCounter receiveMessageCount;
+  @Nullable private final DoubleHistogram clientOperationDurationHistogram;
+  @Nullable private final LongCounter consumedMessagesCounter;
 
   private MessagingConsumerMetrics(Meter meter) {
-    DoubleHistogramBuilder durationBuilder =
-        meter
-            .histogramBuilder("messaging.receive.duration")
-            .setDescription("Measures the duration of receive operation.")
-            .setExplicitBucketBoundariesAdvice(MessagingMetricsAdvice.DURATION_SECONDS_BUCKETS)
-            .setUnit("s");
-    MessagingMetricsAdvice.applyReceiveDurationAdvice(durationBuilder);
-    receiveDurationHistogram = durationBuilder.build();
-
-    LongCounterBuilder longCounterBuilder =
-        meter
-            .counterBuilder("messaging.receive.messages")
-            .setDescription("Measures the number of received messages.")
-            .setUnit("{message}");
-    MessagingMetricsAdvice.applyReceiveMessagesAdvice(longCounterBuilder);
-    receiveMessageCount = longCounterBuilder.build();
+    receiveDurationHistogram = emitOldMessagingSemconv() ? buildReceiveDuration(meter) : null;
+    receiveMessageCount = emitOldMessagingSemconv() ? buildReceiveMessages(meter) : null;
+    clientOperationDurationHistogram =
+        emitStableMessagingSemconv() ? buildClientOperationDuration(meter) : null;
+    consumedMessagesCounter = emitStableMessagingSemconv() ? buildConsumedMessages(meter) : null;
   }
 
   public static OperationMetrics get() {
@@ -85,11 +79,25 @@ public void onEnd(Context context, Attributes endAttributes, long endNanos) {
     }
 
     Attributes attributes = state.startAttributes().toBuilder().putAll(endAttributes).build();
-    receiveDurationHistogram.record(
-        (endNanos - state.startTimeNanos()) / NANOS_PER_S, attributes, context);
+    double duration = (endNanos - state.startTimeNanos()) / NANOS_PER_S;
+    if (receiveDurationHistogram != null) {
+      receiveDurationHistogram.record(duration, attributes, context);
+    }
+    Attributes filteredAttributes =
+        clientOperationDurationHistogram != null || consumedMessagesCounter != null
+            ? MessagingMetricsAdvice.filterAttributes(attributes)
+            : attributes;
+    if (clientOperationDurationHistogram != null) {
+      clientOperationDurationHistogram.record(duration, filteredAttributes, context);
+    }
 
     long receiveMessagesCount = getReceiveMessagesCount(state.startAttributes(), endAttributes);
-    receiveMessageCount.add(receiveMessagesCount, attributes, context);
+    if (receiveMessageCount != null) {
+      receiveMessageCount.add(receiveMessagesCount, attributes, context);
+    }
+    if (consumedMessagesCounter != null) {
+      consumedMessagesCounter.add(receiveMessagesCount, filteredAttributes, context);
+    }
   }
 
   private static long getReceiveMessagesCount(Attributes... attributesList) {
@@ -102,6 +110,49 @@ private static long getReceiveMessagesCount(Attributes... attributesList) {
     return 1;
   }
 
+  private static DoubleHistogram buildReceiveDuration(Meter meter) {
+    DoubleHistogramBuilder builder =
+        meter
+            .histogramBuilder("messaging.receive.duration")
+            .setDescription("Measures the duration of receive operation.")
+            .setExplicitBucketBoundariesAdvice(MessagingMetricsAdvice.DURATION_SECONDS_BUCKETS)
+            .setUnit("s");
+    MessagingMetricsAdvice.applyOldDurationAdvice(builder);
+    return builder.build();
+  }
+
+  private static LongCounter buildReceiveMessages(Meter meter) {
+    LongCounterBuilder builder =
+        meter
+            .counterBuilder("messaging.receive.messages")
+            .setDescription("Measures the number of received messages.")
+            .setUnit("{message}");
+    MessagingMetricsAdvice.applyOldMessagesAdvice(builder);
+    return builder.build();
+  }
+
+  private static DoubleHistogram buildClientOperationDuration(Meter meter) {
+    DoubleHistogramBuilder builder =
+        meter
+            .histogramBuilder("messaging.client.operation.duration")
+            .setDescription(
+                "Duration of messaging operation initiated by a producer or consumer client.")
+            .setExplicitBucketBoundariesAdvice(MessagingMetricsAdvice.DURATION_SECONDS_BUCKETS)
+            .setUnit("s");
+    MessagingMetricsAdvice.applyClientOperationDurationAdvice(builder);
+    return builder.build();
+  }
+
+  private static LongCounter buildConsumedMessages(Meter meter) {
+    LongCounterBuilder builder =
+        meter
+            .counterBuilder("messaging.client.consumed.messages")
+            .setDescription("Number of messages that were delivered to the application.")
+            .setUnit("{message}");
+    MessagingMetricsAdvice.applyConsumedMessagesAdvice(builder);
+    return builder.build();
+  }
+
   @AutoValue
   abstract static class State {