feat: Implement Pulsar 4.0 monitoring support with messaging.client.sent.messages

.fossa.yml

@@ -772,6 +772,12 @@ targets:
     - type: gradle
       path: ./
       target: ':instrumentation:pulsar:pulsar-2.8:javaagent'
+    - type: gradle
+      path: ./
+      target: ':instrumentation:pulsar:pulsar-4.0:javaagent'
+    - type: gradle
+      path: ./
+      target: ':instrumentation:pulsar:pulsar-common:javaagent'
     - type: gradle
       path: ./
       target: ':instrumentation:ratpack:ratpack-1.4:javaagent'

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

@@ -66,5 +66,12 @@ static void applyReceiveMessagesAdvice(LongCounterBuilder builder) {
     ((ExtendedLongCounterBuilder) builder).setAttributesAdvice(MESSAGING_ATTRIBUTES);
   }
 
+  static void applyProduceMessagesAdvice(LongCounterBuilder builder) {
+    if (!(builder instanceof ExtendedLongCounterBuilder)) {
+      return;
+    }
+    ((ExtendedLongCounterBuilder) builder).setAttributesAdvice(MESSAGING_ATTRIBUTES);
+  }
+
   private MessagingMetricsAdvice() {}
 }

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

@@ -9,9 +9,12 @@
 
 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;
