Commit 5458306c06cfcfedb728ecadfe249dc6f6ff8af5 - golang-github-go-kit-kit

+13

-0

.gitignore less more

29	29	_testmain.go
30	30
31	31	*.exe
	32
	33	# https://github.com/github/gitignore/blob/master/Global/Vim.gitignore
	34	# swap
	35	[._]*.s[a-w][a-z]
	36	[._]s[a-w][a-z]
	37	# session
	38	Session.vim
	39	# temporary
	40	.netrwhist
	41	*~
	42	# auto-generated tag files
	43	tags
	44

+30

-17

examples/README.md less more

639	639
640	640	## Advanced topics
641	641
	642	### Threading a context
	643
	644	The context object is used to carry information across conceptual boundaries in the scope of a single request.
	645	In our example, we haven't yet threaded the context through our business logic.
	646	But that's almost always a good idea.
	647	It allows you to pass request-scoped information between business logic and middlewares,
	648	and is necessary for more sophisticated tasks like granular distributed tracing annotations.
	649
	650	Concretely, this means your business logic interfaces will look like
	651
	652	```go
	653	type MyService interface {
	654	Foo(context.Context, string, int) (string, error)
	655	Bar(context.Context, string) error
	656	Baz(context.Context) (int, error)
	657	}
	658	```
	659
642	660	### Request tracing
643	661
644	662	Once your infrastructure grows beyond a certain size, it becomes important to trace requests through multiple services, so you can identify and troubleshoot hotspots.

