Add Zipkin v2 API reporter

This reporter wraps an instance of a Zipkin reporter, which closely
matches the interface for Jaeger reporters, so the adaptation is very
simple.

Most of the complexity lies in converting from a Jaeger span to a Zipkin
v2 span.

The existing Zipkin v1 Thrift code is left untouched.

Signed-off-by: Ben Keith <[email protected]>

Gopkg.toml

@@ -25,3 +25,7 @@
 [[constraint]]
   name = "go.uber.org/zap"
   version = "^1"
+
+[[constraint]]
+  name = "github.com/openzipkin/zipkin-go"
+  version = "0.1.0"

README.md

@@ -150,6 +150,8 @@ The remote reporter uses "transports" to actually send the spans out
 of process. Currently the supported transports include:
   * [Jaeger Thrift](https://github.com/jaegertracing/jaeger-idl/blob/master/thrift/agent.thrift) over UDP or HTTP,
   * [Zipkin Thrift](https://github.com/jaegertracing/jaeger-idl/blob/master/thrift/zipkincore.thrift) over HTTP.
+  * [Zipkin v2 Reporter](./zipkin/reporter.go) that can send spans using any
+	  Zipkin Reporter instance, which support a variety of transports
 
 ### Sampling
 

glide.yaml

@@ -12,6 +12,11 @@ import:
   - metrics
 - package: github.com/pkg/errors
   version: ~0.8.0
+- package: github.com/openzipkin/zipkin-go
+  version: ~0.1.0
+  subpackages:
+  - reporter
+  - model
 testImport:
 - package: github.com/stretchr/testify
   subpackages:
@@ -20,3 +25,7 @@ testImport:
   - suite
 - package: github.com/prometheus/client_golang
   version: v0.8.0
+- package: github.com/openzipkin/zipkin-go
+  version: ~0.1.0
+  subpackages:
+  - reporter/http

reporter.go

@@ -245,10 +245,10 @@ func (r *remoteReporter) sendCloseEvent() {
 	wg.Wait()
 }
 
-// processQueue reads spans from the queue, converts them to Thrift, and stores them in an internal buffer.
-// When the buffer length reaches batchSize, it is flushed by submitting the accumulated spans to Jaeger.
-// Buffer also gets flushed automatically every batchFlushInterval seconds, just in case the tracer stopped
-// reporting new spans.
+// processQueue reads spans from the queue, converts them, and stores them in an internal buffer.
+// When the buffer length reaches batchSize, it is flushed by submitting the accumulated spans to Jaeger
+// or Zipkin. Buffer also gets flushed automatically every batchFlushInterval seconds, just in case the
+// tracer stopped reporting new spans.
 func (r *remoteReporter) processQueue() {
 	// flush causes the Sender to flush its accumulated spans and clear the buffer
 	flush := func() {

utils/utils.go

@@ -75,6 +75,13 @@ func PackIPAsUint32(ip net.IP) uint32 {
 	return 0
 }
 
+// UnpackUint32AsIP does the reverse of PackIPAsUint32
+func UnpackUint32AsIP(ip uint32) net.IP {
+	localIP := make(net.IP, 4)
+	binary.BigEndian.PutUint32(localIP, ip)
+	return localIP
+}
+
 // TimeToMicrosecondsSinceEpochInt64 converts Go time.Time to a long
 // representing time since epoch in microseconds, which is used expected
 // in the Jaeger spans encoded as Thrift.

utils/utils_test.go

@@ -89,3 +89,18 @@ func TestPackIPAsUint32(t *testing.T) {
 		assert.Equal(t, test.out, ip)
 	}
 }
+
+func TestUnpackUint32AsIP(t *testing.T) {
+	tests := []struct {
+		expected net.IP
+		in       uint32
+	}{
+		{net.IPv4(1, 2, 3, 4), 1<<24 | 2<<16 | 3<<8 | 4},
+		{net.IPv4(127, 0, 0, 1), 127<<24 | 1},
+	}
+
+	for _, test := range tests {
+		ip := UnpackUint32AsIP(test.in)
+		assert.Equal(t, test.expected, ip.To16())
+	}
+}

zipkin/reporter.go

@@ -0,0 +1,47 @@
+// Copyright (c) 2018 The Jaeger 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
+
+import (
+	zipkinReporter "github.com/openzipkin/zipkin-go/reporter"
+	jaeger "github.com/uber/jaeger-client-go"
+)
+
+// Reporter adapts a Zipkin Reporter to a Jaeger reporter
+//
+// Example:
+//
+//     import zipkinHttp "github.com/openzipkin/zipkin-go/reporter/http"
+//     ...
+//
+//     zipkinReporter := Reporter{
+//         zipkinHttp.NewReporter(zipkinUrl)
+//     }
+//
+// Will give you a Jaeger reporter that can be used with the tracer.
+type Reporter struct {
+	zipkinReporter.Reporter
+}
+
+// Report converts the Jaeger span to a Zipkin Span and reports it to the
+// Zipkin reporter.
+func (r *Reporter) Report(span *jaeger.Span) {
+	r.Reporter.Send(*jaeger.BuildZipkinV2Span(span))
+}
+
+// Close proxies to the Zipkin reporter's close method
+func (r *Reporter) Close() {
+	r.Reporter.Close()
+}

zipkin/reporter_test.go

@@ -0,0 +1,72 @@
+// Copyright (c) 2018 The Jaeger 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
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/crossdock/crossdock-go/assert"
+	zipkinModel "github.com/openzipkin/zipkin-go/model"
+	zipkinHttp "github.com/openzipkin/zipkin-go/reporter/http"
+	jaeger "github.com/uber/jaeger-client-go"
+)
+
+func TestZipkinReporter(t *testing.T) {
+	var spansReceived []zipkinModel.SpanModel
+
+	// Inspired by
+	// https://github.com/openzipkin/zipkin-go/blob/master/reporter/http/http_test.go
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != "POST" {
+			t.Errorf("expected 'POST' request, got '%s'", r.Method)
+		}
+
+		if spansReceived != nil {
+			t.Fatalf("received more than one set of spans")
+		}
+
+		err := json.NewDecoder(r.Body).Decode(&spansReceived)
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err.Error())
+		}
+	}))
+	defer ts.Close()
+
+	rep := zipkinHttp.NewReporter(ts.URL)
+	defer rep.Close()
+	zipkinReporter := &Reporter{
+		zipkinHttp.NewReporter(ts.URL),
+	}
+
+	tracer, closer := jaeger.NewTracer("DOOP",
+		jaeger.NewConstSampler(true),
+		zipkinReporter)
+
+	sp := tracer.StartSpan("s1").(*jaeger.Span)
+	sp.SetTag("test", "abc")
+	sp.Finish()
+
+	sp2 := tracer.StartSpan("s2").(*jaeger.Span)
+	sp2.Finish()
+
+	closer.Close()
+
+	assert.Len(t, spansReceived, 2)
+	assert.Equal(t, "abc", spansReceived[0].Tags["test"])
+	assert.Equal(t, "s2", spansReceived[1].Name)
+}