Add messaging wrappers to support lightweight manual instrumentation

messaging-wrappers/README.md

@@ -0,0 +1,118 @@
+# 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,55 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+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;
+
+import javax.annotation.Nullable;
+
+public final class MnsHelper {
+
+  public static <REQUEST extends MnsProcessRequest>
+      MnsProcessWrapperBuilder<REQUEST> processWrapperBuilder() {
+    return new MnsProcessWrapperBuilder<>();
+  }
+
+  @Nullable
+  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} */
+  @Nullable
+  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,18 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+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,78 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.messaging.wrappers.mns;
+
+import static java.util.Collections.emptyList;
+
+import com.aliyun.mns.model.Message;
+import com.aliyun.mns.model.MessagePropertyValue;
+import com.aliyun.mns.model.MessageSystemPropertyName;
+import com.aliyun.mns.model.MessageSystemPropertyValue;
+import io.opentelemetry.context.propagation.TextMapGetter;
+import io.opentelemetry.contrib.messaging.wrappers.mns.semconv.MnsProcessRequest;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+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;
+  }
+
+  @Nullable
+  @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/package-info.java

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

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

@@ -0,0 +1,97 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.messaging.wrappers.mns.semconv;
+
+import com.aliyun.mns.model.Message;
+import io.opentelemetry.contrib.messaging.wrappers.mns.MnsHelper;
+import io.opentelemetry.contrib.messaging.wrappers.semconv.MessagingProcessRequest;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+
+public class MnsProcessRequest implements MessagingProcessRequest {
+
+  private final Message message;
+
+  @Nullable private final String destination;
+
+  public static MnsProcessRequest of(Message message) {
+    return of(message, null);
+  }
+
+  public static MnsProcessRequest of(Message message, @Nullable String destination) {
+    return new MnsProcessRequest(message, destination);
+  }
+
+  @Override
+  public String getSystem() {
+    return "smq";
+  }
+
+  @Nullable
+  @Override
+  public String getDestination() {
+    return this.destination;
+  }
+
+  @Nullable
+  @Override
+  public String getDestinationTemplate() {
+    return null;
+  }
+
+  @Override
+  public boolean isTemporaryDestination() {
+    return false;
+  }
+
+  @Override
+  public boolean isAnonymousDestination() {
+    return false;
+  }
+
+  @Nullable
+  @Override
+  public String getConversationId() {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public Long getMessageBodySize() {
+    return (long) message.getMessageBodyAsBytes().length;
+  }
+
+  @Nullable
+  @Override
+  public Long getMessageEnvelopeSize() {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public String getMessageId() {
+    return message.getMessageId();
+  }
+
+  @Override
+  public List<String> getMessageHeader(String name) {
+    String header = MnsHelper.getMessageHeader(message, name);
+    if (header == null) {
+      return Collections.emptyList();
+    }
+    return Collections.singletonList(header);
+  }
+
+  public Message getMessage() {
+    return message;
+  }
+
+  private MnsProcessRequest(Message message, @Nullable String destination) {
+    this.message = message;
+    this.destination = destination;
+  }
+}