Commit 4b50a0b1a8c03782d77d0dcb95a3a9a03501ad22 - golang-github-go-kit-kit

+2

-2

addsvc/main.go less more

97	97	*zipkinCollectorTimeout,
98	98	*zipkinCollectorBatchSize,
99	99	*zipkinCollectorBatchInterval,
	100	logger,
100	101	); err != nil {
101	102	logger.Log("err", err)
102	103	os.Exit(1)

104	105	}
105	106	zipkinMethodName := "add"
106	107	zipkinSpanFunc := zipkin.MakeNewSpanFunc(zipkinHostPort, *zipkinServiceName, zipkinMethodName)
107		zipkin.Log.Swap(logger) // log diagnostic/error details
108	108
109	109	// Our business and operational domain
110	110	var a Add = pureAdd

141	141	go func() {
142	142	ctx, cancel := context.WithCancel(root)
143	143	defer cancel()
144		before := []httptransport.BeforeFunc{zipkin.ToContext(zipkinSpanFunc)}
	144	before := []httptransport.BeforeFunc{zipkin.ToContext(zipkinSpanFunc, logger)}
145	145	after := []httptransport.AfterFunc{}
146	146	handler := makeHTTPBinding(ctx, e, before, after)
147	147	logger.Log("addr", *httpAddr, "transport", "HTTP/JSON")

+11

-3

tracing/zipkin/collector.go less more

9	9
10	10	"github.com/apache/thrift/lib/go/thrift"
11	11
	12	"github.com/go-kit/kit/log"
12	13	"github.com/go-kit/kit/tracing/zipkin/_thrift/gen-go/scribe"
13	14	)
14	15

30	31	nextSend time.Time
31	32	batchInterval time.Duration
32	33	batchSize int
	34	logger log.Logger
