Merge pull request 'tracer: tweaks for span tags and naming' (#239) from tracing into v3

Reviewed-on: #239

metadata/metadata.go

@@ -19,6 +19,8 @@ var (
 	HeaderTimeout = "Micro-Timeout"
 	// HeaderAuthorization specifies Authorization header
 	HeaderAuthorization = "Authorization"
+	// HeaderXRequestID specifies request id
+	HeaderXRequestID = "X-Request-Id"
 )
 
 // Metadata is our way of representing request headers internally.

tracer/options.go

@@ -91,11 +91,19 @@ type SpanOptions struct {
 type SpanOption func(o *SpanOptions)
 
 // EventOptions contains event options
-type EventOptions struct{}
+type EventOptions struct {
+	Labels []interface{}
+}
 
 // EventOption func signature
 type EventOption func(o *EventOptions)
 
+func WithEventLabels(labels ...interface{}) EventOption {
+	return func(o *EventOptions) {
+		o.Labels = labels
+	}
+}
+
 func WithSpanLabels(labels ...interface{}) SpanOption {
 	return func(o *SpanOptions) {
 		o.Labels = labels

tracer/tracer.go

@@ -3,6 +3,8 @@ package tracer // import "go.unistack.org/micro/v3/tracer"
 
 import (
 	"context"
+	"fmt"
+	"sort"
 )
 
 // DefaultTracer is the global default tracer
@@ -42,3 +44,37 @@ type Span interface {
 	// Kind returns span kind
 	Kind() SpanKind
 }
+
+// sort labels alphabeticaly by label name
+type byKey []interface{}
+
+func (k byKey) Len() int           { return len(k) / 2 }
+func (k byKey) Less(i, j int) bool { return fmt.Sprintf("%s", k[i*2]) < fmt.Sprintf("%s", k[j*2]) }
+func (k byKey) Swap(i, j int) {
+	k[i*2], k[j*2] = k[j*2], k[i*2]
+	k[i*2+1], k[j*2+1] = k[j*2+1], k[i*2+1]
+}
+
+func UniqLabels(labels []interface{}) []interface{} {
+	if len(labels)%2 == 1 {
+		labels = labels[:len(labels)-1]
+	}
+
+	if len(labels) > 2 {
+		sort.Sort(byKey(labels))
+
+		idx := 0
+		for {
+			if labels[idx] == labels[idx+2] {
+				copy(labels[idx:], labels[idx+2:])
+				labels = labels[:len(labels)-2]
+			} else {
+				idx += 2
+			}
+			if idx+2 >= len(labels) {
+				break
+			}
+		}
+	}
+	return labels
+}

tracer/tracer_test.go

@@ -0,0 +1,13 @@
+package tracer
+
+import (
+	"testing"
+)
+
+func TestUniqLabels(t *testing.T) {
+	labels := []interface{}{"key1", "val1", "key1", "val2"}
+	labels = UniqLabels(labels)
+	if labels[1] != "val2" {
+		t.Fatalf("UniqLabels not works")
+	}
+}

tracer/wrapper/wrapper.go

@@ -4,6 +4,7 @@ package wrapper // import "go.unistack.org/micro/v3/tracer/wrapper"
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	"go.unistack.org/micro/v3/client"
 	"go.unistack.org/micro/v3/metadata"
@@ -11,107 +12,87 @@ import (
 	"go.unistack.org/micro/v3/tracer"
 )
 
+var DefaultHeadersExctract = []string{metadata.HeaderXRequestID}
+
+func ExtractDefaultLabels(md metadata.Metadata) []interface{} {
+	labels := make([]interface{}, 0, len(DefaultHeadersExctract))
+	for _, k := range DefaultHeadersExctract {
+		if v, ok := md.Get(k); ok {
+			labels = append(labels, strings.ToLower(k), v)
+		}
+	}
+	return labels
+}
+
 var (
 	DefaultClientCallObserver = func(ctx context.Context, req client.Request, rsp interface{}, opts []client.CallOption, sp tracer.Span, err error) {
-		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)+1)
-			for k, v := range md {
-				labels = append(labels, k, v)
-			}
+			labels = append(labels, ExtractDefaultLabels(md)...)
 		}
 		if err != nil {
-			labels = append(labels, "error", err.Error())
 			sp.SetStatus(tracer.SpanStatusError, err.Error())
 		}
-		labels = append(labels, "kind", sp.Kind())
-		sp.SetLabels(labels...)
+		sp.AddLabels(labels...)
 	}
 
 	DefaultClientStreamObserver = func(ctx context.Context, req client.Request, opts []client.CallOption, stream client.Stream, sp tracer.Span, err error) {
-		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))
-			for k, v := range md {
-				labels = append(labels, k, v)
-			}
+			labels = append(labels, ExtractDefaultLabels(md)...)
 		}
 		if err != nil {
-			labels = append(labels, "error", err.Error())
 			sp.SetStatus(tracer.SpanStatusError, err.Error())
 		}
