Merge pull request #169 from unistack-org/tracer

tracer/wrapper: fix observers

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/options.go

@@ -1,6 +1,10 @@
 package tracer
 
-import "go.unistack.org/micro/v3/logger"
+import (
+	"context"
+
+	"go.unistack.org/micro/v3/logger"
+)
 
 // SpanOptions contains span option
 type SpanOptions struct {
@@ -28,6 +32,8 @@ type Options struct {
 	Logger logger.Logger
 	// Name of the tracer
 	Name string
+	// Context used to store custome tracer options
+	Context context.Context
 }
 
 // Option func signature

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, "type", "client")
 		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, "type", "client")
 		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, "type", "publisher")
 		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, "type", "server")
 		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, "type", "subscriber")
 		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, "type", "client")
 		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, "")
 	}
 	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, "")
 	}
 	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, "")
 	}
 	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, "")
 	}
 	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, "")
 	}
 	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, "")
 	}
 	defer sp.Finish()