Adds Scribe transport (#19)

This adds a Scribe transport, based on scrooge and zipkin-sender-libthrift.

This variant allocates exact-size arrays for thrift messages, so avoids some
of the complexity needed to deal with pooled buffers.

README.md

@@ -1,7 +1,7 @@
 [![Gitter chat](http://img.shields.io/badge/gitter-join%20chat%20%E2%86%92-brightgreen.svg)](https://gitter.im/openzipkin/zipkin) [![Build Status](https://travis-ci.org/openzipkin/zipkin-finagle.svg?branch=master)](https://travis-ci.org/openzipkin/zipkin-finagle) [![Download](https://api.bintray.com/packages/openzipkin/maven/zipkin-finagle/images/download.svg) ](https://bintray.com/openzipkin/maven/zipkin-finagle/_latestVersion)
 
 # zipkin-finagle
-Integration between Finagle tracing to Zipkin transports including http and kafka.
+Integration between Finagle tracing to Zipkin transports including http, kafka and scribe.
 
 ## Quick start
 Finagle will use a tracer that it detects in the classpath. For example, depending on `io.zipkin.finagle:zipkin-finagle-http_2.11` will send to Zipkin over Http.
@@ -78,6 +78,21 @@ Ex. Here's how to configure the Kafka server with a system property:
 $ java -Dzipkin.kafka.bootstrapServers=192.168.99.100 ...
 ```
 
+### Scribe Configuration
+Adding `io.zipkin.finagle:zipkin-finagle-scribe_2.11` to your classpath will configure Finagle
+to report trace data to the zipkin category of Scribe.
+
+Here are the flags that apply to Scribe:
+
+Flag | Default | Description
+--- | --- | ---
+zipkin.scribe.host | localhost:1463 | The network location of the Scribe service. See http://twitter.github.io/finagle/guide/Names.html
+
+Ex. Here's how to configure the Scribe host with a system property. In this case pointing directly to a zipkin server:
+```bash
+$ java -Dzipkin.scribe.host=zipkinhost:9410 ...
+```
+
 ### Programmatic
 
 You can also configure zipkin tracers programmatically. Here's an example:

pom.xml

@@ -26,6 +26,7 @@
     <module>core</module>
     <module>kafka</module>
     <module>http</module>
+    <module>scribe</module>
   </modules>
 
   <properties>
@@ -45,7 +46,7 @@
     <!-- version numbers that are used more than once -->
     <scala.binary.version>2.11</scala.binary.version>
     <finagle.version>6.38.0</finagle.version>
-    <zipkin-reporter.version>0.5.0</zipkin-reporter.version>
+    <zipkin-reporter.version>0.5.1</zipkin-reporter.version>
     <license-maven-plugin.version>2.11</license-maven-plugin.version>
   </properties>
 

scribe/pom.xml

@@ -0,0 +1,71 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright 2016 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
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software distributed under the License
+    is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+    or implied. See the License for the specific language governing permissions and limitations under
+    the License.
+
+-->
+<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>
+    <groupId>io.zipkin.finagle</groupId>
+    <artifactId>zipkin-finagle-parent_2.11</artifactId>
+    <version>0.2.2-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>zipkin-finagle-scribe_2.11</artifactId>
+  <name>Zipkin Finagle: Scribe</name>
+
+  <properties>
+    <main.basedir>${project.basedir}/..</main.basedir>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>zipkin-finagle_${scala.binary.version}</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.thrift</groupId>
+      <artifactId>libthrift</artifactId>
+      <!-- avoids transitive dep on org.apache.thrift:libthrift:jar:0.5.0-1 -->
+      <version>0.9.3</version>
+    </dependency>
+
+    <dependency>
+      <groupId>com.twitter</groupId>
+      <artifactId>finagle-thrift_${scala.binary.version}</artifactId>
+      <version>${finagle.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>io.zipkin.reporter</groupId>
+      <artifactId>zipkin-sender-libthrift</artifactId>
+      <version>${zipkin-reporter.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>zipkin-finagle_${scala.binary.version}</artifactId>
+      <type>test-jar</type>
+    </dependency>
+
+    <dependency>
+      <groupId>io.zipkin.java</groupId>
+      <artifactId>zipkin-collector-scribe</artifactId>
+      <version>1.11.1</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+</project>

scribe/src/main/java/zipkin/finagle/scribe/ScribeClient.java

@@ -0,0 +1,102 @@
+/**
+ * Copyright 2016 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package zipkin.finagle.scribe;
+
+import com.twitter.finagle.Service;
+import com.twitter.finagle.Thrift;
+import com.twitter.finagle.context.Contexts;
+import com.twitter.finagle.thrift.DeserializeCtx;
+import com.twitter.finagle.thrift.ThriftClientRequest;
+import com.twitter.util.AbstractClosable;
+import com.twitter.util.Function;
+import com.twitter.util.Function0;
+import com.twitter.util.Future;
+import com.twitter.util.Return;
+import com.twitter.util.Throw;
+import com.twitter.util.Time;
+import com.twitter.util.Try;
+import java.util.List;
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.transport.TMemoryBuffer;
+import org.apache.thrift.transport.TMemoryInputTransport;
+import scala.runtime.BoxedUnit;
+import zipkin.reporter.libthrift.InternalScribeCodec;
+
+/**
+ * This class's functional composition was derived from how scrooge does it. For example, scrooge
+ * doesn't use thrift sequence ids.
+ */
+final class ScribeClient extends AbstractClosable {
+  static final byte[] category = "zipkin".getBytes();
+
+  private final Service<ThriftClientRequest, byte[]> service;
+
+  ScribeClient(ScribeZipkinTracer.Config config) {
+    service = Thrift.client().newService(config.host(), "zipkin-scribe");
+  }
+
+  static int messageSizeInBytes(List<byte[]> encodedSpans) {
+    return InternalScribeCodec.messageSizeInBytes(category, encodedSpans);
+  }
+
+  Future<Boolean> sendMessage(List<byte[]> encodedSpans) throws TException {
+    final ThriftClientRequest request = parseRequest(encodedSpans);
+    DeserializeCtx ctx = new DeserializeCtx<>(request, READ_TRY);
+    return Contexts.local().let(DeserializeCtx.Key(), ctx, new Function0<Future<Boolean>>() {
+      @Override
+      public Future<Boolean> apply() {
+        Future<byte[]> done = service.apply(request);
+        return done.flatMap(READ_FUTURE);
+      }
+    });
+  }
+
+  @Override public Future<BoxedUnit> close(Time time) {
+    return service.close();
+  }
+
+  // Avoids needing to keep track of pooled buffers by allocating an exact size array.
+  ThriftClientRequest parseRequest(List<byte[]> encodedSpans) throws TException {
+    int encodedSize = InternalScribeCodec.messageSizeInBytes(category, encodedSpans);
+    TMemoryBuffer mem = new TMemoryBuffer(encodedSize);
+    TBinaryProtocol prot = new TBinaryProtocol(mem);
+    InternalScribeCodec.writeLogRequest(category, encodedSpans, 0, prot);
+    return new ThriftClientRequest(mem.getArray(), false);
+  }
+
+  static final Function<byte[], Try<Boolean>> READ_TRY =
+      new Function<byte[], Try<Boolean>>() {
+        public Try<Boolean> apply(byte[] responseBytes) {
+          TBinaryProtocol iprot = new TBinaryProtocol(new TMemoryInputTransport(responseBytes));
+          try {
+            return new Return(InternalScribeCodec.readLogResponse(0, iprot));
+          } catch (Exception e) {
+            return new Throw(e);
+          }
+        }
+      };
+
+  static final Function<byte[], Future<Boolean>> READ_FUTURE =
+      new Function<byte[], Future<Boolean>>() {
+        public Future<Boolean> apply(byte[] responseBytes) {
+          TBinaryProtocol iprot = new TBinaryProtocol(new TMemoryInputTransport(responseBytes));
+          try {
+            return Future.value(InternalScribeCodec.readLogResponse(0, iprot));
+          } catch (Exception e) {
+            return Future.exception(e);
+          }
+        }
+      };
+}

scribe/src/main/java/zipkin/finagle/scribe/ScribeSender.java

@@ -0,0 +1,88 @@
+/**
+ * Copyright 2016 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package zipkin.finagle.scribe;
+
+import com.twitter.finagle.tracing.Trace;
+import com.twitter.util.Try;
+import java.util.List;
+import org.apache.thrift.TException;
+import scala.runtime.AbstractFunction0;
+import scala.runtime.AbstractFunction1;
+import scala.runtime.BoxedUnit;
+import zipkin.reporter.Callback;
+import zipkin.reporter.Encoding;
+import zipkin.reporter.Sender;
+
+/**
+ * Receives the Finagle generated traces and sends them off to Zipkin via Scribe.
+ */
+final class ScribeSender implements Sender {
+  final ScribeClient client;
+  final ScribeZipkinTracer.Config config;
+
+  ScribeSender(ScribeZipkinTracer.Config config) {
+    this.client = new ScribeClient(config);
+    this.config = config;
+  }
+
+  @Override public int messageMaxBytes() {
+    return 5 * 1024 * 1024;
+  }
+
+  @Override public int messageSizeInBytes(List<byte[]> encodedSpans) {
+    return ScribeClient.messageSizeInBytes(encodedSpans);
+  }
+
+  @Override public Encoding encoding() {
+    return Encoding.THRIFT;
+  }
+
+  @Override public void sendSpans(final List<byte[]> spans, final Callback callback) {
+    Trace.letClear(new AbstractFunction0<Void>() {
+      @Override public Void apply() {
+        try {
+          doSendSpans(spans, callback);
+        } catch (TException | RuntimeException e) {
+          callback.onError(e);
+        }
+        return null;
+      }
+    });
+  }
+
+  void doSendSpans(List<byte[]> spans, final Callback callback) throws TException {
+    client.sendMessage(spans).respond(new AbstractFunction1<Try<Boolean>, BoxedUnit>() {
+      @Override public BoxedUnit apply(Try<Boolean> result) {
+        if (result.isReturn()) {
+          if (result.get()) {
+            callback.onComplete();
+          } else {
+            callback.onError(new IllegalStateException("try later"));
+          }
+        } else {
+          callback.onError(result.throwable());
+        }
+        return BoxedUnit.UNIT;
+      }
+    });
+  }
+
+  @Override public CheckResult check() {
+    return CheckResult.OK; // TODO
+  }
+
+  @Override public void close() {
+    client.close();
+  }
+}