-		labels = append(labels, "kind", sp.Kind())
-		sp.SetLabels(labels...)
+		sp.AddLabels(labels...)
 	}
 
 	DefaultClientPublishObserver = func(ctx context.Context, msg client.Message, opts []client.PublishOption, sp tracer.Span, err error) {
-		sp.SetName(fmt.Sprintf("Publish %s", msg.Topic()))
 		var labels []interface{}
 		if md, ok := metadata.FromOutgoingContext(ctx); ok {
-			labels = make([]interface{}, 0, len(md))
-			for k, v := range md {
-				labels = append(labels, k, v)
-			}
+			labels = append(labels, ExtractDefaultLabels(md)...)
 		}
+		labels = append(labels, ExtractDefaultLabels(msg.Metadata())...)
 		if err != nil {
-			labels = append(labels, "error", err.Error())
 			sp.SetStatus(tracer.SpanStatusError, err.Error())
 		}
-		labels = append(labels, "kind", sp.Kind())
-		sp.SetLabels(labels...)
+		sp.AddLabels(labels...)
 	}
 
 	DefaultServerHandlerObserver = func(ctx context.Context, req server.Request, rsp interface{}, sp tracer.Span, err error) {
-		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))
-			for k, v := range md {
-				labels = append(labels, k, v)
-			}
+			labels = append(labels, ExtractDefaultLabels(md)...)
 		}
 		if err != nil {
-			labels = append(labels, "error", err.Error())
 			sp.SetStatus(tracer.SpanStatusError, err.Error())
 		}
-		labels = append(labels, "kind", sp.Kind())
-		sp.SetLabels(labels...)
+		sp.AddLabels(labels...)
 	}
 
 	DefaultServerSubscriberObserver = func(ctx context.Context, msg server.Message, sp tracer.Span, err error) {
-		sp.SetName(fmt.Sprintf("Subscriber %s", msg.Topic()))
 		var labels []interface{}
 		if md, ok := metadata.FromIncomingContext(ctx); ok {
-			labels = make([]interface{}, 0, len(md))
-			for k, v := range md {
-				labels = append(labels, k, v)
-			}
+			labels = append(labels, ExtractDefaultLabels(md)...)
 		}
+		labels = append(labels, ExtractDefaultLabels(msg.Header())...)
+
 		if err != nil {
-			labels = append(labels, "error", err.Error())
 			sp.SetStatus(tracer.SpanStatusError, err.Error())
 		}
-		labels = append(labels, "kind", sp.Kind())
-		sp.SetLabels(labels...)
+		sp.AddLabels(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("Call %s.%s", req.Service(), req.Method()))
+		sp.SetName(fmt.Sprintf("%s.%s call", req.Service(), req.Method()))
 		var labels []interface{}
 		if md, ok := metadata.FromOutgoingContext(ctx); ok {
-			labels = make([]interface{}, 0, len(md))
-			for k, v := range md {
-				labels = append(labels, k, v)
-			}
+			labels = append(labels, ExtractDefaultLabels(md)...)
 		}
 		if err != nil {
-			labels = append(labels, "error", err.Error())
 			sp.SetStatus(tracer.SpanStatusError, err.Error())
 		}
-		labels = append(labels, "kind", sp.Kind())
-		sp.SetLabels(labels...)
+		sp.AddLabels(labels...)
 	}
 
 	DefaultSkipEndpoints = []string{"Meter.Metrics", "Health.Live", "Health.Ready", "Health.Version"}
@@ -241,16 +222,22 @@ 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, "", tracer.WithSpanKind(tracer.SpanKindClient))
-	}
+	nctx, sp := ot.opts.Tracer.Start(ctx, fmt.Sprintf("%s.%s rpc-client", req.Service(), req.Method()),
+		tracer.WithSpanKind(tracer.SpanKindClient),
+		tracer.WithSpanLabels(
+			"rpc.service", req.Service(),
+			"rpc.method", req.Method(),
+			"rpc.flavor", "rpc",
+			"rpc.call", "/"+req.Service()+"/"+req.Endpoint(),
+			"rpc.call_type", "unary",
+		),
+	)
 	defer sp.Finish()
 
-	err := ot.Client.Call(ctx, req, rsp, opts...)
+	err := ot.Client.Call(nctx, req, rsp, opts...)
 
 	for _, o := range ot.opts.ClientCallObservers {
-		o(ctx, req, rsp, opts, sp, err)
+		o(nctx, req, rsp, opts, sp, err)
 	}
 
 	return err
