[enhance][pulsar] add apache-pulsar client support

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

@@ -0,0 +1,19 @@
+plugins {
+  id("otel.javaagent-instrumentation")
+}
+
+muzzle {
+  pass {
+    group.set("org.apache.pulsar")
+    module.set("pulsar-client")
+    versions.set("[2.8.0,)")
+    assertInverse.set(true)
+  }
+}
+
+dependencies {
+  library("org.apache.pulsar:pulsar-client:2.8.0")
+
+  testImplementation("javax.annotation:javax.annotation-api:1.3.2")
+  testImplementation("org.testcontainers:pulsar:1.17.1")
+}

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

@@ -0,0 +1,131 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.pulsar;
+
+import static io.opentelemetry.javaagent.instrumentation.pulsar.PulsarTelemetry.CONSUMER_NAME;
+import static io.opentelemetry.javaagent.instrumentation.pulsar.PulsarTelemetry.PROPAGATOR;
+import static io.opentelemetry.javaagent.instrumentation.pulsar.PulsarTelemetry.SERVICE_URL;
+import static io.opentelemetry.javaagent.instrumentation.pulsar.PulsarTelemetry.SUBSCRIPTION;
+import static io.opentelemetry.javaagent.instrumentation.pulsar.PulsarTelemetry.TOPIC;
+import static io.opentelemetry.javaagent.instrumentation.pulsar.PulsarTelemetry.TRACER;
+import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.isProtected;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.context.Context;
+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.info.ClientEnhanceInfo;
+import io.opentelemetry.javaagent.instrumentation.pulsar.info.MessageEnhanceInfo;
+import io.opentelemetry.javaagent.instrumentation.pulsar.textmap.MessageTextMapGetter;
+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.ConsumerImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+
+public class ConsumerImplInstrumentation implements TypeInstrumentation {
+
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    return named("org.apache.pulsar.client.impl.ConsumerImpl");
+  }
+
+  @Override
+  public void transform(TypeTransformer transformer) {
+    String klassName = ConsumerImplInstrumentation.class.getName();
+
+    transformer.applyAdviceToMethod(isConstructor(), klassName + "$ConsumerImplConstructorAdviser");
+
+    transformer.applyAdviceToMethod(
+        isMethod()
+            .and(isProtected())
+            .and(named("messageProcessed"))
+            .and(takesArgument(0, named("org.apache.pulsar.client.api.Message"))),
+        klassName + "$ConsumerImplMethodAdviser");
+  }
+
+  @SuppressWarnings("unused")
+  public static class ConsumerImplConstructorAdviser {
+
+    @Advice.OnMethodExit
+    public static void before(
+        @Advice.This ConsumerImpl<?> consumer,
+        @Advice.Argument(value = 0) PulsarClient client,
+        @Advice.Argument(value = 1) String topic) {
+
+      PulsarClientImpl pulsarClient = (PulsarClientImpl) client;
+      String url = pulsarClient.getLookup().getServiceUrl();
+
+      ClientEnhanceInfo info = new ClientEnhanceInfo(topic, url);
+      ClientEnhanceInfo.virtualField(consumer, info);
+    }
+  }
+
+  @SuppressWarnings("unused")
+  public static class ConsumerImplMethodAdviser {
+
+    @Advice.OnMethodEnter
+    public static Scope before(
+        @Advice.This ConsumerImpl<?> consumer, @Advice.Argument(value = 0) Message<?> message) {
+      ClientEnhanceInfo info = ClientEnhanceInfo.virtualField(consumer);
+      if (null == info) {
+        return Scope.noop();
+      }
+
+      MessageImpl<?> messageImpl = (MessageImpl<?>) message;
+      Context context =
+          PROPAGATOR.extract(Context.current(), messageImpl, MessageTextMapGetter.INSTANCE);
+
+      return TRACER
+          .spanBuilder("Pulsar://ConsumerImpl/messageProcessed")
+          .setParent(context)
+          .setSpanKind(SpanKind.CONSUMER)
+          .setAttribute(TOPIC, info.topic)
+          .setAttribute(SERVICE_URL, info.brokerUrl)
+          .setAttribute(SUBSCRIPTION, consumer.getSubscription())
+          .setAttribute(CONSUMER_NAME, consumer.getConsumerName())
+          .startSpan()
+          .makeCurrent();
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class)
+    public static void after(
+        @Advice.This ConsumerImpl<?> consumer,
+        @Advice.Argument(value = 0) Message<?> message,
+        @Advice.Thrown Throwable t,
+        @Advice.Enter Scope scope) {
+      ClientEnhanceInfo info = ClientEnhanceInfo.virtualField(consumer);
+      if (null == info || scope == null) {
+        if (null != scope) {
+          scope.close();
+        }
+        return;
+      }
+
+      MessageEnhanceInfo messageInfo = MessageEnhanceInfo.virtualField(message);
+      if (null != messageInfo) {
+        messageInfo.setFields(Context.current(), consumer.getTopic(), message.getMessageId());
+      }
+
+      Span span = Span.current();
+      if (t != null) {
+        span.recordException(t);
+      }
+
+      span.end();
+      scope.close();
+    }
+  }
+}

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

