tracer: add missing UniqLabels function

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

config/options.go

@@ -197,7 +197,7 @@ func NewWatchOptions(opts ...options.Option) WatchOptions {
 }
 
 // Coalesce controls watch event combining
-func Coalesce(b bool) options.Option {
+func WatchCoalesce(b bool) options.Option {
 	return func(src interface{}) error {
 		return options.Set(src, b, ".Coalesce")
 	}

flow/options.go

@@ -134,3 +134,10 @@ func Fallback(step string) options.Option {
 		return options.Set(src, step, ".Fallback")
 	}
 }
+
+// ID sets the step ID
+func StepID(id string) options.Option {
+	return func(src interface{}) error {
+		return options.Set(src, id, ".ID")
+	}
+}

go.mod

@@ -8,13 +8,13 @@ require (
 	github.com/imdario/mergo v0.3.15
 	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/silas/dag v0.0.0-20220518035006-a7e85ada93c5
-	golang.org/x/sync v0.1.0
+	golang.org/x/sync v0.3.0
 	golang.org/x/sys v0.7.0
-	google.golang.org/grpc v1.54.0
-	google.golang.org/protobuf v1.30.0
+	google.golang.org/grpc v1.57.0
+	google.golang.org/protobuf v1.31.0
 )
 
 require (
 	github.com/golang/protobuf v1.5.3 // indirect
-	google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 // indirect
 )

go.sum

@@ -13,21 +13,21 @@ github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaR
 github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/silas/dag v0.0.0-20220518035006-a7e85ada93c5 h1:G/FZtUu7a6NTWl3KUHMV9jkLAh/Rvtf03NWMHaEDl+E=
 github.com/silas/dag v0.0.0-20220518035006-a7e85ada93c5/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
-golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ=
-golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
-golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/net v0.9.0 h1:aWJ/m6xSmxWBx+V0XRHTlrYrPG56jKsLdTFmsSsCzOM=
+golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E=
+golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
 golang.org/x/sys v0.7.0 h1:3jlCCIQZPdOYu1h8BkNvLz8Kgwtae2cagcG/VamtZRU=
 golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/text v0.8.0 h1:57P1ETyNKtuIjB4SRd15iJxuhj8Gc416Y78H3qgMh68=
+golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 h1:KpwkzHKEF7B9Zxg18WzOa7djJ+Ha5DzthMyZYQfEn2A=
-google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1/go.mod h1:nKE/iIaLqn2bQwXBg8f1g2Ylh6r5MN5CmZvuzZCgsCU=
-google.golang.org/grpc v1.54.0 h1:EhTqbhiYeixwWQtAEZAxmV9MGqcjEU2mFx52xCzNyag=
-google.golang.org/grpc v1.54.0/go.mod h1:PUSEXI6iWghWaB6lXM4knEgpJNu2qUcKfDtNci3EC2g=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 h1:0nDDozoAU19Qb2HwhXadU8OcsiO/09cnTqhUtq2MEOM=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19/go.mod h1:66JfowdXAEgad5O9NnYcsNPLCPZJD++2L9X0PCMODrA=
+google.golang.org/grpc v1.57.0 h1:kfzNeI/klCGD2YPMUlaGNT3pxvYfga7smW3Vth8Zsiw=
+google.golang.org/grpc v1.57.0/go.mod h1:Sd+9RMTACXwmub0zcNY2c4arhtrbBYD1AUHI/dt16Mo=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
-google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
+google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
+google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

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.

options/options.go

@@ -187,10 +187,3 @@ func Timeout(td time.Duration) Option {
 		return Set(src, td, ".Timeout")
 	}
 }
-
-// ID sets the step ID
-func StepID(id string) Option {
-	return func(src interface{}) error {
-		return Set(src, id, ".ID")
-	}
-}

tracer/options.go

