Add Pulsar MessagingProducerMetrics

instrumentation-api-incubator/build.gradle.kts

@@ -21,6 +21,7 @@ dependencies {
   testImplementation(project(":testing-common"))
   testImplementation("io.opentelemetry:opentelemetry-sdk")
   testImplementation("io.opentelemetry:opentelemetry-sdk-testing")
+  testImplementation("io.opentelemetry.semconv:opentelemetry-semconv-incubating")
 }
 
 tasks {

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

@@ -0,0 +1,50 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.api.incubator.semconv.messaging;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.unmodifiableList;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.incubator.metrics.ExtendedDoubleHistogramBuilder;
+import io.opentelemetry.api.metrics.DoubleHistogramBuilder;
+import io.opentelemetry.semconv.ErrorAttributes;
+import io.opentelemetry.semconv.ServerAttributes;
+import java.util.List;
+
+final class MessagingMetricsAdvice {
+  static final List<Double> DURATION_SECONDS_BUCKETS =
+      unmodifiableList(
+          asList(0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0, 7.5, 10.0));
+
+  // copied from MessagingIncubatingAttributes
+  private static final AttributeKey<String> MESSAGING_SYSTEM =
+      AttributeKey.stringKey("messaging.system");
+  private static final AttributeKey<String> MESSAGING_DESTINATION_NAME =
+      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");
+
+  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));
+  }
+
+  private MessagingMetricsAdvice() {}
+}

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

@@ -0,0 +1,85 @@
+/*
+ * 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.Attributes;
+import io.opentelemetry.api.metrics.DoubleHistogram;
+import io.opentelemetry.api.metrics.DoubleHistogramBuilder;
+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#metric-messagingpublishduration">Producer
+ * metrics</a>.
+ */
+public final class MessagingProducerMetrics implements OperationListener {
+  private static final double NANOS_PER_S = TimeUnit.SECONDS.toNanos(1);
+
+  private static final ContextKey<MessagingProducerMetrics.State> MESSAGING_PRODUCER_METRICS_STATE =
+      ContextKey.named("messaging-producer-metrics-state");
+  private static final Logger logger = Logger.getLogger(MessagingProducerMetrics.class.getName());
+
+  private final DoubleHistogram publishDurationHistogram;
+
+  private MessagingProducerMetrics(Meter meter) {
+    DoubleHistogramBuilder durationBuilder =
+        meter
+            .histogramBuilder("messaging.publish.duration")
+            .setDescription("Measures the duration of publish operation.")
+            .setExplicitBucketBoundariesAdvice(MessagingMetricsAdvice.DURATION_SECONDS_BUCKETS)
+            .setUnit("s");
+    MessagingMetricsAdvice.applyPublishDurationAdvice(durationBuilder);
+    publishDurationHistogram = durationBuilder.build();
+  }
+
+  public static OperationMetrics get() {
+    return OperationMetricsUtil.create("messaging produce", MessagingProducerMetrics::new);
+  }
+
+  @Override
+  @CanIgnoreReturnValue
+  public Context onStart(Context context, Attributes startAttributes, long startNanos) {
+    return context.with(
+        MESSAGING_PRODUCER_METRICS_STATE,
+        new AutoValue_MessagingProducerMetrics_State(startAttributes, startNanos));
+  }
+
+  @Override
+  public void onEnd(Context context, Attributes endAttributes, long endNanos) {
+    MessagingProducerMetrics.State state = context.get(MESSAGING_PRODUCER_METRICS_STATE);
+    if (state == null) {
+      logger.log(
+          FINE,
+          "No state present when ending context {0}. Cannot record produce publish metrics.",
+          context);
+      return;
+    }
+
+    Attributes attributes = state.startAttributes().toBuilder().putAll(endAttributes).build();
+
+    publishDurationHistogram.record(
+        (endNanos - state.startTimeNanos()) / NANOS_PER_S, attributes, context);
+  }
+
+  @AutoValue
+  abstract static class State {
+
+    abstract Attributes startAttributes();
+
+    abstract long startTimeNanos();
+  }
+}

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

