Adds support for json encoding

Before the AWS transports only supported thrift encoding. This adds json support, which will notably be used for zipkin2 format. Pending a release of zipkin 1.30.1 and a new zipkin-reporter
openzipkin · Aug 12, 2017 · 15ea81e · 15ea81e
1 parent 733c277
commit 15ea81e
Show file tree

Hide file tree

Showing 8 changed files with 55 additions and 32 deletions.
diff --git a/aws-junit/src/main/java/zipkin/junit/aws/AmazonSQSRule.java b/aws-junit/src/main/java/zipkin/junit/aws/AmazonSQSRule.java
@@ -1,5 +1,5 @@
 /**
- * Copyright 2016 The OpenZipkin Authors
+ * Copyright 2016-2017 The OpenZipkin Authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
  * in compliance with the License. You may obtain a copy of the License at
@@ -30,6 +30,7 @@
 import org.junit.rules.ExternalResource;
 import zipkin.Codec;
 import zipkin.Span;
+import zipkin.SpanDecoder;
 
 import static java.util.Collections.singletonList;
 
@@ -136,7 +137,6 @@ private void sendSpansInternal(List<Span> spans) {
 
   private static List<Span> fromBase64(String base64) {
     byte[] bytes = Base64.decode(base64);
-    return Codec.THRIFT.readSpans(bytes);
+    return SpanDecoder.DETECTING_DECODER.readSpans(bytes);
   }
-
 }
diff --git a/collector-kinesis/src/main/java/zipkin/collector/kinesis/KinesisSpanProcessor.java b/collector-kinesis/src/main/java/zipkin/collector/kinesis/KinesisSpanProcessor.java
@@ -18,13 +18,11 @@
 import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
 import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownInput;
 import com.amazonaws.services.kinesis.model.Record;
-import java.util.ArrayList;
-import java.util.List;
-import zipkin.Codec;
 import zipkin.collector.Collector;
-import zipkin.internal.Nullable;
 import zipkin.storage.Callback;
 
+import static zipkin.SpanDecoder.DETECTING_DECODER;
+
 public class KinesisSpanProcessor implements IRecordProcessor {
 
   private final Collector collector;
@@ -40,7 +38,7 @@ public void initialize(InitializationInput initializationInput) {
   @Override
   public void processRecords(ProcessRecordsInput processRecordsInput) {
     for (Record record : processRecordsInput.getRecords()) {
-      collector.acceptSpans(record.getData().array(), Codec.THRIFT, Callback.NOOP);
+      collector.acceptSpans(record.getData().array(), DETECTING_DECODER, Callback.NOOP);
     }
   }
 

diff --git a/collector-sqs/src/main/java/zipkin/collector/sqs/SQSSpanProcessor.java b/collector-sqs/src/main/java/zipkin/collector/sqs/SQSSpanProcessor.java
@@ -27,12 +27,12 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
 import java.util.logging.Logger;
-import zipkin.Codec;
 import zipkin.Component;
 import zipkin.collector.Collector;
 import zipkin.internal.Nullable;
 import zipkin.storage.Callback;
 
+import static zipkin.SpanDecoder.DETECTING_DECODER;
 
 final class SQSSpanProcessor implements Runnable, Component {
 
@@ -101,7 +101,7 @@ private void process(final List<Message> messages) {
       final String deleteId = String.valueOf(count++);
       try {
         byte[] spans = Base64.decode(message.getBody());
-        collector.acceptSpans(spans, Codec.THRIFT, new Callback<Void>() {
+        collector.acceptSpans(spans, DETECTING_DECODER, new Callback<Void>() {
           @Override public void onSuccess(@Nullable Void value) {
             toDelete.add(new DeleteMessageBatchRequestEntry(deleteId, message.getReceiptHandle()));
           }

diff --git a/pom.xml b/pom.xml
@@ -44,7 +44,7 @@
 
     <license-maven-plugin.version>2.11</license-maven-plugin.version>
 
-    <zipkin.version>1.29.4</zipkin.version>
+    <zipkin.version>1.30.1-SNAPSHOT</zipkin.version>
     <zipkin-reporter.version>1.0.1</zipkin-reporter.version>
 
     <slf4j.version>1.7.25</slf4j.version>

diff --git a/sender-kinesis/src/main/java/zipkin/reporter/kinesis/KinesisSender.java b/sender-kinesis/src/main/java/zipkin/reporter/kinesis/KinesisSender.java
@@ -49,6 +49,7 @@ public static KinesisSender create(String streamName) {
   public static Builder builder() {
     return new AutoValue_KinesisSender.Builder()
         .credentialsProvider(new DefaultAWSCredentialsProviderChain())
+        .encoding(Encoding.THRIFT)
         .messageMaxBytes(1024 * 1024); // 1MB Kinesis limit.
   }
 
@@ -68,6 +69,9 @@ public interface Builder {
     /** Maximum size of a message. Kinesis max message size is 1MB */
     Builder messageMaxBytes(int messageMaxBytes);
 
+    /** Allows you to change to json format. Default is thrift */
+    Builder encoding(Encoding encoding);
+
     KinesisSender build();
   }
 
