Merge pull request #142 from openzipkin/compression

Adds HttpSpanCollector.Config.compressionEnabled
openzipkin · Mar 3, 2016 · 003cd69 · 003cd69
2 parents d20b703 + 834ee0f
commit 003cd69
Show file tree

Hide file tree

Showing 3 changed files with 56 additions and 4 deletions.
diff --git a/brave-spancollector-http/src/main/java/com/github/kristofa/brave/http/HttpSpanCollector.java b/brave-spancollector-http/src/main/java/com/github/kristofa/brave/http/HttpSpanCollector.java
@@ -1,10 +1,13 @@
 package com.github.kristofa.brave.http;
 
-import com.github.kristofa.brave.*;
+import com.github.kristofa.brave.EmptySpanCollectorMetricsHandler;
+import com.github.kristofa.brave.SpanCollector;
+import com.github.kristofa.brave.SpanCollectorMetricsHandler;
 import com.github.kristofa.brave.internal.Nullable;
 import com.google.auto.value.AutoValue;
-import com.twitter.zipkin.gen.SpanCodec;
 import com.twitter.zipkin.gen.Span;
+import com.twitter.zipkin.gen.SpanCodec;
+import java.io.ByteArrayOutputStream;
 import java.io.Closeable;
 import java.io.Flushable;
 import java.io.IOException;
@@ -18,6 +21,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.zip.GZIPOutputStream;
 
 import static java.util.concurrent.TimeUnit.SECONDS;
 
@@ -32,6 +36,7 @@ public static Builder builder() {
       return new AutoValue_HttpSpanCollector_Config.Builder()
           .connectTimeout(10 * 1000)
           .readTimeout(60 * 1000)
+          .compressionEnabled(false)
           .flushInterval(1);
     }
 
@@ -41,6 +46,8 @@ public static Builder builder() {
 
     abstract int flushInterval();
 
+    abstract boolean compressionEnabled();
+
     @AutoValue.Builder
     public interface Builder {
       /** Default 10 * 1000 milliseconds. 0 implies no timeout. */
@@ -52,6 +59,13 @@ public interface Builder {
       /** Default 1 second. 0 implies spans are {@link #flush() flushed} externally. */
       Builder flushInterval(int flushInterval);
 
+      /**
+       * Default false. true implies that spans will be gzipped before transport.
+       *
+       * <p>Note: This feature requires zipkin-scala 1.34+ or zipkin-java 0.6+
+       */
+      Builder compressionEnabled(boolean compressSpans);
+
       Config build();
     }
   }
@@ -162,6 +176,14 @@ void postSpans(byte[] json) throws IOException {
     connection.setReadTimeout(config.readTimeout());
     connection.setRequestMethod("POST");
     connection.addRequestProperty("Content-Type", "application/json");
+    if (config.compressionEnabled()) {
+      connection.addRequestProperty("Content-Encoding", "gzip");
+      ByteArrayOutputStream gzipped = new ByteArrayOutputStream();
+      try (GZIPOutputStream compressor = new GZIPOutputStream(gzipped)) {
+        compressor.write(json);
+      }
+      json = gzipped.toByteArray();
+    }
     connection.setDoOutput(true);
     connection.setFixedLengthStreamingMode(json.length);
     connection.getOutputStream().write(json);

diff --git a/...pancollector-http/src/test/java/com/github/kristofa/brave/http/HttpSpanCollectorTest.java b/...pancollector-http/src/test/java/com/github/kristofa/brave/http/HttpSpanCollectorTest.java
@@ -1,7 +1,9 @@
 package com.github.kristofa.brave.http;
 
 import com.github.kristofa.brave.SpanCollectorMetricsHandler;
+import com.twitter.zipkin.gen.Annotation;
 import com.twitter.zipkin.gen.Span;
+import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Rule;
 import org.junit.Test;
@@ -41,7 +43,9 @@ public void dropsWhenQueueIsFull() throws Exception {
     for (int i = 0; i < 1001; i++)
       collector.collect(span(1L, "foo"));
 
-    assertThat(metrics.acceptedSpans.get()).isEqualTo(1001);
+    collector.flush(); // manually flush the spans
+
+    assertThat(zipkin.receivedSpanCount()).isEqualTo(1000);
     assertThat(metrics.droppedSpans.get()).isEqualTo(1);
   }
 
@@ -62,6 +66,32 @@ public void postsSpans() throws Exception {
     );
   }
 
+  @Test
+  public void postsCompressedSpans() throws Exception {
+    char[] annotation2K = new char[2048];
+    Arrays.fill(annotation2K, 'a');
+
+    ZipkinRule zipkin = new ZipkinRule();
+    try {
+      zipkin.start(0);
+
+      HttpSpanCollector.Config config = HttpSpanCollector.Config.builder()
+          .flushInterval(0).compressionEnabled(true).build();
+
+      HttpSpanCollector collector = new HttpSpanCollector(zipkin.httpUrl(), config, metrics);
+
+      collector.collect(span(1L, "foo")
+          .addToAnnotations(Annotation.create(1111L, new String(annotation2K), null)));
+
+      collector.flush(); // manually flush the span
+
+      // Ensure the span was compressed
+      assertThat(zipkin.receivedSpanBytes()).isLessThan(annotation2K.length);
+    } finally {
+      zipkin.shutdown();
+    }
+  }
+
   @Test
   public void incrementsDroppedSpansWhenServerErrors() throws Exception {
     zipkin.enqueueFailure(HttpFailure.sendErrorResponse(500, "Server Error!"));

diff --git a/pom.xml b/pom.xml
@@ -37,7 +37,7 @@
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <spring.version>4.1.6.RELEASE</spring.version>
-    <zipkin.version>0.6.0</zipkin.version>
+    <zipkin.version>0.6.1</zipkin.version>
     <log4j.version>2.3</log4j.version>
     <httpcomponents.version>4.4.1</httpcomponents.version>
     <maven-release-plugin.version>2.5.2</maven-release-plugin.version>