Merge pull request #169 from unistack-org/tracer

tracer/wrapper: fix observers
unistack-org · Jan 17, 2023 · 6f17fd8 · 6f17fd8
2 parents ccbf236 + fd93308
commit 6f17fd8
Showing 1 changed file with 21 additions and 15 deletions.
diff --git a/tracer/wrapper/wrapper.go b/tracer/wrapper/wrapper.go
@@ -13,22 +13,23 @@ 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)
 			}
 		}
 		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()