@@ -83,6 +87,8 @@ public interface Builder {
   @Nullable
   abstract EndpointConfiguration endpointConfiguration();
 
+  abstract Builder toBuilder();
+
   private final AtomicBoolean closeCalled = new AtomicBoolean(false);
 
   private final AtomicReference<String> partitionKey = new AtomicReference<>("");
@@ -128,14 +134,9 @@ protected AmazonKinesisAsync compute() {
     return builder.build();
   }
 
-  @Override
-  public Encoding encoding() {
-    return Encoding.THRIFT;
-  }
-
   @Override
   public int messageSizeInBytes(List<byte[]> list) {
-    return Encoding.THRIFT.listSizeInBytes(list);
+    return encoding().listSizeInBytes(list);
   }
 
   @Override

diff --git a/sender-kinesis/src/test/java/zipkin/reporter/kinesis/KinesisSenderTest.java b/sender-kinesis/src/test/java/zipkin/reporter/kinesis/KinesisSenderTest.java
@@ -31,11 +31,12 @@
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
-import zipkin.Codec;
 import zipkin.Component;
 import zipkin.Span;
+import zipkin.SpanDecoder;
 import zipkin.TestObjects;
 import zipkin.reporter.Encoder;
+import zipkin.reporter.Encoding;
 import zipkin.reporter.internal.AwaitableCallback;
 
 import static java.util.stream.Collectors.toList;
@@ -58,10 +59,21 @@ public void setup() throws Exception {
   }
 
   @Test
-  public void sendsSpans() throws Exception {
+  public void sendsSpans_thrift() throws Exception {
+    sendSpans(Encoder.THRIFT);
+  }
+
+  @Test
+  public void sendsSpans_json() throws Exception {
+    sender.close();
+    sender = sender.toBuilder().encoding(Encoding.JSON).build();
+    sendSpans(Encoder.JSON);
+  }
+
+  void sendSpans(Encoder<Span> encoder) throws InterruptedException, IOException {
     server.enqueue(new MockResponse());
 
-    send(TestObjects.TRACE);
+    send(encoder, TestObjects.TRACE);
 
     RecordedRequest request = server.takeRequest();
     assertThat(extractSpans(request.getBody()))
@@ -111,13 +123,13 @@ void enqueueCborResponse(JsonNode document) throws JsonProcessingException {
   }
 
   List<Span> extractSpans(Buffer body) throws IOException {
-    byte[] thriftEncodedSpans = mapper.readTree(body.inputStream()).get("Data").binaryValue();
-    return Codec.THRIFT.readSpans(thriftEncodedSpans);
+    byte[] encodedSpans = mapper.readTree(body.inputStream()).get("Data").binaryValue();
+    return SpanDecoder.DETECTING_DECODER.readSpans(encodedSpans);
   }
 
-  void send(List<Span> spans) {
+  void send(Encoder<Span> encoder, List<Span> spans) {
     AwaitableCallback callback = new AwaitableCallback();
-    sender.sendSpans(spans.stream().map(Encoder.THRIFT::encode).collect(toList()), callback);
+    sender.sendSpans(spans.stream().map(encoder::encode).collect(toList()), callback);
     callback.await();
   }
 }
diff --git a/sender-sqs/src/main/java/zipkin/reporter/sqs/SQSSender.java b/sender-sqs/src/main/java/zipkin/reporter/sqs/SQSSender.java
@@ -57,6 +57,7 @@ public static SQSSender create(String url) {
   public static Builder builder() {
     return new AutoValue_SQSSender.Builder()
         .credentialsProvider(new DefaultAWSCredentialsProviderChain())
+        .encoding(Encoding.THRIFT)
         .messageMaxBytes(256 * 1024); // 256KB SQS limit.
   }
 
@@ -75,6 +76,9 @@ public interface Builder {
     /** Maximum size of a message. SQS max message size is 256KB including attributes. */
     Builder messageMaxBytes(int messageMaxBytes);
 
+    /** Allows you to change to json format. Default is thrift */
+    Builder encoding(Encoding encoding);
+
     SQSSender build();
   }
 
@@ -105,10 +109,6 @@ public interface Builder {
     return (listSize + 2) * 4 / 3; // account for base64 encoding
   }
 
-  @Override public Encoding encoding() {
-    return Encoding.THRIFT;
-  }
-
   @Override public void sendSpans(List<byte[]> list, Callback callback) {
     if (closeCalled.get()) throw new IllegalStateException("closed");
 

diff --git a/sender-sqs/src/test/java/zipkin/reporter/sqs/SQSSenderTest.java b/sender-sqs/src/test/java/zipkin/reporter/sqs/SQSSenderTest.java
@@ -25,6 +25,7 @@
 import zipkin.TestObjects;
 import zipkin.junit.aws.AmazonSQSRule;
 import zipkin.reporter.Encoder;
+import zipkin.reporter.Encoding;
 import zipkin.reporter.internal.AwaitableCallback;
 
 import static java.util.stream.Collectors.toList;
@@ -44,8 +45,19 @@ public class SQSSenderTest {
       .build();
 
   @Test
-  public void sendsSpans() throws Exception {
-    send(TestObjects.TRACE);
+  public void sendsSpans_thrift() throws Exception {
+    sendSpans(Encoder.THRIFT);
+  }
+
+  @Test
+  public void sendsSpans_json() throws Exception {
+    sender.close();
+    sender = sender.toBuilder().encoding(Encoding.JSON).build();
+    sendSpans(Encoder.JSON);
+  }
+
+  void sendSpans(Encoder<Span> encoder) {
+    send(encoder, TestObjects.TRACE);
 
     assertThat(sqsRule.queueCount()).isEqualTo(1);
 
@@ -61,9 +73,9 @@ public void checkOk() throws Exception {
     assertThat(sender.check()).isEqualTo(Component.CheckResult.OK);
   }
 
-  private void send(List<Span> spans) {
+  private void send(Encoder<Span> encoder, List<Span> spans) {
     AwaitableCallback callback = new AwaitableCallback();
-    sender.sendSpans(spans.stream().map(Encoder.THRIFT::encode).collect(toList()), callback);
+    sender.sendSpans(spans.stream().map(encoder::encode).collect(toList()), callback);
     callback.await();
   }
 }