Add messaging wrappers to support lightweight manual instrumentation

messaging-wrappers/README.md

@@ -0,0 +1,117 @@
+# OpenTelemetry Messaging Wrappers
+
+This is a lightweight messaging wrappers API designed to help you quickly add instrumentation to any
+type of messaging system client. To further ease the burden of instrumentation, we will also provide
+predefined implementations for certain messaging systems, helping you seamlessly address the issue 
+of broken traces.
+
+## Overview
+
+The primary goal of this API is to simplify the process of adding instrumentation to your messaging 
+systems, thereby enhancing observability without introducing significant overhead. Inspired by 
+[#13340](https://github.com/open-telemetry/opentelemetry-java-instrumentation/issues/13340) and 
+[opentelemetry-java-instrumentation](https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/main/instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/messaging/MessagingAttributesExtractor.java), 
+this tool aims to streamline the tracing and monitoring process.
+
+## Predefined Implementations
+
+| Messaging system  | Version Scope | Wrapper type |
+|-------------------|---------------|--------------|
+| kafka-clients     | `[0.11.0.0,)` | process      |
+| aliyun mns-client | `[1.3.0,)`    | process      |
+
+## Quickstart
+
+### Step 1 Add dependencies
+
+To use OpenTelemetry in your project, you need to add the necessary dependencies. Below are the configurations for both
+Gradle and Maven.
+
+#### Gradle
+
+```kotlin
+dependencies {
+    implementation("io.opentelemetry.contrib:opentelemetry-messaging-wrappers-api")
+}
+```
+
+#### Maven
+
+```xml
+<dependency>
+    <groupId>io.opentelemetry.contrib</groupId>
+    <artifactId>opentelemetry-messaging-wrappers-api</artifactId>
+</dependency>
+```
+
+### Step 2 Initializing MessagingWrappers
+
+Below is an example of how to initialize a messaging wrapper.
+
+```java
+public class Demo {
+
+  public static MessagingProcessWrapper<MyMessagingProcessRequest> createWrapper(
+      OpenTelemetry openTelemetry,
+      MyTextMapGetter textMapGetter,
+      List<AttributesExtractor<MyMessagingProcessRequest, Void>> additionalExtractor) {
+
+    return MessagingProcessWrapper.<MyMessagingProcessRequest>defaultBuilder()
+        .openTelemetry(openTelemetry)
+        .textMapGetter(textMapGetter)
+        .addAttributesExtractors(additionalExtractor)
+        .build();
+  }
+}
+
+public class MyMessagingProcessRequest implements MessagingProcessRequest {
+  // your implementation here
+}
+
+public class MyTextMapGetter implements TextMapGetter<MyMessagingProcessRequest> {
+  // your implementation here
+}
+```
+
+For arbitrary messaging systems, you need to manually define `MessagingProcessRequest` and the corresponding `TextMapGetter`.
+You can also customize your messaging spans by adding an AttributesExtractor.
+
+For popular messaging systems, we provide pre-implemented wrappers that allow for out-of-the-box integration. We provide
+an implementation based on the OpenTelemetry semantic convention by default.
+
+```java
+public class KafkaDemo {
+
+  public static MessagingProcessWrapper<KafkaProcessRequest> createWrapper() {
+    return KafkaHelper.processWrapperBuilder().build();
+  }
+}
+```
+
+### Step 3 Wrapping your process
+
+Once the MessagingWrapper are initialized, you can wrap your message processing logic to ensure that tracing spans are
+properly created and propagated.
+
+**P.S.** Some instrumentations may also [generate process spans](https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/main/docs/supported-libraries.md).
+If both are enabled, it might result in duplicate nested process spans. It is recommended to disable one of them.
+
+```java
+public class Demo {
+
+  private static final MessagingProcessWrapper<MyMessagingProcessRequest> WRAPPER = createWrapper();
+
+  public String consume(Message message) {
+    WRAPPER.doProcess(new MyMessagingProcessRequest(message), () -> {
+      // your processing logic
+    });
+  }
+}
+```
+
+## Component owners
+
+- [Minghui Zhang](https://github.com/Cirilla-zmh), Alibaba
+- [Steve Rao](https://github.com/steverao), Alibaba
+
+Learn more about component owners in [component_owners.yml](../.github/component_owners.yml).

messaging-wrappers/aliyun-mns-sdk/build.gradle.kts

@@ -0,0 +1,21 @@
+plugins {
+  id("otel.java-conventions")
+
+  id("otel.publish-conventions")
+}
+
+description = "OpenTelemetry Messaging Wrappers - aliyun-mns-sdk implementation"
+otelJava.moduleName.set("io.opentelemetry.contrib.messaging.wrappers.aliyun-mns-sdk")
+
+dependencies {
+  api(project(":messaging-wrappers:api"))
+
+  compileOnly("com.aliyun.mns:aliyun-sdk-mns:1.3.0")
+
+  testImplementation("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure")
+  testImplementation("io.opentelemetry:opentelemetry-sdk-trace")
+  testImplementation("io.opentelemetry:opentelemetry-sdk-testing")
+
+  testImplementation("io.opentelemetry:opentelemetry-sdk-extension-incubator")
+  testImplementation("uk.org.webcompere:system-stubs-jupiter:2.0.3")
+}

messaging-wrappers/aliyun-mns-sdk/gradle.properties

@@ -0,0 +1,2 @@
+# TODO: uncomment when ready to mark as stable
+# otel.stable=true

messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MNSHelper.java

@@ -0,0 +1,48 @@
+package io.opentelemetry.contrib.messaging.wrappers.mns;
+
+import com.aliyun.mns.model.BaseMessage;
+import com.aliyun.mns.model.MessagePropertyValue;
+import com.aliyun.mns.model.MessageSystemPropertyName;
+import com.aliyun.mns.model.MessageSystemPropertyValue;
+import io.opentelemetry.contrib.messaging.wrappers.mns.semconv.MNSProcessRequest;
+
+public final class MNSHelper {
+
+  public static <REQUEST extends MNSProcessRequest> MNSProcessWrapperBuilder<REQUEST> processWrapperBuilder() {
+    return new MNSProcessWrapperBuilder<>();
+  }
+
+  public static String getMessageHeader(BaseMessage message, String name) {
+    MessageSystemPropertyName key = convert2SystemPropertyName(name);
+    if (key != null) {
+      MessageSystemPropertyValue systemProperty = message.getSystemProperty(key);
+      if (systemProperty != null) {
+        return systemProperty.getStringValueByType();
+      }
+    }
+    MessagePropertyValue messagePropertyValue = message.getUserProperties().get(name);
+    if (messagePropertyValue != null) {
+      return messagePropertyValue.getStringValueByType();
+    }
+    return null;
+  }
+
+  /**
+   * see {@link MessageSystemPropertyName}
+   * */
+  public static MessageSystemPropertyName convert2SystemPropertyName(String name) {
+    if (name == null) {
+      return null;
+    } else if (name.equals(MessageSystemPropertyName.TRACE_PARENT.getValue())) {
+      return MessageSystemPropertyName.TRACE_PARENT;
+    } else if (name.equals(MessageSystemPropertyName.BAGGAGE.getValue())) {
+      return MessageSystemPropertyName.BAGGAGE;
+    } else if (name.equals(MessageSystemPropertyName.TRACE_STATE.getValue())) {
+      return MessageSystemPropertyName.TRACE_STATE;
+    }
+    return null;
+  }
+
+  private MNSHelper() {
+  }
+}

messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MNSProcessWrapperBuilder.java

@@ -0,0 +1,13 @@
+package io.opentelemetry.contrib.messaging.wrappers.mns;
+
+import io.opentelemetry.contrib.messaging.wrappers.DefaultMessagingProcessWrapperBuilder;
+import io.opentelemetry.contrib.messaging.wrappers.mns.semconv.MNSProcessRequest;
+
+public class MNSProcessWrapperBuilder<REQUEST extends MNSProcessRequest>
+    extends DefaultMessagingProcessWrapperBuilder<REQUEST> {
+
+  MNSProcessWrapperBuilder() {
+    super();
+    super.textMapGetter = MNSTextMapGetter.create();
+  }
+}

messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MNSTextMapGetter.java

@@ -0,0 +1,71 @@
+package io.opentelemetry.contrib.messaging.wrappers.mns;
+
+import static java.util.Collections.emptyList;
+
+import com.aliyun.mns.model.*;
+import io.opentelemetry.context.propagation.TextMapGetter;
+import io.opentelemetry.contrib.messaging.wrappers.mns.semconv.MNSProcessRequest;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class MNSTextMapGetter<REQUEST extends MNSProcessRequest> implements TextMapGetter<REQUEST> {
+
+  public static <REQUEST extends MNSProcessRequest> TextMapGetter<REQUEST> create() {
+    return new MNSTextMapGetter<>();
+  }
+
+  @Override
+  public Iterable<String> keys(@Nullable REQUEST carrier) {
+    if (carrier == null || carrier.getMessage() == null) {
+      return emptyList();
+    }
+    Message message = carrier.getMessage();
+
+    Map<String, MessageSystemPropertyValue> systemProperties = message.getSystemProperties();
+    if (systemProperties == null) {
+      systemProperties = Collections.emptyMap();
+    }
+    Map<String, MessagePropertyValue> userProperties = message.getUserProperties();
+    if (userProperties == null) {
+      userProperties = Collections.emptyMap();
+    }
+    List<String> keys = new ArrayList<>(systemProperties.size() + userProperties.size());
+    keys.addAll(systemProperties.keySet());
+    keys.addAll(userProperties.keySet());
+    return keys;
+  }
+
+  @Override
+  public String get(@Nullable REQUEST carrier, String key) {
+    if (carrier == null || carrier.getMessage() == null) {
+      return null;
+    }
+    Message message = carrier.getMessage();
+
+    // the system property should always take precedence over the user property
+    MessageSystemPropertyName systemPropertyName = MNSHelper.convert2SystemPropertyName(key);
+    if (systemPropertyName != null) {
+      MessageSystemPropertyValue value = message.getSystemProperty(systemPropertyName);
+      if (value != null) {
+        return value.getStringValueByType();
+      }
+    }
+
+    Map<String, MessagePropertyValue> userProperties = message.getUserProperties();
+    if (userProperties == null || userProperties.isEmpty()) {
+      return null;
+    }
+    MessagePropertyValue value = userProperties.getOrDefault(key, null);
+    if (value != null) {
+      return value.getStringValueByType();
+    }
+    return null;
+  }
+
+  private MNSTextMapGetter() {
+  }
+}

messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/example/MNSConsumer.java

@@ -0,0 +1,72 @@
+package io.opentelemetry.contrib.messaging.wrappers.mns.example;
+
+import com.aliyun.mns.client.CloudAccount;
+import com.aliyun.mns.client.CloudQueue;
+import com.aliyun.mns.client.MNSClient;
+import com.aliyun.mns.common.ClientException;
+import com.aliyun.mns.common.ServiceException;
+import com.aliyun.mns.model.Message;
+import com.aliyun.mns.model.MessageSystemPropertyValue;
+import io.opentelemetry.contrib.messaging.wrappers.MessagingProcessWrapper;
+import io.opentelemetry.contrib.messaging.wrappers.mns.MNSHelper;
+import io.opentelemetry.contrib.messaging.wrappers.mns.semconv.MNSProcessRequest;
+
+import static com.aliyun.mns.model.MessageSystemPropertyName.TRACE_PARENT;
+
+public class MNSConsumer {
+
+  public static void main(String[] args) {
+    // 1. create wrapper by default
+    MessagingProcessWrapper<MNSProcessRequest> wrapper = MNSHelper.processWrapperBuilder().build();
+    CloudAccount account = new CloudAccount("my-ak", "my-sk", "endpoint");
+    MNSClient client = account.getMNSClient();
+
+    final String queueName = "test-queue";
+    try {
+      CloudQueue queue = client.getQueueRef(queueName);
+      while (true) {
+        Message popMsg = queue.popMessage(5);
+        if (popMsg != null) {
+          // 2. wrap your consume block
+          String result = wrapper.doProcess(MNSProcessRequest.of(popMsg, queueName), () -> {
+            System.out.println("message handle: " + popMsg.getReceiptHandle());
+            System.out.println("message body: " + popMsg.getMessageBodyAsString());
+            System.out.println("message id: " + popMsg.getMessageId());
+            System.out.println("message dequeue count:" + popMsg.getDequeueCount());
+            MessageSystemPropertyValue systemProperty = popMsg.getSystemProperty(TRACE_PARENT);
+            if (systemProperty != null) {
+              System.out.println("message trace parent: " + systemProperty.getStringValueByType());
+            } else {
+              System.out.println("empty system property");
+            }
+            //<<to add your special logic.>>
+
+            queue.deleteMessage(popMsg.getReceiptHandle());
+            System.out.println("delete message successfully\n");
+            return "success";
+          });
+        }
+      }
+    } catch (ClientException ce) {
+      System.out.println("Something wrong with the network connection between client and MNS service."
+          + "Please check your network and DNS availablity.");
+      ce.printStackTrace();
+    } catch (ServiceException se) {
+      if (se.getErrorCode().equals("QueueNotExist")) {
+        System.out.println("Queue is not exist.Please create queue before use");
+      } else if (se.getErrorCode().equals("TimeExpired")) {
+        System.out.println("The request is time expired. Please check your local machine timeclock");
+      }
+      /*
+      you can get more MNS service error code in following link.
+      https://help.aliyun.com/document_detail/mns/api_reference/error_code/error_code.html?spm=5176.docmns/api_reference/error_code/error_response
+      */
+      se.printStackTrace();
+    } catch (Exception e) {
+      System.out.println("Unknown exception happened!");
+      e.printStackTrace();
+    }
+
+    client.close();
+  }
+}

messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/package-info.java

@@ -0,0 +1,2 @@
+/** OpenTelemetry messaging wrappers extension - mns implementation. */
+package io.opentelemetry.contrib.messaging.wrappers.mns;