Add instrumentation for rocketmq

.github/workflows/ci.yaml

@@ -81,4 +81,4 @@ jobs:
         env:
           S3_BUILD_CACHE_ACCESS_KEY_ID: ${{ secrets.S3_BUILD_CACHE_ACCESS_KEY_ID }}
           S3_BUILD_CACHE_SECRET_KEY: ${{ secrets.S3_BUILD_CACHE_SECRET_KEY }}
-        run: ./gradlew test -PtestJavaVersion=${{ matrix.java }} --stacktrace -Porg.gradle.java.installations.paths=${{ steps.setup-test-java.outputs.path }} -Porg.gradle.java.installations.auto-download=false
+        run: ./gradlew test -PtestJavaVersion=${{ matrix.java }} --stacktrace -Porg.gradle.java.installations.paths=${{ steps.setup-test-java.outputs.path }} -Porg.gradle.java.installations.auto-download=false

instrumentation/rocketmq-client-4.8/javaagent/rocketmq-client-4.8-javaagent.gradle

@@ -0,0 +1,21 @@
+apply from: "$rootDir/gradle/instrumentation.gradle"
+
+muzzle {
+  pass {
+    group = "org.apache.rocketmq"
+    module = 'rocketmq-client'
+    versions = "[4.8.0,)"
+    assertInverse = true
+  }
+}
+
+dependencies {
+  library group: 'org.apache.rocketmq', name: 'rocketmq-client', version: '4.8.0'
+  implementation project(':instrumentation:rocketmq-client-4.8:library')
+  testImplementation project(':instrumentation:rocketmq-client-4.8:testing')
+
+}
+
+tasks.withType(Test) {
+  jvmArgs "-Dotel.instrumentation.rocketmq.client.experimental-span-attributes=true"
+}

instrumentation/rocketmq-client-4.8/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rocketmq/RocketMqConsumerInstrumentation.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rocketmq;
+
+import static io.opentelemetry.javaagent.tooling.bytebuddy.matcher.ClassLoaderMatcher.hasClassesNamed;
+import static java.util.Collections.singletonMap;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+import io.opentelemetry.instrumentation.rocketmq.TracingConsumeMessageHookImpl;
+import io.opentelemetry.javaagent.tooling.TypeInstrumentation;
+import java.util.Map;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
+
+public class RocketMqConsumerInstrumentation implements TypeInstrumentation {
+
+  @Override
+  public ElementMatcher<ClassLoader> classLoaderOptimization() {
+    return hasClassesNamed("org.apache.rocketmq.client.consumer.DefaultMQPushConsumer");
+  }
+
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    return named("org.apache.rocketmq.client.consumer.DefaultMQPushConsumer");
+  }
+
+  @Override
+  public Map<? extends ElementMatcher<? super MethodDescription>, String> transformers() {
+    return singletonMap(
+        isMethod().and(named("start")).and(takesArguments(0)),
+        RocketMqConsumerInstrumentation.class.getName() + "$AdviceStart");
+  }
+
+  public static class AdviceStart {
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static void onEnter(
+        @Advice.FieldValue(
+                value = "defaultMQPushConsumerImpl",
+                declaringType = DefaultMQPushConsumer.class)
+            DefaultMQPushConsumerImpl defaultMqPushConsumerImpl) {
+      defaultMqPushConsumerImpl.registerConsumeMessageHook(new TracingConsumeMessageHookImpl());
+    }
+  }
+}

instrumentation/rocketmq-client-4.8/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rocketmq/RocketMqInstrumentationModule.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rocketmq;
+
+import static java.util.Arrays.asList;
+
+import com.google.auto.service.AutoService;
+import io.opentelemetry.javaagent.tooling.InstrumentationModule;
+import io.opentelemetry.javaagent.tooling.TypeInstrumentation;
+import java.util.List;
+
+@AutoService(InstrumentationModule.class)
+public class RocketMqInstrumentationModule extends InstrumentationModule {
+  public RocketMqInstrumentationModule() {
+    super("rocketmq-client", "rocketmq-client-4.8");
+  }
+
+  @Override
+  public List<TypeInstrumentation> typeInstrumentations() {
+    return asList(new RocketMqProducerInstrumentation(), new RocketMqConsumerInstrumentation());
+  }
+}