@@ -29,11 +32,15 @@
 public final class MessagingProducerMetrics 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<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 final LongCounter produceMessageCount;
 
   private MessagingProducerMetrics(Meter meter) {
     DoubleHistogramBuilder durationBuilder =
@@ -44,6 +51,14 @@ private MessagingProducerMetrics(Meter meter) {
             .setUnit("s");
     MessagingMetricsAdvice.applyPublishDurationAdvice(durationBuilder);
     publishDurationHistogram = durationBuilder.build();
+
+    LongCounterBuilder longCounterBuilder =
+        meter
+            .counterBuilder("messaging.client.sent.messages")
+            .setDescription("Number of messages producer attempted to send to the broker.")
+            .setUnit("{message}");
+    MessagingMetricsAdvice.applyProduceMessagesAdvice(longCounterBuilder);
+    produceMessageCount = longCounterBuilder.build();
   }
 
   public static OperationMetrics get() {
@@ -73,6 +88,21 @@ public void onEnd(Context context, Attributes endAttributes, long endNanos) {
 
     publishDurationHistogram.record(
         (endNanos - state.startTimeNanos()) / NANOS_PER_S, attributes, context);
+
+    long receiveMessagesCount = getProduceMessagesCount(state.startAttributes(), endAttributes);
+    if (receiveMessagesCount > 0) {
+      produceMessageCount.add(receiveMessagesCount, attributes, context);
+    }
+  }
+
+  private static long getProduceMessagesCount(Attributes... attributesList) {
+    for (Attributes attributes : attributesList) {
+      Long value = attributes.get(MESSAGING_BATCH_MESSAGE_COUNT);
+      if (value != null) {
+        return value;
+      }
+    }
+    return 0;
   }
 
   @AutoValue

instrumentation/pulsar/pulsar-2.8/javaagent-unit-tests/build.gradle.kts

@@ -4,4 +4,5 @@ plugins {
 
 dependencies {
   testImplementation(project(":instrumentation:pulsar:pulsar-2.8:javaagent"))
+  testImplementation(project(":instrumentation:pulsar:pulsar-common:javaagent"))
 }

instrumentation/pulsar/pulsar-2.8/javaagent-unit-tests/src/test/java/io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/UrlParserTest.java

@@ -7,7 +7,8 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-import io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.UrlParser.UrlData;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.UrlParser;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.UrlParser.UrlData;
 import org.junit.jupiter.api.Test;
 
 public class UrlParserTest {

instrumentation/pulsar/pulsar-2.8/javaagent/build.gradle.kts

@@ -13,6 +13,7 @@ muzzle {
 
 dependencies {
   library("org.apache.pulsar:pulsar-client:2.8.0")
+  implementation(project(":instrumentation:pulsar:pulsar-common:javaagent"))
 
   testImplementation("javax.annotation:javax.annotation-api:1.3.2")
   testImplementation("org.testcontainers:pulsar")

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

@@ -11,7 +11,7 @@
 
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
-import io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.telemetry.MessageListenerContext;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.MessageListenerContext;
 import net.bytebuddy.asm.Advice;
 import net.bytebuddy.description.type.TypeDescription;
 import net.bytebuddy.matcher.ElementMatcher;

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

@@ -20,6 +20,7 @@
 import io.opentelemetry.instrumentation.api.internal.Timer;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.VirtualFieldStore;
 import java.util.concurrent.CompletableFuture;
 import net.bytebuddy.asm.Advice;
 import net.bytebuddy.description.type.TypeDescription;

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

@@ -12,6 +12,7 @@
 
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.VirtualFieldStore;
 import net.bytebuddy.asm.Advice;
 import net.bytebuddy.description.type.TypeDescription;
 import net.bytebuddy.matcher.ElementMatcher;

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

@@ -15,7 +15,8 @@
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
-import io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.telemetry.PulsarRequest;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.VirtualFieldStore;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.PulsarRequest;
 import net.bytebuddy.asm.Advice;
 import net.bytebuddy.description.type.TypeDescription;
 import net.bytebuddy.implementation.bytecode.assign.Assigner;

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

@@ -16,7 +16,8 @@
 import io.opentelemetry.context.Context;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
-import io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.telemetry.PulsarRequest;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.VirtualFieldStore;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.PulsarRequest;
 import net.bytebuddy.asm.Advice;
 import net.bytebuddy.description.type.TypeDescription;
 import net.bytebuddy.matcher.ElementMatcher;

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

@@ -15,7 +15,9 @@
 import io.opentelemetry.context.Scope;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
-import io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.telemetry.PulsarRequest;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.SendCallbackData;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.VirtualFieldStore;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.PulsarRequest;
 import net.bytebuddy.asm.Advice;
 import net.bytebuddy.description.type.TypeDescription;
 import net.bytebuddy.matcher.ElementMatcher;

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

@@ -5,9 +5,10 @@
 
 package io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.telemetry;
 
-import static io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.UrlParser.parseUrl;
+import static io.opentelemetry.javaagent.instrumentation.pulsar.common.UrlParser.parseUrl;
 
-import io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.UrlParser.UrlData;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.UrlParser.UrlData;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.BasePulsarRequest;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.Messages;
 import org.apache.pulsar.common.naming.TopicName;

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

@@ -10,6 +10,8 @@
 import io.opentelemetry.instrumentation.api.instrumenter.SpanLinksBuilder;
 import io.opentelemetry.instrumentation.api.instrumenter.SpanLinksExtractor;
 import io.opentelemetry.instrumentation.api.internal.PropagatorBasedSpanLinksExtractor;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.MessageTextMapGetter;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.PulsarRequest;
 import org.apache.pulsar.client.api.Message;
 
 final class PulsarBatchRequestSpanLinksExtractor implements SpanLinksExtractor<PulsarBatchRequest> {

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

@@ -9,6 +9,7 @@
 import static java.util.Collections.singletonList;
 
 import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.PulsarRequest;
 import java.util.List;
 import javax.annotation.Nullable;
 import org.apache.pulsar.client.api.Message;

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

@@ -27,7 +27,13 @@
 import io.opentelemetry.instrumentation.api.semconv.network.ServerAttributesExtractor;
 import io.opentelemetry.javaagent.bootstrap.internal.AgentInstrumentationConfig;
 import io.opentelemetry.javaagent.bootstrap.internal.ExperimentalConfig;
-import io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.VirtualFieldStore;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.VirtualFieldStore;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.ExperimentalProducerAttributesExtractor;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.MessageListenerContext;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.MessageTextMapGetter;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.MessageTextMapSetter;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.PulsarNetClientAttributesGetter;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.PulsarRequest;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import org.apache.pulsar.client.api.Consumer;

instrumentation/pulsar/pulsar-4.0/javaagent/build.gradle.kts

@@ -0,0 +1,20 @@
+plugins {
+  id("otel.javaagent-instrumentation")
+}
+
+muzzle {
+  pass {
+    group.set("org.apache.pulsar")
+    module.set("pulsar-client")
+    versions.set("[4.0.0,)")
+    assertInverse.set(true)
+  }
+}
+
+dependencies {
+  library("org.apache.pulsar:pulsar-client:4.0.0")
+  implementation(project(":instrumentation:pulsar:pulsar-common:javaagent"))
+
+  testImplementation("org.testcontainers:pulsar")
+  testImplementation("org.apache.pulsar:pulsar-client-admin:4.0.0")
+}

instrumentation/pulsar/pulsar-4.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pulsar/v4_0/ProducerImplInstrumentation.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.pulsar.v4_0;
+
+import static io.opentelemetry.javaagent.instrumentation.pulsar.v4_0.telemetry.PulsarSingletons.producerInstrumenter;
+import static net.bytebuddy.matcher.ElementMatchers.hasSuperType;
+import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.isPublic;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.VirtualFieldStore;
+import io.opentelemetry.javaagent.instrumentation.pulsar.common.telemetry.PulsarRequest;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.ProducerImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.SendCallback;
+
+public class ProducerImplInstrumentation implements TypeInstrumentation {
+
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    return named("org.apache.pulsar.client.impl.ProducerImpl");
+  }
+
+  @Override
+  public void transform(TypeTransformer transformer) {
+    transformer.applyAdviceToMethod(
+        isConstructor()
+            .and(isPublic())
+            .and(
+                takesArgument(0, hasSuperType(named("org.apache.pulsar.client.api.PulsarClient")))),
+        ProducerImplInstrumentation.class.getName() + "$ProducerImplConstructorAdvice");
+    transformer.applyAdviceToMethod(
+        isMethod()
+            .and(named("sendAsync"))
+            .and(takesArgument(1, named("org.apache.pulsar.client.impl.SendCallback"))),
+        ProducerImplInstrumentation.class.getName() + "$ProducerSendAsyncMethodAdvice");
+  }
+
+  @SuppressWarnings("unused")
+  public static class ProducerImplConstructorAdvice {
+    @Advice.OnMethodExit(suppress = Throwable.class)
+    public static void intercept(
+        @Advice.This ProducerImpl<?> producer, @Advice.Argument(value = 0) PulsarClient client) {
+      PulsarClientImpl pulsarClient = (PulsarClientImpl) client;
+      String brokerUrl = pulsarClient.getLookup().getServiceUrl();
+      String topic = producer.getTopic();
+      VirtualFieldStore.inject(producer, brokerUrl, topic);
+    }
+  }
+
+  @SuppressWarnings("unused")
+  public static class ProducerSendAsyncMethodAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static void before(
+        @Advice.This ProducerImpl<?> producer,
+        @Advice.Argument(value = 0) Message<?> message,
+        @Advice.Argument(value = 1) SendCallback callback) {
+      Context parent = Context.current();
+      PulsarRequest request = PulsarRequest.create(message, VirtualFieldStore.extract(producer));
+
+      if (!producerInstrumenter().shouldStart(parent, request)) {
+        return;
+      }
+
+      Context context = producerInstrumenter().start(parent, request);
+      // Inject the context/request into the SendCallback. This will be extracted and used when the
+      // message is sent and the callback is invoked. see `SendCallbackInstrumentation`.
+      VirtualFieldStore.inject(callback, context, request);
+    }
+  }
+}

instrumentation/pulsar/pulsar-4.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pulsar/v4_0/PulsarInstrumentationModule.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.pulsar.v4_0;
+
+import com.google.auto.service.AutoService;
+import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import java.util.Arrays;
+import java.util.List;
+
+@AutoService(InstrumentationModule.class)
+public class PulsarInstrumentationModule extends InstrumentationModule {
+  public PulsarInstrumentationModule() {
+    super("pulsar", "pulsar-4.0");
+  }
+
+  @Override
+  public List<TypeInstrumentation> typeInstrumentations() {
+    return Arrays.asList(new ProducerImplInstrumentation(), new SendCallbackInstrumentation());
+  }
+}