@@ -0,0 +1,52 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.pulsar;
+
+import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import io.opentelemetry.javaagent.instrumentation.pulsar.info.MessageEnhanceInfo;
+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.impl.MessageImpl;
+import org.apache.pulsar.client.impl.TopicMessageImpl;
+
+public class MessageInstrumentation implements TypeInstrumentation {
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    return named("org.apache.pulsar.client.impl.MessageImpl")
+        .or(named("org.apache.pulsar.client.impl.TopicMessageImpl"));
+  }
+
+  @Override
+  public void transform(TypeTransformer transformer) {
+    transformer.applyAdviceToMethod(
+        isConstructor(), MessageInstrumentation.class.getName() + "$MessageConstructorAdviser");
+  }
+
+  @SuppressWarnings("unused")
+  public static class MessageConstructorAdviser {
+
+    @Advice.OnMethodExit
+    public static void before(
+        @Advice.This Message<?> message, @Advice.AllArguments Object[] allArguments) {
+
+      if (message instanceof MessageImpl) {
+        MessageEnhanceInfo.virtualField(message, new MessageEnhanceInfo());
+      } else {
+        Object argument2 = allArguments[2];
+        if (message instanceof TopicMessageImpl && argument2 instanceof MessageImpl) {
+          MessageImpl<?> impl = (MessageImpl<?>) argument2;
+          MessageEnhanceInfo.virtualField(message, MessageEnhanceInfo.virtualField(impl));
+        }
+      }
+    }
+  }
+}

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

@@ -0,0 +1,107 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.pulsar;
+
+import static io.opentelemetry.javaagent.instrumentation.pulsar.PulsarTelemetry.CONSUMER_NAME;
+import static io.opentelemetry.javaagent.instrumentation.pulsar.PulsarTelemetry.MESSAGE_ID;
+import static io.opentelemetry.javaagent.instrumentation.pulsar.PulsarTelemetry.SUBSCRIPTION;
+import static io.opentelemetry.javaagent.instrumentation.pulsar.PulsarTelemetry.TOPIC;
+import static io.opentelemetry.javaagent.instrumentation.pulsar.PulsarTelemetry.TRACER;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.isPublic;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.context.Context;
+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.info.MessageEnhanceInfo;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.implementation.bytecode.assign.Assigner;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageListener;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+
+public class MessageListenerInstrumentation implements TypeInstrumentation {
+
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    // return hasSuperType(named("org.apache.pulsar.client.api.MessageListener"));
+    // can't enhance MessageListener here like above due to jvm can't enhance lambda.
+    return named("org.apache.pulsar.client.impl.conf.ConsumerConfigurationData");
+  }
+
+  @Override
+  public void transform(TypeTransformer transformer) {
+    transformer.applyAdviceToMethod(
+        isMethod().and(isPublic()).and(named("getMessageListener")),
+        MessageListenerInstrumentation.class.getName() + "$ConsumerConfigurationDataMethodAdviser");
+  }
+
+  @SuppressWarnings("unused")
+  public static class ConsumerConfigurationDataMethodAdviser {
+
+    @Advice.OnMethodExit
+    public static void after(
+        @Advice.This ConsumerConfigurationData<?> data,
+        @Advice.Return(readOnly = false, typing = Assigner.Typing.DYNAMIC)
+            MessageListener<?> listener) {
+      if (null == listener) {
+        return;
+      }
+
+      listener = new MessageListenerWrapper<>(listener);
+    }
+  }
+
+  public static class MessageListenerWrapper<T> implements MessageListener<T> {
+    private static final long serialVersionUID = 1L;
+
+    private final MessageListener<T> delegator;
+
+    public MessageListenerWrapper(MessageListener<T> messageListener) {
+      this.delegator = messageListener;
+    }
+
+    @Override
+    public void received(Consumer<T> consumer, Message<T> msg) {
+      MessageEnhanceInfo info = MessageEnhanceInfo.virtualField(msg);
+      Context parent = info == null ? Context.current() : info.getContext();
+      String topic = null == info ? consumer.getTopic() : info.getTopic();
+      String mid = null == info ? "unknown" : info.getMessageId();
+
+      Span span =
+          TRACER
+              .spanBuilder("Pulsar://MessageListener/received")
+              .setParent(parent)
+              .setSpanKind(SpanKind.CONSUMER)
+              .setAttribute(TOPIC, topic)
+              .setAttribute(MESSAGE_ID, mid)
+              .setAttribute(SUBSCRIPTION, consumer.getSubscription())
+              .setAttribute(CONSUMER_NAME, consumer.getConsumerName())
+              .startSpan();
+
+      try (Scope scope = span.makeCurrent()) {
+        this.delegator.received(consumer, msg);
+      } catch (Throwable t) {
+        span.recordException(t);
+        throw t;
+      } finally {
+        span.end();
+      }
+    }
+
+    @Override
+    public void reachedEndOfTopic(Consumer<T> consumer) {
+      this.delegator.reachedEndOfTopic(consumer);
+    }
+  }
+}