@@ -0,0 +1,122 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.api.incubator.semconv.messaging;
+
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.api.trace.TraceFlags;
+import io.opentelemetry.api.trace.TraceState;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.api.instrumenter.OperationListener;
+import io.opentelemetry.sdk.metrics.SdkMeterProvider;
+import io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions;
+import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader;
+import io.opentelemetry.semconv.ServerAttributes;
+import io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+
+class MessagingProducerMetricsTest {
+
+  private static final double[] DURATION_BUCKETS =
+      MessagingMetricsAdvice.DURATION_SECONDS_BUCKETS.stream().mapToDouble(d -> d).toArray();
+
+  @Test
+  void collectsMetrics() {
+    InMemoryMetricReader metricReader = InMemoryMetricReader.create();
+    SdkMeterProvider meterProvider =
+        SdkMeterProvider.builder().registerMetricReader(metricReader).build();
+
+    OperationListener listener = MessagingProducerMetrics.get().create(meterProvider.get("test"));
+
+    Attributes requestAttributes =
+        Attributes.builder()
+            .put(MessagingIncubatingAttributes.MESSAGING_SYSTEM, "pulsar")
+            .put(
+                MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME,
+                "persistent://public/default/topic")
+            .put(MessagingIncubatingAttributes.MESSAGING_OPERATION, "publish")
+            .put(ServerAttributes.SERVER_PORT, 6650)
+            .put(ServerAttributes.SERVER_ADDRESS, "localhost")
+            .build();
+
+    Attributes responseAttributes =
+        Attributes.builder()
+            .put(MessagingIncubatingAttributes.MESSAGING_MESSAGE_ID, "1:1:0:0")
+            .put(MessagingIncubatingAttributes.MESSAGING_BATCH_MESSAGE_COUNT, 2)
+            .build();
+
+    Context parent =
+        Context.root()
+            .with(
+                Span.wrap(
+                    SpanContext.create(
+                        "ff01020304050600ff0a0b0c0d0e0f00",
+                        "090a0b0c0d0e0f00",
+                        TraceFlags.getSampled(),
+                        TraceState.getDefault())));
+
+    Context context1 = listener.onStart(parent, requestAttributes, nanos(100));
+
+    assertThat(metricReader.collectAllMetrics()).isEmpty();
+
+    Context context2 = listener.onStart(Context.root(), requestAttributes, nanos(150));
+
+    assertThat(metricReader.collectAllMetrics()).isEmpty();
+
+    listener.onEnd(context1, responseAttributes, nanos(250));
+
+    assertThat(metricReader.collectAllMetrics())
+        .satisfiesExactlyInAnyOrder(
+            metric ->
+                OpenTelemetryAssertions.assertThat(metric)
+                    .hasName("messaging.publish.duration")
+                    .hasUnit("s")
+                    .hasDescription("Measures the duration of publish operation.")
+                    .hasHistogramSatisfying(
+                        histogram ->
+                            histogram.hasPointsSatisfying(
+                                point ->
+                                    point
+                                        .hasSum(0.15 /* seconds */)
+                                        .hasAttributesSatisfying(
+                                            equalTo(
+                                                MessagingIncubatingAttributes.MESSAGING_SYSTEM,
+                                                "pulsar"),
+                                            equalTo(
+                                                MessagingIncubatingAttributes
+                                                    .MESSAGING_DESTINATION_NAME,
+                                                "persistent://public/default/topic"),
+                                            equalTo(ServerAttributes.SERVER_PORT, 6650),
+                                            equalTo(ServerAttributes.SERVER_ADDRESS, "localhost"))
+                                        .hasExemplarsSatisfying(
+                                            exemplar ->
+                                                exemplar
+                                                    .hasTraceId("ff01020304050600ff0a0b0c0d0e0f00")
+                                                    .hasSpanId("090a0b0c0d0e0f00"))
+                                        .hasBucketBoundaries(DURATION_BUCKETS))));
+
+    listener.onEnd(context2, responseAttributes, nanos(300));
+
+    assertThat(metricReader.collectAllMetrics())
+        .satisfiesExactlyInAnyOrder(
+            metric ->
+                OpenTelemetryAssertions.assertThat(metric)
+                    .hasName("messaging.publish.duration")
+                    .hasHistogramSatisfying(
+                        histogram ->
+                            histogram.hasPointsSatisfying(
+                                point -> point.hasSum(0.3 /* seconds */))));
+  }
+
+  private static long nanos(int millis) {
+    return TimeUnit.MILLISECONDS.toNanos(millis);
+  }
+}

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.MessagingProducerMetrics;
 import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanNameExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
