Merge pull request 'client/noop: fix metadata overwrite' (#284) from noopfix into v3

Reviewed-on: #284

client/noop.go

@@ -486,10 +486,18 @@ func (n *noopClient) publish(ctx context.Context, ps []Message, opts ...PublishO
 	msgs := make([]*broker.Message, 0, len(ps))
 
 	for _, p := range ps {
-		md, ok := metadata.FromOutgoingContext(ctx)
+		omd, ok := metadata.FromOutgoingContext(ctx)
 		if !ok {
-			md = metadata.New(0)
+			omd = metadata.New(0)
 		}
+
+		md := metadata.Copy(omd)
+		iter := p.Metadata().Iterator()
+		var k, v string
+		for iter.Next(&k, &v) {
+			md.Set(k, v)
+		}
+
 		md[metadata.HeaderContentType] = p.ContentType()
 
 		topic := p.Topic()

database/dsn.go

@@ -0,0 +1,157 @@
+package database
+
+import (
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"net/url"
+	"strings"
+)
+
+var (
+	ErrInvalidDSNAddr      = errors.New("invalid dsn addr")
+	ErrInvalidDSNUnescaped = errors.New("dsn must be escaped")
+	ErrInvalidDSNNoSlash   = errors.New("dsn must contains slash")
+)
+
+type Config struct {
+	TLSConfig *tls.Config
+	Username  string
+	Password  string
+	Scheme    string
+	Host      string
+	Port      string
+	Database  string
+	Params    []string
+}
+
+func (cfg *Config) FormatDSN() string {
+	var s strings.Builder
+
+	if len(cfg.Scheme) > 0 {
+		s.WriteString(cfg.Scheme + "://")
+	}
+	// [username[:password]@]
+	if len(cfg.Username) > 0 {
+		s.WriteString(cfg.Username)
+		if len(cfg.Password) > 0 {
+			s.WriteByte(':')
+			s.WriteString(url.PathEscape(cfg.Password))
+		}
+		s.WriteByte('@')
+	}
+
+	// [host:port]
+	if len(cfg.Host) > 0 {
+		s.WriteString(cfg.Host)
+		if len(cfg.Port) > 0 {
+			s.WriteByte(':')
+			s.WriteString(cfg.Port)
+		}
+	}
+
+	// /dbname
+	s.WriteByte('/')
+	s.WriteString(url.PathEscape(cfg.Database))
+
+	for i := 0; i < len(cfg.Params); i += 2 {
+		if i == 0 {
+			s.WriteString("?")
+		} else {
+			s.WriteString("&")
+		}
+		s.WriteString(cfg.Params[i])
+		s.WriteString("=")
+		s.WriteString(cfg.Params[i+1])
+	}
+
+	return s.String()
+}
+
+func ParseDSN(dsn string) (*Config, error) {
+	cfg := &Config{}
+
+	// [user[:password]@][net[(addr)]]/dbname[?param1=value1&paramN=valueN]
+	// Find last '/' that goes before dbname
+	foundSlash := false
+	for i := len(dsn) - 1; i >= 0; i-- {
+		if dsn[i] == '/' {
+			foundSlash = true
+			var j, k int
+
+			// left part is empty if i <= 0
+			if i > 0 {
+				// Find the first ':' in dsn
+				for j = i; j >= 0; j-- {
+					if dsn[j] == ':' {
+						cfg.Scheme = dsn[0:j]
+					}
+				}
+
+				// [username[:password]@][host]
+				// Find the last '@' in dsn[:i]
+				for j = i; j >= 0; j-- {
+					if dsn[j] == '@' {
+						// username[:password]
+						// Find the second ':' in dsn[:j]
+						for k = 0; k < j; k++ {
+							if dsn[k] == ':' {
+								if cfg.Scheme == dsn[:k] {
+									continue
+								}
+								var err error
+								cfg.Password, err = url.PathUnescape(dsn[k+1 : j])
+								if err != nil {
+									return nil, err
+								}
+								break
+							}
+						}
+						cfg.Username = dsn[len(cfg.Scheme)+3 : k]
+						break
+					}
+				}
+
+				for k = j + 1; k < i; k++ {
+					if dsn[k] == ':' {
+						cfg.Host = dsn[j+1 : k]
+						cfg.Port = dsn[k+1 : i]
+						break
+					}
+				}
+
+			}
+
+			// dbname[?param1=value1&...&paramN=valueN]
+			// Find the first '?' in dsn[i+1:]
+			for j = i + 1; j < len(dsn); j++ {
+				if dsn[j] == '?' {
+					parts := strings.Split(dsn[j+1:], "&")
+					cfg.Params = make([]string, 0, len(parts)*2)
+					for _, p := range parts {
+						k, v, found := strings.Cut(p, "=")
+						if !found {
+							continue
+						}
+						cfg.Params = append(cfg.Params, k, v)
+					}
+
+					break
+				}
+			}
+			var err error
+			dbname := dsn[i+1 : j]
+			if cfg.Database, err = url.PathUnescape(dbname); err != nil {
+				return nil, fmt.Errorf("invalid dbname %q: %w", dbname, err)
+			}
+
+			break
+		}
+	}
+
+	if !foundSlash && len(dsn) > 0 {
+		return nil, ErrInvalidDSNNoSlash
+	}
+
+	return cfg, nil
+}

database/dsn_test.go

@@ -0,0 +1,31 @@
+package database
+
+import (
+	"net/url"
+	"testing"
+)
+
+func TestParseDSN(t *testing.T) {
+	cfg, err := ParseDSN("postgres://username:p@ssword#@host:12345/dbname?key1=val2&key2=val2")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if cfg.Password != "p@ssword#" {
+		t.Fatalf("parsing error")
+	}
+}
+
+func TestFormatDSN(t *testing.T) {
+	src := "postgres://username:p@ssword#@host:12345/dbname?key1=val2&key2=val2"
+	cfg, err := ParseDSN(src)
+	if err != nil {
+		t.Fatal(err)
+	}
+	dst, err := url.PathUnescape(cfg.FormatDSN())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if src != dst {
+		t.Fatalf("\n%s\n%s", src, dst)
+	}
+}

errors/errors.go

@@ -4,11 +4,17 @@ package errors // import "go.unistack.org/micro/v3/errors"
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"strconv"
 	"strings"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 var (
@@ -340,3 +346,135 @@ func addslashes(str string) string {
 	}
 	return buf.String()
 }
+
+type retryableError struct {
+	err error
+}
+
+// Retryable returns error that can be retried later
+func Retryable(err error) error {
+	return &retryableError{err: err}
+}
+
+type IsRetryableFunc func(error) bool
+
+var (
+	RetrayableOracleErrors = []IsRetryableFunc{
+		func(err error) bool {
+			errmsg := err.Error()
+			switch {
+			case strings.Contains(errmsg, `ORA-`):
+				return true
+			case strings.Contains(errmsg, `can not assign`):
+				return true
+			case strings.Contains(errmsg, `can't assign`):
+				return true
+			}
+			return false
+		},
+	}
+	RetrayablePostgresErrors = []IsRetryableFunc{
+		func(err error) bool {
+			errmsg := err.Error()
+			switch {
+			case strings.Contains(errmsg, `number of field descriptions must equal number of`):
+				return true
+			case strings.Contains(errmsg, `not a pointer`):
+				return true
+			case strings.Contains(errmsg, `values, but dst struct has only`):
+				return true
+			case strings.Contains(errmsg, `struct doesn't have corresponding row field`):
+				return true
+			case strings.Contains(errmsg, `cannot find field`):
+				return true
+			case strings.Contains(errmsg, `cannot scan`) || strings.Contains(errmsg, `cannot convert`):
+				return true
+			case strings.Contains(errmsg, `failed to connect to`):
+				return true
+			}
+			return false
+		},
+	}
+	RetryableMicroErrors = []IsRetryableFunc{
+		func(err error) bool {
+			switch verr := err.(type) {
+			case *Error:
+				switch verr.Code {
+				case 401, 403, 408, 500, 501, 502, 503, 504:
+					return true
+				default:
+					return false
+				}
+			case *retryableError:
+				return true
+			}
+			return false
+		},
+	}
+	RetryableGoErrors = []IsRetryableFunc{
+		func(err error) bool {
+			switch verr := err.(type) {
+			case interface{ SafeToRetry() bool }:
+				return verr.SafeToRetry()
+			case interface{ Timeout() bool }:
+				return verr.Timeout()
+			}
+			switch {
+			case errors.Is(err, io.EOF), errors.Is(err, io.ErrUnexpectedEOF):
+				return true
+			case errors.Is(err, context.DeadlineExceeded):
+				return true
+			case errors.Is(err, io.ErrClosedPipe), errors.Is(err, io.ErrShortBuffer), errors.Is(err, io.ErrShortWrite):
+				return true
+			}
+			return false
+		},
+	}
+	RetryableGrpcErrors = []IsRetryableFunc{
+		func(err error) bool {
+			st, ok := status.FromError(err)
+			if !ok {
+				return false
+			}
+			switch st.Code() {
+			case codes.Unavailable, codes.ResourceExhausted:
+				return true
+			case codes.DeadlineExceeded:
+				return true
+			case codes.Internal:
+				switch {
+				case strings.Contains(st.Message(), `transport: received the unexpected content-type "text/html; charset=UTF-8"`):
+					return true
+				case strings.Contains(st.Message(), io.ErrUnexpectedEOF.Error()):
+					return true
+				case strings.Contains(st.Message(), `stream terminated by RST_STREAM with error code: INTERNAL_ERROR`):
+					return true
+				}
+			}
+			return false
+		},
+	}
+)
+
+// Unwrap provides error wrapping
+func (e *retryableError) Unwrap() error {
+	return e.err
+}
+
+// Error returns the error string
+func (e *retryableError) Error() string {
+	if e.err == nil {
+		return ""
+	}
+	return e.err.Error()
+}
+
+// IsRetryable checks error for ability to retry later
+func IsRetryable(err error, fns ...IsRetryableFunc) bool {
+	for _, fn := range fns {
+		if ok := fn(err); ok {
+			return true
+		}
+	}
+	return false
+}

errors/errors_test.go

@@ -8,6 +8,13 @@ import (
 	"testing"
 )
 
+func TestIsRetrayable(t *testing.T) {
+	err := fmt.Errorf("ORA-")
+	if !IsRetryable(err, RetrayableOracleErrors...) {
+		t.Fatalf("IsRetrayable not works")
+	}
+}
+
 func TestMarshalJSON(t *testing.T) {
 	e := InternalServerError("id", "err: %v", fmt.Errorf("err: %v", `xxx: "UNIX_TIMESTAMP": invalid identifier`))
 	_, err := json.Marshal(e)

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/google/uuid v1.3.0
 	github.com/imdario/mergo v0.3.15
 	github.com/patrickmn/go-cache v2.1.0+incompatible
-	github.com/silas/dag v0.0.0-20211117232152-9d50aa809f35
+	github.com/silas/dag v0.0.0-20220518035006-a7e85ada93c5
 	golang.org/x/sync v0.3.0
 	google.golang.org/grpc v1.57.0
 	google.golang.org/protobuf v1.31.0
@@ -15,5 +15,6 @@ require (
 
 require (
 	github.com/golang/protobuf v1.5.3 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 // indirect
+	golang.org/x/net v0.14.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20230526203410-71b5a4ffd15e // indirect
 )

go.sum

@@ -11,16 +11,17 @@ github.com/imdario/mergo v0.3.15 h1:M8XP7IuFNsqUx6VPK2P9OSmsYsI/YFaGil0uD21V3dM=
 github.com/imdario/mergo v0.3.15/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+hD27ysY=
 github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
 github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
-github.com/silas/dag v0.0.0-20211117232152-9d50aa809f35 h1:4mohWoM/UGg1BvFFiqSPRl5uwJY3rVV0HQX0ETqauqQ=
-github.com/silas/dag v0.0.0-20211117232152-9d50aa809f35/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
-golang.org/x/net v0.9.0 h1:aWJ/m6xSmxWBx+V0XRHTlrYrPG56jKsLdTFmsSsCzOM=
+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.14.0 h1:BONx9s002vGdD9umnlX1Po8vOZmrgH34qlHcD1MfK14=
+golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI=
 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/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
+golang.org/x/sys v0.11.0 h1:eG7RXZHdqOJ1i+0lgLgCpSXAp6M3LYlAo6osgSi0xOM=
+golang.org/x/text v0.12.0 h1:k+n5B8goJNdU7hSvEtMUz3d1Q6D/XW4COJSJR6fN0mc=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-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/genproto/googleapis/rpc v0.0.0-20230526203410-71b5a4ffd15e h1:NumxXLPfHSndr3wBBdeKiVHjGVFzi9RX2HwwQke94iY=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20230526203410-71b5a4ffd15e/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=

server/subscriber.go

@@ -342,9 +342,6 @@ func (n *noopServer) createSubHandler(sb *subscriber, opts Options) broker.Handl
 
 		hdr := metadata.New(len(msg.Header))
 		for k, v := range msg.Header {
-			if k == "Content-Type" {
-				continue
-			}
 			hdr.Set(k, v)
 		}
 

service.go

@@ -88,6 +88,7 @@ func (s *service) Name() string {
 // Init initialises options. Additionally it calls cmd.Init
 // which parses command line flags. cmd.Init is only called
 // on first Init.
+//
 //nolint:gocyclo
 func (s *service) Init(opts ...Option) error {
 	var err error
@@ -375,19 +376,71 @@ func (s *service) Run() error {
 	return s.Stop()
 }
 
-type nameIface interface {
-	Name() string
-}
-
 func getNameIndex(n string, ifaces interface{}) int {
-	values, ok := ifaces.([]interface{})
-	if !ok {
-		return 0
-	}
-	for idx, iface := range values {
-		if ifc, ok := iface.(nameIface); ok && ifc.Name() == n {
-			return idx
+	switch values := ifaces.(type) {
+	case []router.Router:
+		for idx, iface := range values {
+			if iface.Name() == n {
+				return idx
+			}
 		}
+	case []register.Register:
+		for idx, iface := range values {
+			if iface.Name() == n {
+				return idx
+			}
+		}
+	case []store.Store:
+		for idx, iface := range values {
+			if iface.Name() == n {
+				return idx
+			}
+		}
+	case []tracer.Tracer:
+		for idx, iface := range values {
+			if iface.Name() == n {
+				return idx
+			}
+		}
+	case []server.Server:
+		for idx, iface := range values {
+			if iface.Name() == n {
+				return idx
+			}
+		}
+	case []config.Config:
+		for idx, iface := range values {
+			if iface.Name() == n {
+				return idx
+			}
+		}
+	case []meter.Meter:
+		for idx, iface := range values {
+			if iface.Name() == n {
+				return idx
+			}
+		}
+	case []broker.Broker:
+		for idx, iface := range values {
+			if iface.Name() == n {
+				return idx
+			}
+		}
+	case []client.Client:
+		for idx, iface := range values {
+			if iface.Name() == n {
+				return idx
+			}
+		}
+		/*
+			case []logger.Logger:
+					for idx, iface := range values {
+						if iface.Name() == n {
+							return idx
+						}
+					}
+		*/
 	}
+
 	return 0
 }

service_test.go

@@ -17,6 +17,21 @@ import (
 	"go.unistack.org/micro/v3/tracer"
 )
 
+func TestClient(t *testing.T) {
+	c1 := client.NewClient(client.Name("test1"))
+	c2 := client.NewClient(client.Name("test2"))
+
+	svc := NewService(Client(c1, c2))
+	if err := svc.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	x1 := svc.Client("test2")
+	if x1.Name() != "test2" {
+		t.Fatal("invalid client")
+	}
+}
+
 type testItem struct {
 	name string
 }

tracer/noop.go

@@ -7,19 +7,27 @@ import (
 var _ Tracer = (*noopTracer)(nil)
 
 type noopTracer struct {
-	opts Options
+	opts  Options
+	spans []Span
+}
+
+func (t *noopTracer) Spans() []Span {
+	return t.spans
 }
 
 func (t *noopTracer) Start(ctx context.Context, name string, opts ...SpanOption) (context.Context, Span) {
+	options := NewSpanOptions(opts...)
 	span := &noopSpan{
 		name:   name,
 		ctx:    ctx,
 		tracer: t,
-		opts:   NewSpanOptions(opts...),
+		labels: options.Labels,
+		kind:   options.Kind,
 	}
 	if span.ctx == nil {
 		span.ctx = context.Background()
 	}
+	t.spans = append(t.spans, span)
 	return NewSpanContext(ctx, span), span
 }
 
@@ -38,13 +46,21 @@ func (t *noopTracer) Name() string {
 	return t.opts.Name
 }
 
+type noopEvent struct {
+	name   string
+	labels []interface{}
+}
+
 type noopSpan struct {
 	ctx       context.Context
 	tracer    Tracer
 	name      string
-	opts      SpanOptions
-	status    SpanStatus
 	statusMsg string
+	events    []*noopEvent
+	labels    []interface{}
+	logs      []interface{}
+	kind      SpanKind
+	status    SpanStatus
 }
 
 func (s *noopSpan) Finish(opts ...SpanOption) {
@@ -59,22 +75,24 @@ func (s *noopSpan) Tracer() Tracer {
 }
 
 func (s *noopSpan) AddEvent(name string, opts ...EventOption) {
+	options := NewEventOptions(opts...)
+	s.events = append(s.events, &noopEvent{name: name, labels: options.Labels})
 }
 
 func (s *noopSpan) SetName(name string) {
 	s.name = name
 }
 
-func (s *noopSpan) SetLabels(labels ...interface{}) {
-	s.opts.Labels = labels
+func (s *noopSpan) AddLogs(kv ...interface{}) {
+	s.logs = append(s.logs, kv...)
 }
 
-func (s *noopSpan) AddLabels(labels ...interface{}) {
-	s.opts.Labels = append(s.opts.Labels, labels...)
+func (s *noopSpan) AddLabels(kv ...interface{}) {
+	s.labels = append(s.labels, kv...)
 }
 
 func (s *noopSpan) Kind() SpanKind {
-	return s.opts.Kind
+	return s.kind
 }
 
 func (s *noopSpan) Status() (SpanStatus, string) {

tracer/options.go

@@ -98,15 +98,15 @@ type EventOptions struct {
 // EventOption func signature
 type EventOption func(o *EventOptions)
 
-func WithEventLabels(labels ...interface{}) EventOption {
+func WithEventLabels(kv ...interface{}) EventOption {
 	return func(o *EventOptions) {
-		o.Labels = labels
+		o.Labels = kv
 	}
 }
 
-func WithSpanLabels(labels ...interface{}) SpanOption {
+func WithSpanLabels(kv ...interface{}) SpanOption {
 	return func(o *SpanOptions) {
-		o.Labels = labels
+		o.Labels = kv
 	}
 }
 
@@ -136,6 +136,15 @@ func Logger(l logger.Logger) Option {
 	}
 }
 
+// NewEventOptions returns default EventOptions
+func NewEventOptions(opts ...EventOption) EventOptions {
+	options := EventOptions{}
+	for _, o := range opts {
+		o(&options)
+	}
+	return options
+}
+
 // NewSpanOptions returns default SpanOptions
 func NewSpanOptions(opts ...SpanOption) SpanOptions {
 	options := SpanOptions{

tracer/tracer.go

@@ -27,8 +27,6 @@ type Span interface {
 	Tracer() Tracer
 	// Finish complete and send span
 	Finish(opts ...SpanOption)
-	// AddEvent add event to span
-	AddEvent(name string, opts ...EventOption)
 	// Context return context with span
 	Context() context.Context
 	// SetName set the span name
@@ -37,10 +35,12 @@ type Span interface {
 	SetStatus(status SpanStatus, msg string)
 	// Status returns span status and msg
 	Status() (SpanStatus, string)
-	// SetLabels set the span labels
-	SetLabels(labels ...interface{})
-	// AddLabels append the span labels
-	AddLabels(labels ...interface{})
+	// AddLabels append labels to span
+	AddLabels(kv ...interface{})
+	// AddEvent append event to span
+	AddEvent(name string, opts ...EventOption)
+	// AddEvent append event to span
+	AddLogs(kv ...interface{})
 	// Kind returns span kind
 	Kind() SpanKind
 }

util/grpc/tracer.go

@@ -0,0 +1,271 @@
+package grpc_util
+
+import (
+	"context"
+	"net"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"go.unistack.org/micro/v3/tracer"
+	grpc_codes "google.golang.org/grpc/codes"
+	"google.golang.org/grpc/peer"
+	"google.golang.org/grpc/stats"
+	"google.golang.org/grpc/status"
+)
+
+type gRPCContextKey struct{}
+
+type gRPCContext struct {
+	messagesReceived int64
+	messagesSent     int64
+}
+
+type Options struct {
+	Tracer tracer.Tracer
+}
+
+type Option func(*Options)
+
+func Tracer(tr tracer.Tracer) Option {
+	return func(o *Options) {
+		o.Tracer = tr
+	}
+}
+
+// NewServerHandler creates a stats.Handler for gRPC server.
+func NewServerHandler(opts ...Option) stats.Handler {
+	options := Options{Tracer: tracer.DefaultTracer}
+	for _, o := range opts {
+		o(&options)
+	}
+	h := &serverHandler{
+		opts: options,
+	}
+	return h
+}
+
+type serverHandler struct {
+	opts Options
+}
+
+// TagRPC can attach some information to the given context.
+func (h *serverHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context {
+	name, attrs := parseFullMethod(info.FullMethodName)
+	attrs = append(attrs, "rpc.system", "grpc")
+	ctx, _ = h.opts.Tracer.Start(
+		ctx,
+		name,
+		tracer.WithSpanKind(tracer.SpanKindServer),
+		tracer.WithSpanLabels(attrs...),
+	)
+
+	gctx := gRPCContext{}
+	return context.WithValue(ctx, gRPCContextKey{}, &gctx)
+}
+
+// HandleRPC processes the RPC stats.
+func (h *serverHandler) HandleRPC(ctx context.Context, rs stats.RPCStats) {
+	handleRPC(ctx, rs)
+}
+
+// TagConn can attach some information to the given context.
+func (h *serverHandler) TagConn(ctx context.Context, info *stats.ConnTagInfo) context.Context {
+	if span, ok := tracer.SpanFromContext(ctx); ok {
+		attrs := peerAttr(peerFromCtx(ctx))
+		span.AddLabels(attrs...)
+	}
+	return ctx
+}
+
+// HandleConn processes the Conn stats.
+func (h *serverHandler) HandleConn(ctx context.Context, info stats.ConnStats) {
+}
+
+type clientHandler struct {
+	opts Options
+}
+
+// NewClientHandler creates a stats.Handler for gRPC client.
+func NewClientHandler(opts ...Option) stats.Handler {
+	options := Options{Tracer: tracer.DefaultTracer}
+	for _, o := range opts {
+		o(&options)
+	}
+	h := &clientHandler{
+		opts: options,
+	}
+	return h
+}
+
+// TagRPC can attach some information to the given context.
+func (h *clientHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context {
+	name, attrs := parseFullMethod(info.FullMethodName)
+	attrs = append(attrs, "rpc.system", "grpc", "rpc.flavor", "grpc", "rpc.call", info.FullMethodName)
+	ctx, _ = h.opts.Tracer.Start(
+		ctx,
+		name,
+		tracer.WithSpanKind(tracer.SpanKindClient),
+		tracer.WithSpanLabels(attrs...),
+	)
+
+	gctx := gRPCContext{}
+
+	return context.WithValue(ctx, gRPCContextKey{}, &gctx)
+}
+
+// HandleRPC processes the RPC stats.
+func (h *clientHandler) HandleRPC(ctx context.Context, rs stats.RPCStats) {
+	handleRPC(ctx, rs)
+}
+
+// TagConn can attach some information to the given context.
+func (h *clientHandler) TagConn(ctx context.Context, cti *stats.ConnTagInfo) context.Context {
+	// TODO
+	if span, ok := tracer.SpanFromContext(ctx); ok {
+		attrs := peerAttr(cti.RemoteAddr.String())
+		span.AddLabels(attrs...)
+	}
+	return ctx
+}
+
+// HandleConn processes the Conn stats.
+func (h *clientHandler) HandleConn(context.Context, stats.ConnStats) {
+	// no-op
+}
+
+func handleRPC(ctx context.Context, rs stats.RPCStats) {
+	span, ok := tracer.SpanFromContext(ctx)
+	gctx, _ := ctx.Value(gRPCContextKey{}).(*gRPCContext)
+	var messageID int64
+	if rs.IsClient() {
+		span.AddLabels("span.kind", "client")
+	} else {
+		span.AddLabels("span.kind", "server")
+	}
+
+	switch rs := rs.(type) {
+	case *stats.Begin:
+		if rs.IsClientStream || rs.IsServerStream {
+			span.AddLabels("rpc.call_type", "stream")
+		} else {
+			span.AddLabels("rpc.call_type", "unary")
+		}
+		span.AddEvent("message",
+			tracer.WithEventLabels(
+				"message.begin_time", rs.BeginTime.Format(time.RFC3339),
+			),
+		)
+	case *stats.InPayload:
+		if gctx != nil {
+			messageID = atomic.AddInt64(&gctx.messagesReceived, 1)
+		}
+		if ok {
+			span.AddEvent("message",
+				tracer.WithEventLabels(
+					"message.recv_time", rs.RecvTime.Format(time.RFC3339),
+					"message.type", "RECEIVED",
+					"message.id", messageID,
+					"message.compressed_size", rs.CompressedLength,
+					"message.uncompressed_size", rs.Length,
+				),
+			)
+		}
+	case *stats.OutPayload:
+		if gctx != nil {
+			messageID = atomic.AddInt64(&gctx.messagesSent, 1)
+		}
+		if ok {
+			span.AddEvent("message",
+				tracer.WithEventLabels(
+					"message.sent_time", rs.SentTime.Format(time.RFC3339),
+					"message.type", "SENT",
+					"message.id", messageID,
+					"message.compressed_size", rs.CompressedLength,
+					"message.uncompressed_size", rs.Length,
+				),
+			)
+		}
+	case *stats.End:
+		if ok {
+			span.AddEvent("message",
+				tracer.WithEventLabels(
+					"message.begin_time", rs.BeginTime.Format(time.RFC3339),
+					"message.end_time", rs.EndTime.Format(time.RFC3339),
+				),
+			)
+			if rs.Error != nil {
+				s, _ := status.FromError(rs.Error)
+				span.SetStatus(tracer.SpanStatusError, s.Message())
+				span.AddLabels("rpc.grpc.status_code", s.Code())
+			} else {
+				span.AddLabels("rpc.grpc.status_code", grpc_codes.OK)
+			}
+			span.Finish()
+		}
+	default:
+		return
+	}
+}
+
+func parseFullMethod(fullMethod string) (string, []interface{}) {
+	if !strings.HasPrefix(fullMethod, "/") {
+		// Invalid format, does not follow `/package.service/method`.
+		return fullMethod, nil
+	}
+	name := fullMethod[1:]
+	pos := strings.LastIndex(name, "/")
+	if pos < 0 {
+		// Invalid format, does not follow `/package.service/method`.
+		return name, nil
+	}
+	service, method := name[:pos], name[pos+1:]
+
+	var attrs []interface{}
+	if service != "" {
+		attrs = append(attrs, "rpc.service", service)
+	}
+	if method != "" {
+		attrs = append(attrs, "rpc.method", method)
+	}
+	return name, attrs
+}
+
+func peerAttr(addr string) []interface{} {
+	host, p, err := net.SplitHostPort(addr)
+	if err != nil {
+		return nil
+	}
+
+	if host == "" {
+		host = "127.0.0.1"
+	}
+	port, err := strconv.Atoi(p)
+	if err != nil {
+		return nil
+	}
+
+	var attr []interface{}
+	if ip := net.ParseIP(host); ip != nil {
+		attr = []interface{}{
+			"net.sock.peer.addr", host,
+			"net.sock.peer.port", port,
+		}
+	} else {
+		attr = []interface{}{
+			"net.peer.name", host,
+			"net.peer.port", port,
+		}
+	}
+
+	return attr
+}
+
+func peerFromCtx(ctx context.Context) string {
+	p, ok := peer.FromContext(ctx)
+	if !ok {
+		return ""
+	}
+	return p.Addr.String()
+}

util/http/clienttracer.go

@@ -0,0 +1,253 @@
+//
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package http
+
+import (
+	"context"
+	"crypto/tls"
+	"net/http/httptrace"
+	"net/textproto"
+	"strings"
+	"sync"
+
+	"go.unistack.org/micro/v3/tracer"
+)
+
+const (
+	httpStatus     = "http.status"
+	httpHeaderMIME = "http.mime"
+	httpRemoteAddr = "http.remote"
+	httpLocalAddr  = "http.local"
+	httpHost       = "http.host"
+)
+
+var hookMap = map[string]string{
+	"http.dns":     "http.getconn",
+	"http.connect": "http.getconn",
+	"http.tls":     "http.getconn",
+}
+
+func parentHook(hook string) string {
+	if strings.HasPrefix(hook, "http.connect") {
+		return hookMap["http.connect"]
+	}
+	return hookMap[hook]
+}
+
+type clientTracer struct {
+	context.Context
+	tr          tracer.Tracer
+	activeHooks map[string]context.Context
+	root        tracer.Span
+	mtx         sync.Mutex
+}
+
+func NewClientTrace(ctx context.Context, tr tracer.Tracer) *httptrace.ClientTrace {
+	ct := &clientTracer{
+		Context:     ctx,
+		activeHooks: make(map[string]context.Context),
+		tr:          tr,
+	}
+
+	return &httptrace.ClientTrace{
+		GetConn:              ct.getConn,
+		GotConn:              ct.gotConn,
+		PutIdleConn:          ct.putIdleConn,
+		GotFirstResponseByte: ct.gotFirstResponseByte,
+		Got100Continue:       ct.got100Continue,
+		Got1xxResponse:       ct.got1xxResponse,
+		DNSStart:             ct.dnsStart,
+		DNSDone:              ct.dnsDone,
+		ConnectStart:         ct.connectStart,
+		ConnectDone:          ct.connectDone,
+		TLSHandshakeStart:    ct.tlsHandshakeStart,
+		TLSHandshakeDone:     ct.tlsHandshakeDone,
+		WroteHeaderField:     ct.wroteHeaderField,
+		WroteHeaders:         ct.wroteHeaders,
+		Wait100Continue:      ct.wait100Continue,
+		WroteRequest:         ct.wroteRequest,
+	}
+}
+
+func (ct *clientTracer) start(hook, spanName string, attrs ...interface{}) {
+	ct.mtx.Lock()
+	defer ct.mtx.Unlock()
+
+	if hookCtx, found := ct.activeHooks[hook]; !found {
+		var sp tracer.Span
+		ct.activeHooks[hook], sp = ct.tr.Start(ct.getParentContext(hook), spanName, tracer.WithSpanLabels(attrs...), tracer.WithSpanKind(tracer.SpanKindClient))
+		if ct.root == nil {
+			ct.root = sp
+		}
+	} else {
+		// end was called before start finished, add the start attributes and end the span here
+		if span, ok := tracer.SpanFromContext(hookCtx); ok {
+			span.AddLabels(attrs...)
+			span.Finish()
+		}
+
+		delete(ct.activeHooks, hook)
+	}
+}
+
+func (ct *clientTracer) end(hook string, err error, attrs ...interface{}) {
+	ct.mtx.Lock()
+	defer ct.mtx.Unlock()
+	if ctx, ok := ct.activeHooks[hook]; ok { // nolint:nestif
+		if span, ok := tracer.SpanFromContext(ctx); ok {
+			if err != nil {
+				span.SetStatus(tracer.SpanStatusError, err.Error())
+			}
+			span.AddLabels(attrs...)
+			span.Finish()
+		}
+		delete(ct.activeHooks, hook)
+	} else {
+		// start is not finished before end is called.
+		// Start a span here with the ending attributes that will be finished when start finishes.
+		// Yes, it's backwards. v0v
+		ctx, span := ct.tr.Start(ct.getParentContext(hook), hook, tracer.WithSpanLabels(attrs...), tracer.WithSpanKind(tracer.SpanKindClient))
+		if err != nil {
+			span.SetStatus(tracer.SpanStatusError, err.Error())
+		}
+		ct.activeHooks[hook] = ctx
+	}
+}
+
+func (ct *clientTracer) getParentContext(hook string) context.Context {
+	ctx, ok := ct.activeHooks[parentHook(hook)]
+	if !ok {
+		return ct.Context
+	}
+	return ctx
+}
+
+func (ct *clientTracer) span(hook string) (tracer.Span, bool) {
+	ct.mtx.Lock()
+	defer ct.mtx.Unlock()
+	if ctx, ok := ct.activeHooks[hook]; ok {
+		return tracer.SpanFromContext(ctx)
+	}
+	return nil, false
+}
+
+func (ct *clientTracer) getConn(host string) {
+	ct.start("http.getconn", "http.getconn", httpHost, host)
+}
+
+func (ct *clientTracer) gotConn(info httptrace.GotConnInfo) {
+	ct.end("http.getconn",
+		nil,
+		httpRemoteAddr, info.Conn.RemoteAddr().String(),
+		httpLocalAddr, info.Conn.LocalAddr().String(),
+	)
+}
+
+func (ct *clientTracer) putIdleConn(err error) {
+	ct.end("http.receive", err)
+}
+
+func (ct *clientTracer) gotFirstResponseByte() {
+	ct.start("http.receive", "http.receive")
+}
+
+func (ct *clientTracer) dnsStart(info httptrace.DNSStartInfo) {
+	ct.start("http.dns", "http.dns", httpHost, info.Host)
+}
+
+func (ct *clientTracer) dnsDone(info httptrace.DNSDoneInfo) {
+	ct.end("http.dns", info.Err)
+}
+
+func (ct *clientTracer) connectStart(network, addr string) {
+	_ = network
+	ct.start("http.connect."+addr, "http.connect", httpRemoteAddr, addr)
+}
+
+func (ct *clientTracer) connectDone(network, addr string, err error) {
+	_ = network
+	ct.end("http.connect."+addr, err)
+}
+
+func (ct *clientTracer) tlsHandshakeStart() {
+	ct.start("http.tls", "http.tls")
+}
+
+func (ct *clientTracer) tlsHandshakeDone(_ tls.ConnectionState, err error) {
+	ct.end("http.tls", err)
+}
+
+func (ct *clientTracer) wroteHeaderField(k string, v []string) {
+	if sp, ok := ct.span("http.headers"); !ok || sp == nil {
+		ct.start("http.headers", "http.headers")
+	}
+	ct.root.AddLabels("http."+strings.ToLower(k), sliceToString(v))
+}
+
+func (ct *clientTracer) wroteHeaders() {
+	ct.start("http.send", "http.send")
+}
+
+func (ct *clientTracer) wroteRequest(info httptrace.WroteRequestInfo) {
+	if info.Err != nil {
+		ct.root.SetStatus(tracer.SpanStatusError, info.Err.Error())
+	}
+	ct.end("http.send", info.Err)
+}
+
+func (ct *clientTracer) got100Continue() {
+	if sp, ok := ct.span("http.receive"); ok && sp != nil {
+		sp.AddEvent("GOT 100 - Continue")
+	}
+}
+
+func (ct *clientTracer) wait100Continue() {
+	if sp, ok := ct.span("http.receive"); ok && sp != nil {
+		sp.AddEvent("GOT 100 - Wait")
+	}
+}
+
+func (ct *clientTracer) got1xxResponse(code int, header textproto.MIMEHeader) error {
+	if sp, ok := ct.span("http.receive"); ok && sp != nil {
+		sp.AddEvent("GOT 1xx",
+			tracer.WithEventLabels(
+				httpStatus, code,
+				httpHeaderMIME, sm2s(header),
+			),
+		)
+	}
+	return nil
+}
+
+func sliceToString(value []string) string {
+	if len(value) == 0 {
+		return "undefined"
+	}
+	return strings.Join(value, ",")
+}
+
+func sm2s(value map[string][]string) string {
+	var buf strings.Builder
+	for k, v := range value {
+		if buf.Len() != 0 {
+			buf.WriteString(",")
+		}
+		buf.WriteString(k)
+		buf.WriteString("=")
+		buf.WriteString(sliceToString(v))
+	}
+	return buf.String()
+}

util/reflect/reflect.go

@@ -508,3 +508,74 @@ func FieldName(name string) string {
 
 	return string(newstr)
 }
+
+func Equal(src interface{}, dst interface{}, excptFields ...string) bool {
+	srcVal := reflect.ValueOf(src)
+	dstVal := reflect.ValueOf(dst)
+
+	switch srcVal.Kind() {
+	case reflect.Array, reflect.Slice:
+		for i := 0; i < srcVal.Len(); i++ {
+			e := srcVal.Index(i).Interface()
+			a := dstVal.Index(i).Interface()
+			if !Equal(e, a, excptFields...) {
+				return false
+			}
+		}
+		return true
+	case reflect.Map:
+		for i := 0; i < len(srcVal.MapKeys()); i++ {
+			key := srcVal.MapKeys()[i]
+			keyStr := fmt.Sprintf("%v", key.Interface())
+			if stringContains(keyStr, excptFields) {
+				continue
+			}
+			s := srcVal.MapIndex(key)
+			d := dstVal.MapIndex(key)
+			if !Equal(s.Interface(), d.Interface(), excptFields...) {
+				return false
+			}
+		}
+		return true
+	case reflect.Struct, reflect.Interface:
+		for i := 0; i < srcVal.NumField(); i++ {
+			typeField := srcVal.Type().Field(i)
+			if stringContains(typeField.Name, excptFields) {
+				continue
+			}
+			s := srcVal.Field(i)
+			d := dstVal.FieldByName(typeField.Name)
+			if s.CanInterface() && d.CanInterface() {
+				if !Equal(s.Interface(), d.Interface(), excptFields...) {
+					return false
+				}
+			} else {
+				return false
+			}
+		}
+		return true
+	case reflect.Ptr:
+		if srcVal.IsNil() {
+			return dstVal.IsNil()
+		}
+		s := srcVal.Elem()
+		d := reflect.Indirect(dstVal)
+		if s.CanInterface() && d.CanInterface() {
+			return Equal(s.Interface(), d.Interface(), excptFields...)
+		}
+		return false
+	case reflect.String, reflect.Int, reflect.Int64, reflect.Float32, reflect.Float64, reflect.Bool:
+		return src == dst
+	default:
+		return srcVal.Interface() == dstVal.Interface()
+	}
+}
+
+func stringContains(a string, list []string) bool {
+	for _, b := range list {
+		if b == a {
+			return true
+		}
+	}
+	return false
+}

util/reflect/reflect_test.go

@@ -133,3 +133,16 @@ func TestMergeNested(t *testing.T) {
 		t.Fatalf("merge error: %#+v", dst.Nested)
 	}
 }
+
+func TestEqual(t *testing.T) {
+	type tstr struct {
+		Key1 string
+		Key2 string
+	}
+
+	src := &tstr{Key1: "val1", Key2: "micro:generate"}
+	dst := &tstr{Key1: "val1", Key2: "val2"}
+	if !Equal(src, dst, "Key2") {
+		t.Fatal("invalid Equal test")
+	}
+}