648	666
649	667	It's possible to use Go kit to create a client package to your service, to make consuming your service easier from other Go programs.
650	668	Effectively, your client package will provide an implementation of your service interface, which invokes a remote service instance using a specific transport.
651		An example is in the works. Stay tuned.
652
653		### Threading a context
654
655		The context object is used to carry information across conceptual boundaries in the scope of a single request.
656		In our example, we haven't threaded the context through our business logic.
657		But it may be useful to do so, to get access to request-scoped information like trace IDs.
658		It may also be possible to pass things like loggers and metrics objects through the context, although this is not currently recommended.
	669	See [package addsvc/client](https://github.com/go-kit/kit/tree/master/examples/addsvc/client) for an example.
659	670
660	671	## Other examples
661	672
662		### Transport-specific
663
664		We plan to have small example services that bind to each of our supported transports.
665		Stay tuned.
666
667	673	### addsvc
668	674
669	675	[addsvc](https://github.com/go-kit/kit/blob/master/examples/addsvc) was the original example application.
670		It exposes a set of operations over all supported transports.
	676	It exposes a set of operations over all supported transports.
671	677	It's fully logged, instrumented, and uses Zipkin request tracing.
672	678	It also demonstrates how to create and use client packages.
673	679	It's a good example of a fully-featured Go kit service.

675	681	### profilesvc
676	682
677	683	[profilesvc](https://github.com/go-kit/kit/blob/master/examples/profilesvc)
678		demonstrates how to use Go kit to build a REST-ish microservice.
	684	demonstrates how to use Go kit to build a REST-ish microservice.
679	685
680	686	### apigateway
681
682		Track [issue #202](https://github.com/go-kit/kit/issues/202) for progress.
	687	[apigateway](https://github.com/go-kit/kit/blob/master/examples/apigateway/main.go)
	688	demonstrates how to implement the API gateway pattern,
	689	backed by a Consul service discovery system.
	690
	691	### shipping
	692
	693	[shipping](https://github.com/go-kit/kit/tree/master/examples/shipping)
	694	is a complete, "real-world" application composed of multiple microservices,
	695	based on Domain Driven Design principles.

+35

-23

examples/addsvc/client/grpc/factory.go less more

2	2	import (
3	3	"io"
4	4
	5	kitot "github.com/go-kit/kit/tracing/opentracing"
	6	"github.com/opentracing/opentracing-go"
5	7	"google.golang.org/grpc"
6	8
7	9	"github.com/go-kit/kit/endpoint"
8	10	"github.com/go-kit/kit/examples/addsvc/pb"
	11	"github.com/go-kit/kit/loadbalancer"
	12	"github.com/go-kit/kit/log"
9	13	grpctransport "github.com/go-kit/kit/transport/grpc"
10	14	)
11	15
12		// SumEndpointFactory transforms GRPC host:port strings into Endpoints that call the Sum method on a GRPC server
	16	// MakeSumEndpointFactory returns a loadbalancer.Factory that transforms GRPC
	17	// host:port strings into Endpoints that call the Sum method on a GRPC server
13	18	// at that address.
14		func SumEndpointFactory(instance string) (endpoint.Endpoint, io.Closer, error) {
15		cc, err := grpc.Dial(instance, grpc.WithInsecure())
16		return grpctransport.NewClient(
17		cc,
18		"Add",
19		"Sum",
20		encodeSumRequest,
21		decodeSumResponse,
22		pb.SumReply{},
23		).Endpoint(), cc, err
	19	func MakeSumEndpointFactory(tracer opentracing.Tracer, tracingLogger log.Logger) loadbalancer.Factory {
	20	return func(instance string) (endpoint.Endpoint, io.Closer, error) {
	21	cc, err := grpc.Dial(instance, grpc.WithInsecure())
	22	return grpctransport.NewClient(
	23	cc,
	24	"Add",
	25	"Sum",
	26	encodeSumRequest,
	27	decodeSumResponse,
	28	pb.SumReply{},
	29	grpctransport.SetClientBefore(kitot.ToGRPCRequest(tracer, tracingLogger)),
	30	).Endpoint(), cc, err
	31	}
24	32	}
25	33
26		// ConcatEndpointFactory transforms GRPC host:port strings into Endpoints that call the Concat method on a GRPC server
27		// at that address.
28		func ConcatEndpointFactory(instance string) (endpoint.Endpoint, io.Closer, error) {
29		cc, err := grpc.Dial(instance, grpc.WithInsecure())
30		return grpctransport.NewClient(
31		cc,
32		"Add",
33		"Concat",
34		encodeConcatRequest,
35		decodeConcatResponse,
36		pb.ConcatReply{},
37		).Endpoint(), cc, err
	34	// MakeConcatEndpointFactory returns a loadbalancer.Factory that transforms
	35	// GRPC host:port strings into Endpoints that call the Concat method on a GRPC
	36	// server at that address.
	37	func MakeConcatEndpointFactory(tracer opentracing.Tracer, tracingLogger log.Logger) loadbalancer.Factory {
	38	return func(instance string) (endpoint.Endpoint, io.Closer, error) {
	39	cc, err := grpc.Dial(instance, grpc.WithInsecure())
	40	return grpctransport.NewClient(
	41	cc,
	42	"Add",
	43	"Concat",
	44	encodeConcatRequest,
	45	decodeConcatResponse,
	46	pb.ConcatReply{},
	47	grpctransport.SetClientBefore(kitot.ToGRPCRequest(tracer, tracingLogger)),
	48	).Endpoint(), cc, err
	49	}
38	50	}

+47

-32

examples/addsvc/client/httpjson/factory.go less more

3	3	"io"
4	4	"net/url"
5	5
	6	"github.com/opentracing/opentracing-go"
	7
6	8	"github.com/go-kit/kit/endpoint"
7	9	"github.com/go-kit/kit/examples/addsvc/server"
	10	"github.com/go-kit/kit/loadbalancer"
	11	"github.com/go-kit/kit/log"
	12	kitot "github.com/go-kit/kit/tracing/opentracing"
8	13	httptransport "github.com/go-kit/kit/transport/http"
9	14	)
10	15
11		// SumEndpointFactory transforms a http url into an Endpoint.
	16	// MakeSumEndpointFactory generates a Factory that transforms an http url into
	17	// an Endpoint.
	18	//
12	19	// The path of the url is reset to /sum.
13		func SumEndpointFactory(instance string) (endpoint.Endpoint, io.Closer, error) {
14		sumURL, err := url.Parse(instance)
15		if err != nil {
16		return nil, nil, err
	20	func MakeSumEndpointFactory(tracer opentracing.Tracer, tracingLogger log.Logger) loadbalancer.Factory {
	21	return func(instance string) (endpoint.Endpoint, io.Closer, error) {
	22	sumURL, err := url.Parse(instance)
	23	if err != nil {
	24	return nil, nil, err
	25	}
	26	sumURL.Path = "/sum"
	27
	28	client := httptransport.NewClient(
	29	"GET",
	30	sumURL,
	31	server.EncodeSumRequest,
	32	server.DecodeSumResponse,
	33	httptransport.SetClient(nil),
	34	httptransport.SetClientBefore(kitot.ToHTTPRequest(tracer, tracingLogger)),
	35	)
	36
	37	return client.Endpoint(), nil, nil
17	38	}
18		sumURL.Path = "/sum"
19
20		client := httptransport.NewClient(
21		"GET",
22		sumURL,
23		server.EncodeSumRequest,
24		server.DecodeSumResponse,
25		httptransport.SetClient(nil),
26		)
27
28		return client.Endpoint(), nil, nil
29	39	}
30	40
31		// ConcatEndpointFactory transforms a http url into an Endpoint.
	41	// MakeConcatEndpointFactory generates a Factory that transforms an http url
	42	// into an Endpoint.
	43	//
32	44	// The path of the url is reset to /concat.
33		func ConcatEndpointFactory(instance string) (endpoint.Endpoint, io.Closer, error) {
34		concatURL, err := url.Parse(instance)
35		if err != nil {
36		return nil, nil, err
	45	func MakeConcatEndpointFactory(tracer opentracing.Tracer, tracingLogger log.Logger) loadbalancer.Factory {
	46	return func(instance string) (endpoint.Endpoint, io.Closer, error) {
	47	concatURL, err := url.Parse(instance)
	48	if err != nil {
	49	return nil, nil, err
	50	}
	51	concatURL.Path = "/concat"
	52
	53	client := httptransport.NewClient(
	54	"GET",
	55	concatURL,
	56	server.EncodeConcatRequest,
	57	server.DecodeConcatResponse,
	58	httptransport.SetClient(nil),
	59	httptransport.SetClientBefore(kitot.ToHTTPRequest(tracer, tracingLogger)),
	60	)
	61
	62	return client.Endpoint(), nil, nil
37	63	}
38		concatURL.Path = "/concat"
39
40		client := httptransport.NewClient(
41		"GET",
42		concatURL,
43		server.EncodeConcatRequest,
44		server.DecodeConcatResponse,
45		httptransport.SetClient(nil),
46		)
47
48		return client.Endpoint(), nil, nil
49	64	}

+37

-8

examples/addsvc/client/main.go less more

8	8	"strings"
9	9	"time"
10	10
	11	"github.com/lightstep/lightstep-tracer-go"
	12	"github.com/opentracing/opentracing-go"
	13	appdashot "github.com/sourcegraph/appdash/opentracing"
11	14	"golang.org/x/net/context"
	15	"sourcegraph.com/sourcegraph/appdash"
12	16
13	17	"github.com/go-kit/kit/endpoint"
14	18	grpcclient "github.com/go-kit/kit/examples/addsvc/client/grpc"

18	22	"github.com/go-kit/kit/loadbalancer"
19	23	"github.com/go-kit/kit/loadbalancer/static"
20	24	"github.com/go-kit/kit/log"
	25	kitot "github.com/go-kit/kit/tracing/opentracing"
21	26	)
22	27
23	28	func main() {

30	35	thriftProtocol = flag.String("thrift.protocol", "binary", "binary, compact, json, simplejson")
31	36	thriftBufferSize = flag.Int("thrift.buffer.size", 0, "0 for unbuffered")
32	37	thriftFramed = flag.Bool("thrift.framed", false, "true to enable framing")
	38
	39	// Two OpenTracing backends (to demonstrate how they can be interchanged):
	40	appdashAddr = flag.String("appdash.addr", "", "Enable Appdash tracing via an Appdash server host:port")
	41	lightstepAccessToken = flag.String("lightstep.token", "", "Enable LightStep tracing via a LightStep access token")
33	42	)
34	43	flag.Parse()
35	44	if len(os.Args) < 4 {

47	56	logger = log.NewLogfmtLogger(os.Stdout)
48	57	logger = log.NewContext(logger).With("caller", log.DefaultCaller)
49	58	logger = log.NewContext(logger).With("transport", *transport)
	59	tracingLogger := log.NewContext(logger).With("component", "tracing")
	60
	61	// Set up OpenTracing
	62	var tracer opentracing.Tracer
	63	{
	64	switch {
	65	case appdashAddr != "" && lightstepAccessToken == "":
	66	tracer = appdashot.NewTracer(appdash.NewRemoteCollector(*appdashAddr))
	67	case appdashAddr == "" && lightstepAccessToken != "":
	68	tracer = lightstep.NewTracer(lightstep.Options{
	69	AccessToken: *lightstepAccessToken,
	70	})
	71	defer lightstep.FlushLightStepTracer(tracer)
	72	case appdashAddr == "" && lightstepAccessToken == "":
	73	tracer = opentracing.GlobalTracer() // no-op
	74	default:
	75	panic("specify either -appdash.addr or -lightstep.access.token, not both")
	76	}
	77	}
50	78
51	79	var (
52	80	instances []string

56	84	switch *transport {
57	85	case "grpc":
58	86	instances = strings.Split(*grpcAddrs, ",")
59		sumFactory = grpcclient.SumEndpointFactory
60		concatFactory = grpcclient.ConcatEndpointFactory
	87	sumFactory = grpcclient.MakeSumEndpointFactory(tracer, tracingLogger)
	88	concatFactory = grpcclient.MakeConcatEndpointFactory(tracer, tracingLogger)
61	89
62	90	case "httpjson":
63	91	instances = strings.Split(*httpAddrs, ",")

66	94	instances[i] = "http://" + rawurl
67	95	}
68	96	}
69		sumFactory = httpjsonclient.SumEndpointFactory
70		concatFactory = httpjsonclient.ConcatEndpointFactory
	97	sumFactory = httpjsonclient.MakeSumEndpointFactory(tracer, tracingLogger)
	98	concatFactory = httpjsonclient.MakeConcatEndpointFactory(tracer, tracingLogger)
71	99
72	100	case "netrpc":
73	101	instances = strings.Split(*netrpcAddrs, ",")

85	113	os.Exit(1)
86	114	}
87	115
88		sum := buildEndpoint(instances, sumFactory, randomSeed, logger)
89		concat := buildEndpoint(instances, concatFactory, randomSeed, logger)
	116	sum := buildEndpoint(tracer, "sum", instances, sumFactory, randomSeed, logger)
	117	concat := buildEndpoint(tracer, "concat", instances, concatFactory, randomSeed, logger)
90	118
91	119	svc := newClient(root, sum, concat, logger)
92	120

109	137	}
110	138	}
111	139
112		func buildEndpoint(instances []string, factory loadbalancer.Factory, seed int64, logger log.Logger) endpoint.Endpoint {
	140	func buildEndpoint(tracer opentracing.Tracer, operationName string, instances []string, factory loadbalancer.Factory, seed int64, logger log.Logger) endpoint.Endpoint {
113	141	publisher := static.NewPublisher(instances, factory, logger)
114	142	random := loadbalancer.NewRandom(publisher, seed)
115		return loadbalancer.Retry(10, 10*time.Second, random)
	143	endpoint := loadbalancer.Retry(10, 10*time.Second, random)
	144	return kitot.TraceClient(tracer, operationName)(endpoint)
116	145	}

+19

-3

examples/addsvc/grpc_binding.go less more

2	2	import (
3	3	"golang.org/x/net/context"
4	4
	5	"github.com/opentracing/opentracing-go"
	6
5	7	"github.com/go-kit/kit/examples/addsvc/pb"
6	8	"github.com/go-kit/kit/examples/addsvc/server"
7	9	servergrpc "github.com/go-kit/kit/examples/addsvc/server/grpc"
	10	"github.com/go-kit/kit/log"
	11	kitot "github.com/go-kit/kit/tracing/opentracing"
8	12	"github.com/go-kit/kit/transport/grpc"
9	13	)
10	14

12	16	sum, concat grpc.Handler
13	17	}
14	18
15		func newGRPCBinding(ctx context.Context, svc server.AddService) grpcBinding {
	19	func newGRPCBinding(ctx context.Context, tracer opentracing.Tracer, svc server.AddService, tracingLogger log.Logger) grpcBinding {
16	20	return grpcBinding{
17		sum: grpc.NewServer(ctx, makeSumEndpoint(svc), servergrpc.DecodeSumRequest, servergrpc.EncodeSumResponse),
18		concat: grpc.NewServer(ctx, makeConcatEndpoint(svc), servergrpc.DecodeConcatRequest, servergrpc.EncodeConcatResponse),
	21	sum: grpc.NewServer(
	22	ctx,
	23	kitot.TraceServer(tracer, "sum")(makeSumEndpoint(svc)),
	24	servergrpc.DecodeSumRequest,
	25	servergrpc.EncodeSumResponse,
	26	grpc.ServerBefore(kitot.FromGRPCRequest(tracer, "", tracingLogger)),
	27	),
	28	concat: grpc.NewServer(
	29	ctx,
	30	kitot.TraceServer(tracer, "concat")(makeConcatEndpoint(svc)),
	31	servergrpc.DecodeConcatRequest,
	32	servergrpc.EncodeConcatResponse,
	33	grpc.ServerBefore(kitot.FromGRPCRequest(tracer, "", tracingLogger)),
	34	),
19	35	}
20	36	}
21	37

+38

-34

examples/addsvc/main.go less more

14	14	"time"
15	15
16	16	"github.com/apache/thrift/lib/go/thrift"
	17	"github.com/lightstep/lightstep-tracer-go"
	18	"github.com/opentracing/opentracing-go"
17	19	stdprometheus "github.com/prometheus/client_golang/prometheus"
	20	appdashot "github.com/sourcegraph/appdash/opentracing"
18	21	"golang.org/x/net/context"
19	22	"google.golang.org/grpc"
	23	"sourcegraph.com/sourcegraph/appdash"
20	24
21	25	"github.com/go-kit/kit/endpoint"
22	26	"github.com/go-kit/kit/examples/addsvc/pb"

26	30	"github.com/go-kit/kit/metrics"
27	31	"github.com/go-kit/kit/metrics/expvar"
28	32	"github.com/go-kit/kit/metrics/prometheus"
	33	kitot "github.com/go-kit/kit/tracing/opentracing"
29	34	"github.com/go-kit/kit/tracing/zipkin"
30	35	httptransport "github.com/go-kit/kit/transport/http"
31	36	)

35	40	// of glog. So, we define a new flag set, to keep those domains distinct.
36	41	fs := flag.NewFlagSet("", flag.ExitOnError)
37	42	var (
38		debugAddr = fs.String("debug.addr", ":8000", "Address for HTTP debug/instrumentation server")
39		httpAddr = fs.String("http.addr", ":8001", "Address for HTTP (JSON) server")
40		grpcAddr = fs.String("grpc.addr", ":8002", "Address for gRPC server")
41		netrpcAddr = fs.String("netrpc.addr", ":8003", "Address for net/rpc server")
42		thriftAddr = fs.String("thrift.addr", ":8004", "Address for Thrift server")
43		thriftProtocol = fs.String("thrift.protocol", "binary", "binary, compact, json, simplejson")
44		thriftBufferSize = fs.Int("thrift.buffer.size", 0, "0 for unbuffered")
45		thriftFramed = fs.Bool("thrift.framed", false, "true to enable framing")
46		zipkinHostPort = fs.String("zipkin.host.port", "my.service.domain:12345", "Zipkin host:port")
47		zipkinServiceName = fs.String("zipkin.service.name", "addsvc", "Zipkin service name")
48		zipkinCollectorAddr = fs.String("zipkin.collector.addr", "", "Zipkin Kafka collector address (empty will log spans)")
	43	debugAddr = fs.String("debug.addr", ":8000", "Address for HTTP debug/instrumentation server")
	44	httpAddr = fs.String("http.addr", ":8001", "Address for HTTP (JSON) server")
	45	grpcAddr = fs.String("grpc.addr", ":8002", "Address for gRPC server")
	46	netrpcAddr = fs.String("netrpc.addr", ":8003", "Address for net/rpc server")
	47	thriftAddr = fs.String("thrift.addr", ":8004", "Address for Thrift server")
	48	thriftProtocol = fs.String("thrift.protocol", "binary", "binary, compact, json, simplejson")
	49	thriftBufferSize = fs.Int("thrift.buffer.size", 0, "0 for unbuffered")
	50	thriftFramed = fs.Bool("thrift.framed", false, "true to enable framing")
	51
	52	// Supported OpenTracing backends
	53	appdashAddr = fs.String("appdash.addr", "", "Enable Appdash tracing via an Appdash server host:port")
	54	lightstepAccessToken = fs.String("lightstep.token", "", "Enable LightStep tracing via a LightStep access token")
49	55	)
50	56	flag.Usage = fs.Usage // only show our flags
51	57	if err := fs.Parse(os.Args[1:]); err != nil {

77	83	))
78	84	}
79	85
80		// package tracing
81		var collector zipkin.Collector
82		{
83		zipkinLogger := log.NewContext(logger).With("component", "zipkin")
84		collector = loggingCollector{zipkinLogger} // TODO(pb)
85		if *zipkinCollectorAddr != "" {
86		var err error
87		if collector, err = zipkin.NewKafkaCollector(
88		[]string{*zipkinCollectorAddr},
89		zipkin.KafkaLogger(zipkinLogger),
90		); err != nil {
91		zipkinLogger.Log("err", err)
92		os.Exit(1)
93		}
	86	// Set up OpenTracing
	87	var tracer opentracing.Tracer
	88	{
	89	switch {
	90	case appdashAddr != "" && lightstepAccessToken == "":
	91	tracer = appdashot.NewTracer(appdash.NewRemoteCollector(*appdashAddr))
	92	case appdashAddr == "" && lightstepAccessToken != "":
	93	tracer = lightstep.NewTracer(lightstep.Options{
	94	AccessToken: *lightstepAccessToken,
	95	})
	96	defer lightstep.FlushLightStepTracer(tracer)
	97	case appdashAddr == "" && lightstepAccessToken == "":
	98	tracer = opentracing.GlobalTracer() // no-op
	99	default:
	100	panic("specify either -appdash.addr or -lightstep.access.token, not both")
94	101	}
95	102	}
96	103

123	130	var (
124	131	transportLogger = log.NewContext(logger).With("transport", "HTTP/JSON")
125	132	tracingLogger = log.NewContext(transportLogger).With("component", "tracing")
126		newSumSpan = zipkin.MakeNewSpanFunc(zipkinHostPort, zipkinServiceName, "sum")
127		newConcatSpan = zipkin.MakeNewSpanFunc(zipkinHostPort, zipkinServiceName, "concat")
128		traceSum = zipkin.ToContext(newSumSpan, tracingLogger)
129		traceConcat = zipkin.ToContext(newConcatSpan, tracingLogger)
130	133	mux = http.NewServeMux()
131	134	sum, concat endpoint.Endpoint
132	135	)
133	136
134	137	sum = makeSumEndpoint(svc)
135		sum = zipkin.AnnotateServer(newSumSpan, collector)(sum)
	138	sum = kitot.TraceServer(tracer, "sum")(sum)
136	139	mux.Handle("/sum", httptransport.NewServer(
137	140	root,
138	141	sum,
139	142	server.DecodeSumRequest,
140	143	server.EncodeSumResponse,
141		httptransport.ServerBefore(traceSum),
142	144	httptransport.ServerErrorLogger(transportLogger),
	145	httptransport.ServerBefore(kitot.FromHTTPRequest(tracer, "sum", tracingLogger)),
143	146	))
144	147
145	148	concat = makeConcatEndpoint(svc)
146		concat = zipkin.AnnotateServer(newConcatSpan, collector)(concat)
	149	concat = kitot.TraceServer(tracer, "concat")(concat)
147	150	mux.Handle("/concat", httptransport.NewServer(
148	151	root,
149	152	concat,
150	153	server.DecodeConcatRequest,
151	154	server.EncodeConcatResponse,
152		httptransport.ServerBefore(traceConcat),
153	155	httptransport.ServerErrorLogger(transportLogger),
	156	httptransport.ServerBefore(kitot.FromHTTPRequest(tracer, "concat", tracingLogger)),
154	157	))
155	158
156	159	transportLogger.Log("addr", *httpAddr)

160	163	// Transport: gRPC
161	164	go func() {
162	165	transportLogger := log.NewContext(logger).With("transport", "gRPC")
	166	tracingLogger := log.NewContext(transportLogger).With("component", "tracing")
163	167	ln, err := net.Listen("tcp", *grpcAddr)
164	168	if err != nil {
165	169	errc <- err
166	170	return
167	171	}
168	172	s := grpc.NewServer() // uses its own, internal context
169		pb.RegisterAddServer(s, newGRPCBinding(root, svc))
	173	pb.RegisterAddServer(s, newGRPCBinding(root, tracer, svc, tracingLogger))
170	174	transportLogger.Log("addr", *grpcAddr)
171	175	errc <- s.Serve(ln)
172	176	}()

+3

-2

examples/apigateway/main.go less more

16	16
17	17	"github.com/gorilla/mux"
18	18	"github.com/hashicorp/consul/api"
	19	"github.com/opentracing/opentracing-go"
19	20	"golang.org/x/net/context"
20	21
21	22	"github.com/go-kit/kit/endpoint"

76	77	factory loadbalancer.Factory
77	78	}{
78	79	"addsvc": {
79		{path: "/api/addsvc/concat", factory: grpc.ConcatEndpointFactory},
80		{path: "/api/addsvc/sum", factory: grpc.SumEndpointFactory},
	80	{path: "/api/addsvc/concat", factory: grpc.MakeConcatEndpointFactory(opentracing.GlobalTracer(), nil)},
	81	{path: "/api/addsvc/sum", factory: grpc.MakeSumEndpointFactory(opentracing.GlobalTracer(), nil)},
81	82	},
82	83	"stringsvc": {
83	84	{path: "/api/stringsvc/uppercase", factory: httpFactory(ctx, "GET", "uppercase/")},

+35

-8

log/logfmt_logger.go less more

0	0	package log
1	1
2	2	import (
	3	"bytes"
3	4	"io"
	5	"sync"
4	6
5	7	"github.com/go-logfmt/logfmt"
6	8	)
	9
	10	type logfmtEncoder struct {
	11	*logfmt.Encoder
	12	buf bytes.Buffer
	13	}
	14
	15	func (l *logfmtEncoder) Reset() {
	16	l.Encoder.Reset()
	17	l.buf.Reset()
	18	}
	19
	20	var logfmtEncoderPool = sync.Pool{
	21	New: func() interface{} {
	22	var enc logfmtEncoder
	23	enc.Encoder = logfmt.NewEncoder(&enc.buf)
	24	return &enc
	25	},
	26	}
7	27
8	28	type logfmtLogger struct {
9	29	w io.Writer

17	37	}
18	38
19	39	func (l logfmtLogger) Log(keyvals ...interface{}) error {
	40	enc := logfmtEncoderPool.Get().(*logfmtEncoder)
	41	enc.Reset()
	42	defer logfmtEncoderPool.Put(enc)
	43
	44	if err := enc.EncodeKeyvals(keyvals...); err != nil {
	45	return err
	46	}
	47
	48	// Add newline to the end of the buffer
	49	if err := enc.EndRecord(); err != nil {
	50	return err
	51	}
	52
20	53	// The Logger interface requires implementations to be safe for concurrent
21	54	// use by multiple goroutines. For this implementation that means making
22		// only one call to l.w.Write() for each call to Log. We first collect all
23		// of the bytes into b, and then call l.w.Write(b).
24		b, err := logfmt.MarshalKeyvals(keyvals...)
25		if err != nil {
26		return err
27		}
28		b = append(b, '\n')
29		if _, err := l.w.Write(b); err != nil {
	55	// only one call to l.w.Write() for each call to Log.
	56	if _, err := l.w.Write(enc.buf.Bytes()); err != nil {
30	57	return err
31	58	}
32	59	return nil

+43

-0

metrics/discard/discard.go less more

	0	// Package discard implements a backend for package metrics that succeeds
	1	// without doing anything.
	2	package discard
	3
	4	import "github.com/go-kit/kit/metrics"
	5
	6	type counter struct {
	7	name string
	8	}
	9
	10	// NewCounter returns a Counter that does nothing.
	11	func NewCounter(name string) metrics.Counter { return &counter{name} }
	12
	13	func (c *counter) Name() string { return c.name }
	14	func (c *counter) With(metrics.Field) metrics.Counter { return c }
	15	func (c *counter) Add(delta uint64) {}
	16
	17	type gauge struct {
	18	name string
	19	}
	20
	21	// NewGauge returns a Gauge that does nothing.
	22	func NewGauge(name string) metrics.Gauge { return &gauge{name} }
	23
	24	func (g *gauge) Name() string { return g.name }
	25	func (g *gauge) With(metrics.Field) metrics.Gauge { return g }
	26	func (g *gauge) Set(value float64) {}
	27	func (g *gauge) Add(delta float64) {}
	28	func (g *gauge) Get() float64 { return 0 }
	29
	30	type histogram struct {
	31	name string
	32	}
	33
	34	// NewHistogram returns a Histogram that does nothing.
	35	func NewHistogram(name string) metrics.Histogram { return &histogram{name} }
	36
	37	func (h *histogram) Name() string { return h.name }
	38	func (h *histogram) With(metrics.Field) metrics.Histogram { return h }
	39	func (h *histogram) Observe(value int64) {}
	40	func (h *histogram) Distribution() ([]metrics.Bucket, []metrics.Quantile) {
	41	return []metrics.Bucket{}, []metrics.Quantile{}
	42	}

+25

-25

metrics/dogstatsd/dogstatsd.go less more

23	23
24	24	const maxBufferSize = 1400 // bytes
25	25
26		type dogstatsdCounter struct {
	26	type counter struct {
27	27	key string
28	28	c chan string
29	29	tags []metrics.Field

41	41	// NewCounterTick is the same as NewCounter, but allows the user to pass in a
42	42	// ticker channel instead of invoking time.Tick.
43	43	func NewCounterTick(w io.Writer, key string, reportTicker <-chan time.Time, tags []metrics.Field) metrics.Counter {
44		c := &dogstatsdCounter{
	44	c := &counter{
45	45	key: key,
46	46	c: make(chan string),
47	47	tags: tags,

50	50	return c
51	51	}
52	52
53		func (c *dogstatsdCounter) Name() string { return c.key }
54
55		func (c *dogstatsdCounter) With(f metrics.Field) metrics.Counter {
56		return &dogstatsdCounter{
	53	func (c *counter) Name() string { return c.key }
	54
	55	func (c *counter) With(f metrics.Field) metrics.Counter {
	56	return &counter{
57	57	key: c.key,
58	58	c: c.c,
59	59	tags: append(c.tags, f),
60	60	}
61	61	}
62	62
63		func (c *dogstatsdCounter) Add(delta uint64) { c.c <- applyTags(fmt.Sprintf("%d\|c", delta), c.tags) }
64
65		type dogstatsdGauge struct {
	63	func (c *counter) Add(delta uint64) { c.c <- applyTags(fmt.Sprintf("%d\|c", delta), c.tags) }
	64
	65	type gauge struct {
66	66	key string
67	67	lastValue uint64 // math.Float64frombits
68	68	g chan string

81	81	// NewGaugeTick is the same as NewGauge, but allows the user to pass in a ticker
82	82	// channel instead of invoking time.Tick.
83	83	func NewGaugeTick(w io.Writer, key string, reportTicker <-chan time.Time, tags []metrics.Field) metrics.Gauge {
84		g := &dogstatsdGauge{
	84	g := &gauge{
85	85	key: key,
86	86	g: make(chan string),
87	87	tags: tags,

90	90	return g
91	91	}
92	92
93		func (g *dogstatsdGauge) Name() string { return g.key }
94
95		func (g *dogstatsdGauge) With(f metrics.Field) metrics.Gauge {
96		return &dogstatsdGauge{
	93	func (g *gauge) Name() string { return g.key }
	94
	95	func (g *gauge) With(f metrics.Field) metrics.Gauge {
	96	return &gauge{
97	97	key: g.key,
98	98	lastValue: g.lastValue,
99	99	g: g.g,

101	101	}
102	102	}
103	103
104		func (g *dogstatsdGauge) Add(delta float64) {
	104	func (g *gauge) Add(delta float64) {
105	105	// https://github.com/etsy/statsd/blob/master/docs/metric_types.md#gauges
106	106	sign := "+"
107	107	if delta < 0 {

110	110	g.g <- applyTags(fmt.Sprintf("%s%f\|g", sign, delta), g.tags)
111	111	}
112	112
113		func (g *dogstatsdGauge) Set(value float64) {
	113	func (g *gauge) Set(value float64) {
114	114	atomic.StoreUint64(&g.lastValue, math.Float64bits(value))
115	115	g.g <- applyTags(fmt.Sprintf("%f\|g", value), g.tags)
116	116	}
117	117
118		func (g *dogstatsdGauge) Get() float64 {
	118	func (g *gauge) Get() float64 {
119	119	return math.Float64frombits(atomic.LoadUint64(&g.lastValue))
120	120	}
121	121

146	146	return c
147	147	}
148	148
149		type dogstatsdHistogram struct {
	149	type histogram struct {
150	150	key string
151	151	h chan string
152	152	tags []metrics.Field

176	176	// NewHistogramTick is the same as NewHistogram, but allows the user to pass a
177	177	// ticker channel instead of invoking time.Tick.
178	178	func NewHistogramTick(w io.Writer, key string, reportTicker <-chan time.Time, tags []metrics.Field) metrics.Histogram {
179		h := &dogstatsdHistogram{
	179	h := &histogram{
180	180	key: key,
181	181	h: make(chan string),
182	182	tags: tags,

185	185	return h
186	186	}
187	187
188		func (h *dogstatsdHistogram) Name() string { return h.key }
189
190		func (h *dogstatsdHistogram) With(f metrics.Field) metrics.Histogram {
191		return &dogstatsdHistogram{
	188	func (h *histogram) Name() string { return h.key }
	189
	190	func (h *histogram) With(f metrics.Field) metrics.Histogram {
	191	return &histogram{
192	192	key: h.key,
193	193	h: h.h,
194	194	tags: append(h.tags, f),
195	195	}
196	196	}
197	197
198		func (h *dogstatsdHistogram) Observe(value int64) {
	198	func (h *histogram) Observe(value int64) {
199	199	h.h <- applyTags(fmt.Sprintf("%d\|ms", value), h.tags)
200	200	}
201	201
202		func (h *dogstatsdHistogram) Distribution() ([]metrics.Bucket, []metrics.Quantile) {
	202	func (h *histogram) Distribution() ([]metrics.Bucket, []metrics.Quantile) {
203	203	// TODO(pb): no way to do this without introducing e.g. codahale/hdrhistogram
204	204	return []metrics.Bucket{}, []metrics.Quantile{}
205	205	}

+32

-32

metrics/prometheus/prometheus.go less more

13	13	// PrometheusLabelValueUnknown.
14	14	var PrometheusLabelValueUnknown = "unknown"
15	15
16		type prometheusCounter struct {
	16	type counter struct {
17	17	*prometheus.CounterVec
18	18	name string
19	19	Pairs map[string]string

28	28	for _, fieldName := range fieldKeys {
29	29	p[fieldName] = PrometheusLabelValueUnknown
30	30	}
31		return prometheusCounter{
	31	return counter{
32	32	CounterVec: m,
33	33	name: opts.Name,
34	34	Pairs: p,
35	35	}
36	36	}
37	37
38		func (c prometheusCounter) Name() string { return c.name }
39
40		func (c prometheusCounter) With(f metrics.Field) metrics.Counter {
41		return prometheusCounter{
	38	func (c counter) Name() string { return c.name }
	39
	40	func (c counter) With(f metrics.Field) metrics.Counter {
	41	return counter{
42	42	CounterVec: c.CounterVec,
43	43	name: c.name,
44	44	Pairs: merge(c.Pairs, f),
45	45	}
46	46	}
47	47
48		func (c prometheusCounter) Add(delta uint64) {
	48	func (c counter) Add(delta uint64) {
49	49	c.CounterVec.With(prometheus.Labels(c.Pairs)).Add(float64(delta))
50	50	}
51	51
52		type prometheusGauge struct {
	52	type gauge struct {
53	53	*prometheus.GaugeVec
54	54	name string
55	55	Pairs map[string]string

60	60	func NewGauge(opts prometheus.GaugeOpts, fieldKeys []string) metrics.Gauge {
61	61	m := prometheus.NewGaugeVec(opts, fieldKeys)
62	62	prometheus.MustRegister(m)
63		return prometheusGauge{
	63	return gauge{
64	64	GaugeVec: m,
65	65	name: opts.Name,
66	66	Pairs: pairsFrom(fieldKeys),
67	67	}
68	68	}
69	69
70		func (g prometheusGauge) Name() string { return g.name }
71
72		func (g prometheusGauge) With(f metrics.Field) metrics.Gauge {
73		return prometheusGauge{
	70	func (g gauge) Name() string { return g.name }
	71
	72	func (g gauge) With(f metrics.Field) metrics.Gauge {
	73	return gauge{
74	74	GaugeVec: g.GaugeVec,
75	75	name: g.name,
76	76	Pairs: merge(g.Pairs, f),
77	77	}
78	78	}
79	79
80		func (g prometheusGauge) Set(value float64) {
	80	func (g gauge) Set(value float64) {
81	81	g.GaugeVec.With(prometheus.Labels(g.Pairs)).Set(value)
82	82	}
83	83
84		func (g prometheusGauge) Add(delta float64) {
	84	func (g gauge) Add(delta float64) {
85	85	g.GaugeVec.With(prometheus.Labels(g.Pairs)).Add(delta)
86	86	}
87	87
88		func (g prometheusGauge) Get() float64 {
	88	func (g gauge) Get() float64 {
89	89	// TODO(pb): see https://github.com/prometheus/client_golang/issues/58
90	90	return 0.0
91	91	}

98	98	prometheus.MustRegister(prometheus.NewGaugeFunc(opts, callback))
99	99	}
100	100
101		type prometheusSummary struct {
	101	type summary struct {
102	102	*prometheus.SummaryVec
103	103	name string
104	104	Pairs map[string]string

112	112	func NewSummary(opts prometheus.SummaryOpts, fieldKeys []string) metrics.Histogram {
113	113	m := prometheus.NewSummaryVec(opts, fieldKeys)
114	114	prometheus.MustRegister(m)
115		return prometheusSummary{
	115	return summary{
116	116	SummaryVec: m,
117	117	name: opts.Name,
118	118	Pairs: pairsFrom(fieldKeys),
119	119	}
120	120	}
121	121
122		func (s prometheusSummary) Name() string { return s.name }
123
124		func (s prometheusSummary) With(f metrics.Field) metrics.Histogram {
125		return prometheusSummary{
	122	func (s summary) Name() string { return s.name }
	123
	124	func (s summary) With(f metrics.Field) metrics.Histogram {
	125	return summary{
126	126	SummaryVec: s.SummaryVec,
127	127	name: s.name,
128	128	Pairs: merge(s.Pairs, f),
129	129	}
130	130	}
131	131
132		func (s prometheusSummary) Observe(value int64) {
	132	func (s summary) Observe(value int64) {
133	133	s.SummaryVec.With(prometheus.Labels(s.Pairs)).Observe(float64(value))
134	134	}
135	135
136		func (s prometheusSummary) Distribution() ([]metrics.Bucket, []metrics.Quantile) {
	136	func (s summary) Distribution() ([]metrics.Bucket, []metrics.Quantile) {
137	137	// TODO(pb): see https://github.com/prometheus/client_golang/issues/58
138	138	return []metrics.Bucket{}, []metrics.Quantile{}
139	139	}
140	140
141		type prometheusHistogram struct {
	141	type histogram struct {
142	142	*prometheus.HistogramVec
143	143	name string
144	144	Pairs map[string]string

152	152	func NewHistogram(opts prometheus.HistogramOpts, fieldKeys []string) metrics.Histogram {
153	153	m := prometheus.NewHistogramVec(opts, fieldKeys)
154	154	prometheus.MustRegister(m)
155		return prometheusHistogram{
	155	return histogram{
156	156	HistogramVec: m,
157	157	name: opts.Name,
158	158	Pairs: pairsFrom(fieldKeys),
159	159	}
160	160	}
161	161
162		func (h prometheusHistogram) Name() string { return h.name }
163
164		func (h prometheusHistogram) With(f metrics.Field) metrics.Histogram {
165		return prometheusHistogram{
	162	func (h histogram) Name() string { return h.name }
	163
	164	func (h histogram) With(f metrics.Field) metrics.Histogram {
	165	return histogram{
166	166	HistogramVec: h.HistogramVec,
167	167	name: h.name,
168	168	Pairs: merge(h.Pairs, f),
169	169	}
170	170	}
171	171
172		func (h prometheusHistogram) Observe(value int64) {
	172	func (h histogram) Observe(value int64) {
173	173	h.HistogramVec.With(prometheus.Labels(h.Pairs)).Observe(float64(value))
174	174	}
175	175
176		func (h prometheusHistogram) Distribution() ([]metrics.Bucket, []metrics.Quantile) {
	176	func (h histogram) Distribution() ([]metrics.Bucket, []metrics.Quantile) {
177	177	// TODO(pb): see https://github.com/prometheus/client_golang/issues/58
178	178	return []metrics.Bucket{}, []metrics.Quantile{}
179	179	}

+25

-25

metrics/statsd/statsd.go less more

29	29
30	30	const maxBufferSize = 1400 // bytes
31	31
32		type statsdCounter struct {
	32	type counter struct {
33	33	key string
34	34	c chan string
35	35	}

47	47	// NewCounterTick is the same as NewCounter, but allows the user to pass in a
48	48	// ticker channel instead of invoking time.Tick.
49	49	func NewCounterTick(w io.Writer, key string, reportTicker <-chan time.Time) metrics.Counter {
50		c := &statsdCounter{
	50	c := &counter{
51	51	key: key,
52	52	c: make(chan string),
53	53	}

55	55	return c
56	56	}
57	57
58		func (c *statsdCounter) Name() string { return c.key }
59
60		func (c *statsdCounter) With(metrics.Field) metrics.Counter { return c }
61
62		func (c *statsdCounter) Add(delta uint64) { c.c <- fmt.Sprintf("%d\|c", delta) }
63
64		type statsdGauge struct {
	58	func (c *counter) Name() string { return c.key }
	59
	60	func (c *counter) With(metrics.Field) metrics.Counter { return c }
	61
	62	func (c *counter) Add(delta uint64) { c.c <- fmt.Sprintf("%d\|c", delta) }
	63
	64	type gauge struct {
65	65	key string
66	66	lastValue uint64 // math.Float64frombits
67	67	g chan string

80	80	// NewGaugeTick is the same as NewGauge, but allows the user to pass in a ticker
81	81	// channel instead of invoking time.Tick.
82	82	func NewGaugeTick(w io.Writer, key string, reportTicker <-chan time.Time) metrics.Gauge {
83		g := &statsdGauge{
	83	g := &gauge{
84	84	key: key,
85	85	g: make(chan string),
86	86	}

88	88	return g
89	89	}
90	90
91		func (g *statsdGauge) Name() string { return g.key }
92
93		func (g *statsdGauge) With(metrics.Field) metrics.Gauge { return g }
94
95		func (g *statsdGauge) Add(delta float64) {
	91	func (g *gauge) Name() string { return g.key }
	92
	93	func (g *gauge) With(metrics.Field) metrics.Gauge { return g }
	94
	95	func (g *gauge) Add(delta float64) {
96	96	// https://github.com/etsy/statsd/blob/master/docs/metric_types.md#gauges
97	97	sign := "+"
98	98	if delta < 0 {

101	101	g.g <- fmt.Sprintf("%s%f\|g", sign, delta)
102	102	}
103	103
104		func (g *statsdGauge) Set(value float64) {
	104	func (g *gauge) Set(value float64) {
105	105	atomic.StoreUint64(&g.lastValue, math.Float64bits(value))
106	106	g.g <- fmt.Sprintf("%f\|g", value)
107	107	}
108	108
109		func (g *statsdGauge) Get() float64 {
	109	func (g *gauge) Get() float64 {
110	110	return math.Float64frombits(atomic.LoadUint64(&g.lastValue))
111	111	}
112	112

137	137	return c
138	138	}
139	139
140		type statsdHistogram struct {
	140	type histogram struct {
141	141	key string
142	142	h chan string
143	143	}

166	166	// NewHistogramTick is the same as NewHistogram, but allows the user to pass a
167	167	// ticker channel instead of invoking time.Tick.
168	168	func NewHistogramTick(w io.Writer, key string, reportTicker <-chan time.Time) metrics.Histogram {
169		h := &statsdHistogram{
	169	h := &histogram{
170	170	key: key,
171	171	h: make(chan string),
172	172	}

174	174	return h
175	175	}
176	176
177		func (h *statsdHistogram) Name() string { return h.key }
178
179		func (h *statsdHistogram) With(metrics.Field) metrics.Histogram { return h }
180
181		func (h *statsdHistogram) Observe(value int64) {
	177	func (h *histogram) Name() string { return h.key }
	178
	179	func (h *histogram) With(metrics.Field) metrics.Histogram { return h }
	180
	181	func (h *histogram) Observe(value int64) {
182	182	h.h <- fmt.Sprintf("%d\|ms", value)
183	183	}
184	184
185		func (h *statsdHistogram) Distribution() ([]metrics.Bucket, []metrics.Quantile) {
	185	func (h *histogram) Distribution() ([]metrics.Bucket, []metrics.Quantile) {
186	186	// TODO(pb): no way to do this without introducing e.g. codahale/hdrhistogram
187	187	return []metrics.Bucket{}, []metrics.Quantile{}
188	188	}

+50

-0

tracing/opentracing/endpoint.go less more

	0	package opentracing
	1
	2	import (
	3	"github.com/opentracing/opentracing-go"
	4	otext "github.com/opentracing/opentracing-go/ext"
	5	"golang.org/x/net/context"
	6
	7	"github.com/go-kit/kit/endpoint"
	8	)
	9
	10	// TraceServer returns a Middleware that wraps the `next` Endpoint in an
	11	// OpenTracing Span called `operationName`.
	12	//
	13	// If `ctx` already has a Span, it is re-used and the operation name is
	14	// overwritten. If `ctx` does not yet have a Span, one is created here.
	15	func TraceServer(tracer opentracing.Tracer, operationName string) endpoint.Middleware {
	16	return func(next endpoint.Endpoint) endpoint.Endpoint {
	17	return func(ctx context.Context, request interface{}) (interface{}, error) {
	18	serverSpan := opentracing.SpanFromContext(ctx)
	19	if serverSpan == nil {
	20	// All we can do is create a new root span.
	21	serverSpan = tracer.StartSpan(operationName)
	22	} else {
	23	serverSpan.SetOperationName(operationName)
	24	}
	25	defer serverSpan.Finish()
	26	otext.SpanKind.Set(serverSpan, otext.SpanKindRPCServer)
	27	ctx = opentracing.ContextWithSpan(ctx, serverSpan)
	28	return next(ctx, request)
	29	}
	30	}
	31	}
	32
	33	// TraceClient returns a Middleware that wraps the `next` Endpoint in an
	34	// OpenTracing Span called `operationName`.
	35	func TraceClient(tracer opentracing.Tracer, operationName string) endpoint.Middleware {
	36	return func(next endpoint.Endpoint) endpoint.Endpoint {
	37	return func(ctx context.Context, request interface{}) (interface{}, error) {
	38	parentSpan := opentracing.SpanFromContext(ctx)
	39	clientSpan := tracer.StartSpanWithOptions(opentracing.StartSpanOptions{
	40	OperationName: operationName,
	41	Parent: parentSpan, // may be nil
	42	})
	43	defer clientSpan.Finish()
	44	otext.SpanKind.Set(clientSpan, otext.SpanKindRPCClient)
	45	ctx = opentracing.ContextWithSpan(ctx, clientSpan)
	46	return next(ctx, request)
	47	}
	48	}
	49	}

+96

-0

tracing/opentracing/endpoint_test.go less more

	0	package opentracing_test
	1
	2	import (
	3	"testing"
	4
	5	"github.com/opentracing/opentracing-go"
	6	"github.com/opentracing/opentracing-go/mocktracer"
	7	"golang.org/x/net/context"
	8
	9	"github.com/go-kit/kit/endpoint"
	10	kitot "github.com/go-kit/kit/tracing/opentracing"
	11	)
	12
	13	func TestTraceServer(t *testing.T) {
	14	tracer := mocktracer.New()
	15
	16	// Initialize the ctx with a nameless Span.
	17	contextSpan := tracer.StartSpan("").(*mocktracer.MockSpan)
	18	ctx := opentracing.ContextWithSpan(context.Background(), contextSpan)
	19
	20	var innerEndpoint endpoint.Endpoint
	21	innerEndpoint = func(context.Context, interface{}) (interface{}, error) {
	22	return struct{}{}, nil
	23	}
	24	tracedEndpoint := kitot.TraceServer(tracer, "testOp")(innerEndpoint)
	25	if _, err := tracedEndpoint(ctx, struct{}{}); err != nil {
	26	t.Fatal(err)
	27	}
	28	if want, have := 1, len(tracer.FinishedSpans); want != have {
	29	t.Fatalf("Want %v span(s), found %v", want, have)
	30	}
	31
	32	endpointSpan := tracer.FinishedSpans[0]
	33	// Test that the op name is updated
	34	if want, have := "testOp", endpointSpan.OperationName; want != have {
	35	t.Fatalf("Want %q, have %q", want, have)
	36	}
	37	// ... and that the ID is unmodified.
	38	if want, have := contextSpan.SpanID, endpointSpan.SpanID; want != have {
	39	t.Errorf("Want SpanID %q, have %q", want, have)
	40	}
	41	}
	42
	43	func TestTraceServerNoContextSpan(t *testing.T) {
	44	tracer := mocktracer.New()
	45
	46	var innerEndpoint endpoint.Endpoint
	47	innerEndpoint = func(context.Context, interface{}) (interface{}, error) {
	48	return struct{}{}, nil
	49	}
	50	tracedEndpoint := kitot.TraceServer(tracer, "testOp")(innerEndpoint)
	51	// Empty/background context:
	52	if _, err := tracedEndpoint(context.Background(), struct{}{}); err != nil {
	53	t.Fatal(err)
	54	}
	55	// tracedEndpoint created a new Span:
	56	if want, have := 1, len(tracer.FinishedSpans); want != have {
	57	t.Fatalf("Want %v span(s), found %v", want, have)
	58	}
	59
	60	endpointSpan := tracer.FinishedSpans[0]
	61	if want, have := "testOp", endpointSpan.OperationName; want != have {
	62	t.Fatalf("Want %q, have %q", want, have)
	63	}
	64	}
	65
	66	func TestTraceClient(t *testing.T) {
	67	tracer := mocktracer.New()
	68
	69	// Initialize the ctx with a parent Span.
	70	parentSpan := tracer.StartSpan("parent").(*mocktracer.MockSpan)
	71	defer parentSpan.Finish()
	72	ctx := opentracing.ContextWithSpan(context.Background(), parentSpan)
	73
	74	var innerEndpoint endpoint.Endpoint
	75	innerEndpoint = func(context.Context, interface{}) (interface{}, error) {
	76	return struct{}{}, nil
	77	}
	78	tracedEndpoint := kitot.TraceClient(tracer, "testOp")(innerEndpoint)
	79	if _, err := tracedEndpoint(ctx, struct{}{}); err != nil {
	80	t.Fatal(err)
	81	}
	82	// tracedEndpoint created a new Span:
	83	if want, have := 1, len(tracer.FinishedSpans); want != have {
	84	t.Fatalf("Want %v span(s), found %v", want, have)
	85	}
	86
	87	endpointSpan := tracer.FinishedSpans[0]
	88	if want, have := "testOp", endpointSpan.OperationName; want != have {
	89	t.Fatalf("Want %q, have %q", want, have)
	90	}
	91	// ... and that the parent ID is set appropriately.
	92	if want, have := parentSpan.SpanID, endpointSpan.ParentID; want != have {
	93	t.Errorf("Want ParentID %q, have %q", want, have)
	94	}
	95	}

+75

-0

tracing/opentracing/grpc.go less more

	0	package opentracing
	1
	2	import (
	3	"encoding/base64"
	4	"strings"
	5
	6	"github.com/opentracing/opentracing-go"
	7	"golang.org/x/net/context"
	8	"google.golang.org/grpc/metadata"
	9
	10	"github.com/go-kit/kit/log"
	11	)
	12
	13	// ToGRPCRequest returns a grpc RequestFunc that injects an OpenTracing Span
	14	// found in `ctx` into the grpc Metadata. If no such Span can be found, the
	15	// RequestFunc is a noop.
	16	//
	17	// The logger is used to report errors and may be nil.
	18	func ToGRPCRequest(tracer opentracing.Tracer, logger log.Logger) func(ctx context.Context, md *metadata.MD) context.Context {
	19	return func(ctx context.Context, md *metadata.MD) context.Context {
	20	if span := opentracing.SpanFromContext(ctx); span != nil {
	21	// There's nothing we can do with an error here.
	22	err := tracer.Inject(span, opentracing.TextMap, metadataReaderWriter{md})
	23	if err != nil && logger != nil {
	24	logger.Log("msg", "Inject failed", "err", err)
	25	}
	26	}
	27	return ctx
	28	}
	29	}
	30
	31	// FromGRPCRequest returns a grpc RequestFunc that tries to join with an
	32	// OpenTracing trace found in `req` and starts a new Span called
	33	// `operationName` accordingly. If no trace could be found in `req`, the Span
	34	// will be a trace root. The Span is incorporated in the returned Context and
	35	// can be retrieved with opentracing.SpanFromContext(ctx).
	36	//
	37	// The logger is used to report errors and may be nil.
	38	func FromGRPCRequest(tracer opentracing.Tracer, operationName string, logger log.Logger) func(ctx context.Context, md *metadata.MD) context.Context {
	39	return func(ctx context.Context, md *metadata.MD) context.Context {
	40	span, err := tracer.Join(operationName, opentracing.TextMap, metadataReaderWriter{md})
	41	if err != nil && logger != nil {
	42	logger.Log("msg", "Join failed", "err", err)
	43	}
	44	if span == nil {
	45	span = tracer.StartSpan(operationName)
	46	}
	47	return opentracing.ContextWithSpan(ctx, span)
	48	}
	49	}
	50
	51	// A type that conforms to opentracing.TextMapReader and
	52	// opentracing.TextMapWriter.
	53	type metadataReaderWriter struct {
	54	*metadata.MD
	55	}
	56
	57	func (w metadataReaderWriter) Set(key, val string) {
	58	key = strings.ToLower(key)
	59	if strings.HasSuffix(key, "-bin") {
	60	val = string(base64.StdEncoding.EncodeToString([]byte(val)))
	61	}
	62	(w.MD)[key] = append((w.MD)[key], val)
	63	}
	64
	65	func (w metadataReaderWriter) ForeachKey(handler func(key, val string) error) error {
	66	for k, vals := range *w.MD {
	67	for _, v := range vals {
	68	if err := handler(k, v); err != nil {
	69	return err
	70	}
	71	}
	72	}
	73	return nil
	74	}

+59

-0

tracing/opentracing/grpc_test.go less more

	0	package opentracing_test
	1
	2	import (
	3	"testing"
	4
	5	"github.com/opentracing/opentracing-go"
	6	"github.com/opentracing/opentracing-go/mocktracer"
	7	"golang.org/x/net/context"
	8	"google.golang.org/grpc/metadata"
	9
	10	kitot "github.com/go-kit/kit/tracing/opentracing"
	11	"github.com/go-kit/kit/transport/grpc"
	12	)
	13
	14	func TestTraceGRPCRequestRoundtrip(t *testing.T) {
	15	tracer := mocktracer.New()
	16
	17	// Initialize the ctx with a Span to inject.
	18	beforeSpan := tracer.StartSpan("to_inject").(*mocktracer.MockSpan)
	19	defer beforeSpan.Finish()
	20	beforeSpan.SetBaggageItem("baggage", "check")
	21	beforeCtx := opentracing.ContextWithSpan(context.Background(), beforeSpan)
	22
	23	var toGRPCFunc grpc.RequestFunc = kitot.ToGRPCRequest(tracer, nil)
	24	md := metadata.Pairs()
	25	// Call the RequestFunc.
	26	afterCtx := toGRPCFunc(beforeCtx, &md)
	27
	28	// The Span should not have changed.
	29	afterSpan := opentracing.SpanFromContext(afterCtx)
	30	if beforeSpan != afterSpan {
	31	t.Errorf("Should not swap in a new span")
	32	}
	33
	34	// No spans should have finished yet.
	35	if want, have := 0, len(tracer.FinishedSpans); want != have {
	36	t.Errorf("Want %v span(s), found %v", want, have)
	37	}
	38
	39	// Use FromGRPCRequest to verify that we can join with the trace given MD.
	40	var fromGRPCFunc grpc.RequestFunc = kitot.FromGRPCRequest(tracer, "joined", nil)
	41	joinCtx := fromGRPCFunc(afterCtx, &md)
	42	joinedSpan := opentracing.SpanFromContext(joinCtx).(*mocktracer.MockSpan)
	43
	44	if joinedSpan.SpanID == beforeSpan.SpanID {
	45	t.Error("SpanID should have changed", joinedSpan.SpanID, beforeSpan.SpanID)
	46	}
	47
	48	// Check that the parent/child relationship is as expected for the joined span.
	49	if want, have := beforeSpan.SpanID, joinedSpan.ParentID; want != have {
	50	t.Errorf("Want ParentID %q, have %q", want, have)
	51	}
	52	if want, have := "joined", joinedSpan.OperationName; want != have {
	53	t.Errorf("Want %q, have %q", want, have)
	54	}
	55	if want, have := "check", joinedSpan.BaggageItem("baggage"); want != have {
	56	t.Errorf("Want %q, have %q", want, have)
	57	}
	58	}

+75

-0

tracing/opentracing/http.go less more

	0	package opentracing
	1
	2	import (
	3	"net"
	4	"net/http"
	5	"strconv"
	6
	7	"github.com/opentracing/opentracing-go"
	8	"github.com/opentracing/opentracing-go/ext"
	9	"golang.org/x/net/context"
	10
	11	"github.com/go-kit/kit/log"
	12	kithttp "github.com/go-kit/kit/transport/http"
	13	)
	14
	15	// ToHTTPRequest returns an http RequestFunc that injects an OpenTracing Span
	16	// found in `ctx` into the http headers. If no such Span can be found, the
	17	// RequestFunc is a noop.
	18	//
	19	// The logger is used to report errors and may be nil.
	20	func ToHTTPRequest(tracer opentracing.Tracer, logger log.Logger) kithttp.RequestFunc {
	21	return func(ctx context.Context, req *http.Request) context.Context {
	22	// Try to find a Span in the Context.
	23	if span := opentracing.SpanFromContext(ctx); span != nil {
	24	// Add standard OpenTracing tags.
	25	ext.HTTPMethod.Set(span, req.URL.RequestURI())
	26	host, portString, err := net.SplitHostPort(req.URL.Host)
	27	if err == nil {
	28	ext.PeerHostname.Set(span, host)
	29	if port, err := strconv.Atoi(portString); err != nil {
	30	ext.PeerPort.Set(span, uint16(port))
	31	}
	32	} else {
	33	ext.PeerHostname.Set(span, req.URL.Host)
	34	}
	35
	36	// There's nothing we can do with any errors here.
	37	err = tracer.Inject(
	38	span,
	39	opentracing.TextMap,
	40	opentracing.HTTPHeaderTextMapCarrier(req.Header),
	41	)
	42	if err != nil && logger != nil {
	43	logger.Log("msg", "Join failed", "err", err)
	44	}
	45	}
	46	return ctx
	47	}
	48	}
	49
	50	// FromHTTPRequest returns an http RequestFunc that tries to join with an
	51	// OpenTracing trace found in `req` and starts a new Span called
	52	// `operationName` accordingly. If no trace could be found in `req`, the Span
	53	// will be a trace root. The Span is incorporated in the returned Context and
	54	// can be retrieved with opentracing.SpanFromContext(ctx).
	55	//
	56	// The logger is used to report errors and may be nil.
	57	func FromHTTPRequest(tracer opentracing.Tracer, operationName string, logger log.Logger) kithttp.RequestFunc {
	58	return func(ctx context.Context, req *http.Request) context.Context {
	59	// Try to join to a trace propagated in `req`. There's nothing we can
	60	// do with any errors here, so we ignore them.
	61	span, err := tracer.Join(
	62	operationName,
	63	opentracing.TextMap,
	64	opentracing.HTTPHeaderTextMapCarrier(req.Header),
	65	)
	66	if err != nil && logger != nil {
	67	logger.Log("msg", "Join failed", "err", err)
	68	}
	69	if span == nil {
	70	span = opentracing.StartSpan(operationName)
	71	}
	72	return opentracing.ContextWithSpan(ctx, span)
	73	}
	74	}

+59

-0

tracing/opentracing/http_test.go less more

	0	package opentracing_test
	1
	2	import (
	3	"net/http"
	4	"testing"
	5
	6	"github.com/opentracing/opentracing-go"
	7	"github.com/opentracing/opentracing-go/mocktracer"
	8	"golang.org/x/net/context"
	9
	10	kitot "github.com/go-kit/kit/tracing/opentracing"
	11	kithttp "github.com/go-kit/kit/transport/http"
	12	)
	13
	14	func TestTraceHTTPRequestRoundtrip(t *testing.T) {
	15	tracer := mocktracer.New()
	16
	17	// Initialize the ctx with a Span to inject.
	18	beforeSpan := tracer.StartSpan("to_inject").(*mocktracer.MockSpan)
	19	defer beforeSpan.Finish()
	20	beforeSpan.SetBaggageItem("baggage", "check")
	21	beforeCtx := opentracing.ContextWithSpan(context.Background(), beforeSpan)
	22
	23	var toHTTPFunc kithttp.RequestFunc = kitot.ToHTTPRequest(tracer, nil)
	24	req, _ := http.NewRequest("GET", "http://test.biz/url", nil)
	25	// Call the RequestFunc.
	26	afterCtx := toHTTPFunc(beforeCtx, req)
	27
	28	// The Span should not have changed.
	29	afterSpan := opentracing.SpanFromContext(afterCtx)
	30	if beforeSpan != afterSpan {
	31	t.Errorf("Should not swap in a new span")
	32	}
	33
	34	// No spans should have finished yet.
	35	if want, have := 0, len(tracer.FinishedSpans); want != have {
	36	t.Errorf("Want %v span(s), found %v", want, have)
	37	}
	38
	39	// Use FromHTTPRequest to verify that we can join with the trace given a req.
	40	var fromHTTPFunc kithttp.RequestFunc = kitot.FromHTTPRequest(tracer, "joined", nil)
	41	joinCtx := fromHTTPFunc(afterCtx, req)
	42	joinedSpan := opentracing.SpanFromContext(joinCtx).(*mocktracer.MockSpan)
	43
	44	if joinedSpan.SpanID == beforeSpan.SpanID {
	45	t.Error("SpanID should have changed", joinedSpan.SpanID, beforeSpan.SpanID)
	46	}
	47
	48	// Check that the parent/child relationship is as expected for the joined span.
	49	if want, have := beforeSpan.SpanID, joinedSpan.ParentID; want != have {
	50	t.Errorf("Want ParentID %q, have %q", want, have)
	51	}
	52	if want, have := "joined", joinedSpan.OperationName; want != have {
	53	t.Errorf("Want %q, have %q", want, have)
	54	}
	55	if want, have := "check", joinedSpan.BaggageItem("baggage"); want != have {
	56	t.Errorf("Want %q, have %q", want, have)
	57	}
	58	}

+2

-2

transport/http/request_response_funcs.go less more

6	6	)
7	7
8	8	// RequestFunc may take information from an HTTP request and put it into a
9		// request context. In Servers, BeforeFuncs are executed prior to invoking the
10		// endpoint. In Clients, BeforeFuncs are executed after creating the request
	9	// request context. In Servers, RequestFuncs are executed prior to invoking the
	10	// endpoint. In Clients, RequestFuncs are executed after creating the request
11	11	// but prior to invoking the HTTP client.
12	12	type RequestFunc func(context.Context, *http.Request) context.Context
13	13