Add Pulsar Consumer metrics

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

@@ -36,6 +36,9 @@ public final class MessagingAttributesExtractor<REQUEST, RESPONSE>
       AttributeKey.booleanKey("messaging.destination.anonymous");
   private static final AttributeKey<String> MESSAGING_DESTINATION_NAME =
       AttributeKey.stringKey("messaging.destination.name");
+
+  private static final AttributeKey<String> MESSAGING_DESTINATION_PARTITION_ID =
+      AttributeKey.stringKey("messaging.destination.partition.id");
   private static final AttributeKey<String> MESSAGING_DESTINATION_TEMPLATE =
       AttributeKey.stringKey("messaging.destination.template");
   private static final AttributeKey<Boolean> MESSAGING_DESTINATION_TEMPORARY =
@@ -122,6 +125,8 @@ public void onEnd(
     internalSet(attributes, MESSAGING_MESSAGE_ID, getter.getMessageId(request, response));
     internalSet(
         attributes, MESSAGING_BATCH_MESSAGE_COUNT, getter.getBatchMessageCount(request, response));
+    internalSet(
+        attributes, MESSAGING_DESTINATION_PARTITION_ID, getter.getDestinationPartitionId(request));
 
     for (String name : capturedHeaders) {
       List<String> values = getter.getMessageHeader(request, name);

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

@@ -62,6 +62,11 @@ default Long getMessagePayloadCompressedSize(REQUEST request) {
   @Nullable
   Long getBatchMessageCount(REQUEST request, @Nullable RESPONSE response);
 
+  @Nullable
+  default String getDestinationPartitionId(REQUEST request) {
+    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/MessagingConsumerMetrics.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.api.incubator.semconv.messaging;
+
+import static java.util.logging.Level.FINE;
+
+import com.google.auto.value.AutoValue;
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.DoubleHistogram;
+import io.opentelemetry.api.metrics.DoubleHistogramBuilder;
+import io.opentelemetry.api.metrics.LongCounter;
+import io.opentelemetry.api.metrics.LongCounterBuilder;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.ContextKey;
+import io.opentelemetry.instrumentation.api.instrumenter.OperationListener;
+import io.opentelemetry.instrumentation.api.instrumenter.OperationMetrics;
+import io.opentelemetry.instrumentation.api.internal.OperationMetricsUtil;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Logger;
+
+/**
+ * {@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
+ * metrics</a>.
+ */
+public final class MessagingConsumerMetrics implements OperationListener {
+  private static final double NANOS_PER_S = TimeUnit.SECONDS.toNanos(1);
+
+  // copied from MessagingIncubatingAttributes
+  private static final AttributeKey<Long> MESSAGING_BATCH_MESSAGE_COUNT =
+      AttributeKey.longKey("messaging.batch.message_count");
+  private static final ContextKey<MessagingConsumerMetrics.State> MESSAGING_CONSUMER_METRICS_STATE =
+      ContextKey.named("messaging-consumer-metrics-state");
+  private static final Logger logger = Logger.getLogger(MessagingConsumerMetrics.class.getName());
+
+  private final DoubleHistogram receiveDurationHistogram;
+  private final LongCounter receiveMessageCount;
+
+  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();
+  }
+
+  public static OperationMetrics get() {
+    return OperationMetricsUtil.create("messaging consumer", MessagingConsumerMetrics::new);
+  }
+
+  @Override
+  @CanIgnoreReturnValue
+  public Context onStart(Context context, Attributes startAttributes, long startNanos) {
+    return context.with(
+        MESSAGING_CONSUMER_METRICS_STATE,
+        new AutoValue_MessagingConsumerMetrics_State(startAttributes, startNanos));
+  }
+
+  @Override
+  public void onEnd(Context context, Attributes endAttributes, long endNanos) {
+    MessagingConsumerMetrics.State state = context.get(MESSAGING_CONSUMER_METRICS_STATE);
+    if (state == null) {
+      logger.log(
+          FINE,
+          "No state present when ending context {0}. Cannot record consumer receive metrics.",
+          context);
+      return;
+    }
+
+    Attributes attributes = state.startAttributes().toBuilder().putAll(endAttributes).build();
+    receiveDurationHistogram.record(
+        (endNanos - state.startTimeNanos()) / NANOS_PER_S, attributes, context);
+
+    Long receiveMessagesCount = getReceiveMessagesCount(state.startAttributes(), endAttributes);
+    if (receiveMessagesCount != null) {
+      receiveMessageCount.add(receiveMessagesCount, attributes, context);
+    }
+  }
+
+  private static Long getReceiveMessagesCount(Attributes... attributesList) {
+    for (Attributes attributes : attributesList) {
+      Long value = attributes.get(MESSAGING_BATCH_MESSAGE_COUNT);
+      if (value != null) {
+        return value;
+      }
+    }
+    return null;
+  }
+
+  @AutoValue
+  abstract static class State {
+
+    abstract Attributes startAttributes();
+
+    abstract long startTimeNanos();
+  }
+}

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

@@ -10,7 +10,9 @@
 
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.incubator.metrics.ExtendedDoubleHistogramBuilder;
+import io.opentelemetry.api.incubator.metrics.ExtendedLongCounterBuilder;
 import io.opentelemetry.api.metrics.DoubleHistogramBuilder;
+import io.opentelemetry.api.metrics.LongCounterBuilder;
 import io.opentelemetry.semconv.ErrorAttributes;
 import io.opentelemetry.semconv.ServerAttributes;
 import java.util.List;
@@ -27,23 +29,41 @@ final class MessagingMetricsAdvice {
       AttributeKey.stringKey("messaging.destination.name");
   private static final AttributeKey<String> MESSAGING_OPERATION =
       AttributeKey.stringKey("messaging.operation");
-  private static final AttributeKey<Long> MESSAGING_BATCH_MESSAGE_COUNT =
-      AttributeKey.longKey("messaging.batch.message_count");
+  private static final AttributeKey<String> MESSAGING_DESTINATION_PARTITION_ID =
+      AttributeKey.stringKey("messaging.destination.partition.id");
+  private static final AttributeKey<String> MESSAGING_DESTINATION_TEMPLATE =
+      AttributeKey.stringKey("messaging.destination.template");
+
+  private static final List<AttributeKey<?>> MESSAGING_ATTRIBUTES =
+      asList(
+          MESSAGING_SYSTEM,
+          MESSAGING_DESTINATION_NAME,
+          MESSAGING_OPERATION,
+          MESSAGING_DESTINATION_PARTITION_ID,
+          MESSAGING_DESTINATION_TEMPLATE,
+          ErrorAttributes.ERROR_TYPE,
+          ServerAttributes.SERVER_PORT,
+          ServerAttributes.SERVER_ADDRESS);
 
   static void applyPublishDurationAdvice(DoubleHistogramBuilder builder) {
     if (!(builder instanceof ExtendedDoubleHistogramBuilder)) {
       return;
     }
-    ((ExtendedDoubleHistogramBuilder) builder)
-        .setAttributesAdvice(
-            asList(
-                MESSAGING_SYSTEM,
-                MESSAGING_DESTINATION_NAME,
-                MESSAGING_OPERATION,
-                MESSAGING_BATCH_MESSAGE_COUNT,
-                ErrorAttributes.ERROR_TYPE,
-                ServerAttributes.SERVER_PORT,
-                ServerAttributes.SERVER_ADDRESS));
+    ((ExtendedDoubleHistogramBuilder) builder).setAttributesAdvice(MESSAGING_ATTRIBUTES);
+  }
+
+  static void applyReceiveDurationAdvice(DoubleHistogramBuilder builder) {
+    if (!(builder instanceof ExtendedDoubleHistogramBuilder)) {
+      return;
+    }
+    ((ExtendedDoubleHistogramBuilder) builder).setAttributesAdvice(MESSAGING_ATTRIBUTES);
+  }
+
+  static void applyReceiveMessagesAdvice(LongCounterBuilder builder) {
+    if (!(builder instanceof ExtendedLongCounterBuilder)) {
+      return;
+    }
+    ((ExtendedLongCounterBuilder) builder).setAttributesAdvice(MESSAGING_ATTRIBUTES);
   }
 
   private MessagingMetricsAdvice() {}

instrumentation-api-incubator/src/test/java/io/opentelemetry/instrumentation/api/incubator/semconv/messaging/MessagingProducerMetricsTest.java

@@ -50,7 +50,7 @@ void collectsMetrics() {
     Attributes responseAttributes =
         Attributes.builder()
             .put(MessagingIncubatingAttributes.MESSAGING_MESSAGE_ID, "1:1:0:0")
-            .put(MessagingIncubatingAttributes.MESSAGING_BATCH_MESSAGE_COUNT, 2)
+            .put(MessagingIncubatingAttributes.MESSAGING_DESTINATION_PARTITION_ID, "1")
             .build();
 
     Context parent =
@@ -90,6 +90,10 @@ void collectsMetrics() {
                                             equalTo(
                                                 MessagingIncubatingAttributes.MESSAGING_SYSTEM,
                                                 "pulsar"),
+                                            equalTo(
+                                                MessagingIncubatingAttributes
+                                                    .MESSAGING_DESTINATION_PARTITION_ID,
+                                                "1"),
                                             equalTo(
                                                 MessagingIncubatingAttributes
                                                     .MESSAGING_DESTINATION_NAME,

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

@@ -11,6 +11,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 import javax.annotation.Nullable;
+import org.apache.pulsar.common.naming.TopicName;
 
 enum PulsarBatchMessagingAttributesGetter
     implements MessagingAttributesGetter<PulsarBatchRequest, Void> {
@@ -81,6 +82,16 @@ public Long getBatchMessageCount(PulsarBatchRequest request, @Nullable Void unus
     return (long) request.getMessages().size();
   }
 
+  @Nullable
+  @Override
+  public String getDestinationPartitionId(PulsarBatchRequest request) {
+    int partitionIndex = TopicName.getPartitionIndex(request.getDestination());
+    if (partitionIndex == -1) {
+      return null;
+    }
+    return String.valueOf(partitionIndex);
+  }
+
   @Override
   public List<String> getMessageHeader(PulsarBatchRequest request, String name) {
     return StreamSupport.stream(request.getMessages().spliterator(), false)

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

@@ -10,6 +10,7 @@
 import io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.UrlParser.UrlData;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.Messages;
+import org.apache.pulsar.common.naming.TopicName;
 
 public final class PulsarBatchRequest extends BasePulsarRequest {
   private final Messages<?> messages;
@@ -30,7 +31,10 @@ private static String getTopicName(Messages<?> messages) {
       if (topicName == null) {
         topicName = name;
       } else if (!topicName.equals(name)) {
-        return null;
+        // this is a partitioned topic
+        // persistent://public/default/test-partition-0 persistent://public/default/test-partition-1
+        // return persistent://public/default/test
+        return TopicName.get(topicName).getPartitionedTopicName();
       }
     }
     return topicName;

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

@@ -12,6 +12,7 @@
 import java.util.List;
 import javax.annotation.Nullable;
 import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.common.naming.TopicName;
 
 enum PulsarMessagingAttributesGetter implements MessagingAttributesGetter<PulsarRequest, Void> {
   INSTANCE;
@@ -83,6 +84,16 @@ public Long getBatchMessageCount(PulsarRequest request, @Nullable Void unused) {
     return null;
   }
 
+  @Nullable
+  @Override
+  public String getDestinationPartitionId(PulsarRequest request) {
+    int partitionIndex = TopicName.getPartitionIndex(request.getDestination());
+    if (partitionIndex == -1) {
+      return null;
+    }
+    return String.valueOf(partitionIndex);
+  }
+
   @Override
   public List<String> getMessageHeader(PulsarRequest request, String name) {
     String value = request.getMessage().getProperty(name);

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

@@ -13,6 +13,7 @@
 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.MessagingConsumerMetrics;
 import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingProducerMetrics;
 import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanNameExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
@@ -76,6 +77,7 @@ private static Instrumenter<PulsarRequest, Void> createConsumerReceiveInstrument
                 MessagingSpanNameExtractor.create(getter, MessageOperation.RECEIVE))
             .addAttributesExtractor(
                 createMessagingAttributesExtractor(getter, MessageOperation.RECEIVE))
+            .addOperationMetrics(MessagingConsumerMetrics.get())
             .addAttributesExtractor(
                 ServerAttributesExtractor.create(new PulsarNetClientAttributesGetter()));
 
@@ -101,6 +103,7 @@ private static Instrumenter<PulsarBatchRequest, Void> createConsumerBatchReceive
         .addAttributesExtractor(
             ServerAttributesExtractor.create(new PulsarNetClientAttributesGetter()))
         .addSpanLinksExtractor(new PulsarBatchRequestSpanLinksExtractor(PROPAGATOR))
+        .addOperationMetrics(MessagingConsumerMetrics.get())
         .buildInstrumenter(SpanKindExtractor.alwaysConsumer());
   }
 
@@ -189,7 +192,7 @@ private static Context startAndEndConsumerReceive(
       Timer timer,
       Consumer<?> consumer,
       Throwable throwable) {
-    if (messages == null) {
+    if (messages == null || messages.size() == 0) {
       return null;
     }
     String brokerUrl = VirtualFieldStore.extract(consumer);