Aws sdk2 sqs context propagation
ADD
instrumentation/aws-sdk/aws-sdk-2.2/javaagent/src/test/groovy/Aws2SqsTracingTest.groovy
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import io.opentelemetry.instrumentation.awssdk.v2_2.AbstractAws2SqsTracingTest
+import io.opentelemetry.instrumentation.test.AgentTestTrait
+import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration
+
+class Aws2SqsTracingTest extends AbstractAws2SqsTracingTest implements AgentTestTrait {
+ @Override
+ ClientOverrideConfiguration.Builder createOverrideConfigurationBuilder() {
+ return ClientOverrideConfiguration.builder()
+ }
+}

MODIFY
instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/AwsSdkInstrumenterFactory.java
instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/AwsSdkInstrumenterFactory.java
@@ -32,6 +32,7 @@ final class AwsSdkInstrumenterFactory {
netAttributesExtractor = NetClientAttributesExtractor.create(netAttributesGetter);
private static final AwsSdkExperimentalAttributesExtractor experimentalAttributesExtractor =
new AwsSdkExperimentalAttributesExtractor();
+ private static final AwsSdkSpanKindExtractor spanKindExtractor = new AwsSdkSpanKindExtractor();
private static final List<AttributesExtractor<ExecutionAttributes, SdkHttpResponse>>
defaultAttributesExtractors =
@@ -45,16 +46,34 @@ final class AwsSdkInstrumenterFactory {
netAttributesExtractor,
experimentalAttributesExtractor);
- static Instrumenter<ExecutionAttributes, SdkHttpResponse> createInstrumenter(
+ static Instrumenter<ExecutionAttributes, SdkHttpResponse> requestInstrumenter(
OpenTelemetry openTelemetry, boolean captureExperimentalSpanAttributes) {
+ return createInstrumenter(
+ openTelemetry,
+ captureExperimentalSpanAttributes,
+ AwsSdkInstrumenterFactory.spanKindExtractor);
+ }
+
+ static Instrumenter<ExecutionAttributes, SdkHttpResponse> consumerInstrumenter(
+ OpenTelemetry openTelemetry, boolean captureExperimentalSpanAttributes) {
+
+ return createInstrumenter(
+ openTelemetry, captureExperimentalSpanAttributes, SpanKindExtractor.alwaysConsumer());
+ }
+
+ private static Instrumenter<ExecutionAttributes, SdkHttpResponse> createInstrumenter(
+ OpenTelemetry openTelemetry,
+ boolean captureExperimentalSpanAttributes,
+ SpanKindExtractor<ExecutionAttributes> spanKindExtractor) {
+
return Instrumenter.<ExecutionAttributes, SdkHttpResponse>builder(
openTelemetry, INSTRUMENTATION_NAME, AwsSdkInstrumenterFactory::spanName)
.addAttributesExtractors(
captureExperimentalSpanAttributes
? extendedAttributesExtractors
: defaultAttributesExtractors)
- .newInstrumenter(SpanKindExtractor.alwaysClient());
+ .newInstrumenter(spanKindExtractor);
}
private static String spanName(ExecutionAttributes attributes) {

ADD
instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/AwsSdkSpanKindExtractor.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.awssdk.v2_2;
+
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.instrumentation.api.instrumenter.SpanKindExtractor;
+import software.amazon.awssdk.core.SdkRequest;
+import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
+
+class AwsSdkSpanKindExtractor implements SpanKindExtractor<ExecutionAttributes> {
+ @Override
+ public SpanKind extract(ExecutionAttributes request) {
+ return isSqsProducer(request) ? SpanKind.PRODUCER : SpanKind.CLIENT;
+ }
+
+ private static boolean isSqsProducer(ExecutionAttributes executionAttributes) {
+ SdkRequest request =
+ executionAttributes.getAttribute(TracingExecutionInterceptor.SDK_REQUEST_ATTRIBUTE);
+ return request
+ .getClass()
+ .getName()
+ .equals("software.amazon.awssdk.services.sqs.model.SendMessageRequest");
+ }
+}

MODIFY
instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/AwsSdkTelemetry.java
instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/AwsSdkTelemetry.java
@@ -39,12 +39,16 @@ public class AwsSdkTelemetry {
return new AwsSdkTelemetryBuilder(openTelemetry);
}
- private final Instrumenter<ExecutionAttributes, SdkHttpResponse> tracer;
+ private final Instrumenter<ExecutionAttributes, SdkHttpResponse> requestInstrumenter;
+ private final Instrumenter<ExecutionAttributes, SdkHttpResponse> consumerInstrumenter;
private final boolean captureExperimentalSpanAttributes;
AwsSdkTelemetry(OpenTelemetry openTelemetry, boolean captureExperimentalSpanAttributes) {
- this.tracer =
- AwsSdkInstrumenterFactory.createInstrumenter(
+ this.requestInstrumenter =
+ AwsSdkInstrumenterFactory.requestInstrumenter(
+ openTelemetry, captureExperimentalSpanAttributes);
+ this.consumerInstrumenter =
+ AwsSdkInstrumenterFactory.consumerInstrumenter(
openTelemetry, captureExperimentalSpanAttributes);
this.captureExperimentalSpanAttributes = captureExperimentalSpanAttributes;
}
@@ -54,6 +58,7 @@ public class AwsSdkTelemetry {
* ClientOverrideConfiguration.Builder#addExecutionInterceptor(ExecutionInterceptor)}.
*/
public ExecutionInterceptor newExecutionInterceptor() {
- return new TracingExecutionInterceptor(tracer, captureExperimentalSpanAttributes);
+ return new TracingExecutionInterceptor(
+ requestInstrumenter, consumerInstrumenter, captureExperimentalSpanAttributes);
}
}

ADD
instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/SqsMessageAccess.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.awssdk.v2_2;
+
+import static java.lang.invoke.MethodType.methodType;
+
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.util.Collections;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+/**
+ * Reflective access to aws-sdk-java-sqs class Message.
+ *
+ * <p>We currently don't have a good pattern of instrumenting a core library with various plugins
+ * that need plugin-specific instrumentation - if we accessed the class directly, Muzzle would
+ * prevent the entire instrumentation from loading when the plugin isn't available. We need to
+ * carefully check this class has all reflection errors result in no-op, and in the future we will
+ * hopefully come up with a better pattern.
+ */
+final class SqsMessageAccess {
+
+ @Nullable private static final MethodHandle GET_ATTRIBUTES;
+
+ static {
+ Class<?> messageClass = null;
+ try {
+ messageClass = Class.forName("software.amazon.awssdk.services.sqs.model.Message");
+ } catch (Throwable t) {
+ // Ignore.
+ }
+ if (messageClass != null) {
+ MethodHandles.Lookup lookup = MethodHandles.publicLookup();
+ MethodHandle getAttributes = null;
+ try {
+ getAttributes =
+ lookup.findVirtual(messageClass, "attributesAsStrings", methodType(Map.class));
+ } catch (NoSuchMethodException | IllegalAccessException e) {
+ // Ignore
+ }
+ GET_ATTRIBUTES = getAttributes;
+ } else {
+ GET_ATTRIBUTES = null;
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ static Map<String, String> getAttributes(Object message) {
+ if (GET_ATTRIBUTES == null) {
+ return Collections.emptyMap();
+ }
+ try {
+ return (Map<String, String>) GET_ATTRIBUTES.invoke(message);
+ } catch (Throwable t) {
+ return Collections.emptyMap();
+ }
+ }
+
+ private SqsMessageAccess() {}
+}

ADD
instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/SqsParentContext.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.awssdk.v2_2;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapGetter;
+import io.opentelemetry.extension.aws.AwsXrayPropagator;
+import java.util.Collections;
+import java.util.Map;
+
+final class SqsParentContext {
+
+ enum MapGetter implements TextMapGetter<Map<String, String>> {
+ INSTANCE;
+
+ @Override
+ public Iterable<String> keys(Map<String, String> map) {
+ return map.keySet();
+ }
+
+ @Override
+ public String get(Map<String, String> map, String s) {
+ return map.get(s);
+ }
+ }
+
+ static final String AWS_TRACE_SYSTEM_ATTRIBUTE = "AWSTraceHeader";
+
+ static Context ofSystemAttributes(Map<String, String> systemAttributes) {
+ String traceHeader = systemAttributes.get(AWS_TRACE_SYSTEM_ATTRIBUTE);
+ return AwsXrayPropagator.getInstance()
+ .extract(
+ Context.root(),
+ Collections.singletonMap("X-Amzn-Trace-Id", traceHeader),
+ MapGetter.INSTANCE);
+ }
+
+ private SqsParentContext() {}
+}

ADD
instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/SqsReceiveMessageRequestAccess.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.awssdk.v2_2;
+
+import static java.lang.invoke.MethodType.methodType;
+
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.util.Collection;
+import java.util.List;
+import javax.annotation.Nullable;
+import software.amazon.awssdk.core.SdkRequest;
+
+/**
+ * Reflective access to aws-sdk-java-sqs class ReceiveMessageRequest.
+ *
+ * <p>We currently don't have a good pattern of instrumenting a core library with various plugins
+ * that need plugin-specific instrumentation - if we accessed the class directly, Muzzle would
+ * prevent the entire instrumentation from loading when the plugin isn't available. We need to
+ * carefully check this class has all reflection errors result in no-op, and in the future we will
+ * hopefully come up with a better pattern.
+ */
+final class SqsReceiveMessageRequestAccess {
+
+ @Nullable private static final MethodHandle ATTRIBUTE_NAMES_WITH_STRINGS;
+
+ static {
+ Class<?> receiveMessageRequestClass = null;
+ try {
+ receiveMessageRequestClass =
+ Class.forName("software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest$Builder");
+ } catch (Throwable t) {
+ // Ignore.
+ }
+ if (receiveMessageRequestClass != null) {
+ MethodHandles.Lookup lookup = MethodHandles.publicLookup();
+ MethodHandle withAttributeNames = null;
+ try {
+ withAttributeNames =
+ lookup.findVirtual(
+ receiveMessageRequestClass,
+ "attributeNamesWithStrings",
+ methodType(receiveMessageRequestClass, Collection.class));
+ } catch (NoSuchMethodException | IllegalAccessException e) {
+ // Ignore
+ }
+ ATTRIBUTE_NAMES_WITH_STRINGS = withAttributeNames;
+ } else {
+ ATTRIBUTE_NAMES_WITH_STRINGS = null;
+ }
+ }
+
+ static boolean isInstance(SdkRequest request) {
+ return request
+ .getClass()
+ .getName()
+ .equals("software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest");
+ }
+
+ static void attributeNamesWithStrings(SdkRequest.Builder builder, List<String> attributeNames) {
+ if (ATTRIBUTE_NAMES_WITH_STRINGS == null) {
+ return;
+ }
+ try {
+ ATTRIBUTE_NAMES_WITH_STRINGS.invoke(builder, attributeNames);
+ } catch (Throwable throwable) {
+ // Ignore
+ }
+ }
+
+ private SqsReceiveMessageRequestAccess() {}
+}

MODIFY
instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/TracingExecutionInterceptor.java
instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/TracingExecutionInterceptor.java
@@ -12,7 +12,10 @@ import io.opentelemetry.context.Scope;
import io.opentelemetry.extension.aws.AwsXrayPropagator;
import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Optional;
import software.amazon.awssdk.awscore.AwsResponse;
import software.amazon.awssdk.core.ClientType;
import software.amazon.awssdk.core.SdkRequest;
@@ -38,15 +41,20 @@ final class TracingExecutionInterceptor implements ExecutionInterceptor {
new ExecutionAttribute<>(TracingExecutionInterceptor.class.getName() + ".AwsSdkRequest");
static final ExecutionAttribute<SdkHttpRequest> SDK_HTTP_REQUEST_ATTRIBUTE =
new ExecutionAttribute<>(TracingExecutionInterceptor.class.getName() + ".SdkHttpRequest");
+ static final ExecutionAttribute<SdkRequest> SDK_REQUEST_ATTRIBUTE =
+ new ExecutionAttribute<>(TracingExecutionInterceptor.class.getName() + ".SdkRequest");
- private final Instrumenter<ExecutionAttributes, SdkHttpResponse> instrumenter;
+ private final Instrumenter<ExecutionAttributes, SdkHttpResponse> requestInstrumenter;
+ private final Instrumenter<ExecutionAttributes, SdkHttpResponse> consumerInstrumenter;
private final boolean captureExperimentalSpanAttributes;
private final FieldMapper fieldMapper;
TracingExecutionInterceptor(
- Instrumenter<ExecutionAttributes, SdkHttpResponse> instrumenter,
+ Instrumenter<ExecutionAttributes, SdkHttpResponse> requestInstrumenter,
+ Instrumenter<ExecutionAttributes, SdkHttpResponse> consumerInstrumenter,
boolean captureExperimentalSpanAttributes) {
- this.instrumenter = instrumenter;
+ this.requestInstrumenter = requestInstrumenter;
+ this.consumerInstrumenter = consumerInstrumenter;
this.captureExperimentalSpanAttributes = captureExperimentalSpanAttributes;
this.fieldMapper = new FieldMapper();
}
@@ -56,8 +64,9 @@ final class TracingExecutionInterceptor implements ExecutionInterceptor {
Context.AfterMarshalling context, ExecutionAttributes executionAttributes) {
io.opentelemetry.context.Context parentOtelContext = io.opentelemetry.context.Context.current();
+ executionAttributes.putAttribute(SDK_REQUEST_ATTRIBUTE, context.request());
- if (!instrumenter.shouldStart(parentOtelContext, executionAttributes)) {
+ if (!requestInstrumenter.shouldStart(parentOtelContext, executionAttributes)) {
return;
}
@@ -65,7 +74,7 @@ final class TracingExecutionInterceptor implements ExecutionInterceptor {
executionAttributes.putAttribute(SDK_HTTP_REQUEST_ATTRIBUTE, httpRequest);
io.opentelemetry.context.Context otelContext =
- instrumenter.start(parentOtelContext, executionAttributes);
+ requestInstrumenter.start(parentOtelContext, executionAttributes);
executionAttributes.putAttribute(CONTEXT_ATTRIBUTE, otelContext);
if (executionAttributes
.getAttribute(SdkExecutionAttribute.CLIENT_TYPE)
@@ -84,12 +93,36 @@ final class TracingExecutionInterceptor implements ExecutionInterceptor {
populateRequestAttributes(span, awsSdkRequest, context.request(), executionAttributes);
}
} catch (Throwable throwable) {
- instrumenter.end(otelContext, executionAttributes, null, throwable);
+ requestInstrumenter.end(otelContext, executionAttributes, null, throwable);
clearAttributes(executionAttributes);
throw throwable;
}
}
+ @Override
+ public SdkRequest modifyRequest(
+ Context.ModifyRequest context, ExecutionAttributes executionAttributes) {
+ SdkRequest request = context.request();
+ if (SqsReceiveMessageRequestAccess.isInstance(request)) {
+ List<String> existingAttributeNames = getAttributeNames(request);
+ if (!existingAttributeNames.contains(SqsParentContext.AWS_TRACE_SYSTEM_ATTRIBUTE)) {
+ List<String> attributeNames = new ArrayList<>();
+ attributeNames.addAll(existingAttributeNames);
+ attributeNames.add(SqsParentContext.AWS_TRACE_SYSTEM_ATTRIBUTE);
+ SdkRequest.Builder builder = request.toBuilder();
+ SqsReceiveMessageRequestAccess.attributeNamesWithStrings(builder, attributeNames);
+ return builder.build();
+ }
+ }
+ return request;
+ }
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static List<String> getAttributeNames(SdkRequest request) {
+ Optional<List> optional = request.getValueForField("AttributeNames", List.class);
+ return optional.isPresent() ? (List<String>) optional.get() : Collections.emptyList();
+ }
+
@Override
public SdkHttpRequest modifyHttpRequest(
Context.ModifyHttpRequest context, ExecutionAttributes executionAttributes) {
@@ -125,16 +158,44 @@ final class TracingExecutionInterceptor implements ExecutionInterceptor {
@Override
public void afterExecution(
Context.AfterExecution context, ExecutionAttributes executionAttributes) {
+ if (SqsReceiveMessageRequestAccess.isInstance(context.request())) {
+ afterConsumerResponse(executionAttributes, context.response(), context.httpResponse());
+ }
+
// http request has been changed
executionAttributes.putAttribute(SDK_HTTP_REQUEST_ATTRIBUTE, context.httpRequest());
io.opentelemetry.context.Context otelContext = getContext(executionAttributes);
Span span = Span.fromContext(otelContext);
onUserAgentHeaderAvailable(span, executionAttributes);
onSdkResponse(span, context.response(), executionAttributes);
- instrumenter.end(otelContext, executionAttributes, context.httpResponse(), null);
+ requestInstrumenter.end(otelContext, executionAttributes, context.httpResponse(), null);
clearAttributes(executionAttributes);
}
+ /** Create and close CONSUMER span for each message consumed. */
+ private void afterConsumerResponse(
+ ExecutionAttributes executionAttributes, SdkResponse response, SdkHttpResponse httpResponse) {
+ List<Object> messages = getMessages(response);
+ for (Object message : messages) {
+ createConsumerSpan(message, executionAttributes, httpResponse);
+ }
+ }
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static List<Object> getMessages(SdkResponse response) {
+ Optional<List> optional = response.getValueForField("Messages", List.class);
+ return optional.isPresent() ? optional.get() : Collections.emptyList();
+ }
+
+ private void createConsumerSpan(
+ Object message, ExecutionAttributes executionAttributes, SdkHttpResponse httpResponse) {
+ io.opentelemetry.context.Context parentContext =
+ SqsParentContext.ofSystemAttributes(SqsMessageAccess.getAttributes(message));
+ io.opentelemetry.context.Context context =
+ consumerInstrumenter.start(parentContext, executionAttributes);
+ consumerInstrumenter.end(context, executionAttributes, httpResponse, null);
+ }
+
// Certain headers in the request like User-Agent are only available after execution.
private static void onUserAgentHeaderAvailable(Span span, ExecutionAttributes request) {
List<String> userAgent =
@@ -161,7 +222,7 @@ final class TracingExecutionInterceptor implements ExecutionInterceptor {
public void onExecutionFailure(
Context.FailedExecution context, ExecutionAttributes executionAttributes) {
io.opentelemetry.context.Context otelContext = getContext(executionAttributes);
- instrumenter.end(otelContext, executionAttributes, null, context.exception());
+ requestInstrumenter.end(otelContext, executionAttributes, null, context.exception());
clearAttributes(executionAttributes);
}

ADD
instrumentation/aws-sdk/aws-sdk-2.2/library/src/test/groovy/io/opentelemetry/instrumentation/awssdk/v2_2/Aws2SqsTracingTest.groovy
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.awssdk.v2_2
+
+import io.opentelemetry.instrumentation.test.LibraryTestTrait
+import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration
+
+class Aws2SqsTracingTest extends AbstractAws2SqsTracingTest implements LibraryTestTrait {
+ @Override
+ ClientOverrideConfiguration.Builder createOverrideConfigurationBuilder() {
+ return ClientOverrideConfiguration.builder()
+ .addExecutionInterceptor(
+ AwsSdkTelemetry.builder(getOpenTelemetry())
+ .setCaptureExperimentalSpanAttributes(true)
+ .build()
+ .newExecutionInterceptor())
+ }
+}

MODIFY
instrumentation/aws-sdk/aws-sdk-2.2/testing/build.gradle.kts
instrumentation/aws-sdk/aws-sdk-2.2/testing/build.gradle.kts
@@ -17,6 +17,9 @@ dependencies {
api("software.amazon.awssdk:s3:2.2.0")
api("software.amazon.awssdk:sqs:2.2.0")
+ // needed for SQS - using emq directly as localstack references emq v0.15.7 ie WITHOUT AWS trace header propagation
+ implementation("org.elasticmq:elasticmq-rest-sqs_2.12:1.0.0")
+
implementation("com.google.guava:guava")
implementation("org.apache.groovy:groovy")

MODIFY
instrumentation/aws-sdk/aws-sdk-2.2/testing/src/main/groovy/io/opentelemetry/instrumentation/awssdk/v2_2/AbstractAws2ClientTest.groovy
instrumentation/aws-sdk/aws-sdk-2.2/testing/src/main/groovy/io/opentelemetry/instrumentation/awssdk/v2_2/AbstractAws2ClientTest.groovy
@@ -57,6 +57,7 @@ import java.util.concurrent.Future
import static com.google.common.collect.ImmutableMap.of
import static io.opentelemetry.api.trace.SpanKind.CLIENT
+import static io.opentelemetry.api.trace.SpanKind.PRODUCER
import static io.opentelemetry.api.trace.StatusCode.ERROR
import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.NetTransportValues.IP_TCP
@@ -341,7 +342,7 @@ abstract class AbstractAws2ClientTest extends InstrumentationSpecification {
trace(0, 1) {
span(0) {
name "$service.$operation"
- kind CLIENT
+ kind operation != "SendMessage" ? CLIENT : PRODUCER
hasNoParent()
attributes {
"$SemanticAttributes.NET_TRANSPORT" IP_TCP
@@ -431,7 +432,7 @@ abstract class AbstractAws2ClientTest extends InstrumentationSpecification {
trace(0, 1) {
span(0) {
name "$service.$operation"
- kind CLIENT
+ kind operation != "SendMessage" ? CLIENT : PRODUCER
hasNoParent()
attributes {
"$SemanticAttributes.NET_TRANSPORT" IP_TCP

ADD
instrumentation/aws-sdk/aws-sdk-2.2/testing/src/main/groovy/io/opentelemetry/instrumentation/awssdk/v2_2/AbstractAws2SqsTracingTest.groovy
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.awssdk.v2_2
+
+import io.opentelemetry.instrumentation.test.InstrumentationSpecification
+import io.opentelemetry.instrumentation.test.utils.PortUtils
+import org.elasticmq.rest.sqs.SQSRestServerBuilder
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider
+import software.amazon.awssdk.core.client.builder.SdkClientBuilder
+import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.sqs.SqsClient
+import software.amazon.awssdk.services.sqs.model.CreateQueueRequest
+import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest
+import software.amazon.awssdk.services.sqs.model.SendMessageRequest
+import spock.lang.Shared
+
+import static io.opentelemetry.api.trace.SpanKind.CLIENT
+import static io.opentelemetry.api.trace.SpanKind.CONSUMER
+import static io.opentelemetry.api.trace.SpanKind.PRODUCER
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.NetTransportValues.IP_TCP
+
+abstract class AbstractAws2SqsTracingTest extends InstrumentationSpecification {
+
+ private static final StaticCredentialsProvider CREDENTIALS_PROVIDER = StaticCredentialsProvider
+ .create(AwsBasicCredentials.create("my-access-key", "my-secret-key"))
+
+ @Shared
+ def sqs
+ @Shared
+ SqsClient client
+ @Shared
+ int sqsPort
+
+ void configureSdkClient(SdkClientBuilder builder) {
+ builder.overrideConfiguration(createOverrideConfigurationBuilder().build())
+ }
+
+ abstract ClientOverrideConfiguration.Builder createOverrideConfigurationBuilder()
+
+ def setupSpec() {
+
+ sqsPort = PortUtils.findOpenPort()
+ sqs = SQSRestServerBuilder.withPort(sqsPort).withInterface("localhost").start()
+ println getClass().name + " SQS server started at: localhost:$sqsPort/"
+
+ def builder = SqsClient.builder()
+ configureSdkClient(builder)
+ client = builder
+ .endpointOverride(new URI("http://localhost:" + sqsPort))
+ .region(Region.AP_NORTHEAST_1)
+ .credentialsProvider(CREDENTIALS_PROVIDER)
+ .build()
+ }
+
+ def cleanupSpec() {
+ if (sqs != null) {
+ sqs.stopAndWait()
+ }
+ }
+
+ def "simple sqs producer-consumer services"() {
+ setup:
+ CreateQueueRequest createQueueRequest = CreateQueueRequest.builder()
+ .queueName("testSdkSqs")
+ .build()
+ client.createQueue(createQueueRequest)
+
+ when:
+ SendMessageRequest sendMessageRequest = SendMessageRequest.builder()
+ .queueUrl("http://localhost:$sqsPort/000000000000/testSdkSqs")
+ .messageBody("{\"type\": \"hello\"}")
+ .build()
+ client.sendMessage(sendMessageRequest)
+ ReceiveMessageRequest receiveMessageRequest = ReceiveMessageRequest.builder()
+ .queueUrl("http://localhost:$sqsPort/000000000000/testSdkSqs")
+ .build()
+ client.receiveMessage(receiveMessageRequest)
+
+ then:
+ assertTraces(3) {
+ trace(0, 1) {
+
+ span(0) {
+ name "Sqs.CreateQueue"
+ kind CLIENT
+ hasNoParent()
+ attributes {
+ "aws.agent" "java-aws-sdk"
+ "aws.queue.name" "testSdkSqs"
+ "aws.requestId" "00000000-0000-0000-0000-000000000000"
+ "rpc.system" "aws-api"
+ "rpc.service" "Sqs"
+ "rpc.method" "CreateQueue"
+ "http.flavor" "1.1"
+ "http.method" "POST"
+ "http.status_code" 200
+ "http.url" { it.startsWith("http://localhost:$sqsPort") }
+ "http.user_agent" String
+ "net.peer.name" "localhost"
+ "net.peer.port" sqsPort
+ "net.transport" IP_TCP
+ }
+ }
+ }
+ trace(1, 2) {
+ span(0) {
+ name "Sqs.SendMessage"
+ kind PRODUCER
+ hasNoParent()
+ attributes {
+ "aws.agent" "java-aws-sdk"
+ "aws.queue.url" "http://localhost:$sqsPort/000000000000/testSdkSqs"
+ "aws.requestId" "00000000-0000-0000-0000-000000000000"
+ "rpc.system" "aws-api"
+ "rpc.method" "SendMessage"
+ "rpc.service" "Sqs"
+ "http.flavor" "1.1"
+ "http.method" "POST"
+ "http.status_code" 200
+ "http.url" { it.startsWith("http://localhost:$sqsPort") }
+ "http.user_agent" String
+ "net.peer.name" "localhost"
+ "net.peer.port" sqsPort
+ "net.transport" IP_TCP
+ }
+ }
+ span(1) {
+ name "Sqs.ReceiveMessage"
+ kind CONSUMER
+ childOf span(0)
+ attributes {
+ "aws.agent" "java-aws-sdk"
+ "rpc.method" "ReceiveMessage"
+ "rpc.system" "aws-api"
+ "rpc.service" "Sqs"
+ "http.flavor" "1.1"
+ "http.method" "POST"
+ "http.status_code" 200
+ "http.url" { it.startsWith("http://localhost:$sqsPort") }
+ "net.peer.name" "localhost"
+ "net.peer.port" sqsPort
+ "net.transport" IP_TCP
+ }
+ }
+ }
+ /**
+ * This span represents HTTP "sending of receive message" operation. It's always single, while there can be multiple CONSUMER spans (one per consumed message).
+ * This one could be suppressed (by IF in TracingRequestHandler#beforeRequest but then HTTP instrumentation span would appear
+ */
+ trace(2, 1) {
+ span(0) {
+ name "Sqs.ReceiveMessage"
+ kind CLIENT
+ hasNoParent()
+ attributes {
+ "aws.agent" "java-aws-sdk"
+ "aws.requestId" "00000000-0000-0000-0000-000000000000"
+ "rpc.method" "ReceiveMessage"
+ "aws.queue.url" "http://localhost:$sqsPort/000000000000/testSdkSqs"
+ "rpc.system" "aws-api"
+ "rpc.service" "Sqs"
+ "http.flavor" "1.1"
+ "http.method" "POST"
+ "http.status_code" 200
+ "http.url" { it.startsWith("http://localhost:$sqsPort") }
+ "http.user_agent" String
+ "net.peer.name" "localhost"
+ "net.peer.port" sqsPort
+ "net.transport" IP_TCP
+ }
+ }
+ }
+ }
+ }
+}
\ No newline at end of file

Aws sdk2 sqs context propagation (#6199)