Merge pull request #170 from unistack-org/tracer

tracer: add span kind

tracer/context.go

@@ -46,3 +46,13 @@ func NewSpanContext(ctx context.Context, span Span) context.Context {
 	}
 	return context.WithValue(ctx, spanKey{}, span)
 }
+
+// SetOption returns a function to setup a context with given value
+func SetOption(k, v interface{}) Option {
+	return func(o *Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, k, v)
+	}
+}

tracer/noop.go

@@ -13,6 +13,7 @@ func (t *noopTracer) Start(ctx context.Context, name string, opts ...SpanOption)
 		name:   name,
 		ctx:    ctx,
 		tracer: t,
+		opts:   NewSpanOptions(opts...),
 	}
 	if span.ctx == nil {
 		span.ctx = context.Background()
@@ -64,6 +65,10 @@ func (s *noopSpan) AddLabels(labels ...interface{}) {
 	s.opts.Labels = append(s.opts.Labels, labels...)
 }
 
+func (s *noopSpan) Kind() SpanKind {
+	return s.opts.Kind
+}
+
 // NewTracer returns new memory tracer
 func NewTracer(opts ...Option) Tracer {
 	return &noopTracer{

tracer/options.go

@@ -1,10 +1,63 @@
 package tracer
 
-import "go.unistack.org/micro/v3/logger"
+import (
+	"context"
+
+	"go.unistack.org/micro/v3/logger"
+)
+
+type SpanKind int
+
+const (
+	// SpanKindUnspecified is an unspecified SpanKind and is not a valid
+	// SpanKind. SpanKindUnspecified should be replaced with SpanKindInternal
+	// if it is received.
+	SpanKindUnspecified SpanKind = 0
+	// SpanKindInternal is a SpanKind for a Span that represents an internal
+	// operation within an application.
+	SpanKindInternal SpanKind = 1
+	// SpanKindServer is a SpanKind for a Span that represents the operation
+	// of handling a request from a client.
+	SpanKindServer SpanKind = 2
+	// SpanKindClient is a SpanKind for a Span that represents the operation
+	// of client making a request to a server.
+	SpanKindClient SpanKind = 3
+	// SpanKindProducer is a SpanKind for a Span that represents the operation
+	// of a producer sending a message to a message broker. Unlike
+	// SpanKindClient and SpanKindServer, there is often no direct
+	// relationship between this kind of Span and a SpanKindConsumer kind. A
+	// SpanKindProducer Span will end once the message is accepted by the
+	// message broker which might not overlap with the processing of that
+	// message.
+	SpanKindProducer SpanKind = 4
+	// SpanKindConsumer is a SpanKind for a Span that represents the operation
+	// of a consumer receiving a message from a message broker. Like
+	// SpanKindProducer Spans, there is often no direct relationship between
+	// this Span and the Span that produced the message.
+	SpanKindConsumer SpanKind = 5
+)
+
+func (sk SpanKind) String() string {
+	switch sk {
+	case SpanKindInternal:
+		return "internal"
+	case SpanKindServer:
+		return "server"
+	case SpanKindClient:
+		return "client"
+	case SpanKindProducer:
+		return "producer"
+	case SpanKindConsumer:
+		return "consumer"
+	default:
+		return "unspecified"
+	}
+}
 
 // SpanOptions contains span option
 type SpanOptions struct {
 	Labels []interface{}
+	Kind   SpanKind
 }
 
 // SpanOption func signature
@@ -16,14 +69,22 @@ type EventOptions struct{}
 // EventOption func signature
 type EventOption func(o *EventOptions)
 
-func SpanLabels(labels ...interface{}) SpanOption {
+func WithSpanLabels(labels ...interface{}) SpanOption {
 	return func(o *SpanOptions) {
 		o.Labels = labels
 	}
 }
 
+func WithSpanKind(k SpanKind) SpanOption {
+	return func(o *SpanOptions) {
+		o.Kind = k
+	}
+}
+
 // Options struct
 type Options struct {
+	// Context used to store custome tracer options
+	Context context.Context
 	// Logger used for logging
 	Logger logger.Logger
 	// Name of the tracer
@@ -40,6 +101,17 @@ func Logger(l logger.Logger) Option {
 	}
 }
 
+// NewSpanOptions returns default SpanOptions
+func NewSpanOptions(opts ...SpanOption) SpanOptions {
+	options := SpanOptions{
+		Kind: SpanKindInternal,
+	}
+	for _, o := range opts {
+		o(&options)
+	}
+	return options
+}
+
 // NewOptions returns default options
 func NewOptions(opts ...Option) Options {
 	options := Options{

tracer/tracer.go

@@ -33,4 +33,6 @@ type Span interface {
 	SetLabels(labels ...interface{})
 	// AddLabels append the span labels
 	AddLabels(labels ...interface{})
+	// Kind returns span kind
+	Kind() SpanKind
 }

tracer/wrapper/wrapper.go

@@ -13,10 +13,10 @@ import (
 
 var (
 	DefaultClientCallObserver = func(ctx context.Context, req client.Request, rsp interface{}, opts []client.CallOption, sp tracer.Span, err error) {
-		sp.SetName(fmt.Sprintf("%s.%s", req.Service(), req.Endpoint()))
+		sp.SetName(fmt.Sprintf("Call %s.%s", req.Service(), req.Method()))
 		var labels []interface{}
 		if md, ok := metadata.FromOutgoingContext(ctx); ok {
-			labels = make([]interface{}, 0, len(md))
+			labels = make([]interface{}, 0, len(md)+1)
 			for k, v := range md {
 				labels = append(labels, k, v)
 			}
@@ -24,11 +24,12 @@ var (
 		if err != nil {
 			labels = append(labels, "error", true)
 		}
+		labels = append(labels, "kind", sp.Kind())
 		sp.SetLabels(labels...)
 	}
 
 	DefaultClientStreamObserver = func(ctx context.Context, req client.Request, opts []client.CallOption, stream client.Stream, sp tracer.Span, err error) {
-		sp.SetName(fmt.Sprintf("%s.%s", req.Service(), req.Endpoint()))
+		sp.SetName(fmt.Sprintf("Stream %s.%s", req.Service(), req.Method()))
 		var labels []interface{}
 		if md, ok := metadata.FromOutgoingContext(ctx); ok {
 			labels = make([]interface{}, 0, len(md))
@@ -39,11 +40,12 @@ var (
 		if err != nil {
 			labels = append(labels, "error", true)
 		}
+		labels = append(labels, "kind", sp.Kind())
 		sp.SetLabels(labels...)
 	}
 
 	DefaultClientPublishObserver = func(ctx context.Context, msg client.Message, opts []client.PublishOption, sp tracer.Span, err error) {
-		sp.SetName(fmt.Sprintf("Pub to %s", msg.Topic()))
+		sp.SetName(fmt.Sprintf("Publish %s", msg.Topic()))
 		var labels []interface{}
 		if md, ok := metadata.FromOutgoingContext(ctx); ok {
 			labels = make([]interface{}, 0, len(md))
@@ -54,11 +56,12 @@ var (
 		if err != nil {
 			labels = append(labels, "error", true)
 		}
+		labels = append(labels, "kind", sp.Kind())
 		sp.SetLabels(labels...)
 	}
 
 	DefaultServerHandlerObserver = func(ctx context.Context, req server.Request, rsp interface{}, sp tracer.Span, err error) {
-		sp.SetName(fmt.Sprintf("%s.%s", req.Service(), req.Endpoint()))
+		sp.SetName(fmt.Sprintf("Handler %s.%s", req.Service(), req.Method()))
 		var labels []interface{}
 		if md, ok := metadata.FromIncomingContext(ctx); ok {
 			labels = make([]interface{}, 0, len(md))
@@ -69,11 +72,12 @@ var (
 		if err != nil {
 			labels = append(labels, "error", true)
 		}
+		labels = append(labels, "kind", sp.Kind())
 		sp.SetLabels(labels...)
 	}
 
 	DefaultServerSubscriberObserver = func(ctx context.Context, msg server.Message, sp tracer.Span, err error) {
-		sp.SetName(fmt.Sprintf("Sub from %s", msg.Topic()))
+		sp.SetName(fmt.Sprintf("Subscriber %s", msg.Topic()))
 		var labels []interface{}
 		if md, ok := metadata.FromIncomingContext(ctx); ok {
 			labels = make([]interface{}, 0, len(md))
@@ -84,11 +88,12 @@ var (
 		if err != nil {
 			labels = append(labels, "error", true)
 		}
+		labels = append(labels, "kind", sp.Kind())
 		sp.SetLabels(labels...)
 	}
 
 	DefaultClientCallFuncObserver = func(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions, sp tracer.Span, err error) {
-		sp.SetName(fmt.Sprintf("%s.%s", req.Service(), req.Endpoint()))
+		sp.SetName(fmt.Sprintf("Call %s.%s", req.Service(), req.Method()))
 		var labels []interface{}
 		if md, ok := metadata.FromOutgoingContext(ctx); ok {
 			labels = make([]interface{}, 0, len(md))
@@ -99,6 +104,7 @@ var (
 		if err != nil {
 			labels = append(labels, "error", true)
 		}
+		labels = append(labels, "kind", sp.Kind())
 		sp.SetLabels(labels...)
 	}
 
@@ -231,7 +237,7 @@ func (ot *tWrapper) Call(ctx context.Context, req client.Request, rsp interface{
 
 	sp, ok := tracer.SpanFromContext(ctx)
 	if !ok {
-		ctx, sp = ot.opts.Tracer.Start(ctx, endpoint)
+		ctx, sp = ot.opts.Tracer.Start(ctx, "", tracer.WithSpanKind(tracer.SpanKindClient))
 	}
 	defer sp.Finish()
 
@@ -254,7 +260,7 @@ func (ot *tWrapper) Stream(ctx context.Context, req client.Request, opts ...clie
 
 	sp, ok := tracer.SpanFromContext(ctx)
 	if !ok {
-		ctx, sp = ot.opts.Tracer.Start(ctx, endpoint)
+		ctx, sp = ot.opts.Tracer.Start(ctx, "", tracer.WithSpanKind(tracer.SpanKindClient))
 	}
 	defer sp.Finish()
 
@@ -270,7 +276,7 @@ func (ot *tWrapper) Stream(ctx context.Context, req client.Request, opts ...clie
 func (ot *tWrapper) Publish(ctx context.Context, msg client.Message, opts ...client.PublishOption) error {
 	sp, ok := tracer.SpanFromContext(ctx)
 	if !ok {
-		ctx, sp = ot.opts.Tracer.Start(ctx, msg.Topic())
+		ctx, sp = ot.opts.Tracer.Start(ctx, "", tracer.WithSpanKind(tracer.SpanKindProducer))
 	}
 	defer sp.Finish()
 
@@ -284,7 +290,7 @@ func (ot *tWrapper) Publish(ctx context.Context, msg client.Message, opts ...cli
 }
 
 func (ot *tWrapper) ServerHandler(ctx context.Context, req server.Request, rsp interface{}) error {
-	endpoint := req.Endpoint()
+	endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Method())
 	for _, ep := range ot.opts.SkipEndpoints {
 		if ep == endpoint {
 			return ot.serverHandler(ctx, req, rsp)
@@ -293,7 +299,7 @@ func (ot *tWrapper) ServerHandler(ctx context.Context, req server.Request, rsp i
 
 	sp, ok := tracer.SpanFromContext(ctx)
 	if !ok {
-		ctx, sp = ot.opts.Tracer.Start(ctx, fmt.Sprintf("%s.%s", req.Service(), req.Endpoint()))
+		ctx, sp = ot.opts.Tracer.Start(ctx, "", tracer.WithSpanKind(tracer.SpanKindServer))
 	}
 	defer sp.Finish()
 
@@ -309,7 +315,7 @@ func (ot *tWrapper) ServerHandler(ctx context.Context, req server.Request, rsp i
 func (ot *tWrapper) ServerSubscriber(ctx context.Context, msg server.Message) error {
 	sp, ok := tracer.SpanFromContext(ctx)
 	if !ok {
-		ctx, sp = ot.opts.Tracer.Start(ctx, msg.Topic())
+		ctx, sp = ot.opts.Tracer.Start(ctx, "", tracer.WithSpanKind(tracer.SpanKindConsumer))
 	}
 	defer sp.Finish()
 
@@ -347,7 +353,7 @@ func NewClientCallWrapper(opts ...Option) client.CallWrapper {
 }
 
 func (ot *tWrapper) ClientCallFunc(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
-	endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
+	endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Method())
 	for _, ep := range ot.opts.SkipEndpoints {
 		if ep == endpoint {
 			return ot.ClientCallFunc(ctx, addr, req, rsp, opts)
@@ -356,7 +362,7 @@ func (ot *tWrapper) ClientCallFunc(ctx context.Context, addr string, req client.
 
 	sp, ok := tracer.SpanFromContext(ctx)
 	if !ok {
-		ctx, sp = ot.opts.Tracer.Start(ctx, endpoint)
+		ctx, sp = ot.opts.Tracer.Start(ctx, "", tracer.WithSpanKind(tracer.SpanKindClient))
 	}
 	defer sp.Finish()