instrumentation/rocketmq-client-4.8/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rocketmq/RocketMqProducerInstrumentation.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rocketmq;
+
+import static io.opentelemetry.javaagent.tooling.bytebuddy.matcher.ClassLoaderMatcher.hasClassesNamed;
+import static java.util.Collections.singletonMap;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+import io.opentelemetry.instrumentation.rocketmq.TracingSendMessageHookImpl;
+import io.opentelemetry.javaagent.tooling.TypeInstrumentation;
+import java.util.Map;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+
+public class RocketMqProducerInstrumentation implements TypeInstrumentation {
+
+  @Override
+  public ElementMatcher<ClassLoader> classLoaderOptimization() {
+    return hasClassesNamed("org.apache.rocketmq.client.producer.DefaultMQProducer");
+  }
+
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    return named("org.apache.rocketmq.client.producer.DefaultMQProducer");
+  }
+
+  @Override
+  public Map<? extends ElementMatcher<? super MethodDescription>, String> transformers() {
+    return singletonMap(
+        isMethod().and(named("start")).and(takesArguments(0)),
+        RocketMqProducerInstrumentation.class.getName() + "$AdviceStart");
+  }
+
+  public static class AdviceStart {
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static void onEnter(
+        @Advice.FieldValue(value = "defaultMQProducerImpl", declaringType = DefaultMQProducer.class)
+            DefaultMQProducerImpl defaultMqProducerImpl) {
+      defaultMqProducerImpl.registerSendMessageHook(new TracingSendMessageHookImpl());
+    }
+  }
+}

instrumentation/rocketmq-client-4.8/javaagent/src/test/groovy/io/opentelemetry/instrumentation/rocketmq/RocketMqClientTest.groovy

@@ -0,0 +1,13 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.rocketmq
+
+import io.opentelemetery.instrumentation.rocketmq.AbstractRocketMqClientTest
+import io.opentelemetry.instrumentation.test.AgentTestTrait
+
+class RocketMqClientTest extends AbstractRocketMqClientTest implements AgentTestTrait {
+
+}

instrumentation/rocketmq-client-4.8/library/rocketmq-client-4.8-library.gradle

@@ -0,0 +1,11 @@
+apply from: "$rootDir/gradle/instrumentation-library.gradle"
+
+dependencies {
+  library group: 'org.apache.rocketmq', name: 'rocketmq-client', version: '4.8.0'
+  testImplementation project(':instrumentation:rocketmq-client-4.8:testing')
+
+}
+
+tasks.withType(Test) {
+  jvmArgs "-Dotel.instrumentation.rocketmq.client.experimental-span-attributes=true"
+}

instrumentation/rocketmq-client-4.8/library/src/main/java/io/opentelemetry/instrumentation/rocketmq/RocketMqClientConfig.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.rocketmq;
+
+import io.opentelemetry.instrumentation.api.config.Config;
+
+public final class RocketMqClientConfig {
+
+  public static boolean isPropagationEnabled() {
+    return Config.get()
+        .getBooleanProperty("otel.instrumentation.rocketmq.client-propagation", true);
+  }
+
+  public static final boolean CAPTURE_EXPERIMENTAL_SPAN_ATTRIBUTES =
+      Config.get()
+          .getBooleanProperty(
+              "otel.instrumentation.rocketmq.client.experimental-span-attributes", false);
+
+  private RocketMqClientConfig() {}
+}