@@ -264,32 +251,36 @@ 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, "", tracer.WithSpanKind(tracer.SpanKindClient))
-	}
+	nctx, sp := ot.opts.Tracer.Start(ctx, fmt.Sprintf("%s.%s rpc-client", req.Service(), req.Method()),
+		tracer.WithSpanKind(tracer.SpanKindClient),
+		tracer.WithSpanLabels(
+			"rpc.service", req.Service(),
+			"rpc.method", req.Method(),
+			"rpc.flavor", "rpc",
+			"rpc.call", "/"+req.Service()+"/"+req.Endpoint(),
+			"rpc.call_type", "stream",
+		),
+	)
 	defer sp.Finish()
 
-	stream, err := ot.Client.Stream(ctx, req, opts...)
+	stream, err := ot.Client.Stream(nctx, req, opts...)
 
 	for _, o := range ot.opts.ClientStreamObservers {
-		o(ctx, req, opts, stream, sp, err)
+		o(nctx, req, opts, stream, sp, err)
 	}
 
 	return stream, err
 }
 
 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, "", tracer.WithSpanKind(tracer.SpanKindProducer))
-	}
+	nctx, sp := ot.opts.Tracer.Start(ctx, msg.Topic()+" publish", tracer.WithSpanKind(tracer.SpanKindProducer))
 	defer sp.Finish()
-
-	err := ot.Client.Publish(ctx, msg, opts...)
+	sp.AddLabels("messaging.destination.name", msg.Topic())
+	sp.AddLabels("messaging.operation", "publish")
+	err := ot.Client.Publish(nctx, msg, opts...)
 
 	for _, o := range ot.opts.ClientPublishObservers {
-		o(ctx, msg, opts, sp, err)
+		o(nctx, msg, opts, sp, err)
 	}
 
 	return err
@@ -303,32 +294,41 @@ 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, "", tracer.WithSpanKind(tracer.SpanKindServer))
+	callType := "unary"
+	if req.Stream() {
+		callType = "stream"
 	}
+
+	nctx, sp := ot.opts.Tracer.Start(ctx, fmt.Sprintf("%s.%s rpc-server", req.Service(), req.Method()),
+		tracer.WithSpanKind(tracer.SpanKindServer),
+		tracer.WithSpanLabels(
+			"rpc.service", req.Service(),
+			"rpc.method", req.Method(),
+			"rpc.flavor", "rpc",
+			"rpc.call", "/"+req.Service()+"/"+req.Endpoint(),
+			"rpc.call_type", callType,
+		),
+	)
 	defer sp.Finish()
 
-	err := ot.serverHandler(ctx, req, rsp)
+	err := ot.serverHandler(nctx, req, rsp)
 
 	for _, o := range ot.opts.ServerHandlerObservers {
-		o(ctx, req, rsp, sp, err)
+		o(nctx, req, rsp, sp, err)
 	}
 
 	return err
 }
 
 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, "", tracer.WithSpanKind(tracer.SpanKindConsumer))
-	}
+	nctx, sp := ot.opts.Tracer.Start(ctx, msg.Topic()+" process", tracer.WithSpanKind(tracer.SpanKindConsumer))
 	defer sp.Finish()
-
-	err := ot.serverSubscriber(ctx, msg)
+	sp.AddLabels("messaging.operation", "process")
+	sp.AddLabels("messaging.source.name", msg.Topic())
+	err := ot.serverSubscriber(nctx, msg)
 
 	for _, o := range ot.opts.ServerSubscriberObservers {
-		o(ctx, msg, sp, err)
+		o(nctx, msg, sp, err)
 	}
 
 	return err
@@ -366,16 +366,23 @@ 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, "", tracer.WithSpanKind(tracer.SpanKindClient))
-	}
+	nctx, sp := ot.opts.Tracer.Start(ctx, fmt.Sprintf("%s.%s rpc-client", req.Service(), req.Method()),
+		tracer.WithSpanKind(tracer.SpanKindClient),
+		tracer.WithSpanLabels(
+			"rpc.service", req.Service(),
+			"rpc.method", req.Method(),
+			"rpc.flavor", "rpc",
+			"rpc.call", "/"+req.Service()+"/"+req.Endpoint(),
+			"rpc.call_type", "unary",
+		),
+	)
+
 	defer sp.Finish()
 
-	err := ot.clientCallFunc(ctx, addr, req, rsp, opts)
+	err := ot.clientCallFunc(nctx, addr, req, rsp, opts)
 
 	for _, o := range ot.opts.ClientCallFuncObservers {
-		o(ctx, addr, req, rsp, opts, sp, err)
+		o(nctx, addr, req, rsp, opts, sp, err)
 	}
 
 	return err