Implement producer part of RocketMQ new client instrumentation

instrumentation/rocketmq/rocketmq-client/rocketmq-client-4.8/README.md

@@ -1,4 +1,4 @@
-# Settings for the Apache RocketMQ Remoting-based client instrumentation
+# Settings for the Apache RocketMQ remoting-based client instrumentation
 
 | System property | Type | Default | Description |
 |---|---|---|---|

instrumentation/rocketmq/rocketmq-client/rocketmq-client-4.8/library/README.md

@@ -1,4 +1,4 @@
-# Library Instrumentation for Apache RocketMQ Remoting-based Client 4.0.0+
+# Library Instrumentation for Apache RocketMQ remoting-based client 4.0.0+
 
 Provides OpenTelemetry instrumentation for [Apache RocketMQ](https://rocketmq.apache.org/) remoting-based client.
 

instrumentation/rocketmq/rocketmq-client/rocketmq-client-5.0/javaagent/build.gradle.kts

@@ -0,0 +1,18 @@
+plugins {
+  id("otel.javaagent-instrumentation")
+}
+
+muzzle {
+  pass {
+    group.set("org.apache.rocketmq")
+    module.set("rocketmq-client-java")
+    versions.set("[5.0.0,)")
+    assertInverse.set(true)
+  }
+}
+
+dependencies {
+  library("org.apache.rocketmq:rocketmq-client-java:5.0.0")
+
+  testImplementation(project(":instrumentation:rocketmq:rocketmq-client:rocketmq-client-5.0:testing"))
+}

instrumentation/rocketmq/rocketmq-client/rocketmq-client-5.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rocketmqclient/v5_0/FutureConverter.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rocketmqclient.v5_0;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.rocketmq.shaded.com.google.common.util.concurrent.FutureCallback;
+import org.apache.rocketmq.shaded.com.google.common.util.concurrent.Futures;
+import org.apache.rocketmq.shaded.com.google.common.util.concurrent.MoreExecutors;
+import org.apache.rocketmq.shaded.com.google.common.util.concurrent.SettableFuture;
+
+/** Future converter, which covert future of list into list of future. */
+public class FutureConverter {
+  private FutureConverter() {}
+
+  public static <T> List<SettableFuture<T>> convert(SettableFuture<List<T>> future, int num) {
+    List<SettableFuture<T>> futures = new ArrayList<>(num);
+    for (int i = 0; i < num; i++) {
+      SettableFuture<T> f = SettableFuture.create();
+      futures.add(f);
+    }
+    ListFutureCallback<T> futureCallback = new ListFutureCallback<>(futures);
+    Futures.addCallback(future, futureCallback, MoreExecutors.directExecutor());
+    return futures;
+  }
+
+  public static class ListFutureCallback<T> implements FutureCallback<List<T>> {
+    private final List<SettableFuture<T>> futures;
+
+    public ListFutureCallback(List<SettableFuture<T>> futures) {
+      this.futures = futures;
+    }
+
+    @Override
+    public void onSuccess(List<T> result) {
+      for (int i = 0; i < result.size(); i++) {
+        futures.get(i).set(result.get(i));
+      }
+    }
+
+    @Override
+    public void onFailure(Throwable t) {
+      for (SettableFuture<T> future : futures) {
+        future.setException(t);
+      }
+    }
+  }
+}

instrumentation/rocketmq/rocketmq-client/rocketmq-client-5.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rocketmqclient/v5_0/MapSetter.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rocketmqclient.v5_0;
+
+import io.opentelemetry.context.propagation.TextMapSetter;
+import java.util.HashMap;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.rocketmq.client.java.message.PublishingMessageImpl;
+
+enum MapSetter implements TextMapSetter<PublishingMessageImpl> {
+  INSTANCE;
+
+  @Override
+  public void set(@Nullable PublishingMessageImpl message, String key, String value) {
+    if (message == null) {
+      return;
+    }
+    Map<String, String> extraProperties = VirtualFieldStore.getExtraPropertiesByMessage(message);
+    if (extraProperties == null) {
+      extraProperties = new HashMap<>();
+      VirtualFieldStore.setExtraPropertiesByMessage(message, extraProperties);
+    }
+    extraProperties.put(key, value);
+  }
+}

instrumentation/rocketmq/rocketmq-client/rocketmq-client-5.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rocketmqclient/v5_0/RocketMqInstrumentationModule.java

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

instrumentation/rocketmq/rocketmq-client/rocketmq-client-5.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rocketmqclient/v5_0/RocketMqInstrumenterFactory.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rocketmqclient.v5_0;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
+import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
+import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessageOperation;
+import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesExtractor;
+import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesGetter;
+import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingSpanNameExtractor;
+import java.util.List;
+import org.apache.rocketmq.client.java.impl.producer.SendReceiptImpl;
+import org.apache.rocketmq.client.java.message.PublishingMessageImpl;
+
+final class RocketMqInstrumenterFactory {
+  private static final String INSTRUMENTATION_NAME = "io.opentelemetry.rocketmq-client-5.0";
+
+  private RocketMqInstrumenterFactory() {}
+
+  public static Instrumenter<PublishingMessageImpl, SendReceiptImpl> createProducerInstrumenter(
+      OpenTelemetry openTelemetry, List<String> capturedHeaders) {
+
+    RocketMqProducerAttributeGetter getter = RocketMqProducerAttributeGetter.INSTANCE;
+    MessageOperation operation = MessageOperation.SEND;
+
+    AttributesExtractor<PublishingMessageImpl, SendReceiptImpl> attributesExtractor =
+        buildMessagingAttributesExtractor(getter, operation, capturedHeaders);
+
+    InstrumenterBuilder<PublishingMessageImpl, SendReceiptImpl> instrumenterBuilder =
+        Instrumenter.<PublishingMessageImpl, SendReceiptImpl>builder(
+                openTelemetry,
+                INSTRUMENTATION_NAME,
+                MessagingSpanNameExtractor.create(getter, operation))
+            .addAttributesExtractor(attributesExtractor)
+            .addAttributesExtractor(RocketMqProducerAttributeExtractor.INSTANCE)
+            .setSpanStatusExtractor(
+                (spanStatusBuilder, message, sendReceipt, error) -> {
+                  if (null != error) {
+                    spanStatusBuilder.setStatus(StatusCode.ERROR);
+                  }
+                });
+
+    return instrumenterBuilder.buildProducerInstrumenter(MapSetter.INSTANCE);
+  }
+
+  private static <T, R> MessagingAttributesExtractor<T, R> buildMessagingAttributesExtractor(
+      MessagingAttributesGetter<T, R> getter,
+      MessageOperation operation,
+      List<String> capturedHeaders) {
+    return MessagingAttributesExtractor.builder(getter, operation)
+        .setCapturedHeaders(capturedHeaders)
+        .build();
+  }
+}

instrumentation/rocketmq/rocketmq-client/rocketmq-client-5.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rocketmqclient/v5_0/RocketMqProducerAttributeExtractor.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rocketmqclient.v5_0;
+
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.MESSAGING_ROCKETMQ_MESSAGE_KEYS;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.MESSAGING_ROCKETMQ_MESSAGE_TAG;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.MESSAGING_ROCKETMQ_MESSAGE_TYPE;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.MessagingRocketmqMessageTypeValues.DELAY;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.MessagingRocketmqMessageTypeValues.FIFO;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.MessagingRocketmqMessageTypeValues.NORMAL;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.MessagingRocketmqMessageTypeValues.TRANSACTION;
+
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
+import java.util.ArrayList;
+import javax.annotation.Nullable;
+import org.apache.rocketmq.client.java.impl.producer.SendReceiptImpl;
+import org.apache.rocketmq.client.java.message.PublishingMessageImpl;
+
+enum RocketMqProducerAttributeExtractor
+    implements AttributesExtractor<PublishingMessageImpl, SendReceiptImpl> {
+  INSTANCE;
+
+  @Override
+  public void onStart(
+      AttributesBuilder attributes, Context parentContext, PublishingMessageImpl message) {
+    message.getTag().ifPresent(s -> attributes.put(MESSAGING_ROCKETMQ_MESSAGE_TAG, s));
+    attributes.put(MESSAGING_ROCKETMQ_MESSAGE_KEYS, new ArrayList<>(message.getKeys()));
+    switch (message.getMessageType()) {
+      case FIFO:
+        attributes.put(MESSAGING_ROCKETMQ_MESSAGE_TYPE, FIFO);
+        break;
+      case DELAY:
+        attributes.put(MESSAGING_ROCKETMQ_MESSAGE_TYPE, DELAY);
+        break;
+      case TRANSACTION:
+        attributes.put(MESSAGING_ROCKETMQ_MESSAGE_TYPE, TRANSACTION);
+        break;
+      default:
+        attributes.put(MESSAGING_ROCKETMQ_MESSAGE_TYPE, NORMAL);
+    }
+  }
+
+  @Override
+  public void onEnd(
+      AttributesBuilder attributes,
+      Context context,
+      PublishingMessageImpl message,
+      @Nullable SendReceiptImpl sendReceipt,
+      @Nullable Throwable error) {}
+}

instrumentation/rocketmq/rocketmq-client/rocketmq-client-5.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rocketmqclient/v5_0/RocketMqProducerAttributeGetter.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.rocketmqclient.v5_0;
+
+import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesGetter;
+import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.rocketmq.client.java.impl.producer.SendReceiptImpl;
+import org.apache.rocketmq.client.java.message.PublishingMessageImpl;
+
+enum RocketMqProducerAttributeGetter
+    implements MessagingAttributesGetter<PublishingMessageImpl, SendReceiptImpl> {
+  INSTANCE;
+
+  @Nullable
+  @Override
+  public String system(PublishingMessageImpl message) {
+    return "rocketmq";
+  }
+
+  @Nullable
+  @Override
+  public String destinationKind(PublishingMessageImpl message) {
+    return SemanticAttributes.MessagingDestinationKindValues.TOPIC;
+  }
+
+  @Nullable
+  @Override
+  public String destination(PublishingMessageImpl message) {
+    return message.getTopic();
+  }
+
+  @Override
+  public boolean temporaryDestination(PublishingMessageImpl message) {
+    return false;
+  }
+
+  @Nullable
+  @Override
+  public String protocol(PublishingMessageImpl message) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public String protocolVersion(PublishingMessageImpl message) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public String url(PublishingMessageImpl message) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public String conversationId(PublishingMessageImpl message) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public Long messagePayloadSize(PublishingMessageImpl message) {
+    return (long) message.getBody().remaining();
+  }
+
+  @Nullable
+  @Override
+  public Long messagePayloadCompressedSize(PublishingMessageImpl message) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public String messageId(PublishingMessageImpl message, @Nullable SendReceiptImpl sendReceipt) {
+    return message.getMessageId().toString();
+  }
+
+  @Override
+  public List<String> header(PublishingMessageImpl message, String name) {
+    String value = message.getProperties().get(name);
+    if (value != null) {
+      return Collections.singletonList(value);
+    }
+    return Collections.emptyList();
+  }
+}