diff --git a/messaging-wrappers/README.md b/messaging-wrappers/README.md new file mode 100644 index 000000000..06d1ae885 --- /dev/null +++ b/messaging-wrappers/README.md @@ -0,0 +1,219 @@ +# 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. + +
+Table of Contents + +- [Overview](#overview) +- [Predefined Implementations](#predefined-implementations) +- [Quickstart For Given Implementations](#quickstart-for-given-implementations) + - [\[Given\] Step 1 Add dependencies](#given-step-1-add-dependencies) + - [\[Given\] Step 2 Initializing MessagingWrappers](#given-step-2-initializing-messagingwrappers) + - [\[Given\] Step 3 Wrapping the Process](#given-step-3-wrapping-the-process) +- [Manual Implementation](#manual-implementation) + - [\[Manual\] Step 1 Add dependencies](#manual-step-1-add-dependencies) + - [\[Manual\] Step 2 Initializing MessagingWrappers](#manual-step-2-initializing-messagingwrappers) + - [\[Manual\] Step 3 Wrapping the Process](#manual-step-3-wrapping-the-process) +- [Component Owners](#component-owners) + +
+ +## 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 For Given Implementations + +This example will demonstrate how to add automatic instrumentation to your Kafka consumer with process wrapper. For +detailed example, please check out [KafkaClientTest](./kafka-clients/src/test/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaClientTest.java). + +### [Given] 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. + +### [Given] Gradle + +```kotlin +dependencies { + implementation("io.opentelemetry.contrib:opentelemetry-messaging-wrappers-kafka-clients:${latest_version}") +} +``` + +### [Given] Maven + +```xml + + io.opentelemetry.contrib + opentelemetry-messaging-wrappers-kafka-clients + ${latest_version} + +``` + +### [Given] Step 2 Initializing MessagingWrappers + +For `kafka-clients`, 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 createWrapper() { + return KafkaHelper.processWrapperBuilder().build(); + } +} +``` + +### [Given] Step 3 Wrapping the 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 WRAPPER = createWrapper(); + + // please initialize consumer + private Consumer consumer; + + public String consume() { + ConsumerRecords records = consumer.poll(Duration.ofSeconds(5)); + ConsumerRecord record = records.iterator().next(); + + return WRAPPER.doProcess( + KafkaProcessRequest.of(record, groupId, clientId), () -> { + // your processing logic + return "success"; + }); + } + + public void consumeWithoutResult() { + ConsumerRecords records = consumer.poll(Duration.ofSeconds(5)); + ConsumerRecord record = records.iterator().next(); + + WRAPPER.doProcess( + KafkaProcessRequest.of(record, groupId, clientId), () -> { + // your processing logic + }); + } +} +``` + +## Manual Implementation + +You can also build implementations based on the `messaging-wrappers-api` for any messaging system to accommodate your +custom message protocol. For detailed example, please check out [UserDefinedMessageSystemTest](./api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/UserDefinedMessageSystemTest.java). + +### [Manual] Step 1 Add dependencies + +#### [Manual] Gradle + +```kotlin +dependencies { + implementation("io.opentelemetry.contrib:opentelemetry-messaging-wrappers-api:${latest_version}") +} +``` + +#### [Manual] Maven + +```xml + + io.opentelemetry.contrib + opentelemetry-messaging-wrappers-api + ${latest_version} + +``` + +### [Manual] Step 2 Initializing MessagingWrappers + +Below is an example of how to initialize a messaging wrapper. + +```java +public class Demo { + + public static MessagingProcessWrapper createWrapper(OpenTelemetry openTelemetry) { + + return MessagingProcessWrapper.defaultBuilder() + .openTelemetry(openTelemetry) + .textMapGetter(DefaultMessageTextMapGetter.create()) + .attributesExtractors( + Collections.singletonList( + MessagingAttributesExtractor.create( + DefaultMessagingAttributesGetter.create(), MessageOperation.PROCESS))) + .build(); + } +} + +public class MyMessagingProcessRequest implements MessagingProcessRequest { + // your implementation here + + @Override + public List getMessageHeader(String name) { + // impl: how to get specific header from your message + return Collections.singletonList(message.getHeaders().get(name)); + } + + @Override + public Collection getAllMessageHeadersKey() { + // impl: how to get all headers key set from your message + return message.getHeaders().keySet(); + } +} +``` + +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`. + +### [Manual] Step 3 Wrapping the 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 WRAPPER = createWrapper(); + + public String consume(Message message) { + return WRAPPER.doProcess(new MyMessagingProcessRequest(message), () -> { + // your processing logic + return "success"; + }); + } + + public void consumeWithoutReturn(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). diff --git a/messaging-wrappers/aliyun-mns-sdk/build.gradle.kts b/messaging-wrappers/aliyun-mns-sdk/build.gradle.kts new file mode 100644 index 000000000..286662c26 --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/build.gradle.kts @@ -0,0 +1,40 @@ +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("com.aliyun.mns:aliyun-sdk-mns:1.3.0") + testImplementation(project(":messaging-wrappers:testing")) + + testImplementation("org.springframework.boot:spring-boot-starter-web:2.7.18") + testImplementation("org.springframework.boot:spring-boot-starter-test:2.7.18") +} + +tasks { + withType().configureEach { + jvmArgs("-Dotel.java.global-autoconfigure.enabled=true") + // TODO: According to https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#message-creation-context-as-parent-of-process-span, + // process span should be the child of receive span. However, we couldn't access the trace context with receive span + // in wrappers, unless we add a generic accessor for that. + jvmArgs("-Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled=false") + jvmArgs("-Dotel.traces.exporter=logging") + jvmArgs("-Dotel.metrics.exporter=logging") + jvmArgs("-Dotel.logs.exporter=logging") + } +} + +configurations.testRuntimeClasspath { + resolutionStrategy { + force("ch.qos.logback:logback-classic:1.2.12") + force("org.slf4j:slf4j-api:1.7.35") + } +} diff --git a/messaging-wrappers/aliyun-mns-sdk/gradle.properties b/messaging-wrappers/aliyun-mns-sdk/gradle.properties new file mode 100644 index 000000000..a0402e1e2 --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/gradle.properties @@ -0,0 +1,2 @@ +# TODO: uncomment when ready to mark as stable +# otel.stable=true diff --git a/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsHelper.java b/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsHelper.java new file mode 100644 index 000000000..a33d569cc --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsHelper.java @@ -0,0 +1,52 @@ +/* + * 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 javax.annotation.Nullable; + +public final class MnsHelper { + + public static MnsProcessWrapperBuilder 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() {} +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsProcessWrapperBuilder.java b/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsProcessWrapperBuilder.java new file mode 100644 index 000000000..1993b4716 --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsProcessWrapperBuilder.java @@ -0,0 +1,29 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.mns; + +import io.opentelemetry.contrib.messaging.wrappers.MessagingProcessWrapperBuilder; +import io.opentelemetry.contrib.messaging.wrappers.mns.semconv.MnsConsumerAttributesGetter; +import io.opentelemetry.contrib.messaging.wrappers.mns.semconv.MnsProcessRequest; +import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessageOperation; +import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesExtractor; +import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanNameExtractor; +import java.util.ArrayList; + +public class MnsProcessWrapperBuilder extends MessagingProcessWrapperBuilder { + + MnsProcessWrapperBuilder() { + super(); + super.textMapGetter = MnsTextMapGetter.create(); + super.spanNameExtractor = + MessagingSpanNameExtractor.create( + MnsConsumerAttributesGetter.INSTANCE, MessageOperation.PROCESS); + super.attributesExtractors = new ArrayList<>(); + super.attributesExtractors.add( + MessagingAttributesExtractor.create( + MnsConsumerAttributesGetter.INSTANCE, MessageOperation.PROCESS)); + } +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsTextMapGetter.java b/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsTextMapGetter.java new file mode 100644 index 000000000..12691bd6a --- /dev/null +++ b/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 implements TextMapGetter { + + public static TextMapGetter create() { + return new MnsTextMapGetter(); + } + + @Override + public Iterable keys(@Nullable MnsProcessRequest carrier) { + if (carrier == null || carrier.getMessage() == null) { + return emptyList(); + } + Message message = carrier.getMessage(); + + Map systemProperties = message.getSystemProperties(); + if (systemProperties == null) { + systemProperties = Collections.emptyMap(); + } + Map userProperties = message.getUserProperties(); + if (userProperties == null) { + userProperties = Collections.emptyMap(); + } + List keys = new ArrayList<>(systemProperties.size() + userProperties.size()); + keys.addAll(systemProperties.keySet()); + keys.addAll(userProperties.keySet()); + return keys; + } + + @Nullable + @Override + public String get(@Nullable MnsProcessRequest 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 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() {} +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/package-info.java b/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/package-info.java new file mode 100644 index 000000000..1530bdbd7 --- /dev/null +++ b/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; diff --git a/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/semconv/MnsConsumerAttributesGetter.java b/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/semconv/MnsConsumerAttributesGetter.java new file mode 100644 index 000000000..4b94c1ce6 --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/semconv/MnsConsumerAttributesGetter.java @@ -0,0 +1,89 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.mns.semconv; + +import io.opentelemetry.contrib.messaging.wrappers.mns.MnsHelper; +import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter; +import java.util.Collections; +import java.util.List; +import javax.annotation.Nullable; + +public enum MnsConsumerAttributesGetter + implements MessagingAttributesGetter { + INSTANCE; + + @Override + public String getSystem(MnsProcessRequest request) { + return "smq"; + } + + @Nullable + @Override + public String getDestination(MnsProcessRequest request) { + return request.getDestination(); + } + + @Nullable + @Override + public String getDestinationTemplate(MnsProcessRequest request) { + return null; + } + + @Override + public boolean isTemporaryDestination(MnsProcessRequest request) { + return false; + } + + @Override + public boolean isAnonymousDestination(MnsProcessRequest request) { + return false; + } + + @Override + @Nullable + public String getConversationId(MnsProcessRequest request) { + return null; + } + + @Nullable + @Override + public Long getMessageBodySize(MnsProcessRequest request) { + return (long) request.getMessage().getMessageBodyAsBytes().length; + } + + @Nullable + @Override + public Long getMessageEnvelopeSize(MnsProcessRequest request) { + return (long) request.getMessage().getMessageBodyAsRawBytes().length; + } + + @Override + @Nullable + public String getMessageId(MnsProcessRequest request, @Nullable Void unused) { + return request.getMessage().getMessageId(); + } + + @Nullable + @Override + public String getClientId(MnsProcessRequest request) { + return null; + } + + @Nullable + @Override + public Long getBatchMessageCount(MnsProcessRequest request, @Nullable Void unused) { + return null; + } + + @Override + public List getMessageHeader(MnsProcessRequest request, String name) { + String header = MnsHelper.getMessageHeader(request.getMessage(), name); + if (header == null) { + return Collections.emptyList(); + } + return Collections.singletonList(header); + } +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/semconv/MnsProcessRequest.java b/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/semconv/MnsProcessRequest.java new file mode 100644 index 000000000..4c5ec9e42 --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/semconv/MnsProcessRequest.java @@ -0,0 +1,38 @@ +/* + * 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 javax.annotation.Nullable; + +public class MnsProcessRequest { + + 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); + } + + public Message getMessage() { + return message; + } + + @Nullable + public String getDestination() { + return this.destination; + } + + private MnsProcessRequest(Message message, @Nullable String destination) { + this.message = message; + this.destination = destination; + } +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/AliyunMnsSdkTest.java b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/AliyunMnsSdkTest.java new file mode 100644 index 000000000..c772465f6 --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/AliyunMnsSdkTest.java @@ -0,0 +1,164 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.mns; + +import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo; +import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_BODY_SIZE; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_ENVELOPE_SIZE; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_ID; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM; +import static org.assertj.core.api.Assertions.assertThat; + +import com.aliyun.mns.client.CloudAccount; +import com.aliyun.mns.client.CloudQueue; +import com.aliyun.mns.client.MNSClient; +import com.aliyun.mns.common.ServiceHandlingRequiredException; +import com.aliyun.mns.common.http.ClientConfiguration; +import com.aliyun.mns.model.Message; +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import io.opentelemetry.contrib.messaging.wrappers.MessagingProcessWrapper; +import io.opentelemetry.contrib.messaging.wrappers.mns.broker.SmqMockedBroker; +import io.opentelemetry.contrib.messaging.wrappers.mns.semconv.MnsProcessRequest; +import io.opentelemetry.contrib.messaging.wrappers.testing.AbstractBaseTest; +import java.nio.charset.StandardCharsets; +import org.apache.commons.codec.binary.Base64; +import org.assertj.core.api.AbstractAssert; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.SpringBootTest.WebEnvironment; +import org.springframework.boot.test.web.server.LocalServerPort; + +@SuppressWarnings("OtelInternalJavadoc") +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +@SpringBootTest( + classes = {SmqMockedBroker.class}, + webEnvironment = WebEnvironment.RANDOM_PORT) +public class AliyunMnsSdkTest extends AbstractBaseTest { + + private static final String TEST_ENDPOINT = "http://test.mns.cn-hangzhou.aliyuncs.com"; + + private static final String QUEUE = "TEST_QUEUE"; + + private static final String MESSAGE_BODY = "Hello OpenTelemetry"; + + @LocalServerPort private int testApplicationPort; // port at which the spring app is running + + private MNSClient mnsClient; + + private CloudQueue queue; + + private OpenTelemetry otel; + + private Tracer tracer; + + private MessagingProcessWrapper wrapper; + + @BeforeAll + void setupClass() { + otel = GlobalOpenTelemetry.get(); + tracer = otel.getTracer("test-tracer", "1.0.0"); + wrapper = MnsHelper.processWrapperBuilder().openTelemetry(otel).build(); + + ClientConfiguration configuration = new ClientConfiguration(); + configuration.setProxyHost("127.0.0.1"); + configuration.setProxyPort(testApplicationPort); + + CloudAccount account = new CloudAccount("test-ak", "test-sk", TEST_ENDPOINT, configuration); + + mnsClient = account.getMNSClient(); + queue = mnsClient.getQueueRef(QUEUE); + } + + @Test + void testSendAndConsume() throws ServiceHandlingRequiredException { + sendWithParent(); + + consumeWithChild(); + + assertTraces(); + } + + public void sendWithParent() { + // mock a send span + Span parent = tracer.spanBuilder("publish " + QUEUE).setSpanKind(SpanKind.PRODUCER).startSpan(); + + try (Scope scope = parent.makeCurrent()) { + Message message = new Message(MESSAGE_BODY); + otel.getPropagators() + .getTextMapPropagator() + .inject(Context.current(), message, MnsTextMapSetter.INSTANCE); + queue.putMessage(message); + } + + parent.end(); + } + + public void consumeWithChild() throws ServiceHandlingRequiredException { + // check that the message was received + Message message = null; + for (int i = 0; i < 3; i++) { + message = queue.popMessage(3); + if (message != null) { + break; + } + } + + assertThat(message).isNotNull(); + + wrapper.doProcess( + MnsProcessRequest.of(message, QUEUE), + () -> { + tracer.spanBuilder("process child").startSpan().end(); + }); + } + + /** + * Copied from testing-common. + */ + @SuppressWarnings("deprecation") // using deprecated semconv + public void assertTraces() { + waitAndAssertTraces( + sortByRootSpanName("parent", "producer callback"), + trace -> + trace.hasSpansSatisfyingExactly( + span -> + // No need to verify the attribute here because it is generated by + // instrumentation library. + span.hasName("publish " + QUEUE).hasKind(SpanKind.PRODUCER).hasNoParent(), + span -> + span.hasName(QUEUE + " process") + .hasKind(SpanKind.CONSUMER) + .hasParent(trace.getSpan(0)) + .hasAttributesSatisfyingExactly( + equalTo(MESSAGING_SYSTEM, "smq"), + equalTo(MESSAGING_DESTINATION_NAME, QUEUE), + equalTo( + MESSAGING_MESSAGE_BODY_SIZE, + MESSAGE_BODY.getBytes(StandardCharsets.UTF_8).length), + equalTo( + MESSAGING_MESSAGE_ENVELOPE_SIZE, + Base64.encodeBase64(MESSAGE_BODY.getBytes(StandardCharsets.UTF_8)) + .length), + equalTo(MESSAGING_OPERATION, "process"), + satisfies(MESSAGING_MESSAGE_ID, AbstractAssert::isNotNull)), + span -> + span.hasName("process child") + .hasKind(SpanKind.INTERNAL) + .hasParent(trace.getSpan(1)))); + } +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsTextMapSetter.java b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsTextMapSetter.java new file mode 100644 index 000000000..4f5ac86f1 --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsTextMapSetter.java @@ -0,0 +1,36 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.mns; + +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 com.aliyun.mns.model.SystemPropertyType; +import io.opentelemetry.context.propagation.TextMapSetter; +import javax.annotation.Nullable; + +public enum MnsTextMapSetter implements TextMapSetter { + INSTANCE; + + /** + * MNS message trails currently only support the W3C protocol; other protocol headers should be + * injected into userProperties. + */ + @Override + public void set(@Nullable Message message, String key, String value) { + if (message == null) { + return; + } + MessageSystemPropertyName systemPropertyName = MnsHelper.convert2SystemPropertyName(key); + if (systemPropertyName != null) { + message.putSystemProperty( + systemPropertyName, new MessageSystemPropertyValue(SystemPropertyType.STRING, value)); + } else { + message.getUserProperties().put(key, new MessagePropertyValue(value)); + } + } +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/SmqMockedBroker.java b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/SmqMockedBroker.java new file mode 100644 index 000000000..bff2ad8ef --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/SmqMockedBroker.java @@ -0,0 +1,232 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.mns.broker; + +import static com.aliyun.mns.common.MNSConstants.DEFAULT_CHARSET; +import static com.aliyun.mns.common.MNSConstants.X_HEADER_MNS_REQUEST_ID; +import static com.aliyun.mns.common.utils.HttpHeaders.CONTENT_TYPE; +import static io.opentelemetry.contrib.messaging.wrappers.mns.broker.SmqUtils.calculateMessageBodyMD5; +import static io.opentelemetry.contrib.messaging.wrappers.mns.broker.SmqUtils.createErrorMessage; +import static io.opentelemetry.contrib.messaging.wrappers.mns.broker.SmqUtils.generateRandomBase64String; +import static io.opentelemetry.contrib.messaging.wrappers.mns.broker.SmqUtils.generateRandomMessageId; +import static io.opentelemetry.contrib.messaging.wrappers.mns.broker.SmqUtils.generateRandomRequestId; +import static io.opentelemetry.contrib.messaging.wrappers.mns.broker.SmqUtils.inputStreamToByteArray; + +import com.aliyun.mns.model.ErrorMessage; +import com.aliyun.mns.model.Message; +import io.opentelemetry.contrib.messaging.wrappers.mns.broker.ser.ErrorMessageSerializer; +import io.opentelemetry.contrib.messaging.wrappers.mns.broker.ser.MessageDeserializer; +import io.opentelemetry.contrib.messaging.wrappers.mns.broker.ser.MessageListDeserializer; +import io.opentelemetry.contrib.messaging.wrappers.mns.broker.ser.MessageListSerializer; +import io.opentelemetry.contrib.messaging.wrappers.mns.broker.ser.MessageSerializer; +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Date; +import java.util.Deque; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Scope; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.DeleteMapping; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestParam; + +@SpringBootApplication +@SuppressWarnings({ + "JavaUtilDate", + "LockNotBeforeTry", + "SystemOut", + "PrivateConstructorForUtilityClass" +}) +public class SmqMockedBroker { + + public static void main(String[] args) { + SpringApplication.run(SmqMockedBroker.class, args); + } + + @Controller + @Scope("singleton") + public static class BrokerController { + + Lock queuesLock = new ReentrantLock(); + + Map> queues = new HashMap<>(); + + @PostMapping( + value = "/queues/{queueName}/messages", + consumes = MediaType.TEXT_XML_VALUE, + produces = MediaType.TEXT_XML_VALUE) + public ResponseEntity serveSendMessage( + @PathVariable String queueName, @RequestBody byte[] requestBody) throws Exception { + String requestId = generateRandomRequestId(); + + boolean isBatch = true; + + MessageListDeserializer listDeserializer = new MessageListDeserializer(); + List msgList = listDeserializer.deserialize(new ByteArrayInputStream(requestBody)); + + if (msgList == null) { + isBatch = false; + MessageDeserializer deserializer = new MessageDeserializer(); + Message msg = deserializer.deserialize(new ByteArrayInputStream(requestBody)); + msgList = Collections.singletonList(msg); + } + + List responses = new ArrayList<>(msgList.size()); + + for (Message msg : msgList) { + String mid = generateRandomMessageId(); + msg.setMessageId(mid); + msg.setEnqueueTime(new Date()); + msg.setPriority(8); + queuesLock.lock(); + Deque queue = queues.computeIfAbsent(queueName, (k) -> new ArrayDeque<>()); + queue.offerFirst(msg); + queuesLock.unlock(); + + Message response = new Message(); + response.setMessageId(mid); + response.setMessageBodyMD5(calculateMessageBodyMD5(msg.getMessageBody())); + responses.add(response); + } + + if (isBatch) { + MessageListSerializer listSerializer = new MessageListSerializer(); + InputStream stream = listSerializer.serialize(responses, DEFAULT_CHARSET); + return ResponseEntity.status(HttpStatus.CREATED) + .header(CONTENT_TYPE, "text/xml;charset=UTF-8") + .header(X_HEADER_MNS_REQUEST_ID, requestId) + .body(inputStreamToByteArray(stream)); + } else { + MessageSerializer serializer = new MessageSerializer(); + InputStream stream = serializer.serialize(responses.get(0), DEFAULT_CHARSET); + return ResponseEntity.status(HttpStatus.CREATED) + .header(CONTENT_TYPE, "text/xml;charset=UTF-8") + .header(X_HEADER_MNS_REQUEST_ID, requestId) + .body(inputStreamToByteArray(stream)); + } + } + + @PostMapping( + value = "/topics/{topicName}/messages", + consumes = MediaType.TEXT_XML_VALUE, + produces = MediaType.TEXT_XML_VALUE) + public ResponseEntity servePublishMessage( + @PathVariable String topicName, @RequestBody byte[] requestBody) throws Exception { + return serveSendMessage(topicName.replaceAll("topic", "queue"), requestBody); + } + + @GetMapping( + value = "/queues/{queueName}/messages", + consumes = MediaType.TEXT_XML_VALUE, + produces = MediaType.TEXT_XML_VALUE) + public ResponseEntity serveReceiveMessage( + @PathVariable String queueName, + @RequestParam(value = "numOfMessages", required = false, defaultValue = "-1") + int numOfMessages, + @RequestParam(value = "waitseconds", defaultValue = "10") int waitSec) + throws Exception { + String requestId = generateRandomRequestId(); + long timeout = System.currentTimeMillis() + waitSec * 1000L; + + queuesLock.lock(); + Deque queue = queues.computeIfAbsent(queueName, (k) -> new ArrayDeque<>()); + queuesLock.unlock(); + + List msgList = null; + while (System.currentTimeMillis() < timeout) { + if (!queue.isEmpty()) { + queuesLock.lock(); + if (!queue.isEmpty()) { + if (numOfMessages == -1) { + // receive single message + Message msg = queue.pollLast(); + msg.setFirstDequeueTime(new Date()); + msg.setNextVisibleTime(new Date(System.currentTimeMillis() + 24 * 3600 * 1000L)); + msg.setDequeueCount((msg.getDequeueCount() == null ? 0 : msg.getDequeueCount()) + 1); + msg.setReceiptHandle(msg.getPriority() + "-" + generateRandomBase64String(32)); + msgList = Collections.singletonList(msg); + } else { + msgList = new ArrayList<>(); + for (int i = 0; i < numOfMessages; i++) { + if (queue.isEmpty()) { + break; + } + Message msg = queue.pollLast(); + msg.setFirstDequeueTime(new Date()); + msg.setNextVisibleTime(new Date(System.currentTimeMillis() + 24 * 3600 * 1000L)); + msg.setDequeueCount( + (msg.getDequeueCount() == null ? 0 : msg.getDequeueCount()) + 1); + msg.setReceiptHandle(msg.getPriority() + "-" + generateRandomBase64String(32)); + msgList.add(msg); + } + } + queuesLock.unlock(); + break; + } + queuesLock.unlock(); + } + + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + break; + } + } + + if (msgList != null) { + if (numOfMessages == -1) { + // receive single message + Message msg = msgList.get(0); + MessageSerializer serializer = new MessageSerializer(); + return ResponseEntity.status(HttpStatus.OK) + .header(CONTENT_TYPE, "text/xml;charset=UTF-8") + .header(X_HEADER_MNS_REQUEST_ID, requestId) + .body(inputStreamToByteArray(serializer.serialize(msg, DEFAULT_CHARSET))); + } else { + MessageListSerializer listSerializer = new MessageListSerializer(); + return ResponseEntity.status(HttpStatus.OK) + .header(CONTENT_TYPE, "text/xml;charset=UTF-8") + .header(X_HEADER_MNS_REQUEST_ID, requestId) + .body(inputStreamToByteArray(listSerializer.serialize(msgList, DEFAULT_CHARSET))); + } + } else { + ErrorMessage errorMessage = createErrorMessage(requestId); + ErrorMessageSerializer serializer = new ErrorMessageSerializer(); + return ResponseEntity.status(HttpStatus.NOT_FOUND) + .header(CONTENT_TYPE, "text/xml;charset=UTF-8") + .header(X_HEADER_MNS_REQUEST_ID, requestId) + .body(inputStreamToByteArray(serializer.serialize(errorMessage, DEFAULT_CHARSET))); + } + } + + @DeleteMapping(value = "/queues/{queueName}/messages", consumes = MediaType.TEXT_XML_VALUE) + public ResponseEntity serveDeleteMessage( + @PathVariable String queueName, + @RequestParam(value = "ReceiptHandle", defaultValue = "") String receiptHandle, + @RequestBody(required = false) byte[] requestBody) + throws Exception { + String requestId = generateRandomRequestId(); + return ResponseEntity.status(HttpStatus.NO_CONTENT) + .header(X_HEADER_MNS_REQUEST_ID, requestId) + .build(); + } + } +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/SmqUtils.java b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/SmqUtils.java new file mode 100644 index 000000000..5bdc81bf3 --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/SmqUtils.java @@ -0,0 +1,76 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.mns.broker; + +import com.aliyun.mns.model.ErrorMessage; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.Base64; +import java.util.Locale; +import java.util.Random; +import java.util.UUID; + +public final class SmqUtils { + + public static byte[] inputStreamToByteArray(InputStream inputStream) throws IOException { + try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) { + byte[] buffer = new byte[1024]; + int bytesRead; + while ((bytesRead = inputStream.read(buffer)) != -1) { + byteArrayOutputStream.write(buffer, 0, bytesRead); + } + return byteArrayOutputStream.toByteArray(); + } + } + + public static String calculateMessageBodyMD5(String messageBody) throws NoSuchAlgorithmException { + MessageDigest md = MessageDigest.getInstance("MD5"); + byte[] hashBytes = md.digest(messageBody.getBytes(StandardCharsets.UTF_8)); + return bytesToHex(hashBytes).toUpperCase(Locale.ROOT); + } + + public static String generateRandomMessageId() { + return UUID.randomUUID().toString().replaceAll("-", "").toUpperCase(Locale.ROOT); + } + + public static String generateRandomRequestId() { + return UUID.randomUUID() + .toString() + .replaceAll("-", "") + .substring(0, 24) + .toUpperCase(Locale.ROOT); + } + + public static ErrorMessage createErrorMessage(String requestId) { + ErrorMessage errorMessage = new ErrorMessage(); + errorMessage.Code = "MessageNotExist"; + errorMessage.Message = "Message not exist."; + errorMessage.RequestId = requestId; + errorMessage.HostId = "http://test.mns.cn-hangzhou.aliyuncs.com"; + return errorMessage; + } + + public static String generateRandomBase64String(int length) { + byte[] randomBytes = new byte[length]; + new Random().nextBytes(randomBytes); + + return Base64.getEncoder().encodeToString(randomBytes); + } + + private static String bytesToHex(byte[] bytes) { + StringBuilder sb = new StringBuilder(); + for (byte b : bytes) { + sb.append(String.format(Locale.ROOT, "%02X", b)); + } + return sb.toString(); + } + + private SmqUtils() {} +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/ErrorMessageSerializer.java b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/ErrorMessageSerializer.java new file mode 100644 index 000000000..bfea33c2e --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/ErrorMessageSerializer.java @@ -0,0 +1,28 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.mns.broker.ser; + +import com.aliyun.mns.model.ErrorMessage; +import com.aliyun.mns.model.serialize.XmlUtil; +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import org.w3c.dom.Document; +import org.w3c.dom.Element; + +public class ErrorMessageSerializer extends XMLSerializer { + + @Override + public InputStream serialize(ErrorMessage msg, String encoding) throws Exception { + Document doc = getDocumentBuilder().newDocument(); + + Element root = serializeError(doc, msg); + doc.appendChild(root); + + String xml = XmlUtil.xmlNodeToString(doc, encoding); + + return new ByteArrayInputStream(xml.getBytes(encoding)); + } +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/MessageDeserializer.java b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/MessageDeserializer.java new file mode 100644 index 000000000..e3fba1bdc --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/MessageDeserializer.java @@ -0,0 +1,22 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.mns.broker.ser; + +import com.aliyun.mns.model.Message; +import java.io.InputStream; +import org.w3c.dom.Document; +import org.w3c.dom.Element; + +public class MessageDeserializer extends XMLDeserializer { + + @Override + public Message deserialize(InputStream stream) throws Exception { + Document doc = getDocumentBuilder().parse(stream); + + Element root = doc.getDocumentElement(); + return (Message) parseMessage(root); + } +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/MessageListDeserializer.java b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/MessageListDeserializer.java new file mode 100644 index 000000000..be3f38b2f --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/MessageListDeserializer.java @@ -0,0 +1,38 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.mns.broker.ser; + +import static com.aliyun.mns.common.MNSConstants.MESSAGE_TAG; + +import com.aliyun.mns.model.Message; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; +import org.w3c.dom.Document; +import org.w3c.dom.Element; +import org.w3c.dom.NodeList; + +public class MessageListDeserializer extends XMLDeserializer> { + @Override + public List deserialize(InputStream stream) throws Exception { + Document doc = getDocumentBuilder().parse(stream); + return deserialize(doc); + } + + public List deserialize(Document doc) { + NodeList list = doc.getElementsByTagName(MESSAGE_TAG); + if (list != null && list.getLength() > 0) { + List results = new ArrayList(); + + for (int i = 0; i < list.getLength(); i++) { + Message msg = (Message) parseMessage((Element) list.item(i)); + results.add(msg); + } + return results; + } + return new ArrayList<>(); + } +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/MessageListSerializer.java b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/MessageListSerializer.java new file mode 100644 index 000000000..8b08babb8 --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/MessageListSerializer.java @@ -0,0 +1,39 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.mns.broker.ser; + +import static com.aliyun.mns.common.MNSConstants.DEFAULT_XML_NAMESPACE; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_LIST_TAG; + +import com.aliyun.mns.model.Message; +import com.aliyun.mns.model.serialize.XmlUtil; +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.util.List; +import org.w3c.dom.Document; +import org.w3c.dom.Element; + +public class MessageListSerializer extends XMLSerializer> { + + @Override + public InputStream serialize(List msgs, String encoding) throws Exception { + Document doc = getDocumentBuilder().newDocument(); + + Element messages = doc.createElementNS(DEFAULT_XML_NAMESPACE, MESSAGE_LIST_TAG); + + doc.appendChild(messages); + + if (msgs != null) { + for (Message msg : msgs) { + Element root = serializeMessage(doc, msg); + messages.appendChild(root); + } + } + String xml = XmlUtil.xmlNodeToString(doc, encoding); + + return new ByteArrayInputStream(xml.getBytes(encoding)); + } +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/MessageSerializer.java b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/MessageSerializer.java new file mode 100644 index 000000000..935618086 --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/MessageSerializer.java @@ -0,0 +1,32 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.mns.broker.ser; + +import com.aliyun.mns.model.Message; +import com.aliyun.mns.model.serialize.XmlUtil; +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import org.w3c.dom.Document; +import org.w3c.dom.Element; + +public class MessageSerializer extends XMLSerializer { + + public MessageSerializer() { + super(); + } + + @Override + public InputStream serialize(Message msg, String encoding) throws Exception { + Document doc = getDocumentBuilder().newDocument(); + + Element root = serializeMessage(doc, msg); + doc.appendChild(root); + + String xml = XmlUtil.xmlNodeToString(doc, encoding); + + return new ByteArrayInputStream(xml.getBytes(encoding)); + } +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/XMLDeserializer.java b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/XMLDeserializer.java new file mode 100644 index 000000000..72f4117de --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/XMLDeserializer.java @@ -0,0 +1,201 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.mns.broker.ser; + +import static com.aliyun.mns.common.MNSConstants.DEQUEUE_COUNT_TAG; +import static com.aliyun.mns.common.MNSConstants.ENQUEUE_TIME_TAG; +import static com.aliyun.mns.common.MNSConstants.FIRST_DEQUEUE_TIME_TAG; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_BODY_MD5_TAG; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_BODY_TAG; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_ERRORCODE_TAG; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_ERRORMESSAGE_TAG; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_ID_TAG; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_PROPERTY_TAG; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_SYSTEM_PROPERTY_TAG; +import static com.aliyun.mns.common.MNSConstants.NEXT_VISIBLE_TIME_TAG; +import static com.aliyun.mns.common.MNSConstants.PRIORITY_TAG; +import static com.aliyun.mns.common.MNSConstants.PROPERTY_NAME_TAG; +import static com.aliyun.mns.common.MNSConstants.PROPERTY_TYPE_TAG; +import static com.aliyun.mns.common.MNSConstants.PROPERTY_VALUE_TAG; +import static com.aliyun.mns.common.MNSConstants.RECEIPT_HANDLE_TAG; +import static com.aliyun.mns.common.MNSConstants.SYSTEM_PROPERTIES_TAG; +import static com.aliyun.mns.common.MNSConstants.USER_PROPERTIES_TAG; +import static java.nio.charset.StandardCharsets.UTF_8; + +import com.aliyun.mns.model.BaseMessage; +import com.aliyun.mns.model.ErrorMessageResult; +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 com.aliyun.mns.model.PropertyType; +import com.aliyun.mns.model.SystemPropertyType; +import com.aliyun.mns.model.serialize.BaseXMLSerializer; +import com.aliyun.mns.model.serialize.Deserializer; +import java.util.ArrayList; +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.commons.codec.binary.Base64; +import org.w3c.dom.Element; +import org.w3c.dom.Node; +import org.w3c.dom.NodeList; + +@SuppressWarnings("JavaUtilDate") +public abstract class XMLDeserializer extends BaseXMLSerializer implements Deserializer { + + protected String safeGetElementContent(Element root, String tagName, String defaultValue) { + NodeList nodes = root.getElementsByTagName(tagName); + if (nodes != null) { + Node node = nodes.item(0); + if (node == null) { + return defaultValue; + } else { + return node.getTextContent(); + } + } + return defaultValue; + } + + protected Element safeGetElement(Element root, String tagName) { + NodeList nodes = root.getElementsByTagName(tagName); + if (nodes != null) { + Node node = nodes.item(0); + if (node == null) { + return null; + } else { + return (Element) node; + } + } + return null; + } + + protected List safeGetElements(Element parent, String tagName) { + NodeList nodeList = parent.getElementsByTagName(tagName); + List elements = new ArrayList(); + for (int i = 0; i < nodeList.getLength(); i++) { + Node node = nodeList.item(i); + if (node.getNodeType() == Node.ELEMENT_NODE) { + elements.add((Element) node); + } + } + return elements; + } + + protected ErrorMessageResult parseErrorMessageResult(Element root) { + ErrorMessageResult result = new ErrorMessageResult(); + String errorCode = safeGetElementContent(root, MESSAGE_ERRORCODE_TAG, null); + result.setErrorCode(errorCode); + + String errorMessage = safeGetElementContent(root, MESSAGE_ERRORMESSAGE_TAG, null); + result.setErrorMessage(errorMessage); + return result; + } + + protected BaseMessage parseMessage(Element root) { + Message message = new Message(); + + String messageId = safeGetElementContent(root, MESSAGE_ID_TAG, null); + if (messageId != null) { + message.setMessageId(messageId); + } + + String messageBody = safeGetElementContent(root, MESSAGE_BODY_TAG, null); + if (messageBody != null) { + message.setMessageBody(messageBody, Message.MessageBodyType.RAW_STRING); + } + + String messageBodyMD5 = safeGetElementContent(root, MESSAGE_BODY_MD5_TAG, null); + message.setMessageBodyMD5(messageBodyMD5); + + String receiptHandle = safeGetElementContent(root, RECEIPT_HANDLE_TAG, null); + message.setReceiptHandle(receiptHandle); + + String enqueueTime = safeGetElementContent(root, ENQUEUE_TIME_TAG, null); + if (enqueueTime != null) { + message.setEnqueueTime(new Date(Long.parseLong(enqueueTime))); + } + + String nextVisibleTime = safeGetElementContent(root, NEXT_VISIBLE_TIME_TAG, null); + if (nextVisibleTime != null) { + message.setNextVisibleTime(new Date(Long.parseLong(nextVisibleTime))); + } + + String firstDequeueTime = safeGetElementContent(root, FIRST_DEQUEUE_TIME_TAG, null); + if (firstDequeueTime != null) { + message.setFirstDequeueTime(new Date(Long.parseLong(firstDequeueTime))); + } + + String dequeueCount = safeGetElementContent(root, DEQUEUE_COUNT_TAG, null); + if (dequeueCount != null) { + message.setDequeueCount(Integer.parseInt(dequeueCount)); + } + + String priority = safeGetElementContent(root, PRIORITY_TAG, null); + if (priority != null) { + message.setPriority(Integer.parseInt(priority)); + } + + // 解析 userProperties + safeAddPropertiesToMessage(root, message); + + // 解析 systemProperties + safeAddSystemPropertiesToMessage(root, message); + + return message; + } + + protected void safeAddPropertiesToMessage(Element root, Message message) { + Element userPropertiesElement = safeGetElement(root, USER_PROPERTIES_TAG); + if (userPropertiesElement != null) { + Map userProperties = message.getUserProperties(); + if (userProperties == null) { + userProperties = new HashMap(); + message.setUserProperties(userProperties); + } + + for (Element propertyValueElement : + safeGetElements(userPropertiesElement, MESSAGE_PROPERTY_TAG)) { + String name = safeGetElementContent(propertyValueElement, PROPERTY_NAME_TAG, null); + String value = safeGetElementContent(propertyValueElement, PROPERTY_VALUE_TAG, null); + String type = safeGetElementContent(propertyValueElement, PROPERTY_TYPE_TAG, null); + + if (name != null && value != null && type != null) { + PropertyType typeEnum = PropertyType.valueOf(type); + // 如果是二进制类型，需要base64解码 + if (typeEnum == PropertyType.BINARY) { + byte[] decodedBytes = Base64.decodeBase64(value); + value = new String(decodedBytes, UTF_8); + } + MessagePropertyValue propertyValue = + new MessagePropertyValue(PropertyType.valueOf(type), value); + userProperties.put(name, propertyValue); + } + } + } + } + + protected void safeAddSystemPropertiesToMessage(Element root, Message message) { + Element systemPropertiesElement = safeGetElement(root, SYSTEM_PROPERTIES_TAG); + if (systemPropertiesElement != null) { + for (Element propertyValueElement : + safeGetElements(systemPropertiesElement, MESSAGE_SYSTEM_PROPERTY_TAG)) { + String name = safeGetElementContent(propertyValueElement, PROPERTY_NAME_TAG, null); + String value = safeGetElementContent(propertyValueElement, PROPERTY_VALUE_TAG, null); + String type = safeGetElementContent(propertyValueElement, PROPERTY_TYPE_TAG, null); + + if (name != null && value != null && type != null) { + SystemPropertyType systemPropertyType = SystemPropertyType.valueOf(type); + MessageSystemPropertyValue propertyValue = + new MessageSystemPropertyValue(systemPropertyType, value); + MessageSystemPropertyName propertyName = MessageSystemPropertyName.getByValue(name); + message.putSystemProperty(propertyName, propertyValue); + } + } + } + } +} diff --git a/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/XMLSerializer.java b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/XMLSerializer.java new file mode 100644 index 000000000..1dfd8d99a --- /dev/null +++ b/messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/broker/ser/XMLSerializer.java @@ -0,0 +1,200 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.mns.broker.ser; + +import static com.aliyun.mns.common.MNSConstants.DEFAULT_XML_NAMESPACE; +import static com.aliyun.mns.common.MNSConstants.DELAY_SECONDS_TAG; +import static com.aliyun.mns.common.MNSConstants.DEQUEUE_COUNT_TAG; +import static com.aliyun.mns.common.MNSConstants.ENQUEUE_TIME_TAG; +import static com.aliyun.mns.common.MNSConstants.ERROR_CODE_TAG; +import static com.aliyun.mns.common.MNSConstants.ERROR_HOST_ID_TAG; +import static com.aliyun.mns.common.MNSConstants.ERROR_REQUEST_ID_TAG; +import static com.aliyun.mns.common.MNSConstants.ERROR_TAG; +import static com.aliyun.mns.common.MNSConstants.FIRST_DEQUEUE_TIME_TAG; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_BODY_MD5_TAG; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_BODY_TAG; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_ID_TAG; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_PROPERTY_TAG; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_SYSTEM_PROPERTY_TAG; +import static com.aliyun.mns.common.MNSConstants.MESSAGE_TAG; +import static com.aliyun.mns.common.MNSConstants.NEXT_VISIBLE_TIME_TAG; +import static com.aliyun.mns.common.MNSConstants.PRIORITY_TAG; +import static com.aliyun.mns.common.MNSConstants.PROPERTY_NAME_TAG; +import static com.aliyun.mns.common.MNSConstants.PROPERTY_TYPE_TAG; +import static com.aliyun.mns.common.MNSConstants.PROPERTY_VALUE_TAG; +import static com.aliyun.mns.common.MNSConstants.RECEIPT_HANDLE_TAG; +import static com.aliyun.mns.common.MNSConstants.SYSTEM_PROPERTIES_TAG; +import static com.aliyun.mns.common.MNSConstants.USER_PROPERTIES_TAG; + +import com.aliyun.mns.model.AbstractMessagePropertyValue; +import com.aliyun.mns.model.ErrorMessage; +import com.aliyun.mns.model.Message; +import com.aliyun.mns.model.serialize.BaseXMLSerializer; +import com.aliyun.mns.model.serialize.Serializer; +import java.util.Date; +import java.util.Map; +import org.w3c.dom.Document; +import org.w3c.dom.Element; + +@SuppressWarnings("JavaUtilDate") +public abstract class XMLSerializer extends BaseXMLSerializer implements Serializer { + + public Element safeCreateContentElement( + Document doc, String tagName, Object value, String defaultValue) { + if (value == null && defaultValue == null) { + return null; + } + + Element node = doc.createElement(tagName); + if (value instanceof Date) { + node.setTextContent(String.valueOf(((Date) value).getTime())); + } else if (value != null) { + node.setTextContent(value.toString()); + } else { + node.setTextContent(defaultValue); + } + return node; + } + + public Element serializeError(Document doc, ErrorMessage msg) { + Element root = doc.createElementNS(DEFAULT_XML_NAMESPACE, ERROR_TAG); + + Element node = safeCreateContentElement(doc, ERROR_CODE_TAG, msg.Code, "MessageNotExist"); + + if (node != null) { + root.appendChild(node); + } + + node = safeCreateContentElement(doc, MESSAGE_TAG, msg.Message, null); + + if (node != null) { + root.appendChild(node); + } + + node = safeCreateContentElement(doc, ERROR_REQUEST_ID_TAG, msg.RequestId, null); + + if (node != null) { + root.appendChild(node); + } + + node = safeCreateContentElement(doc, ERROR_HOST_ID_TAG, msg.HostId, null); + + if (node != null) { + root.appendChild(node); + } + + return root; + } + + public Element serializeMessage(Document doc, Message msg) { + Element root = doc.createElementNS(DEFAULT_XML_NAMESPACE, MESSAGE_TAG); + + Element node = safeCreateContentElement(doc, MESSAGE_ID_TAG, msg.getMessageId(), null); + + if (node != null) { + root.appendChild(node); + } + + node = safeCreateContentElement(doc, MESSAGE_BODY_TAG, msg.getOriginalMessageBody(), ""); + + if (node != null) { + root.appendChild(node); + } + + node = safeCreateContentElement(doc, MESSAGE_BODY_MD5_TAG, msg.getMessageBodyMD5(), null); + + if (node != null) { + root.appendChild(node); + } + + node = safeCreateContentElement(doc, DELAY_SECONDS_TAG, msg.getDelaySeconds(), null); + if (node != null) { + root.appendChild(node); + } + + node = safeCreateContentElement(doc, PRIORITY_TAG, msg.getPriority(), null); + if (node != null) { + root.appendChild(node); + } + + node = + safeCreatePropertiesNode( + doc, msg.getUserProperties(), USER_PROPERTIES_TAG, MESSAGE_PROPERTY_TAG); + if (node != null) { + root.appendChild(node); + } + + node = + safeCreatePropertiesNode( + doc, msg.getSystemProperties(), SYSTEM_PROPERTIES_TAG, MESSAGE_SYSTEM_PROPERTY_TAG); + if (node != null) { + root.appendChild(node); + } + + node = safeCreateContentElement(doc, RECEIPT_HANDLE_TAG, msg.getReceiptHandle(), null); + if (node != null) { + root.appendChild(node); + } + + node = safeCreateContentElement(doc, ENQUEUE_TIME_TAG, msg.getEnqueueTime(), null); + if (node != null) { + root.appendChild(node); + } + + node = safeCreateContentElement(doc, NEXT_VISIBLE_TIME_TAG, msg.getNextVisibleTime(), null); + if (node != null) { + root.appendChild(node); + } + + node = safeCreateContentElement(doc, FIRST_DEQUEUE_TIME_TAG, msg.getFirstDequeueTime(), null); + if (node != null) { + root.appendChild(node); + } + + node = safeCreateContentElement(doc, DEQUEUE_COUNT_TAG, msg.getDequeueCount(), null); + if (node != null) { + root.appendChild(node); + } + + return root; + } + + public Element safeCreatePropertiesNode( + Document doc, + Map map, + String nodeName, + String propertyNodeName) { + if (map == null || map.isEmpty()) { + return null; + } + Element propertiesNode = doc.createElement(nodeName); + for (Map.Entry entry : map.entrySet()) { + Element propNode = doc.createElement(propertyNodeName); + + Element nameNode = safeCreateContentElement(doc, PROPERTY_NAME_TAG, entry.getKey(), null); + if (nameNode != null) { + propNode.appendChild(nameNode); + } + + Element valueNode = + safeCreateContentElement( + doc, PROPERTY_VALUE_TAG, entry.getValue().getStringValueByType(), null); + if (valueNode != null) { + propNode.appendChild(valueNode); + } + + Element typeNode = + safeCreateContentElement( + doc, PROPERTY_TYPE_TAG, entry.getValue().getDataTypeString(), null); + if (typeNode != null) { + propNode.appendChild(typeNode); + } + + propertiesNode.appendChild(propNode); + } + return propertiesNode; + } +} diff --git a/messaging-wrappers/api/build.gradle.kts b/messaging-wrappers/api/build.gradle.kts new file mode 100644 index 000000000..cc27cc6e0 --- /dev/null +++ b/messaging-wrappers/api/build.gradle.kts @@ -0,0 +1,29 @@ +plugins { + id("otel.java-conventions") + + id("otel.publish-conventions") +} + +description = "OpenTelemetry Messaging Wrappers" +otelJava.moduleName.set("io.opentelemetry.contrib.messaging.wrappers") + +dependencies { + api("io.opentelemetry.instrumentation:opentelemetry-instrumentation-api-incubator") + api("io.opentelemetry.semconv:opentelemetry-semconv") + api("io.opentelemetry.semconv:opentelemetry-semconv-incubating") + + compileOnly("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure-spi") + compileOnly("io.opentelemetry:opentelemetry-api-incubator") + + testImplementation("com.google.guava:guava:33.4.8-jre") + testImplementation(project(":messaging-wrappers:testing")) +} + +tasks { + withType().configureEach { + jvmArgs("-Dotel.java.global-autoconfigure.enabled=true") + jvmArgs("-Dotel.traces.exporter=logging") + jvmArgs("-Dotel.metrics.exporter=logging") + jvmArgs("-Dotel.logs.exporter=logging") + } +} diff --git a/messaging-wrappers/api/gradle.properties b/messaging-wrappers/api/gradle.properties new file mode 100644 index 000000000..a0402e1e2 --- /dev/null +++ b/messaging-wrappers/api/gradle.properties @@ -0,0 +1,2 @@ +# TODO: uncomment when ready to mark as stable +# otel.stable=true diff --git a/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/MessagingProcessWrapper.java b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/MessagingProcessWrapper.java new file mode 100644 index 000000000..073927507 --- /dev/null +++ b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/MessagingProcessWrapper.java @@ -0,0 +1,112 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers; + +import static io.opentelemetry.api.trace.SpanKind.CONSUMER; + +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanBuilder; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import io.opentelemetry.context.propagation.TextMapGetter; +import io.opentelemetry.context.propagation.TextMapPropagator; +import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor; +import io.opentelemetry.instrumentation.api.instrumenter.SpanNameExtractor; +import java.util.List; +import javax.annotation.Nullable; + +public class MessagingProcessWrapper { + + private static final String INSTRUMENTATION_SCOPE = "messaging-process-wrapper"; + + private static final String INSTRUMENTATION_VERSION = "1.0.0"; + + private final TextMapPropagator textMapPropagator; + + private final Tracer tracer; + + private final TextMapGetter textMapGetter; + + private final SpanNameExtractor spanNameExtractor; + + // no attributes need to be extracted from responses in process operations + private final List> attributesExtractors; + + public static MessagingProcessWrapperBuilder defaultBuilder() { + return new MessagingProcessWrapperBuilder<>(); + } + + public void doProcess(REQUEST request, ThrowingRunnable runnable) + throws E { + Span span = handleStart(request); + Context context = span.storeInContext(Context.current()); + + try (Scope scope = context.makeCurrent()) { + runnable.run(); + } catch (Throwable t) { + handleEnd(span, context, request, t); + throw t; + } + + handleEnd(span, context, request, null); + } + + public R doProcess(REQUEST request, ThrowingSupplier supplier) + throws E { + Span span = handleStart(request); + Context context = span.storeInContext(Context.current()); + + R result = null; + try (Scope scope = context.makeCurrent()) { + result = supplier.get(); + } catch (Throwable t) { + handleEnd(span, context, request, t); + throw t; + } + + handleEnd(span, context, request, null); + return result; + } + + protected Span handleStart(REQUEST request) { + Context context = + this.textMapPropagator.extract(Context.current(), request, this.textMapGetter); + SpanBuilder spanBuilder = this.tracer.spanBuilder(spanNameExtractor.extract(request)); + spanBuilder.setSpanKind(CONSUMER).setParent(context); + + AttributesBuilder builder = Attributes.builder(); + for (AttributesExtractor extractor : this.attributesExtractors) { + extractor.onStart(builder, context, request); + } + return spanBuilder.setAllAttributes(builder.build()).startSpan(); + } + + protected void handleEnd(Span span, Context context, REQUEST request, @Nullable Throwable t) { + AttributesBuilder builder = Attributes.builder(); + for (AttributesExtractor extractor : this.attributesExtractors) { + extractor.onEnd(builder, context, request, null, t); + } + + span.setAllAttributes(builder.build()); + span.end(); + } + + protected MessagingProcessWrapper( + OpenTelemetry openTelemetry, + TextMapGetter textMapGetter, + SpanNameExtractor spanNameExtractor, + List> attributesExtractors) { + this.textMapPropagator = openTelemetry.getPropagators().getTextMapPropagator(); + this.tracer = openTelemetry.getTracer(INSTRUMENTATION_SCOPE, INSTRUMENTATION_VERSION); + this.textMapGetter = textMapGetter; + this.spanNameExtractor = spanNameExtractor; + this.attributesExtractors = attributesExtractors; + } +} diff --git a/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/MessagingProcessWrapperBuilder.java b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/MessagingProcessWrapperBuilder.java new file mode 100644 index 000000000..b8438e6f1 --- /dev/null +++ b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/MessagingProcessWrapperBuilder.java @@ -0,0 +1,70 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers; + +import static java.util.Objects.requireNonNull; + +import com.google.errorprone.annotations.CanIgnoreReturnValue; +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.context.propagation.TextMapGetter; +import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor; +import io.opentelemetry.instrumentation.api.instrumenter.SpanNameExtractor; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import javax.annotation.Nullable; + +public class MessagingProcessWrapperBuilder { + + @Nullable private OpenTelemetry openTelemetry; + + @Nullable protected TextMapGetter textMapGetter; + + @Nullable protected SpanNameExtractor spanNameExtractor; + + @Nullable protected List> attributesExtractors; + + @CanIgnoreReturnValue + public MessagingProcessWrapperBuilder openTelemetry(OpenTelemetry openTelemetry) { + this.openTelemetry = openTelemetry; + return this; + } + + @CanIgnoreReturnValue + public MessagingProcessWrapperBuilder textMapGetter( + TextMapGetter textMapGetter) { + this.textMapGetter = textMapGetter; + return this; + } + + @CanIgnoreReturnValue + public MessagingProcessWrapperBuilder spanNameExtractor( + SpanNameExtractor spanNameExtractor) { + this.spanNameExtractor = spanNameExtractor; + return this; + } + + @CanIgnoreReturnValue + public MessagingProcessWrapperBuilder attributesExtractors( + Collection> attributesExtractors) { + this.attributesExtractors = new ArrayList<>(); + this.attributesExtractors.addAll(attributesExtractors); + return this; + } + + public MessagingProcessWrapper build() { + requireNonNull(this.spanNameExtractor); + requireNonNull(this.attributesExtractors); + return new MessagingProcessWrapper<>( + this.openTelemetry == null ? GlobalOpenTelemetry.get() : this.openTelemetry, + this.textMapGetter == null ? NoopTextMapGetter.create() : this.textMapGetter, + this.spanNameExtractor, + this.attributesExtractors); + } + + protected MessagingProcessWrapperBuilder() {} +} diff --git a/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/NoopTextMapGetter.java b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/NoopTextMapGetter.java new file mode 100644 index 000000000..2237d5a6a --- /dev/null +++ b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/NoopTextMapGetter.java @@ -0,0 +1,30 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers; + +import io.opentelemetry.context.propagation.TextMapGetter; +import java.util.Collections; +import javax.annotation.Nullable; + +public class NoopTextMapGetter implements TextMapGetter { + + public static TextMapGetter create() { + return new NoopTextMapGetter<>(); + } + + @Override + public Iterable keys(REQUEST request) { + return Collections.emptyList(); + } + + @Nullable + @Override + public String get(@Nullable REQUEST request, String s) { + return null; + } + + private NoopTextMapGetter() {} +} diff --git a/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/ThrowingRunnable.java b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/ThrowingRunnable.java new file mode 100644 index 000000000..e1bbd05b0 --- /dev/null +++ b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/ThrowingRunnable.java @@ -0,0 +1,19 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers; + +/** + * A utility interface representing a {@link Runnable} that may throw. + * + *

Inspired from ThrowingRunnable. + * + * @param Thrown exception type. + */ +@FunctionalInterface +public interface ThrowingRunnable { + void run() throws E; +} diff --git a/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/ThrowingSupplier.java b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/ThrowingSupplier.java new file mode 100644 index 000000000..9bec00d4b --- /dev/null +++ b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/ThrowingSupplier.java @@ -0,0 +1,21 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers; + +import java.util.function.Supplier; + +/** + * A utility interface representing a {@link Supplier} that may throw. + * + *

Inspired from ThrowingSupplier. + * + * @param Thrown exception type. + */ +@FunctionalInterface +public interface ThrowingSupplier { + T get() throws E; +} diff --git a/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/package-info.java b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/package-info.java new file mode 100644 index 000000000..502b96ca6 --- /dev/null +++ b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/package-info.java @@ -0,0 +1,2 @@ +/** OpenTelemetry messaging wrappers extension. */ +package io.opentelemetry.contrib.messaging.wrappers; diff --git a/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/semconv/DefaultMessageTextMapGetter.java b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/semconv/DefaultMessageTextMapGetter.java new file mode 100644 index 000000000..5643a87d2 --- /dev/null +++ b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/semconv/DefaultMessageTextMapGetter.java @@ -0,0 +1,31 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.semconv; + +import io.opentelemetry.context.propagation.TextMapGetter; +import java.util.List; +import javax.annotation.Nullable; + +public enum DefaultMessageTextMapGetter implements TextMapGetter { + INSTANCE; + + @Override + public Iterable keys(MessagingProcessRequest carrier) { + return carrier.getAllMessageHeadersKey(); + } + + @Nullable + @Override + public String get(@Nullable MessagingProcessRequest carrier, String key) { + if (carrier != null) { + List messageHeader = carrier.getMessageHeader(key); + if (messageHeader != null && !messageHeader.isEmpty()) { + return messageHeader.get(0); + } + } + return null; + } +} diff --git a/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/semconv/DefaultMessagingAttributesGetter.java b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/semconv/DefaultMessagingAttributesGetter.java new file mode 100644 index 000000000..67f0f543e --- /dev/null +++ b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/semconv/DefaultMessagingAttributesGetter.java @@ -0,0 +1,93 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.semconv; + +import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter; +import java.util.List; +import javax.annotation.Nullable; + +public enum DefaultMessagingAttributesGetter + implements MessagingAttributesGetter { + INSTANCE; + + @Nullable + @Override + public String getDestinationPartitionId(MessagingProcessRequest messagingProcessRequest) { + return messagingProcessRequest.getDestinationPartitionId(); + } + + @Override + public List getMessageHeader( + MessagingProcessRequest messagingProcessRequest, String name) { + return messagingProcessRequest.getMessageHeader(name); + } + + @Nullable + @Override + public String getSystem(MessagingProcessRequest messagingProcessRequest) { + return messagingProcessRequest.getSystem(); + } + + @Nullable + @Override + public String getDestination(MessagingProcessRequest messagingProcessRequest) { + return messagingProcessRequest.getDestination(); + } + + @Nullable + @Override + public String getDestinationTemplate(MessagingProcessRequest messagingProcessRequest) { + return messagingProcessRequest.getDestinationTemplate(); + } + + @Override + public boolean isTemporaryDestination(MessagingProcessRequest messagingProcessRequest) { + return messagingProcessRequest.isTemporaryDestination(); + } + + @Override + public boolean isAnonymousDestination(MessagingProcessRequest messagingProcessRequest) { + return messagingProcessRequest.isAnonymousDestination(); + } + + @Nullable + @Override + public String getConversationId(MessagingProcessRequest messagingProcessRequest) { + return messagingProcessRequest.getConversationId(); + } + + @Nullable + @Override + public Long getMessageBodySize(MessagingProcessRequest messagingProcessRequest) { + return messagingProcessRequest.getMessageBodySize(); + } + + @Nullable + @Override + public Long getMessageEnvelopeSize(MessagingProcessRequest messagingProcessRequest) { + return messagingProcessRequest.getMessageEnvelopeSize(); + } + + @Nullable + @Override + public String getMessageId( + MessagingProcessRequest messagingProcessRequest, @Nullable Void unused) { + return messagingProcessRequest.getMessageId(); + } + + @Nullable + @Override + public String getClientId(MessagingProcessRequest messagingProcessRequest) { + return messagingProcessRequest.getClientId(); + } + + @Nullable + @Override + public Long getBatchMessageCount( + MessagingProcessRequest messagingProcessRequest, @Nullable Void unused) { + return messagingProcessRequest.getBatchMessageCount(); + } +} diff --git a/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/semconv/MessagingProcessRequest.java b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/semconv/MessagingProcessRequest.java new file mode 100644 index 000000000..119b95895 --- /dev/null +++ b/messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/semconv/MessagingProcessRequest.java @@ -0,0 +1,81 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.semconv; + +import static java.util.Collections.emptyList; + +import java.util.Collection; +import java.util.List; +import javax.annotation.Nullable; + +/** + * An interface to expose messaging properties for the pre-defined process wrapper. + * + *

Inspired from MessagingAttributesGetter. + */ +public interface MessagingProcessRequest { + + String getSystem(); + + @Nullable + String getDestination(); + + @Nullable + String getDestinationTemplate(); + + boolean isTemporaryDestination(); + + boolean isAnonymousDestination(); + + @Nullable + String getConversationId(); + + @Nullable + Long getMessageBodySize(); + + @Nullable + Long getMessageEnvelopeSize(); + + @Nullable + String getMessageId(); + + @Nullable + default String getClientId() { + return null; + } + + @Nullable + default Long getBatchMessageCount() { + return null; + } + + @Nullable + default String getDestinationPartitionId() { + return null; + } + + /** + * Extracts all values of header named {@code name} from the request, or an empty list if there + * were none. + * + *

Implementations of this method must not return a null value; an empty list should be + * returned instead. + */ + default List getMessageHeader(String name) { + return emptyList(); + } + + /** + * Extracts all keys of headers from the request, or an empty list/set if there were none. + * + *

Implementations of this method must not return a null value; an empty list should be + * returned instead. + */ + default Collection getAllMessageHeadersKey() { + return emptyList(); + } +} diff --git a/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/TestConstants.java b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/TestConstants.java new file mode 100644 index 000000000..a826c2393 --- /dev/null +++ b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/TestConstants.java @@ -0,0 +1,19 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers; + +public final class TestConstants { + + public static final String MESSAGE_ID = "42"; + + public static final String MESSAGE_BODY = "Hello messaging wrapper!"; + + public static final String EVENTBUS_NAME = "test-eb"; + + public static final String CLIENT_ID = "eventbus-client-0"; + + private TestConstants() {} +} diff --git a/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/UserDefinedMessageSystemTest.java b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/UserDefinedMessageSystemTest.java new file mode 100644 index 000000000..9cbbdf38a --- /dev/null +++ b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/UserDefinedMessageSystemTest.java @@ -0,0 +1,140 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers; + +import static io.opentelemetry.contrib.messaging.wrappers.TestConstants.CLIENT_ID; +import static io.opentelemetry.contrib.messaging.wrappers.TestConstants.EVENTBUS_NAME; +import static io.opentelemetry.contrib.messaging.wrappers.TestConstants.MESSAGE_BODY; +import static io.opentelemetry.contrib.messaging.wrappers.TestConstants.MESSAGE_ID; +import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo; +import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_BODY_SIZE; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_ID; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM; + +import com.google.common.eventbus.EventBus; +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import io.opentelemetry.contrib.messaging.wrappers.model.Message; +import io.opentelemetry.contrib.messaging.wrappers.model.MessageListener; +import io.opentelemetry.contrib.messaging.wrappers.model.MessageTextMapSetter; +import io.opentelemetry.contrib.messaging.wrappers.semconv.DefaultMessageTextMapGetter; +import io.opentelemetry.contrib.messaging.wrappers.semconv.DefaultMessagingAttributesGetter; +import io.opentelemetry.contrib.messaging.wrappers.semconv.MessagingProcessRequest; +import io.opentelemetry.contrib.messaging.wrappers.testing.AbstractBaseTest; +import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessageOperation; +import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesExtractor; +import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanNameExtractor; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.HashMap; +import org.assertj.core.api.AbstractAssert; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; + +@SuppressWarnings("OtelInternalJavadoc") +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class UserDefinedMessageSystemTest extends AbstractBaseTest { + + private OpenTelemetry otel; + + private Tracer tracer; + + private EventBus eventBus; + + @BeforeAll + void setupClass() { + otel = GlobalOpenTelemetry.get(); + tracer = otel.getTracer("test-tracer", "1.0.0"); + MessagingProcessWrapper wrapper = + MessagingProcessWrapper.defaultBuilder() + .openTelemetry(otel) + .textMapGetter(DefaultMessageTextMapGetter.INSTANCE) + .spanNameExtractor( + MessagingSpanNameExtractor.create( + DefaultMessagingAttributesGetter.INSTANCE, MessageOperation.PROCESS)) + .attributesExtractors( + Collections.singletonList( + MessagingAttributesExtractor.create( + DefaultMessagingAttributesGetter.INSTANCE, MessageOperation.PROCESS))) + .build(); + + eventBus = new EventBus(); + + eventBus.register(MessageListener.create(tracer, wrapper)); + } + + @Test + void testSendAndConsume() { + sendWithParent(tracer); + + assertTraces(); + } + + public void sendWithParent(Tracer tracer) { + // mock a send span + Span parent = + tracer.spanBuilder("publish " + EVENTBUS_NAME).setSpanKind(SpanKind.PRODUCER).startSpan(); + + try (Scope scope = parent.makeCurrent()) { + Message message = Message.create(new HashMap<>(), MESSAGE_ID, MESSAGE_BODY); + otel.getPropagators() + .getTextMapPropagator() + .inject(Context.current(), message, MessageTextMapSetter.create()); + eventBus.post(message); + } + + parent.end(); + } + + /** + * Copied from testing-common. + */ + @SuppressWarnings("deprecation") // using deprecated semconv + public void assertTraces() { + waitAndAssertTraces( + sortByRootSpanName("parent", "producer callback"), + trace -> + trace.hasSpansSatisfyingExactly( + span -> + // No need to verify the attribute here because it is generated by + // instrumentation library. + span.hasName("publish " + EVENTBUS_NAME) + .hasKind(SpanKind.PRODUCER) + .hasNoParent(), + span -> + span.hasName(EVENTBUS_NAME + " process") + .hasKind(SpanKind.CONSUMER) + .hasParent(trace.getSpan(0)) + .hasAttributesSatisfyingExactly( + equalTo(MESSAGING_SYSTEM, "guava-eventbus"), + equalTo(MESSAGING_DESTINATION_NAME, EVENTBUS_NAME), + equalTo( + MESSAGING_MESSAGE_BODY_SIZE, + MESSAGE_BODY.getBytes(StandardCharsets.UTF_8).length), + // FIXME: We do have "messaging.client_id" in instrumentation but + // "messaging.client.id" in + // semconv library right now. It should be replaced after semconv + // release. + equalTo(AttributeKey.stringKey("messaging.client_id"), CLIENT_ID), + equalTo(MESSAGING_OPERATION, "process"), + satisfies(MESSAGING_MESSAGE_ID, AbstractAssert::isNotNull)), + span -> + span.hasName("process child") + .hasKind(SpanKind.INTERNAL) + .hasParent(trace.getSpan(1)))); + } +} diff --git a/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/impl/MessageRequest.java b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/impl/MessageRequest.java new file mode 100644 index 000000000..8c42a852f --- /dev/null +++ b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/impl/MessageRequest.java @@ -0,0 +1,106 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.impl; + +import io.opentelemetry.contrib.messaging.wrappers.model.Message; +import io.opentelemetry.contrib.messaging.wrappers.semconv.MessagingProcessRequest; +import java.nio.charset.StandardCharsets; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import javax.annotation.Nullable; + +public class MessageRequest implements MessagingProcessRequest { + + private final Message message; + + @Nullable private final String clientId; + + @Nullable private final String eventBusName; + + public static MessageRequest of(Message message) { + return of(message, null, null); + } + + public static MessageRequest of( + Message message, @Nullable String clientId, @Nullable String eventBusName) { + return new MessageRequest(message, clientId, eventBusName); + } + + @Override + public String getSystem() { + return "guava-eventbus"; + } + + @Nullable + @Override + public String getDestination() { + return eventBusName; + } + + @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.getBody().getBytes(StandardCharsets.UTF_8).length; + } + + @Nullable + @Override + public Long getMessageEnvelopeSize() { + return null; + } + + @Nullable + @Override + public String getMessageId() { + return message.getId(); + } + + @Nullable + @Override + public String getClientId() { + return clientId; + } + + @Override + public List getMessageHeader(String name) { + return Collections.singletonList(message.getHeaders().get(name)); + } + + @Override + public Collection getAllMessageHeadersKey() { + return message.getHeaders().keySet(); + } + + private MessageRequest( + Message message, @Nullable String clientId, @Nullable String eventBusName) { + this.message = message; + this.clientId = clientId; + this.eventBusName = eventBusName; + } +} diff --git a/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/model/Message.java b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/model/Message.java new file mode 100644 index 000000000..e606e214e --- /dev/null +++ b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/model/Message.java @@ -0,0 +1,51 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.model; + +import java.util.Map; + +public class Message { + + private Map headers; + + private String id; + + private String body; + + public static Message create(Map headers, String id, String body) { + return new Message(headers, id, body); + } + + private Message(Map headers, String id, String body) { + this.headers = headers; + this.id = id; + this.body = body; + } + + public Map getHeaders() { + return headers; + } + + public void setHeaders(Map headers) { + this.headers = headers; + } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + + public String getBody() { + return body; + } + + public void setBody(String body) { + this.body = body; + } +} diff --git a/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/model/MessageListener.java b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/model/MessageListener.java new file mode 100644 index 000000000..fccd5c157 --- /dev/null +++ b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/model/MessageListener.java @@ -0,0 +1,48 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.model; + +import static io.opentelemetry.contrib.messaging.wrappers.TestConstants.CLIENT_ID; +import static io.opentelemetry.contrib.messaging.wrappers.TestConstants.EVENTBUS_NAME; + +import com.google.common.eventbus.Subscribe; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.contrib.messaging.wrappers.MessagingProcessWrapper; +import io.opentelemetry.contrib.messaging.wrappers.impl.MessageRequest; +import io.opentelemetry.contrib.messaging.wrappers.semconv.MessagingProcessRequest; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class MessageListener { + + private static final Logger logger = LoggerFactory.getLogger(MessageListener.class); + + private final Tracer tracer; + + private final MessagingProcessWrapper wrapper; + + public static MessageListener create( + Tracer tracer, MessagingProcessWrapper wrapper) { + return new MessageListener(tracer, wrapper); + } + + @Subscribe + public void handleEvent(Message event) { + wrapper.doProcess( + MessageRequest.of(event, CLIENT_ID, EVENTBUS_NAME), + () -> { + Span span = tracer.spanBuilder("process child").startSpan(); + logger.info("Received event from <" + EVENTBUS_NAME + ">: " + event.getId()); + span.end(); + }); + } + + private MessageListener(Tracer tracer, MessagingProcessWrapper wrapper) { + this.tracer = tracer; + this.wrapper = wrapper; + } +} diff --git a/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/model/MessageTextMapSetter.java b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/model/MessageTextMapSetter.java new file mode 100644 index 000000000..77e5a56f4 --- /dev/null +++ b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/model/MessageTextMapSetter.java @@ -0,0 +1,24 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.model; + +import io.opentelemetry.context.propagation.TextMapSetter; +import javax.annotation.Nullable; + +public class MessageTextMapSetter implements TextMapSetter { + + public static TextMapSetter create() { + return new MessageTextMapSetter(); + } + + @Override + public void set(@Nullable Message carrier, String key, String value) { + if (carrier == null) { + return; + } + carrier.getHeaders().put(key, value); + } +} diff --git a/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/package-info.java b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/package-info.java new file mode 100644 index 000000000..502b96ca6 --- /dev/null +++ b/messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/package-info.java @@ -0,0 +1,2 @@ +/** OpenTelemetry messaging wrappers extension. */ +package io.opentelemetry.contrib.messaging.wrappers; diff --git a/messaging-wrappers/kafka-clients/build.gradle.kts b/messaging-wrappers/kafka-clients/build.gradle.kts new file mode 100644 index 000000000..ec07abd94 --- /dev/null +++ b/messaging-wrappers/kafka-clients/build.gradle.kts @@ -0,0 +1,46 @@ +plugins { + id("otel.java-conventions") + + id("otel.publish-conventions") +} + +description = "OpenTelemetry Messaging Wrappers - kafka-clients implementation" +otelJava.moduleName.set("io.opentelemetry.contrib.messaging.wrappers.kafka") + +dependencies { + api(project(":messaging-wrappers:api")) + + // FIXME: We shouldn't depend on the library "opentelemetry-kafka-clients-common" directly because the api in this + // package could be mutable, unless the components were maintained in "opentelemetry-java-instrumentation" project. + // implementation("io.opentelemetry.instrumentation:opentelemetry-kafka-clients-common:2.13.3-alpha") + + compileOnly("org.apache.kafka:kafka-clients:0.11.0.0") + + testImplementation("org.apache.kafka:kafka-clients:0.11.0.0") + testImplementation("io.opentelemetry.instrumentation:opentelemetry-kafka-clients-2.6") + testImplementation(project(":messaging-wrappers:testing")) + + testAnnotationProcessor("com.google.auto.service:auto-service") + testCompileOnly("com.google.auto.service:auto-service-annotations") + testImplementation("org.testcontainers:kafka") + testImplementation("org.testcontainers:junit-jupiter") +} + +tasks { + withType().configureEach { + jvmArgs("-Dotel.java.global-autoconfigure.enabled=true") + // TODO: According to https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#message-creation-context-as-parent-of-process-span, + // process span should be the child of receive span. However, we couldn't access the trace context with receive span + // in wrappers, unless we add a generic accessor for that. + jvmArgs("-Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled=false") + jvmArgs("-Dotel.traces.exporter=logging") + jvmArgs("-Dotel.metrics.exporter=logging") + jvmArgs("-Dotel.logs.exporter=logging") + } +} + +configurations.all { + resolutionStrategy { + force("org.apache.kafka:kafka-clients:0.11.0.0") + } +} diff --git a/messaging-wrappers/kafka-clients/gradle.properties b/messaging-wrappers/kafka-clients/gradle.properties new file mode 100644 index 000000000..a0402e1e2 --- /dev/null +++ b/messaging-wrappers/kafka-clients/gradle.properties @@ -0,0 +1,2 @@ +# TODO: uncomment when ready to mark as stable +# otel.stable=true diff --git a/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaHelper.java b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaHelper.java new file mode 100644 index 000000000..10896cae3 --- /dev/null +++ b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaHelper.java @@ -0,0 +1,15 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.kafka; + +public final class KafkaHelper { + + public static KafkaProcessWrapperBuilder processWrapperBuilder() { + return new KafkaProcessWrapperBuilder(); + } + + private KafkaHelper() {} +} diff --git a/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaProcessWrapperBuilder.java b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaProcessWrapperBuilder.java new file mode 100644 index 000000000..52735c180 --- /dev/null +++ b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaProcessWrapperBuilder.java @@ -0,0 +1,32 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.kafka; + +import io.opentelemetry.contrib.messaging.wrappers.MessagingProcessWrapperBuilder; +import io.opentelemetry.contrib.messaging.wrappers.kafka.semconv.KafkaConsumerAttributesExtractor; +import io.opentelemetry.contrib.messaging.wrappers.kafka.semconv.KafkaConsumerAttributesGetter; +import io.opentelemetry.contrib.messaging.wrappers.kafka.semconv.KafkaProcessRequest; +import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessageOperation; +import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesExtractor; +import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanNameExtractor; +import java.util.ArrayList; + +public class KafkaProcessWrapperBuilder + extends MessagingProcessWrapperBuilder { + + KafkaProcessWrapperBuilder() { + super(); + super.textMapGetter = KafkaTextMapGetter.create(); + super.spanNameExtractor = + MessagingSpanNameExtractor.create( + KafkaConsumerAttributesGetter.INSTANCE, MessageOperation.PROCESS); + super.attributesExtractors = new ArrayList<>(); + super.attributesExtractors.add( + MessagingAttributesExtractor.create( + KafkaConsumerAttributesGetter.INSTANCE, MessageOperation.PROCESS)); + super.attributesExtractors.add(KafkaConsumerAttributesExtractor.INSTANCE); + } +} diff --git a/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaTextMapGetter.java b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaTextMapGetter.java new file mode 100644 index 000000000..820b7bfca --- /dev/null +++ b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaTextMapGetter.java @@ -0,0 +1,53 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.kafka; + +import io.opentelemetry.context.propagation.TextMapGetter; +import io.opentelemetry.contrib.messaging.wrappers.kafka.semconv.KafkaProcessRequest; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.stream.Collectors; +import java.util.stream.StreamSupport; +import javax.annotation.Nullable; +import org.apache.kafka.common.header.Header; + +/** + * Copied from KafkaConsumerRecordGetter. + */ +public class KafkaTextMapGetter implements TextMapGetter { + + public static TextMapGetter create() { + return new KafkaTextMapGetter(); + } + + @Override + public Iterable keys(@Nullable KafkaProcessRequest carrier) { + if (carrier == null || carrier.getRecord() == null) { + return Collections.emptyList(); + } + return StreamSupport.stream(carrier.getRecord().headers().spliterator(), false) + .map(Header::key) + .collect(Collectors.toList()); + } + + @Nullable + @Override + public String get(@Nullable KafkaProcessRequest carrier, String key) { + if (carrier == null || carrier.getRecord() == null) { + return null; + } + Header header = carrier.getRecord().headers().lastHeader(key); + if (header == null) { + return null; + } + byte[] value = header.value(); + if (value == null) { + return null; + } + return new String(value, StandardCharsets.UTF_8); + } +} diff --git a/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/package-info.java b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/package-info.java new file mode 100644 index 000000000..7e988a5b6 --- /dev/null +++ b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/package-info.java @@ -0,0 +1,2 @@ +/** OpenTelemetry messaging wrappers extension - kafka implementation. */ +package io.opentelemetry.contrib.messaging.wrappers.kafka; diff --git a/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/semconv/KafkaConsumerAttributesExtractor.java b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/semconv/KafkaConsumerAttributesExtractor.java new file mode 100644 index 000000000..f037427c1 --- /dev/null +++ b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/semconv/KafkaConsumerAttributesExtractor.java @@ -0,0 +1,72 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.kafka.semconv; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.context.Context; +import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor; +import java.nio.ByteBuffer; +import javax.annotation.Nullable; +import org.apache.kafka.clients.consumer.ConsumerRecord; + +/** + * Copied from KafkaConsumerAttributesExtractor. + */ +public enum KafkaConsumerAttributesExtractor + implements AttributesExtractor { + INSTANCE; + + // copied from MessagingIncubatingAttributes + private static final AttributeKey MESSAGING_DESTINATION_PARTITION_ID = + AttributeKey.stringKey("messaging.destination.partition.id"); + private static final AttributeKey MESSAGING_KAFKA_CONSUMER_GROUP = + AttributeKey.stringKey("messaging.kafka.consumer.group"); + private static final AttributeKey MESSAGING_KAFKA_MESSAGE_KEY = + AttributeKey.stringKey("messaging.kafka.message.key"); + private static final AttributeKey MESSAGING_KAFKA_MESSAGE_OFFSET = + AttributeKey.longKey("messaging.kafka.message.offset"); + private static final AttributeKey MESSAGING_KAFKA_MESSAGE_TOMBSTONE = + AttributeKey.booleanKey("messaging.kafka.message.tombstone"); + + @Override + public void onStart( + AttributesBuilder attributes, Context parentContext, KafkaProcessRequest request) { + + ConsumerRecord record = request.getRecord(); + + attributes.put(MESSAGING_DESTINATION_PARTITION_ID, String.valueOf(record.partition())); + attributes.put(MESSAGING_KAFKA_MESSAGE_OFFSET, record.offset()); + + Object key = record.key(); + if (key != null && canSerialize(key.getClass())) { + attributes.put(MESSAGING_KAFKA_MESSAGE_KEY, key.toString()); + } + if (record.value() == null) { + attributes.put(MESSAGING_KAFKA_MESSAGE_TOMBSTONE, true); + } + + String consumerGroup = request.getConsumerGroup(); + if (consumerGroup != null) { + attributes.put(MESSAGING_KAFKA_CONSUMER_GROUP, consumerGroup); + } + } + + private static boolean canSerialize(Class keyClass) { + // we make a simple assumption here that we can serialize keys by simply calling toString() + // and that does not work for byte[] or ByteBuffer + return !(keyClass.isArray() || keyClass == ByteBuffer.class); + } + + @Override + public void onEnd( + AttributesBuilder attributes, + Context context, + KafkaProcessRequest request, + @Nullable Void unused, + @Nullable Throwable error) {} +} diff --git a/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/semconv/KafkaConsumerAttributesGetter.java b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/semconv/KafkaConsumerAttributesGetter.java new file mode 100644 index 000000000..5797c7cd1 --- /dev/null +++ b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/semconv/KafkaConsumerAttributesGetter.java @@ -0,0 +1,88 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.kafka.semconv; + +import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.StreamSupport; +import javax.annotation.Nullable; + +public enum KafkaConsumerAttributesGetter + implements MessagingAttributesGetter { + INSTANCE; + + @Override + public String getSystem(KafkaProcessRequest request) { + return "kafka"; + } + + @Override + public String getDestination(KafkaProcessRequest request) { + return request.getRecord().topic(); + } + + @Nullable + @Override + public String getDestinationTemplate(KafkaProcessRequest request) { + return null; + } + + @Override + public boolean isTemporaryDestination(KafkaProcessRequest request) { + return false; + } + + @Override + public boolean isAnonymousDestination(KafkaProcessRequest request) { + return false; + } + + @Override + @Nullable + public String getConversationId(KafkaProcessRequest request) { + return null; + } + + @Nullable + @Override + public Long getMessageBodySize(KafkaProcessRequest request) { + long size = request.getRecord().serializedValueSize(); + return size >= 0 ? size : null; + } + + @Nullable + @Override + public Long getMessageEnvelopeSize(KafkaProcessRequest request) { + return null; + } + + @Override + @Nullable + public String getMessageId(KafkaProcessRequest request, @Nullable Void unused) { + return null; + } + + @Nullable + @Override + public String getClientId(KafkaProcessRequest request) { + return request.getClientId(); + } + + @Nullable + @Override + public Long getBatchMessageCount(KafkaProcessRequest request, @Nullable Void unused) { + return null; + } + + @Override + public List getMessageHeader(KafkaProcessRequest request, String name) { + return StreamSupport.stream(request.getRecord().headers().headers(name).spliterator(), false) + .map(header -> new String(header.value(), StandardCharsets.UTF_8)) + .collect(Collectors.toList()); + } +} diff --git a/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/semconv/KafkaProcessRequest.java b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/semconv/KafkaProcessRequest.java new file mode 100644 index 000000000..5990a8aee --- /dev/null +++ b/messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/semconv/KafkaProcessRequest.java @@ -0,0 +1,52 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.kafka.semconv; + +import javax.annotation.Nullable; +import org.apache.kafka.clients.consumer.ConsumerRecord; + +public class KafkaProcessRequest { + + private final ConsumerRecord consumerRecord; + + @Nullable private final String clientId; + + @Nullable private final String consumerGroup; + + public static KafkaProcessRequest of(ConsumerRecord consumerRecord) { + return of(consumerRecord, null, null); + } + + public static KafkaProcessRequest of( + ConsumerRecord consumerRecord, + @Nullable String consumerGroup, + @Nullable String clientId) { + return new KafkaProcessRequest(consumerRecord, consumerGroup, clientId); + } + + public ConsumerRecord getRecord() { + return consumerRecord; + } + + @Nullable + public String getConsumerGroup() { + return this.consumerGroup; + } + + @Nullable + public String getClientId() { + return this.clientId; + } + + private KafkaProcessRequest( + ConsumerRecord consumerRecord, + @Nullable String consumerGroup, + @Nullable String clientId) { + this.consumerRecord = consumerRecord; + this.consumerGroup = consumerGroup; + this.clientId = clientId; + } +} diff --git a/messaging-wrappers/kafka-clients/src/test/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaClientBaseTest.java b/messaging-wrappers/kafka-clients/src/test/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaClientBaseTest.java new file mode 100644 index 000000000..21138dc14 --- /dev/null +++ b/messaging-wrappers/kafka-clients/src/test/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaClientBaseTest.java @@ -0,0 +1,149 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.kafka; + +import io.opentelemetry.contrib.messaging.wrappers.testing.AbstractBaseTest; +import java.time.Duration; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.IntegerDeserializer; +import org.apache.kafka.common.serialization.IntegerSerializer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.TestInstance; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.output.Slf4jLogConsumer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.kafka.KafkaContainer; +import org.testcontainers.utility.DockerImageName; + +/** + * Copied from KafkaClientBaseTest. + */ +@SuppressWarnings("OtelInternalJavadoc") +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public abstract class KafkaClientBaseTest extends AbstractBaseTest { + + private static final Logger logger = LoggerFactory.getLogger(KafkaClientBaseTest.class); + + protected static final String SHARED_TOPIC = "shared.topic"; + + private KafkaContainer kafka; + protected Producer producer; + protected Consumer consumer; + private final CountDownLatch consumerReady = new CountDownLatch(1); + + public static final int partition = 0; + public static final TopicPartition topicPartition = new TopicPartition(SHARED_TOPIC, partition); + + @BeforeAll + void setupClass() throws ExecutionException, InterruptedException, TimeoutException { + kafka = + new KafkaContainer(DockerImageName.parse("apache/kafka:3.8.0")) + .withEnv("KAFKA_HEAP_OPTS", "-Xmx256m") + .withLogConsumer(new Slf4jLogConsumer(logger)) + .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.Kafka.*Server\\).*", 1)) + .withStartupTimeout(Duration.ofMinutes(1)); + kafka.start(); + + // create test topic + HashMap adminProps = new HashMap<>(); + adminProps.put("bootstrap.servers", kafka.getBootstrapServers()); + + try (AdminClient admin = AdminClient.create(adminProps)) { + admin + .createTopics(Collections.singletonList(new NewTopic(SHARED_TOPIC, 1, (short) 1))) + .all() + .get(30, TimeUnit.SECONDS); + } + + producer = new KafkaProducer<>(producerProps()); + + consumer = new KafkaConsumer<>(consumerProps()); + + consumer.subscribe( + Collections.singletonList(SHARED_TOPIC), + new ConsumerRebalanceListener() { + @Override + public void onPartitionsRevoked(Collection collection) {} + + @Override + public void onPartitionsAssigned(Collection collection) { + consumerReady.countDown(); + } + }); + } + + public Map consumerProps() { + HashMap props = new HashMap<>(); + props.put("bootstrap.servers", kafka.getBootstrapServers()); + props.put("enable.auto.commit", "true"); + props.put("auto.commit.interval.ms", 10); + props.put("session.timeout.ms", "30000"); + props.put("key.deserializer", IntegerDeserializer.class.getName()); + props.put("value.deserializer", StringDeserializer.class.getName()); + return props; + } + + public Map producerProps() { + HashMap props = new HashMap<>(); + props.put("bootstrap.servers", kafka.getBootstrapServers()); + props.put("retries", 0); + props.put("batch.size", "16384"); + props.put("linger.ms", 1); + props.put("buffer.memory", "33554432"); + props.put("key.serializer", IntegerSerializer.class.getName()); + props.put("value.serializer", StringSerializer.class.getName()); + return props; + } + + @AfterAll + void cleanupClass() { + if (producer != null) { + producer.close(); + } + if (consumer != null) { + consumer.close(); + } + kafka.stop(); + } + + @SuppressWarnings("PreferJavaTimeOverload") + public void awaitUntilConsumerIsReady() throws InterruptedException { + if (consumerReady.await(0, TimeUnit.SECONDS)) { + return; + } + for (int i = 0; i < 60; i++) { + consumer.poll(0L); + if (consumerReady.await(3, TimeUnit.SECONDS)) { + break; + } + logger.info("Consumer has not been ready for {} time(s).", i); + } + if (consumerReady.getCount() != 0) { + throw new AssertionError("Consumer wasn't assigned any partitions!"); + } + consumer.seekToBeginning(Collections.emptyList()); + } +} diff --git a/messaging-wrappers/kafka-clients/src/test/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaClientTest.java b/messaging-wrappers/kafka-clients/src/test/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaClientTest.java new file mode 100644 index 000000000..329663a07 --- /dev/null +++ b/messaging-wrappers/kafka-clients/src/test/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaClientTest.java @@ -0,0 +1,166 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.kafka; + +import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat; +import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo; +import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_PARTITION_ID; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_KAFKA_CONSUMER_GROUP; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_BODY_SIZE; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION; +import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM; + +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Scope; +import io.opentelemetry.contrib.messaging.wrappers.MessagingProcessWrapper; +import io.opentelemetry.contrib.messaging.wrappers.kafka.semconv.KafkaProcessRequest; +import io.opentelemetry.instrumentation.kafkaclients.v2_6.TracingConsumerInterceptor; +import io.opentelemetry.instrumentation.kafkaclients.v2_6.TracingProducerInterceptor; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Map; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.assertj.core.api.AbstractAssert; +import org.junit.jupiter.api.Test; + +public class KafkaClientTest extends KafkaClientBaseTest { + + static final String greeting = "Hello Kafka!"; + + static final String clientId = "test-consumer-1"; + + static final String groupId = "test"; + + @Override + public Map producerProps() { + Map props = super.producerProps(); + props.put( + ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName()); + return props; + } + + @Override + public Map consumerProps() { + Map props = super.consumerProps(); + props.put( + ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName()); + props.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId); + props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); + return props; + } + + @Test + void testInterceptors() throws InterruptedException { + OpenTelemetry otel = GlobalOpenTelemetry.get(); + Tracer tracer = otel.getTracer("test-tracer", "1.0.0"); + MessagingProcessWrapper wrapper = + KafkaHelper.processWrapperBuilder().openTelemetry(otel).build(); + + sendWithParent(tracer); + + awaitUntilConsumerIsReady(); + + consumeWithChild(tracer, wrapper); + + assertTraces(); + } + + @SuppressWarnings("FutureReturnValueIgnored") + public void sendWithParent(Tracer tracer) { + Span parent = tracer.spanBuilder("parent").startSpan(); + try (Scope scope = parent.makeCurrent()) { + producer.send( + new ProducerRecord<>(SHARED_TOPIC, greeting), + (meta, ex) -> { + if (ex == null) { + tracer.spanBuilder("producer callback").startSpan().end(); + } else { + tracer.spanBuilder("producer exception: " + ex).startSpan().end(); + } + }); + } + parent.end(); + } + + public void consumeWithChild( + Tracer tracer, MessagingProcessWrapper wrapper) { + // check that the message was received + ConsumerRecords records = consumer.poll(Duration.ofSeconds(5).toMillis()); + assertThat(records.count()).isEqualTo(1); + ConsumerRecord record = records.iterator().next(); + assertThat(record.value()).isEqualTo(greeting); + assertThat(record.key()).isNull(); + + wrapper.doProcess( + KafkaProcessRequest.of(record, groupId, clientId), + () -> { + tracer.spanBuilder("process child").startSpan().end(); + }); + } + + /** + * Copied from testing-common. + */ + @SuppressWarnings("deprecation") // using deprecated semconv + public void assertTraces() { + waitAndAssertTraces( + sortByRootSpanName("parent", "producer callback"), + trace -> + trace.hasSpansSatisfyingExactly( + span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(), + span -> + // No need to verify the attribute here because it is generated by + // instrumentation library. + span.hasName(SHARED_TOPIC + " publish") + .hasKind(SpanKind.PRODUCER) + .hasParent(trace.getSpan(0)), + span -> + span.hasName(SHARED_TOPIC + " process") + .hasKind(SpanKind.CONSUMER) + .hasParent(trace.getSpan(1)) + .hasAttributesSatisfyingExactly( + equalTo(MESSAGING_SYSTEM, "kafka"), + equalTo(MESSAGING_DESTINATION_NAME, SHARED_TOPIC), + equalTo( + MESSAGING_MESSAGE_BODY_SIZE, + greeting.getBytes(StandardCharsets.UTF_8).length), + satisfies( + MESSAGING_DESTINATION_PARTITION_ID, + org.assertj.core.api.AbstractStringAssert::isNotEmpty), + // FIXME: We do have "messaging.client_id" in instrumentation but + // "messaging.client.id" in + // semconv library right now. It should be replaced after semconv + // release. + equalTo( + AttributeKey.stringKey("messaging.client_id"), "test-consumer-1"), + satisfies(MESSAGING_KAFKA_MESSAGE_OFFSET, AbstractAssert::isNotNull), + equalTo(MESSAGING_KAFKA_CONSUMER_GROUP, "test"), + equalTo(MESSAGING_OPERATION, "process")), + span -> + span.hasName("process child") + .hasKind(SpanKind.INTERNAL) + .hasParent(trace.getSpan(2))), + // ideally we'd want producer callback to be part of the main trace, we just aren't able to + // instrument that + trace -> + trace.hasSpansSatisfyingExactly( + span -> + span.hasName("producer callback").hasKind(SpanKind.INTERNAL).hasNoParent())); + } +} diff --git a/messaging-wrappers/testing/build.gradle.kts b/messaging-wrappers/testing/build.gradle.kts new file mode 100644 index 000000000..6c0983923 --- /dev/null +++ b/messaging-wrappers/testing/build.gradle.kts @@ -0,0 +1,20 @@ +plugins { + id("otel.java-conventions") +} + +description = "OpenTelemetry Messaging Wrappers testing" + +dependencies { + annotationProcessor("com.google.auto.service:auto-service") + compileOnly("com.google.auto.service:auto-service-annotations") + + api("org.junit.jupiter:junit-jupiter-api") + api("org.junit.jupiter:junit-jupiter-params") + api("io.opentelemetry:opentelemetry-sdk-testing") + + implementation("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure") + implementation("io.opentelemetry:opentelemetry-sdk-trace") + implementation("io.opentelemetry:opentelemetry-sdk-extension-incubator") + implementation("io.opentelemetry:opentelemetry-exporter-logging") + implementation("io.opentelemetry.javaagent:opentelemetry-testing-common") +} diff --git a/messaging-wrappers/testing/src/main/java/io/opentelemetry/contrib/messaging/wrappers/testing/AbstractBaseTest.java b/messaging-wrappers/testing/src/main/java/io/opentelemetry/contrib/messaging/wrappers/testing/AbstractBaseTest.java new file mode 100644 index 000000000..9ed951bce --- /dev/null +++ b/messaging-wrappers/testing/src/main/java/io/opentelemetry/contrib/messaging/wrappers/testing/AbstractBaseTest.java @@ -0,0 +1,77 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.testing; + +import static io.opentelemetry.instrumentation.testing.util.TelemetryDataUtil.orderByRootSpanName; +import static io.opentelemetry.instrumentation.testing.util.TelemetryDataUtil.waitForTraces; +import static org.awaitility.Awaitility.await; + +import io.opentelemetry.contrib.messaging.wrappers.testing.internal.AutoConfiguredDataCapture; +import io.opentelemetry.instrumentation.testing.util.TelemetryDataUtil; +import io.opentelemetry.sdk.testing.assertj.TraceAssert; +import io.opentelemetry.sdk.testing.assertj.TracesAssert; +import io.opentelemetry.sdk.trace.data.SpanData; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.List; +import java.util.concurrent.TimeoutException; +import java.util.function.Consumer; +import java.util.function.Supplier; +import javax.annotation.Nullable; +import org.awaitility.core.ConditionTimeoutException; + +public abstract class AbstractBaseTest { + + public static Comparator> sortByRootSpanName(String... names) { + return orderByRootSpanName(names); + } + + @SafeVarargs + @SuppressWarnings("varargs") + public static void waitAndAssertTraces( + @Nullable Comparator> traceComparator, Consumer... assertions) { + List> assertionsList = new ArrayList<>(Arrays.asList(assertions)); + try { + await() + .untilAsserted( + () -> + doAssertTraces( + traceComparator, AutoConfiguredDataCapture::getSpans, assertionsList)); + } catch (Throwable t) { + // awaitility is doing a jmx call that is not implemented in GraalVM: + // call: + // https://github.com/awaitility/awaitility/blob/fbe16add874b4260dd240108304d5c0be84eabc8/awaitility/src/main/java/org/awaitility/core/ConditionAwaiter.java#L157 + // see https://github.com/oracle/graal/issues/6101 (spring boot graal native image) + if (t.getClass().getName().equals("com.oracle.svm.core.jdk.UnsupportedFeatureError") + || t instanceof ConditionTimeoutException) { + // Don't throw this failure since the stack is the awaitility thread, causing confusion. + // Instead, just assert one more time on the test thread, which will fail with a better + // stack trace. + // TODO(anuraaga): There is probably a better way to do this. + doAssertTraces(traceComparator, AutoConfiguredDataCapture::getSpans, assertionsList); + } else { + throw t; + } + } + } + + public static void doAssertTraces( + @Nullable Comparator> traceComparator, + Supplier> supplier, + List> assertionsList) { + try { + List> traces = waitForTraces(supplier, assertionsList.size()); + TelemetryDataUtil.assertScopeVersion(traces); + if (traceComparator != null) { + traces.sort(traceComparator); + } + TracesAssert.assertThat(traces).hasTracesSatisfyingExactly(assertionsList); + } catch (InterruptedException | TimeoutException e) { + throw new AssertionError("Error waiting for " + assertionsList.size() + " traces", e); + } + } +} diff --git a/messaging-wrappers/testing/src/main/java/io/opentelemetry/contrib/messaging/wrappers/testing/internal/AutoConfiguredDataCapture.java b/messaging-wrappers/testing/src/main/java/io/opentelemetry/contrib/messaging/wrappers/testing/internal/AutoConfiguredDataCapture.java new file mode 100644 index 000000000..9d25fe83f --- /dev/null +++ b/messaging-wrappers/testing/src/main/java/io/opentelemetry/contrib/messaging/wrappers/testing/internal/AutoConfiguredDataCapture.java @@ -0,0 +1,50 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.messaging.wrappers.testing.internal; + +import com.google.auto.service.AutoService; +import io.opentelemetry.exporter.logging.LoggingSpanExporter; +import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizer; +import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider; +import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.export.SpanExporter; +import java.util.List; + +@AutoService(AutoConfigurationCustomizerProvider.class) +public class AutoConfiguredDataCapture implements AutoConfigurationCustomizerProvider { + + private static final InMemorySpanExporter inMemorySpanExporter = InMemorySpanExporter.create(); + + /* + Returns the spans which have been exported by the autoconfigured global OpenTelemetry SDK. + */ + public static List getSpans() { + return inMemorySpanExporter.getFinishedSpanItems(); + } + + @Override + public void customize(AutoConfigurationCustomizer autoConfiguration) { + autoConfiguration.addSpanExporterCustomizer( + (spanExporter, config) -> { + // we piggy-back onto the autoconfigured logging exporter for now, + // because that one uses a SimpleSpanProcessor which does not impose a batching delay + if (spanExporter instanceof LoggingSpanExporter) { + inMemorySpanExporter.reset(); + return SpanExporter.composite(inMemorySpanExporter, spanExporter); + } + return spanExporter; + }); + } + + @Override + public int order() { + // There might be other autoconfigurations wrapping SpanExporters, + // which can result in us failing to detect it + // We avoid this by ensuring that we run first + return Integer.MIN_VALUE; + } +} diff --git a/settings.gradle.kts b/settings.gradle.kts index 27db34c41..45b417279 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -53,6 +53,10 @@ include(":jmx-scraper") include(":jmx-scraper:test-app") include(":jmx-scraper:test-webapp") include(":maven-extension") +include(":messaging-wrappers:aliyun-mns-sdk") +include(":messaging-wrappers:api") +include(":messaging-wrappers:kafka-clients") +include(":messaging-wrappers:testing") include(":micrometer-meter-provider") include(":noop-api") include(":processors")