@@ -2,11 +2,9 @@ package tracer
 
 import (
 	"context"
-	"reflect"
 
 	"go.unistack.org/micro/v4/logger"
 	"go.unistack.org/micro/v4/options"
-	rutil "go.unistack.org/micro/v4/util/reflect"
 )
 
 type SpanStatus int
@@ -91,23 +89,16 @@ type SpanOptions struct {
 }
 
 // EventOptions contains event options
-type EventOptions struct{}
+type EventOptions struct {
+	Labels []interface{}
+}
 
-func WithSpanLabels(ls ...interface{}) options.Option {
-	return func(src interface{}) error {
-		v, err := options.Get(src, ".Labels")
-		if err != nil {
-			return err
-		} else if rutil.IsZero(v) {
-			v = reflect.MakeSlice(reflect.TypeOf(v), 0, len(ls)).Interface()
-		}
-		cv := reflect.ValueOf(v)
-		for _, l := range ls {
-			reflect.Append(cv, reflect.ValueOf(l))
-		}
-		err = options.Set(src, cv, ".Labels")
-		return err
-	}
+func WithEventLabels(labels ...interface{}) options.Option {
+	return options.Labels(labels...)
+}
+
+func WithSpanLabels(labels ...interface{}) options.Option {
+	return options.Labels(labels...)
 }
 
 func WithSpanKind(k SpanKind) options.Option {

tracer/tracer.go

@@ -3,6 +3,8 @@ package tracer // import "go.unistack.org/micro/v4/tracer"
 
 import (
 	"context"
+	"fmt"
+	"sort"
 
 	"go.unistack.org/micro/v4/options"
 )
@@ -44,3 +46,36 @@ 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/wrapper/wrapper.go

@@ -4,80 +4,71 @@ package wrapper // import "go.unistack.org/micro/v4/tracer/wrapper"
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	"go.unistack.org/micro/v4/client"
 	"go.unistack.org/micro/v4/metadata"
+	"go.unistack.org/micro/v4/options"
 	"go.unistack.org/micro/v4/server"
 	"go.unistack.org/micro/v4/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()))
+	DefaultClientCallObserver = func(ctx context.Context, req client.Request, rsp interface{}, opts []options.Option, sp tracer.Span, err error) {
 		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()))
+	DefaultClientStreamObserver = func(ctx context.Context, req client.Request, opts []options.Option, stream client.Stream, sp tracer.Span, err error) {
 		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...)
 	}
 
 	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...)
 	}
 
 	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"}
@@ -91,8 +82,8 @@ type tWrapper struct {
 }
 
 type (
-	ClientCallObserver     func(context.Context, client.Request, interface{}, []client.CallOption, tracer.Span, error)
-	ClientStreamObserver   func(context.Context, client.Request, []client.CallOption, client.Stream, tracer.Span, error)
+	ClientCallObserver     func(context.Context, client.Request, interface{}, []options.Option, tracer.Span, error)
+	ClientStreamObserver   func(context.Context, client.Request, []options.Option, client.Stream, tracer.Span, error)
 	ClientCallFuncObserver func(context.Context, string, client.Request, interface{}, client.CallOptions, tracer.Span, error)
 	ServerHandlerObserver  func(context.Context, server.Request, interface{}, tracer.Span, error)
 )
@@ -176,7 +167,7 @@ func WithServerHandlerObservers(ob ...ServerHandlerObserver) Option {
 	}
 }
 
-func (ot *tWrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
+func (ot *tWrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...options.Option) error {
 	endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
 	for _, ep := range ot.opts.SkipEndpoints {
 		if ep == endpoint {
@@ -184,22 +175,28 @@ 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
 }
 
-func (ot *tWrapper) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
+func (ot *tWrapper) Stream(ctx context.Context, req client.Request, opts ...options.Option) (client.Stream, error) {
 	endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
 	for _, ep := range ot.opts.SkipEndpoints {
 		if ep == endpoint {
@@ -207,16 +204,22 @@ 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
@@ -230,16 +233,27 @@ 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
@@ -277,16 +291,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