Migrates zipkin-aws to v2 SDK; add v2 sender/instrumentation (#237)

Ports collectors (SQS, Kinesis), module/, and aws-junit from AWS SDK v1
to v2 in-place, keeping package names. Adds new instrumentation-awssdk-sqs
and sender-awssdk-kinesis modules. The module jar exclusively uses v2 SDK
with zero com.amazonaws dependencies. Upgrades KCL to 3.4.1 and fixes all
CVEs via jackson-bom overrides.

Signed-off-by: Adrian Cole <adrian@tetrate.io>

Author: codefromthecrypt

.dockerignore

@@ -14,10 +14,12 @@
 !brave/instrumentation-aws-java-sdk-core/src/main/**
 !brave/instrumentation-aws-java-sdk-v2-core/src/main/**
 !brave/instrumentation-aws-java-sdk-sqs/src/main/**
+!brave/instrumentation-awssdk-sqs/src/main/**
 !brave/propagation-aws/src/main/**
 !module/src/main/**
 !reporter/reporter-xray-udp/src/main/**
 !reporter/sender-awssdk-sqs/src/main/**
+!reporter/sender-awssdk-kinesis/src/main/**
 !reporter/sender-kinesis/src/main/**
 !reporter/sender-sqs/src/main/**
 !storage/xray-udp/src/main/**

README.md

@@ -26,6 +26,7 @@ Instrumentation | Description
 [AWS SDK](brave/instrumentation-aws-java-sdk-core) | Traces [AmazonWebServiceClient](https://github.com/aws/aws-sdk-java)
 [AWS SDK V2](brave/instrumentation-aws-java-sdk-v2-core) | Traces [SdkClient](https://github.com/aws/aws-sdk-java-v2)
 [AWS SQS Messaging](brave/instrumentation-aws-java-sdk-sqs) | Traces [AmazonSQS](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/sqs/AmazonSQS.html)
+[AWS SQS V2 Messaging](brave/instrumentation-awssdk-sqs) | Traces [SqsClient](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/sqs/SqsClient.html)
 
 We also have a [library to read Amazon's trace header](brave/propagation-aws).
 
@@ -49,6 +50,7 @@ Sender | Description
 [SQS](reporter/sender-sqs) | Sends tracing data to Zipkin using [SQS](https://aws.amazon.com/sqs/), a message queue service.
 [SQS v2](reporter/sender-awssdk-sqs) | Sends tracing data to Zipkin using [SQS](https://aws.amazon.com/sqs/), a message queue service.
 [Kinesis](reporter/sender-kinesis) | Sends tracing data to Zipkin using [Kinesis](https://aws.amazon.com/kinesis/), an alternative similar to Kafka.
+[Kinesis v2](reporter/sender-awssdk-kinesis) | Sends tracing data to Zipkin using [Kinesis](https://aws.amazon.com/kinesis/) with the [V2 AWS SDK](https://github.com/aws/aws-sdk-java-v2).
 
 ## Collectors
 The component in a zipkin server that receives trace data is called a

aws-junit/pom.xml

@@ -12,7 +12,7 @@
   <parent>
     <groupId>io.zipkin.aws</groupId>
     <artifactId>zipkin-aws-parent</artifactId>
-    <version>1.4.1-SNAPSHOT</version>
+    <version>2.0.0-SNAPSHOT</version>
   </parent>
 
   <artifactId>zipkin-aws-junit</artifactId>
@@ -56,21 +56,14 @@
     </dependency>
 
     <dependency>
-      <groupId>com.amazonaws</groupId>
-      <artifactId>aws-java-sdk-sqs</artifactId>
-      <version>${aws-java-sdk.version}</version>
-      <exclusions>
-        <exclusion>
-          <groupId>com.fasterxml.jackson.core</groupId>
-          <artifactId>jackson-core</artifactId>
-        </exclusion>
-      </exclusions>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>sqs</artifactId>
+      <version>${sdk-core.version}</version>
     </dependency>
-
     <dependency>
-      <groupId>com.fasterxml.jackson.core</groupId>
-      <artifactId>jackson-core</artifactId>
-      <version>${jackson.version}</version>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>url-connection-client</artifactId>
+      <version>${sdk-core.version}</version>
     </dependency>
 
     <dependency>

aws-junit/src/main/java/zipkin2/junit/aws/AmazonSQSExtension.java

@@ -4,17 +4,8 @@
  */
 package zipkin2.junit.aws;
 
-import com.amazonaws.auth.AWSStaticCredentialsProvider;
-import com.amazonaws.auth.BasicAWSCredentials;
-import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration;
-import com.amazonaws.services.sqs.AmazonSQS;
-import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
-import com.amazonaws.services.sqs.model.DeleteMessageRequest;
-import com.amazonaws.services.sqs.model.Message;
-import com.amazonaws.services.sqs.model.PurgeQueueRequest;
-import com.amazonaws.services.sqs.model.ReceiveMessageResult;
-import com.amazonaws.services.sqs.model.SendMessageRequest;
-import com.amazonaws.util.Base64;
+import java.net.URI;
+import java.util.Base64;
 import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -24,6 +15,18 @@
 import org.junit.jupiter.api.extension.AfterEachCallback;
 import org.junit.jupiter.api.extension.BeforeEachCallback;
 import org.junit.jupiter.api.extension.ExtensionContext;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sqs.SqsClient;
+import software.amazon.awssdk.services.sqs.model.DeleteMessageRequest;
+import software.amazon.awssdk.services.sqs.model.Message;
+import software.amazon.awssdk.services.sqs.model.PurgeQueueRequest;
+import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
+import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
+import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
+import software.amazon.awssdk.services.sqs.model.SendMessageRequest;
 import zipkin2.Span;
 import zipkin2.codec.SpanBytesDecoder;
 
@@ -33,7 +36,7 @@
 public class AmazonSQSExtension implements BeforeEachCallback, AfterEachCallback {
   SQSRestServer server;
   int serverPort;
-  AmazonSQS client;
+  SqsClient client;
   String queueUrl;
 
   public AmazonSQSExtension() {
@@ -47,22 +50,24 @@ public AmazonSQSExtension() {
     }
 
     if (client == null) {
-      client = AmazonSQSClientBuilder.standard()
-          .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("x", "x")))
-          .withEndpointConfiguration(
-              new EndpointConfiguration("http://localhost:%d".formatted(serverPort), null))
+      client = SqsClient.builder()
+          .httpClient(UrlConnectionHttpClient.create())
+          .credentialsProvider(
+              StaticCredentialsProvider.create(AwsBasicCredentials.create("x", "x")))
+          .endpointOverride(URI.create("http://localhost:%d".formatted(serverPort)))
+          .region(Region.US_EAST_1)
           .build();
-      queueUrl = client.createQueue("zipkin").getQueueUrl();
+      queueUrl = client.createQueue(b -> b.queueName("zipkin")).queueUrl();
     }
-    
+
     if (client != null && queueUrl != null) {
-      client.purgeQueue(new PurgeQueueRequest(queueUrl));
+      client.purgeQueue(PurgeQueueRequest.builder().queueUrl(queueUrl).build());
     }
   }
 
   @Override public void afterEach(ExtensionContext extensionContext) {
     if (client != null) {
-      client.shutdown();
+      client.close();
       client = null;
     }
 
@@ -77,18 +82,19 @@ public String queueUrl() {
   }
 
   public int queueCount() {
-    String count = client.getQueueAttributes(queueUrl, singletonList("ApproximateNumberOfMessages"))
-        .getAttributes()
-        .get("ApproximateNumberOfMessages");
+    String count = client.getQueueAttributes(b -> b.queueUrl(queueUrl)
+            .attributeNames(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES))
+        .attributes()
+        .get(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES);
 
     return Integer.parseInt(count);
   }
 
   public int notVisibleCount() {
-    String count =
-        client.getQueueAttributes(queueUrl, singletonList("ApproximateNumberOfMessagesNotVisible"))
-            .getAttributes()
-            .get("ApproximateNumberOfMessagesNotVisible");
+    String count = client.getQueueAttributes(b -> b.queueUrl(queueUrl)
+            .attributeNames(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES_NOT_VISIBLE))
+        .attributes()
+        .get(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES_NOT_VISIBLE);
 
     return Integer.parseInt(count);
   }
@@ -101,19 +107,24 @@ public List<Span> getSpans(boolean delete) {
 
     Stream<Span> spans = Stream.empty();
 
-    ReceiveMessageResult result = client.receiveMessage(queueUrl);
+    ReceiveMessageResponse result = client.receiveMessage(
+        ReceiveMessageRequest.builder().queueUrl(queueUrl).build());
 
-    while (result != null && !result.getMessages().isEmpty()) {
+    while (result != null && !result.messages().isEmpty()) {
 
       spans = Stream.concat(spans,
-          result.getMessages().stream().flatMap(AmazonSQSExtension::decodeSpans));
+          result.messages().stream().flatMap(AmazonSQSExtension::decodeSpans));
 
-      result = client.receiveMessage(queueUrl);
+      result = client.receiveMessage(
+          ReceiveMessageRequest.builder().queueUrl(queueUrl).build());
 
       if (delete) {
-        List<DeleteMessageRequest> deletes = result.getMessages()
+        List<DeleteMessageRequest> deletes = result.messages()
             .stream()
-            .map(m -> new DeleteMessageRequest(queueUrl, m.getReceiptHandle()))
+            .map(m -> DeleteMessageRequest.builder()
+                .queueUrl(queueUrl)
+                .receiptHandle(m.receiptHandle())
+                .build())
             .toList();
         deletes.forEach(d -> client.deleteMessage(d));
       }
@@ -123,12 +134,15 @@ public List<Span> getSpans(boolean delete) {
   }
 
   public void send(String body) {
-    client.sendMessage(new SendMessageRequest(queueUrl, body));
+    client.sendMessage(SendMessageRequest.builder()
+        .queueUrl(queueUrl)
+        .messageBody(body)
+        .build());
   }
 
   static Stream<? extends Span> decodeSpans(Message m) {
     byte[] bytes =
-        m.getBody().charAt(0) == '[' ? m.getBody().getBytes(UTF_8) : Base64.decode(m.getBody());
+        m.body().charAt(0) == '[' ? m.body().getBytes(UTF_8) : Base64.getDecoder().decode(m.body());
     if (bytes[0] == '[') {
       return SpanBytesDecoder.JSON_V2.decodeList(bytes).stream();
     }

brave/instrumentation-aws-java-sdk-core/pom.xml

@@ -11,7 +11,7 @@
   <parent>
     <artifactId>zipkin-aws-parent</artifactId>
     <groupId>io.zipkin.aws</groupId>
-    <version>1.4.1-SNAPSHOT</version>
+    <version>2.0.0-SNAPSHOT</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

brave/instrumentation-aws-java-sdk-sqs/pom.xml

@@ -11,7 +11,7 @@
   <parent>
     <artifactId>zipkin-aws-parent</artifactId>
     <groupId>io.zipkin.aws</groupId>
-    <version>1.4.1-SNAPSHOT</version>
+    <version>2.0.0-SNAPSHOT</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

brave/instrumentation-aws-java-sdk-v2-core/pom.xml

@@ -11,7 +11,7 @@
   <parent>
     <artifactId>zipkin-aws-parent</artifactId>
     <groupId>io.zipkin.aws</groupId>
-    <version>1.4.1-SNAPSHOT</version>
+    <version>2.0.0-SNAPSHOT</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

brave/instrumentation-awssdk-sqs/README.md

@@ -0,0 +1,32 @@
+# AWS SQS V2 Messaging Instrumentation
+
+This module contains instrumentation for AWS SDK V2 `SqsClient` `SendMessage*` request types.
+
+The `SqsMessageTracing` type provides an `ExecutionInterceptor` instance that can be added to your
+clients at build time. It adds tracing headers to both `SendMessageRequest` and
+`SendMessageBatchRequest` types. In the case of `SendMessageBatchRequest` each message in the batch
+gets its own `Span` and headers added.
+
+## Usage
+
+You will want to create your SQS client and add the execution interceptor:
+
+```java
+Tracing tracing = ...;
+SqsMessageTracing sqsMessageTracing = SqsMessageTracing.create(tracing);
+
+ClientOverrideConfiguration configuration = ClientOverrideConfiguration.builder()
+    .addExecutionInterceptor(sqsMessageTracing.executionInterceptor())
+    .build();
+
+SqsClient client = SqsClient.builder()
+    .overrideConfiguration(configuration)
+    .build();
+```
+
+Now use your SQS client as you normally would and spans will be attached to outgoing messages.
+
+## Notes
+
+* This is the V2 SDK equivalent of [instrumentation-aws-java-sdk-sqs](../instrumentation-aws-java-sdk-sqs).
+* This can be combined with [instrumentation-aws-java-sdk-v2-core](../instrumentation-aws-java-sdk-v2-core) on the same client to get both HTTP-level CLIENT spans and message-level PRODUCER spans.

brave/instrumentation-awssdk-sqs/pom.xml

@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright The OpenZipkin Authors
+    SPDX-License-Identifier: Apache-2.0
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <artifactId>zipkin-aws-parent</artifactId>
+    <groupId>io.zipkin.aws</groupId>
+    <version>2.0.0-SNAPSHOT</version>
+    <relativePath>../../pom.xml</relativePath>
+  </parent>
+
+  <artifactId>brave-instrumentation-awssdk-sqs</artifactId>
+  <name>Brave Instrumentation: AWS SDK V2 SQS</name>
+
+  <properties>
+    <main.basedir>${project.basedir}/../..</main.basedir>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>io.zipkin.brave</groupId>
+      <artifactId>brave</artifactId>
+      <version>${brave.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>sqs</artifactId>
+      <version>${sdk-core.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>io.zipkin.brave</groupId>
+      <artifactId>brave-context-log4j2</artifactId>
+      <version>${brave.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.zipkin.brave</groupId>
+      <artifactId>brave-tests</artifactId>
+      <version>${brave.version}</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+</project>