instrumentation/rocketmq-client-4.8/library/src/main/java/io/opentelemetry/instrumentation/rocketmq/RocketMqConsumerTracer.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.rocketmq;
+
+import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
+import static io.opentelemetry.instrumentation.rocketmq.TextMapExtractAdapter.GETTER;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanBuilder;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.api.tracer.BaseTracer;
+import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
+import java.util.List;
+import org.apache.rocketmq.common.message.MessageExt;
+
+public class RocketMqConsumerTracer extends BaseTracer {
+
+  private static final RocketMqConsumerTracer TRACER = new RocketMqConsumerTracer();
+
+  public static RocketMqConsumerTracer tracer() {
+    return TRACER;
+  }
+
+  @Override
+  protected String getInstrumentationName() {
+    return "io.opentelemetry.javaagent.rocketmq-client";
+  }
+
+  public Context startSpan(Context parentContext, List<MessageExt> msgs) {
+    MessageExt msg = msgs.get(0);
+    if (msgs.size() == 1) {
+      SpanBuilder spanBuilder = startSpanBuilder(msg).setParent(extractParent(msg));
+      return withClientSpan(parentContext, spanBuilder.startSpan());
+    } else {
+      SpanBuilder spanBuilder =
+          tracer
+              .spanBuilder(msg.getTopic() + " receive")
+              .setSpanKind(CONSUMER)
+              .setAttribute(SemanticAttributes.MESSAGING_SYSTEM, "rocketmq")
+              .setAttribute(SemanticAttributes.MESSAGING_OPERATION, "receive");
+      Context rootContext = withClientSpan(parentContext, spanBuilder.startSpan());
+      for (MessageExt message : msgs) {
+        createChildSpan(rootContext, message);
+      }
+      return rootContext;
+    }
+  }
+
+  public void createChildSpan(Context parentContext, MessageExt msg) {
+    SpanBuilder childSpanBuilder =
+        startSpanBuilder(msg)
+            .setParent(parentContext)
+            .addLink(Span.fromContext(extractParent(msg)).getSpanContext());
+    end(withClientSpan(parentContext, childSpanBuilder.startSpan()));
+  }
+
+  public SpanBuilder startSpanBuilder(MessageExt msg) {
+    SpanBuilder spanBuilder =
+        tracer
+            .spanBuilder(spanNameOnConsume(msg))
+            .setSpanKind(CONSUMER)
+            .setAttribute(SemanticAttributes.MESSAGING_SYSTEM, "rocketmq")
+            .setAttribute(SemanticAttributes.MESSAGING_DESTINATION, msg.getTopic())
+            .setAttribute(SemanticAttributes.MESSAGING_DESTINATION_KIND, "topic")
+            .setAttribute(SemanticAttributes.MESSAGING_OPERATION, "process")
+            .setAttribute(SemanticAttributes.MESSAGING_MESSAGE_ID, msg.getMsgId())
+            .setAttribute(
+                SemanticAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES,
+                (long) msg.getBody().length);
+    onConsume(spanBuilder, msg);
+    return spanBuilder;
+  }
+
+  private Context extractParent(MessageExt msg) {
+    if (RocketMqClientConfig.isPropagationEnabled()) {
+      return extract(msg.getProperties(), GETTER);
+    } else {
+      return Context.current();
+    }
+  }
+
+  void onConsume(SpanBuilder spanBuilder, MessageExt msg) {
+    if (RocketMqClientConfig.CAPTURE_EXPERIMENTAL_SPAN_ATTRIBUTES) {
+      spanBuilder.setAttribute("messaging.rocketmq.tags", msg.getTags());
+      spanBuilder.setAttribute("messaging.rocketmq.queue_id", msg.getQueueId());
+      spanBuilder.setAttribute("messaging.rocketmq.queue_offset", msg.getQueueOffset());
+      spanBuilder.setAttribute("messaging.rocketmq.broker_address", getBrokerHost(msg));
+    }
+  }
+
+  String spanNameOnConsume(MessageExt msg) {
+    return msg.getTopic() + " process";
+  }
+
+  String getBrokerHost(MessageExt msg) {
+    if (msg.getStoreHost() != null) {
+      return msg.getStoreHost().toString().replace("/", "");
+    } else {
+      return null;
+    }
+  }
+}

instrumentation/rocketmq-client-4.8/library/src/main/java/io/opentelemetry/instrumentation/rocketmq/RocketMqProducerTracer.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.rocketmq;
+
+import static io.opentelemetry.api.trace.SpanKind.PRODUCER;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanBuilder;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.api.tracer.BaseTracer;
+import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+
+public class RocketMqProducerTracer extends BaseTracer {
+
+  private static final RocketMqProducerTracer TRACER = new RocketMqProducerTracer();
+
+  public static RocketMqProducerTracer tracer() {
+    return TRACER;
+  }
+
+  @Override
+  protected String getInstrumentationName() {
+    return "io.opentelemetry.javaagent.rocketmq-client";
+  }
+
+  public Context startProducerSpan(String addr, Message msg, Context parentContext) {
+    SpanBuilder spanBuilder = spanBuilder(spanNameOnProduce(msg), PRODUCER);
+    onProduce(spanBuilder, msg, addr);
+    return withClientSpan(parentContext, spanBuilder.startSpan());
+  }
+
+  public void onCallback(Context context, SendResult sendResult) {
+    if (RocketMqClientConfig.CAPTURE_EXPERIMENTAL_SPAN_ATTRIBUTES) {
+      Span span = Span.fromContext(context);
+      span.setAttribute("messaging.rocketmq.callback_result", sendResult.getSendStatus().name());
+    }
+  }
+
+  private void onProduce(SpanBuilder spanBuilder, Message msg, String addr) {
+    spanBuilder.setAttribute(SemanticAttributes.MESSAGING_SYSTEM, "rocketmq");
+    spanBuilder.setAttribute(SemanticAttributes.MESSAGING_DESTINATION_KIND, "topic");
+    spanBuilder.setAttribute(SemanticAttributes.MESSAGING_DESTINATION, msg.getTopic());
+    if (RocketMqClientConfig.CAPTURE_EXPERIMENTAL_SPAN_ATTRIBUTES) {
+      spanBuilder.setAttribute("messaging.rocketmq.tags", msg.getTags());
+      spanBuilder.setAttribute("messaging.rocketmq.broker_address", addr);
+    }
+  }
+
+  public void afterProduce(Context context, SendResult sendResult) {
+    Span span = Span.fromContext(context);
+    span.setAttribute(SemanticAttributes.MESSAGING_MESSAGE_ID, sendResult.getMsgId());
+    if (RocketMqClientConfig.CAPTURE_EXPERIMENTAL_SPAN_ATTRIBUTES) {
+      span.setAttribute("messaging.rocketmq.send_result", sendResult.getSendStatus().name());
+    }
+  }
+
+  private String spanNameOnProduce(Message msg) {
+    return msg.getTopic() + " send";
+  }
+}