33	35	}
34	36
35		// NewScribeCollector returns a new Scribe-backed Collector, ready for use.
36		func NewScribeCollector(addr string, timeout time.Duration, batchSize int, batchInterval time.Duration) (Collector, error) {
	37	// NewScribeCollector returns a new Scribe-backed Collector. addr should be a
	38	// TCP endpoint of the form "host:port". timeout is passed to the Thrift dial
	39	// function NewTSocketFromAddrTimeout. batchSize and batchInterval control the
	40	// maximum size and interval of a batch of spans; as soon as either limit is
	41	// reached, the batch is sent. The logger is used to log errors, such as batch
	42	// send failures; users should provide an appropriate context, if desired.
	43	func NewScribeCollector(addr string, timeout time.Duration, batchSize int, batchInterval time.Duration, logger log.Logger) (Collector, error) {
37	44	factory := scribeClientFactory(addr, timeout)
38	45	client, err := factory()
39	46	if err != nil {

48	55	nextSend: time.Now().Add(batchInterval),
49	56	batchInterval: batchInterval,
50	57	batchSize: batchSize,
	58	logger: logger,
51	59	}
52	60	go c.loop()
53	61	return c, nil

81	89	case <-c.sendc:
82	90	c.nextSend = time.Now().Add(c.batchInterval)
83	91	if err := c.send(c.batch); err != nil {
84		Log.Log("err", err.Error())
	92	c.logger.Log("err", err.Error())
85	93	continue
86	94	}
87	95	c.batch = c.batch[:0]

+3

-1

tracing/zipkin/collector_test.go less more

2	2	import (
3	3	"encoding/base64"
4	4	"fmt"
	5	"io/ioutil"
5	6	"math/rand"
6	7	"net"
7	8	"sync"

10	11
11	12	"github.com/apache/thrift/lib/go/thrift"
12	13
	14	"github.com/go-kit/kit/log"
13	15	"github.com/go-kit/kit/tracing/zipkin"
14	16	"github.com/go-kit/kit/tracing/zipkin/_thrift/gen-go/scribe"
15	17	"github.com/go-kit/kit/tracing/zipkin/_thrift/gen-go/zipkincore"

20	22
21	23	timeout := time.Second
22	24	batchInterval := time.Millisecond
23		c, err := zipkin.NewScribeCollector(server.addr(), timeout, 0, batchInterval)
	25	c, err := zipkin.NewScribeCollector(server.addr(), timeout, 0, batchInterval, log.NewLogfmtLogger(ioutil.Discard))
24	26	if err != nil {
25	27	t.Fatal(err)
26	28	}

+6

-7

tracing/zipkin/span.go less more

29	29	//binaryAnnotations []BinaryAnnotation // TODO
30	30	}
31	31
32		// NewSpan returns a new Span object ready for use.
	32	// NewSpan returns a new Span, which can be annotated and collected by a
	33	// collector. Spans are passed through the request context to each middleware
	34	// under the SpanContextKey.
33	35	func NewSpan(hostport, serviceName, methodName string, traceID, spanID, parentSpanID int64) *Span {
34	36	return &Span{
35	37	host: makeEndpoint(hostport, serviceName),

40	42	}
41	43	}
42	44
43		// makeEndpoint will return a nil Endpoint if the input parameters are
44		// malformed.
	45	// makeEndpoint takes the hostport and service name that represent this Zipkin
	46	// service, and returns an endpoint that's embedded into the Zipkin core Span
	47	// type. It will return a nil endpoint if the input parameters are malformed.
45	48	func makeEndpoint(hostport, serviceName string) *zipkincore.Endpoint {
46	49	host, port, err := net.SplitHostPort(hostport)
47	50	if err != nil {
48		Log.Log("hostport", hostport, "err", err)
49	51	return nil
50	52	}
51	53	addrs, err := net.LookupIP(host)
52	54	if err != nil {
53		Log.Log("host", host, "err", err)
54	55	return nil
55	56	}
56	57	if len(addrs) <= 0 {
57		Log.Log("host", host, "err", "no IPs")
58	58	return nil
59	59	}
60	60	portInt, err := strconv.ParseInt(port, 10, 16)
61	61	if err != nil {
62		Log.Log("port", port, "err", err)
63	62	return nil
64	63	}
65	64	endpoint := zipkincore.NewEndpoint()

+11

-16

tracing/zipkin/zipkin.go less more

20	20	// • http://www.slideshare.net/johanoskarsson/zipkin-runtime-open-house
21	21	// • https://groups.google.com/forum/#!topic/zipkin-user/KilwtSA0g1k
22	22	// • https://gist.github.com/yoavaa/3478d3a0df666f21a98c
23
24		// Log is used to report diagnostic information. To enable it, swap in your
25		// application's logger.
26		var Log log.SwapLogger
27	23
28	24	const (
29	25	// https://github.com/racker/tryfer#headers

94	90	// ToContext returns a function that satisfies transport/http.BeforeFunc. It
95	91	// takes a Zipkin span from the incoming HTTP request, and saves it in the
96	92	// request context. It's designed to be wired into a server's HTTP transport
97		// Before stack.
98		func ToContext(newSpan NewSpanFunc) func(ctx context.Context, r *http.Request) context.Context {
	93	// Before stack. The logger is used to report errors.
	94	func ToContext(newSpan NewSpanFunc, logger log.Logger) func(ctx context.Context, r *http.Request) context.Context {
99	95	return func(ctx context.Context, r *http.Request) context.Context {
100		return context.WithValue(ctx, SpanContextKey, fromHTTP(newSpan, r))
	96	return context.WithValue(ctx, SpanContextKey, fromHTTP(newSpan, r, logger))
101	97	}
102	98	}
103	99

125	121	}
126	122	}
127	123
128		func fromHTTP(newSpan NewSpanFunc, r http.Request) Span {
	124	func fromHTTP(newSpan NewSpanFunc, r http.Request, logger log.Logger) Span {
129	125	traceIDStr := r.Header.Get(traceIDHTTPHeader)
130	126	if traceIDStr == "" {
131		Log.Log("debug", "make new span")
132	127	return newSpan(newID(), newID(), 0) // normal; just make a new one
133	128	}
134	129	traceID, err := strconv.ParseInt(traceIDStr, 16, 64)
135	130	if err != nil {
136		Log.Log(traceIDHTTPHeader, traceIDStr, "err", err)
	131	logger.Log(traceIDHTTPHeader, traceIDStr, "err", err)
137	132	return newSpan(newID(), newID(), 0)
138	133	}
139	134	spanIDStr := r.Header.Get(spanIDHTTPHeader)
140	135	if spanIDStr == "" {
141		Log.Log("msg", "trace ID without span ID") // abnormal
142		spanIDStr = strconv.FormatInt(newID(), 64) // deal with it
	136	logger.Log("msg", "trace ID without span ID") // abnormal
	137	spanIDStr = strconv.FormatInt(newID(), 64) // deal with it
143	138	}
144	139	spanID, err := strconv.ParseInt(spanIDStr, 16, 64)
145	140	if err != nil {
146		Log.Log(spanIDHTTPHeader, spanIDStr, "err", err) // abnormal
147		spanID = newID() // deal with it
	141	logger.Log(spanIDHTTPHeader, spanIDStr, "err", err) // abnormal
	142	spanID = newID() // deal with it
148	143	}
149	144	parentSpanIDStr := r.Header.Get(parentSpanIDHTTPHeader)
150	145	if parentSpanIDStr == "" {

152	147	}
153	148	parentSpanID, err := strconv.ParseInt(parentSpanIDStr, 16, 64)
154	149	if err != nil {
155		Log.Log(parentSpanIDHTTPHeader, parentSpanIDStr, "err", err) // abnormal
156		parentSpanID = 0 // the only way to deal with it
	150	logger.Log(parentSpanIDHTTPHeader, parentSpanIDStr, "err", err) // abnormal
	151	parentSpanID = 0 // the only way to deal with it
157	152	}
158	153	return newSpan(traceID, spanID, parentSpanID)
159	154	}

+3

-1

tracing/zipkin/zipkin_test.go less more

1	1
2	2	import (
3	3	"fmt"
	4	"io/ioutil"
4	5	"net/http"
5	6	"reflect"
6	7	"runtime"

11	12	"golang.org/x/net/context"
12	13
13	14	"github.com/go-kit/kit/endpoint"
	15	"github.com/go-kit/kit/log"
14	16	"github.com/go-kit/kit/tracing/zipkin"
15	17	)
16	18

30	32	r.Header.Set("X-B3-ParentSpanId", strconv.FormatInt(parentSpanID, 16))
31	33
32	34	newSpan := zipkin.MakeNewSpanFunc(hostport, serviceName, methodName)
33		toContext := zipkin.ToContext(newSpan)
	35	toContext := zipkin.ToContext(newSpan, log.NewLogfmtLogger(ioutil.Discard))
34	36
35	37	ctx := toContext(context.Background(), r)
36	38	val := ctx.Value(zipkin.SpanContextKey)