@@ -136,7 +137,8 @@ private static Instrumenter<PulsarRequest, Void> createProducerInstrumenter() {
             .addAttributesExtractor(
                 createMessagingAttributesExtractor(getter, MessageOperation.PUBLISH))
             .addAttributesExtractor(
-                ServerAttributesExtractor.create(new PulsarNetClientAttributesGetter()));
+                ServerAttributesExtractor.create(new PulsarNetClientAttributesGetter()))
+            .addOperationMetrics(MessagingProducerMetrics.get());
 
     if (InstrumentationConfig.get()
         .getBoolean("otel.instrumentation.pulsar.experimental-span-attributes", false)) {

instrumentation/pulsar/pulsar-2.8/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/AbstractPulsarClientTest.java

@@ -15,13 +15,16 @@
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_ID;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
+import static java.util.Arrays.asList;
+import static java.util.Collections.unmodifiableList;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
 import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
 import io.opentelemetry.sdk.testing.assertj.AttributeAssertion;
+import io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions;
 import io.opentelemetry.sdk.trace.data.LinkData;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import java.time.Duration;
@@ -77,6 +80,12 @@ abstract class AbstractPulsarClientTest {
   private static final AttributeKey<String> MESSAGE_TYPE =
       AttributeKey.stringKey("messaging.pulsar.message.type");
 
+  private static final List<Double> DURATION_SECONDS_BUCKETS =
+      unmodifiableList(
+          asList(0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0, 7.5, 10.0));
+  private static final double[] DURATION_BUCKETS =
+      DURATION_SECONDS_BUCKETS.stream().mapToDouble(d -> d).toArray();
+
   @BeforeAll
   static void beforeAll() throws PulsarClientException {
     pulsar =
@@ -163,6 +172,26 @@ void testConsumeNonPartitionedTopicUsingBatchReceive() throws Exception {
                         .hasParent(trace.getSpan(0))
                         .hasAttributesSatisfyingExactly(
                             batchReceiveAttributes(topic, null, false))));
+
+    assertThat(testing.metrics())
+        .satisfiesExactlyInAnyOrder(
+            metric ->
+                OpenTelemetryAssertions.assertThat(metric)
+                    .hasName("messaging.publish.duration")
+                    .hasUnit("s")
+                    .hasDescription("Measures the duration of publish operation.")
+                    .hasHistogramSatisfying(
+                        histogram ->
+                            histogram.hasPointsSatisfying(
+                                point ->
+                                    point
+                                        .hasSumGreaterThan(0.0)
+                                        .hasAttributesSatisfying(
+                                            equalTo(MESSAGING_SYSTEM, "pulsar"),
+                                            equalTo(MESSAGING_DESTINATION_NAME, topic),
+                                            equalTo(SERVER_PORT, brokerPort),
+                                            equalTo(SERVER_ADDRESS, brokerHost))
+                                        .hasBucketBoundaries(DURATION_BUCKETS))));
   }
 
   @Test