AWS SDK - SNS to SQS tracing propagation test
MODIFY
benchmark-e2e/benchmark-e2e.gradle
benchmark-e2e/benchmark-e2e.gradle
@@ -5,7 +5,7 @@ description = 'e2e benchmark'
dependencies {
implementation 'org.junit.jupiter:junit-jupiter-api:5.3.1'
implementation 'org.slf4j:slf4j-simple:1.6.1'
- implementation 'org.testcontainers:testcontainers:1.15.1'
+ implementation 'org.testcontainers:testcontainers:1.15.2'
}
test {

MODIFY
examples/distro/smoke-tests/build.gradle
examples/distro/smoke-tests/build.gradle
@@ -3,7 +3,7 @@ plugins {
}
dependencies {
- testImplementation("org.testcontainers:testcontainers:1.15.1")
+ testImplementation("org.testcontainers:testcontainers:1.15.2")
testImplementation("com.fasterxml.jackson.core:jackson-databind:2.11.2")
testImplementation("com.google.protobuf:protobuf-java-util:3.12.4")
testImplementation("com.squareup.okhttp3:okhttp:3.12.12")

MODIFY
gradle/dependencies.gradle
gradle/dependencies.gradle
@@ -29,7 +29,8 @@ ext {
autoValue : "1.7.4",
systemLambda : "1.1.0",
prometheus : "0.9.0",
- assertj : '3.19.0'
+ assertj : '3.19.0',
+ testcontainers : '1.15.2'
]
deps = [
@@ -84,7 +85,7 @@ ext {
],
groovy : "org.codehaus.groovy:groovy-all:${versions.groovy}",
systemLambda : "com.github.stefanbirkner:system-lambda:${versions.systemLambda}",
- testcontainers : "org.testcontainers:testcontainers:1.15.1",
+ testcontainers : "org.testcontainers:testcontainers:${versions.testcontainers}",
testLogging : [
dependencies.create(group: 'ch.qos.logback', name: 'logback-classic', version: versions.logback),
dependencies.create(group: 'org.slf4j', name: 'log4j-over-slf4j', version: versions.slf4j),

MODIFY
instrumentation/apache-camel-2.20/javaagent/apache-camel-2.20-javaagent.gradle
instrumentation/apache-camel-2.20/javaagent/apache-camel-2.20-javaagent.gradle
@@ -41,7 +41,7 @@ dependencies {
* Temporarily using emq instead of localstack till the latter supports AWS trace propagation
*
* testImplementation deps.testcontainers
- * testImplementation "org.testcontainers:localstack:1.15.1"
+ * testImplementation "org.testcontainers:localstack:${versions.testcontainers}
**/
latestDepTestLibrary group: 'org.apache.camel', name: 'camel-core', version: '2.+'

MODIFY
instrumentation/aws-sdk/aws-sdk-1.11/javaagent/aws-sdk-1.11-javaagent.gradle
instrumentation/aws-sdk/aws-sdk-1.11/javaagent/aws-sdk-1.11-javaagent.gradle
@@ -44,6 +44,7 @@ configurations {
resolutionStrategy.force 'com.amazonaws:aws-java-sdk-ec2:1.11.0'
resolutionStrategy.force 'com.amazonaws:aws-java-sdk-kinesis:1.11.0'
resolutionStrategy.force 'com.amazonaws:aws-java-sdk-sqs:1.11.0'
+ resolutionStrategy.force 'com.amazonaws:aws-java-sdk-sns:1.11.0'
resolutionStrategy.force 'com.amazonaws:aws-java-sdk-dynamodb:1.11.0'
}
}
@@ -58,6 +59,7 @@ dependencies {
testLibrary group: 'com.amazonaws', name: 'aws-java-sdk-ec2', version: '1.11.106'
testLibrary group: 'com.amazonaws', name: 'aws-java-sdk-kinesis', version: '1.11.106'
testLibrary group: 'com.amazonaws', name: 'aws-java-sdk-dynamodb', version: '1.11.106'
+ testLibrary group: 'com.amazonaws', name: 'aws-java-sdk-sns', version: '1.11.106'
testSqsImplementation group: 'com.amazonaws', name: 'aws-java-sdk-sqs', version: '1.11.106'
// needed for SQS - using emq directly as localstack references emq v0.15.7 ie WITHOUT AWS trace header propagation
@@ -69,11 +71,16 @@ dependencies {
// needed for kinesis:
testImplementation group: 'com.fasterxml.jackson.dataformat', name: 'jackson-dataformat-cbor', version: versions.jackson
+ // needed for SNS
+ testImplementation deps.testcontainers
+ testImplementation "org.testcontainers:localstack:${versions.testcontainers}"
+
test_before_1_11_106Implementation(group: 'com.amazonaws', name: 'aws-java-sdk-s3', version: '1.11.0')
test_before_1_11_106Implementation(group: 'com.amazonaws', name: 'aws-java-sdk-rds', version: '1.11.0')
test_before_1_11_106Implementation(group: 'com.amazonaws', name: 'aws-java-sdk-ec2', version: '1.11.0')
test_before_1_11_106Implementation(group: 'com.amazonaws', name: 'aws-java-sdk-kinesis', version: '1.11.0')
test_before_1_11_106Implementation(group: 'com.amazonaws', name: 'aws-java-sdk-dynamodb', version: '1.11.0')
+ test_before_1_11_106Implementation(group: 'com.amazonaws', name: 'aws-java-sdk-sns', version: '1.11.0')
}
test {

ADD
instrumentation/aws-sdk/aws-sdk-1.11/javaagent/src/test/groovy/SnsTracingTest.groovy
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import static io.opentelemetry.api.trace.SpanKind.CLIENT
+import static io.opentelemetry.api.trace.SpanKind.CONSUMER
+
+import com.amazonaws.services.sns.AmazonSNSAsyncClient
+import com.amazonaws.services.sns.model.CreateTopicResult
+import com.amazonaws.services.sqs.AmazonSQSAsyncClient
+import com.amazonaws.services.sqs.model.GetQueueAttributesRequest
+import com.amazonaws.services.sqs.model.ReceiveMessageRequest
+import io.opentelemetry.instrumentation.test.AgentInstrumentationSpecification
+import org.slf4j.LoggerFactory
+import org.testcontainers.containers.localstack.LocalStackContainer
+import org.testcontainers.containers.output.Slf4jLogConsumer
+import org.testcontainers.utility.DockerImageName
+import spock.lang.Ignore
+import spock.lang.Shared
+
+class SnsTracingTest extends AgentInstrumentationSpecification {
+
+ @Shared
+ LocalStackContainer localstack
+ @Shared
+ AmazonSQSAsyncClient sqsClient
+ @Shared
+ AmazonSNSAsyncClient snsClient
+
+ def setupSpec() {
+
+ localstack = new LocalStackContainer(DockerImageName.parse("localstack/localstack:latest"))
+ .withServices(LocalStackContainer.Service.SQS, LocalStackContainer.Service.SNS)
+ .withEnv("DEBUG", "1")
+ .withEnv("SQS_PROVIDER", "elasticmq")
+ localstack.start()
+
+ sqsClient = AmazonSQSAsyncClient.asyncBuilder()
+ .withEndpointConfiguration(localstack.getEndpointConfiguration(LocalStackContainer.Service.SQS))
+ .withCredentials(localstack.getDefaultCredentialsProvider())
+ .build()
+
+ snsClient = AmazonSNSAsyncClient.asyncBuilder()
+ .withEndpointConfiguration(localstack.getEndpointConfiguration(LocalStackContainer.Service.SNS))
+ .withCredentials(localstack.getDefaultCredentialsProvider())
+ .build()
+
+ localstack.followOutput(new Slf4jLogConsumer(LoggerFactory.getLogger("test")))
+ }
+
+ def cleanupSpec() {
+ if (localstack != null) {
+ localstack.stop()
+ }
+ }
+
+ def createQueue(String queueName) {
+ return sqsClient.createQueue(queueName).getQueueUrl()
+ }
+
+ def getQueueArn(String queueUrl) {
+ return sqsClient.getQueueAttributes(
+ new GetQueueAttributesRequest(queueUrl)
+ .withAttributeNames("QueueArn")).getAttributes()
+ .get("QueueArn")
+ }
+
+ def setQueuePolicy(String queueUrl, String queueArn) {
+ sqsClient.setQueueAttributes(queueUrl, Collections.singletonMap("Policy", policy(queueArn)))
+ }
+
+ def createAndSubscribeTopic(String topicName, String queueArn) {
+ CreateTopicResult ctr = snsClient.createTopic(topicName)
+ snsClient.subscribe(ctr.getTopicArn(), "sqs", queueArn)
+ return ctr.getTopicArn()
+ }
+
+ @Ignore("Requires https://github.com/localstack/localstack/issues/3669 to work with localstack")
+ def "simple SNS producer - SQS consumer services"() {
+ setup:
+ String queueName = "snsToSqsTestQueue"
+ String topicName = "snsToSqsTestTopic"
+
+ String queueUrl = createQueue(queueName)
+ String queueArn = getQueueArn(queueUrl)
+ setQueuePolicy(queueUrl, queueArn)
+ String topicArn = createAndSubscribeTopic(topicName, queueArn)
+
+ when:
+ snsClient.publish(topicArn, "Hello There")
+ Thread.sleep(3000)
+ ReceiveMessageRequest rmr = new ReceiveMessageRequest(queueUrl).withMessageAttributeNames("test")
+ sqsClient.receiveMessage(rmr)
+ Thread.sleep(1000)
+
+ then:
+ assertTraces(7) {
+ trace(0, 1) {
+
+ span(0) {
+ name "SQS.CreateQueue"
+ kind CLIENT
+ hasNoParent()
+ attributes {
+ "aws.agent" "java-aws-sdk"
+ "aws.endpoint" String
+ "aws.operation" "CreateQueueRequest"
+ "aws.queue.name" queueName
+ "aws.service" "AmazonSQS"
+ "http.flavor" "1.1"
+ "http.method" "POST"
+ "http.status_code" 200
+ "http.url" String
+ "net.peer.name" String
+ "net.transport" "IP.TCP"
+ "net.peer.port" {it == null || Number}
+ }
+ }
+ }
+ trace(1, 1) {
+
+ span(0) {
+ name "SQS.GetQueueAttributes"
+ kind CLIENT
+ hasNoParent()
+ attributes {
+ "aws.agent" "java-aws-sdk"
+ "aws.endpoint" String
+ "aws.operation" "GetQueueAttributesRequest"
+ "aws.queue.url" queueUrl
+ "aws.service" "AmazonSQS"
+ "http.flavor" "1.1"
+ "http.method" "POST"
+ "http.status_code" 200
+ "http.url" String
+ "net.peer.name" String
+ "net.transport" "IP.TCP"
+ "net.peer.port" {it == null || Number}
+ }
+ }
+ }
+ trace(2, 1) {
+
+ span(0) {
+ name "SQS.SetQueueAttributes"
+ kind CLIENT
+ hasNoParent()
+ attributes {
+ "aws.agent" "java-aws-sdk"
+ "aws.endpoint" String
+ "aws.operation" "SetQueueAttributesRequest"
+ "aws.queue.url" queueUrl
+ "aws.service" "AmazonSQS"
+ "http.flavor" "1.1"
+ "http.method" "POST"
+ "http.status_code" 200
+ "http.url" String
+ "net.peer.name" String
+ "net.transport" "IP.TCP"
+ "net.peer.port" {it == null || Number}
+ }
+ }
+ }
+ trace(3, 1) {
+
+ span(0) {
+ name "SNS.CreateTopic"
+ kind CLIENT
+ hasNoParent()
+ attributes {
+ "aws.agent" "java-aws-sdk"
+ "aws.endpoint" String
+ "aws.operation" "CreateTopicRequest"
+ "aws.service" "AmazonSNS"
+ "http.flavor" "1.1"
+ "http.method" "POST"
+ "http.status_code" 200
+ "http.url" String
+ "net.peer.name" String
+ "net.transport" "IP.TCP"
+ "net.peer.port" {it == null || Number}
+ }
+ }
+ }
+ trace(4, 1) {
+
+ span(0) {
+ name "SNS.Subscribe"
+ kind CLIENT
+ hasNoParent()
+ attributes {
+ "aws.agent" "java-aws-sdk"
+ "aws.endpoint" String
+ "aws.operation" "SubscribeRequest"
+ "aws.service" "AmazonSNS"
+ "http.flavor" "1.1"
+ "http.method" "POST"
+ "http.status_code" 200
+ "http.url" String
+ "net.peer.name" String
+ "net.transport" "IP.TCP"
+ "net.peer.port" {it == null || Number}
+ }
+ }
+ }
+ trace(5, 2) {
+ span(0) {
+ name "SNS.Publish"
+ kind CLIENT
+ hasNoParent()
+ attributes {
+ "aws.agent" "java-aws-sdk"
+ "aws.endpoint" String
+ "aws.operation" "PublishRequest"
+ "aws.service" "AmazonSNS"
+ "http.flavor" "1.1"
+ "http.method" "POST"
+ "http.status_code" 200
+ "http.url" String
+ "net.peer.name" String
+ "net.transport" "IP.TCP"
+ "net.peer.port" {it == null || Number}
+ }
+ }
+ span(1) {
+ name "SQS.ReceiveMessage"
+ kind CONSUMER
+ childOf span(0)
+ attributes {
+ "aws.agent" "java-aws-sdk"
+ "aws.endpoint" String
+ "aws.operation" "ReceiveMessageRequest"
+ "aws.queue.url" queueUrl
+ "aws.service" "AmazonSQS"
+ "http.flavor" "1.1"
+ "http.method" "POST"
+ "http.status_code" 200
+ "http.url" String
+ "http.user_agent" String
+ "net.peer.name" String
+ "net.transport" "IP.TCP"
+ "net.peer.port" {it == null || Number}
+ }
+ }
+ }
+ /**
+ * 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(6, 1) {
+ span(0) {
+ name "SQS.ReceiveMessage"
+ kind CLIENT
+ hasNoParent()
+ attributes {
+ "aws.agent" "java-aws-sdk"
+ "aws.endpoint" String
+ "aws.operation" "ReceiveMessageRequest"
+ "aws.queue.url" queueUrl
+ "aws.service" "AmazonSQS"
+ "http.flavor" "1.1"
+ "http.method" "POST"
+ "http.status_code" 200
+ "http.url" String
+ "net.peer.name" String
+ "net.transport" "IP.TCP"
+ "net.peer.port" {it == null || Number}
+ }
+ }
+ }
+ }
+ }
+
+ def policy(String queueArn) {
+ return String.format(SQS_POLICY, queueArn)
+ }
+
+ private static final String SQS_POLICY = "{" +
+ " \"Statement\": [" +
+ " {" +
+ " \"Effect\": \"Allow\"," +
+ " \"Principal\": \"*\"," +
+ " \"Action\": \"sqs:SendMessage\"," +
+ " \"Resource\": \"%s\"" +
+ " }]" +
+ "}"
+}
\ No newline at end of file

MODIFY
instrumentation/aws-sdk/aws-sdk-1.11/javaagent/src/test/resources/logback-test.xml
instrumentation/aws-sdk/aws-sdk-1.11/javaagent/src/test/resources/logback-test.xml
</layout>
</appender>
- <root level="DEBUG">
+ <root level="INFO">
<appender-ref ref="console"/>
</root>
- <logger name="org.elasticmq" level="debug" />
-
+ <logger name="org.elasticmq" level="INFO" />
+ <logger name="test" level="WARN"/>
+ <logger name="org.testcontainers" level="WARN" />
+ <logger name="com.amazonaws" level="DEBUG" />
</configuration>

AWS SDK - SNS to SQS tracing propagation test (#2483)