Zipkin tracing support (Native v2)

examples/addsvc/cmd/addcli/addcli.go

@@ -14,7 +14,9 @@ import (
 	"github.com/apache/thrift/lib/go/thrift"
 	lightstep "github.com/lightstep/lightstep-tracer-go"
 	stdopentracing "github.com/opentracing/opentracing-go"
-	zipkin "github.com/openzipkin/zipkin-go-opentracing"
+	zipkin "github.com/openzipkin/zipkin-go"
+	zipkinot "github.com/openzipkin/zipkin-go-opentracing"
+	zipkinhttp "github.com/openzipkin/zipkin-go/reporter/http"
 	"sourcegraph.com/sourcegraph/appdash"
 	appdashot "sourcegraph.com/sourcegraph/appdash/opentracing"
 
@@ -41,9 +43,10 @@ func main() {
 		thriftProtocol = fs.String("thrift-protocol", "binary", "binary, compact, json, simplejson")
 		thriftBuffer   = fs.Int("thrift-buffer", 0, "0 for unbuffered")
 		thriftFramed   = fs.Bool("thrift-framed", false, "true to enable framing")
-		zipkinURL      = fs.String("zipkin-url", "", "Enable Zipkin tracing via a collector URL e.g. http://localhost:9411/api/v1/spans")
-		lightstepToken = flag.String("lightstep-token", "", "Enable LightStep tracing via a LightStep access token")
-		appdashAddr    = flag.String("appdash-addr", "", "Enable Appdash tracing via an Appdash server host:port")
+		zipkinV2URL    = fs.String("zipkin-url", "", "Enable Zipkin v2 tracing (zipkin-go) via HTTP Reporter URL e.g. http://localhost:94111/api/v2/spans")
+		zipkinV1URL    = fs.String("zipkin-v1-url", "", "Enable Zipkin v1 tracing (zipkin-go-opentracing) via a collector URL e.g. http://localhost:9411/api/v1/spans")
+		lightstepToken = fs.String("lightstep-token", "", "Enable LightStep tracing via a LightStep access token")
+		appdashAddr    = fs.String("appdash-addr", "", "Enable Appdash tracing via an Appdash server host:port")
 		method         = fs.String("method", "sum", "sum, concat")
 	)
 	fs.Usage = usageFor(fs, os.Args[0]+" [flags] <a> <b>")
@@ -55,35 +58,57 @@ func main() {
 
 	// This is a demonstration client, which supports multiple tracers.
 	// Your clients will probably just use one tracer.
-	var tracer stdopentracing.Tracer
+	var otTracer stdopentracing.Tracer
 	{
-		if *zipkinURL != "" {
-			collector, err := zipkin.NewHTTPCollector(*zipkinURL)
+		if *zipkinV1URL != "" && *zipkinV2URL == "" {
+			collector, err := zipkinot.NewHTTPCollector(*zipkinV1URL)
 			if err != nil {
 				fmt.Fprintln(os.Stderr, err.Error())
 				os.Exit(1)
 			}
 			defer collector.Close()
 			var (
 				debug       = false
-				hostPort    = "localhost:80"
-				serviceName = "addsvc"
+				hostPort    = "localhost:0"
+				serviceName = "addsvc-cli"
 			)
-			recorder := zipkin.NewRecorder(collector, debug, hostPort, serviceName)
-			tracer, err = zipkin.NewTracer(recorder)
+			recorder := zipkinot.NewRecorder(collector, debug, hostPort, serviceName)
+			otTracer, err = zipkinot.NewTracer(recorder)
 			if err != nil {
 				fmt.Fprintln(os.Stderr, err.Error())
 				os.Exit(1)
 			}
 		} else if *lightstepToken != "" {
-			tracer = lightstep.NewTracer(lightstep.Options{
+			otTracer = lightstep.NewTracer(lightstep.Options{
 				AccessToken: *lightstepToken,
 			})
-			defer lightstep.FlushLightStepTracer(tracer)
+			defer lightstep.FlushLightStepTracer(otTracer)
 		} else if *appdashAddr != "" {
-			tracer = appdashot.NewTracer(appdash.NewRemoteCollector(*appdashAddr))
+			otTracer = appdashot.NewTracer(appdash.NewRemoteCollector(*appdashAddr))
 		} else {
-			tracer = stdopentracing.GlobalTracer() // no-op
+			otTracer = stdopentracing.GlobalTracer() // no-op
+		}
+	}
+
+	// This is a demonstration of the native Zipkin tracing client. If using
+	// Zipkin this is the more idiomatic client over OpenTracing.
+	var zipkinTracer *zipkin.Tracer
+	{
+		var (
+			err           error
+			hostPort      = "" // if host:port is unknown we can keep this empty
+			serviceName   = "addsvc-cli"
+			useNoopTracer = (*zipkinV2URL == "")
+			reporter      = zipkinhttp.NewReporter(*zipkinV2URL)
+		)
+		defer reporter.Close()
+		zEP, _ := zipkin.NewEndpoint(serviceName, hostPort)
+		zipkinTracer, err = zipkin.NewTracer(
+			reporter, zipkin.WithLocalEndpoint(zEP), zipkin.WithNoopTracer(useNoopTracer),
+		)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "unable to create zipkin tracer: %s\n", err.Error())
+			os.Exit(1)
 		}
 	}
 
@@ -94,17 +119,17 @@ func main() {
 		err error
 	)
 	if *httpAddr != "" {
-		svc, err = addtransport.NewHTTPClient(*httpAddr, tracer, log.NewNopLogger())
+		svc, err = addtransport.NewHTTPClient(*httpAddr, otTracer, zipkinTracer, log.NewNopLogger())
 	} else if *grpcAddr != "" {
 		conn, err := grpc.Dial(*grpcAddr, grpc.WithInsecure(), grpc.WithTimeout(time.Second))
 		if err != nil {
 			fmt.Fprintf(os.Stderr, "error: %v", err)
 			os.Exit(1)
 		}
 		defer conn.Close()
-		svc = addtransport.NewGRPCClient(conn, tracer, log.NewNopLogger())
+		svc = addtransport.NewGRPCClient(conn, otTracer, zipkinTracer, log.NewNopLogger())
 	} else if *jsonRPCAddr != "" {
-		svc, err = addtransport.NewJSONRPCClient(*jsonRPCAddr, tracer, log.NewNopLogger())
+		svc, err = addtransport.NewJSONRPCClient(*jsonRPCAddr, otTracer, log.NewNopLogger())
 	} else if *thriftAddr != "" {
 		// It's necessary to do all of this construction in the func main,
 		// because (among other reasons) we need to control the lifecycle of the

examples/addsvc/cmd/addsvc/addsvc.go

@@ -14,7 +14,9 @@ import (
 	lightstep "github.com/lightstep/lightstep-tracer-go"
 	"github.com/oklog/oklog/pkg/group"
 	stdopentracing "github.com/opentracing/opentracing-go"
-	zipkin "github.com/openzipkin/zipkin-go-opentracing"
+	zipkin "github.com/openzipkin/zipkin-go"
+	zipkinot "github.com/openzipkin/zipkin-go-opentracing"
+	zipkinhttp "github.com/openzipkin/zipkin-go/reporter/http"
 	stdprometheus "github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"google.golang.org/grpc"
@@ -24,6 +26,7 @@ import (
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/metrics"
 	"github.com/go-kit/kit/metrics/prometheus"
+	kitgrpc "github.com/go-kit/kit/transport/grpc"
 
 	addpb "github.com/go-kit/kit/examples/addsvc/pb"
 	"github.com/go-kit/kit/examples/addsvc/pkg/addendpoint"
@@ -46,9 +49,10 @@ func main() {
 		thriftProtocol = fs.String("thrift-protocol", "binary", "binary, compact, json, simplejson")
 		thriftBuffer   = fs.Int("thrift-buffer", 0, "0 for unbuffered")
 		thriftFramed   = fs.Bool("thrift-framed", false, "true to enable framing")
-		zipkinURL      = fs.String("zipkin-url", "", "Enable Zipkin tracing via a collector URL e.g. http://localhost:9411/api/v1/spans")
-		lightstepToken = flag.String("lightstep-token", "", "Enable LightStep tracing via a LightStep access token")
-		appdashAddr    = flag.String("appdash-addr", "", "Enable Appdash tracing via an Appdash server host:port")
+		zipkinV2URL    = fs.String("zipkin-url", "", "Enable Zipkin v2 tracing (zipkin-go) using a Reporter URL e.g. http://localhost:9411/api/v2/spans")
+		zipkinV1URL    = fs.String("zipkin-v1-url", "", "Enable Zipkin v1 tracing (zipkin-go-opentracing) using a collector URL e.g. http://localhost:9411/api/v1/spans")
+		lightstepToken = fs.String("lightstep-token", "", "Enable LightStep tracing via a LightStep access token")
+		appdashAddr    = fs.String("appdash-addr", "", "Enable Appdash tracing via an Appdash server host:port")
 	)
 	fs.Usage = usageFor(fs, os.Args[0]+" [flags]")
 	fs.Parse(os.Args[1:])
@@ -61,13 +65,13 @@ func main() {
 		logger = log.With(logger, "caller", log.DefaultCaller)
 	}
 
-	// Determine which tracer to use. We'll pass the tracer to all the
+	// Determine which OpenTracing tracer to use. We'll pass the tracer to all the
 	// components that use it, as a dependency.
 	var tracer stdopentracing.Tracer
 	{
-		if *zipkinURL != "" {
-			logger.Log("tracer", "Zipkin", "URL", *zipkinURL)
-			collector, err := zipkin.NewHTTPCollector(*zipkinURL)
+		if *zipkinV1URL != "" && *zipkinV2URL == "" {
+			logger.Log("tracer", "Zipkin", "type", "OpenTracing", "URL", *zipkinV1URL)
+			collector, err := zipkinot.NewHTTPCollector(*zipkinV1URL)
 			if err != nil {
 				logger.Log("err", err)
 				os.Exit(1)
@@ -78,8 +82,8 @@ func main() {
 				hostPort    = "localhost:80"
 				serviceName = "addsvc"
 			)
-			recorder := zipkin.NewRecorder(collector, debug, hostPort, serviceName)
-			tracer, err = zipkin.NewTracer(recorder)
+			recorder := zipkinot.NewRecorder(collector, debug, hostPort, serviceName)
+			tracer, err = zipkinot.NewTracer(recorder)
 			if err != nil {
 				logger.Log("err", err)
 				os.Exit(1)
@@ -94,11 +98,33 @@ func main() {
 			logger.Log("tracer", "Appdash", "addr", *appdashAddr)
 			tracer = appdashot.NewTracer(appdash.NewRemoteCollector(*appdashAddr))
 		} else {
-			logger.Log("tracer", "none")
 			tracer = stdopentracing.GlobalTracer() // no-op
 		}
 	}
 
+	var zipkinTracer *zipkin.Tracer
+	{
+		var (
+			err           error
+			hostPort      = "localhost:80"
+			serviceName   = "addsvc"
+			useNoopTracer = (*zipkinV2URL == "")
+			reporter      = zipkinhttp.NewReporter(*zipkinV2URL)
+		)
+		defer reporter.Close()
+		zEP, _ := zipkin.NewEndpoint(serviceName, hostPort)
+		zipkinTracer, err = zipkin.NewTracer(
+			reporter, zipkin.WithLocalEndpoint(zEP), zipkin.WithNoopTracer(useNoopTracer),
+		)
+		if err != nil {
+			logger.Log("err", err)
+			os.Exit(1)
+		}
+		if !useNoopTracer {
+			logger.Log("tracer", "Zipkin", "type", "Native", "URL", *zipkinV2URL)
+		}
+	}
+
 	// Create the (sparse) metrics we'll use in the service. They, too, are
 	// dependencies that we pass to components that use them.
 	var ints, chars metrics.Counter
@@ -137,9 +163,9 @@ func main() {
 	// them to ports or anything yet; we'll do that next.
 	var (
 		service        = addservice.New(logger, ints, chars)
-		endpoints      = addendpoint.New(service, logger, duration, tracer)
-		httpHandler    = addtransport.NewHTTPHandler(endpoints, tracer, logger)
-		grpcServer     = addtransport.NewGRPCServer(endpoints, tracer, logger)
+		endpoints      = addendpoint.New(service, logger, duration, tracer, zipkinTracer)
+		httpHandler    = addtransport.NewHTTPHandler(endpoints, tracer, zipkinTracer, logger)
+		grpcServer     = addtransport.NewGRPCServer(endpoints, tracer, zipkinTracer, logger)
 		thriftServer   = addtransport.NewThriftServer(endpoints)
 		jsonrpcHandler = addtransport.NewJSONRPCHandler(endpoints, logger)
 	)
@@ -196,7 +222,9 @@ func main() {
 		}
 		g.Add(func() error {
 			logger.Log("transport", "gRPC", "addr", *grpcAddr)
-			baseServer := grpc.NewServer()
+			// we add the Go Kit gRPC Interceptor to our gRPC service as it is used by
+			// the here demonstrated zipkin tracing middleware.
+			baseServer := grpc.NewServer(grpc.UnaryInterceptor(kitgrpc.Interceptor))
 			addpb.RegisterAddServer(baseServer, grpcServer)
 			return baseServer.Serve(grpcListener)
 		}, func(error) {

examples/addsvc/cmd/addsvc/wiring_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 
 	"github.com/opentracing/opentracing-go"
+	zipkin "github.com/openzipkin/zipkin-go"
 
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/metrics/discard"
@@ -18,9 +19,10 @@ import (
 )
 
 func TestHTTP(t *testing.T) {
+	zkt, _ := zipkin.NewTracer(nil, zipkin.WithNoopTracer(true))
 	svc := addservice.New(log.NewNopLogger(), discard.NewCounter(), discard.NewCounter())
-	eps := addendpoint.New(svc, log.NewNopLogger(), discard.NewHistogram(), opentracing.GlobalTracer())
-	mux := addtransport.NewHTTPHandler(eps, opentracing.GlobalTracer(), log.NewNopLogger())
+	eps := addendpoint.New(svc, log.NewNopLogger(), discard.NewHistogram(), opentracing.GlobalTracer(), zkt)
+	mux := addtransport.NewHTTPHandler(eps, opentracing.GlobalTracer(), zkt, log.NewNopLogger())
 	srv := httptest.NewServer(mux)
 	defer srv.Close()
 

examples/addsvc/pkg/addendpoint/set.go

@@ -7,6 +7,7 @@ import (
 	"golang.org/x/time/rate"
 
 	stdopentracing "github.com/opentracing/opentracing-go"
+	stdzipkin "github.com/openzipkin/zipkin-go"
 	"github.com/sony/gobreaker"
 
 	"github.com/go-kit/kit/circuitbreaker"
@@ -15,6 +16,7 @@ import (
 	"github.com/go-kit/kit/metrics"
 	"github.com/go-kit/kit/ratelimit"
 	"github.com/go-kit/kit/tracing/opentracing"
+	"github.com/go-kit/kit/tracing/zipkin"
 
 	"github.com/go-kit/kit/examples/addsvc/pkg/addservice"
 )
@@ -29,13 +31,14 @@ type Set struct {
 
 // New returns a Set that wraps the provided server, and wires in all of the
 // expected endpoint middlewares via the various parameters.
-func New(svc addservice.Service, logger log.Logger, duration metrics.Histogram, trace stdopentracing.Tracer) Set {
+func New(svc addservice.Service, logger log.Logger, duration metrics.Histogram, otTracer stdopentracing.Tracer, zipkinTracer *stdzipkin.Tracer) Set {
 	var sumEndpoint endpoint.Endpoint
 	{
 		sumEndpoint = MakeSumEndpoint(svc)
 		sumEndpoint = ratelimit.NewErroringLimiter(rate.NewLimiter(rate.Every(time.Second), 1))(sumEndpoint)
 		sumEndpoint = circuitbreaker.Gobreaker(gobreaker.NewCircuitBreaker(gobreaker.Settings{}))(sumEndpoint)
-		sumEndpoint = opentracing.TraceServer(trace, "Sum")(sumEndpoint)
+		sumEndpoint = opentracing.TraceServer(otTracer, "Sum")(sumEndpoint)
+		sumEndpoint = zipkin.TraceEndpoint(zipkinTracer, "Sum")(sumEndpoint)
 		sumEndpoint = LoggingMiddleware(log.With(logger, "method", "Sum"))(sumEndpoint)
 		sumEndpoint = InstrumentingMiddleware(duration.With("method", "Sum"))(sumEndpoint)
 	}
@@ -44,7 +47,8 @@ func New(svc addservice.Service, logger log.Logger, duration metrics.Histogram,
 		concatEndpoint = MakeConcatEndpoint(svc)
 		concatEndpoint = ratelimit.NewErroringLimiter(rate.NewLimiter(rate.Every(time.Second), 100))(concatEndpoint)
 		concatEndpoint = circuitbreaker.Gobreaker(gobreaker.NewCircuitBreaker(gobreaker.Settings{}))(concatEndpoint)
-		concatEndpoint = opentracing.TraceServer(trace, "Concat")(concatEndpoint)
+		concatEndpoint = opentracing.TraceServer(otTracer, "Concat")(concatEndpoint)
+		concatEndpoint = zipkin.TraceEndpoint(zipkinTracer, "Concat")(concatEndpoint)
 		concatEndpoint = LoggingMiddleware(log.With(logger, "method", "Concat"))(concatEndpoint)
 		concatEndpoint = InstrumentingMiddleware(duration.With("method", "Concat"))(concatEndpoint)
 	}