Merge pull request 'client/noop: fixup md' (#285) from noopfix into v3

Reviewed-on: #285

client/noop.go

@@ -3,6 +3,7 @@ package client
 import (
 	"context"
 	"fmt"
+	"os"
 	"time"
 
 	"go.unistack.org/micro/v3/broker"
@@ -485,21 +486,34 @@ func (n *noopClient) publish(ctx context.Context, ps []Message, opts ...PublishO
 
 	msgs := make([]*broker.Message, 0, len(ps))
 
+	// get proxy
+	exchange := ""
+	if v, ok := os.LookupEnv("MICRO_PROXY"); ok {
+		exchange = v
+	}
+	// get the exchange
+	if len(options.Exchange) > 0 {
+		exchange = options.Exchange
+	}
+
+	omd, ok := metadata.FromOutgoingContext(ctx)
+	if !ok {
+		omd = metadata.New(0)
+	}
+
 	for _, p := range ps {
-		md, ok := metadata.FromOutgoingContext(ctx)
-		if !ok {
-			md = metadata.New(0)
-		}
+		md := metadata.Copy(omd)
 		md[metadata.HeaderContentType] = p.ContentType()
-
 		topic := p.Topic()
-
-		// get the exchange
-		if len(options.Exchange) > 0 {
-			topic = options.Exchange
+		if len(exchange) > 0 {
+			topic = exchange
 		}
-
 		md[metadata.HeaderTopic] = topic
+		iter := p.Metadata().Iterator()
+		var k, v string
+		for iter.Next(&k, &v) {
+			md.Set(k, v)
+		}
 
 		var body []byte
 

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

@@ -3,8 +3,18 @@ module go.unistack.org/micro/v3
 go 1.19
 
 require (
+	github.com/DATA-DOG/go-sqlmock v1.5.0
 	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
+)
+
+require (
+	github.com/golang/protobuf v1.5.3 // indirect
+	golang.org/x/net v0.14.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20230526203410-71b5a4ffd15e // indirect
 )

go.sum

@@ -1,11 +1,33 @@
+github.com/DATA-DOG/go-sqlmock v1.5.0 h1:Shsta01QNfFxHCfpW6YH2STWB0MudeXXEWMr20OEh60=
+github.com/DATA-DOG/go-sqlmock v1.5.0/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM=
+github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
+github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 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=
+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.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-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=
+google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+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.

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

@@ -91,14 +91,22 @@ 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 WithSpanLabels(labels ...interface{}) SpanOption {
+func WithEventLabels(kv ...interface{}) EventOption {
+	return func(o *EventOptions) {
+		o.Labels = kv
+	}
+}
+
+func WithSpanLabels(kv ...interface{}) SpanOption {
 	return func(o *SpanOptions) {
-		o.Labels = labels
+		o.Labels = kv
 	}
 }
 
@@ -128,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

@@ -3,6 +3,8 @@ package tracer // import "go.unistack.org/micro/v3/tracer"
 
 import (
 	"context"
+	"fmt"
+	"sort"
 )
 
 // DefaultTracer is the global default tracer
@@ -25,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
@@ -35,10 +35,46 @@ 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
 }
+
+// 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

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/http/trie_test.go

@@ -5,8 +5,32 @@ import (
 	"testing"
 )
 
-func TestTrieBackwards(t *testing.T) {
-	_ = &Trie{}
+func TestTrieRPC(t *testing.T) {
+	var err error
+	type handler struct {
+		name string
+	}
+	tr := NewTrie()
+	if err = tr.Insert([]string{"helloworld"}, "Call", &handler{name: "helloworld.Call"}); err != nil {
+		t.Fatal(err)
+	}
+	if err = tr.Insert([]string{"helloworld"}, "Stream", &handler{name: "helloworld.Stream"}); err != nil {
+		t.Fatal(err)
+	}
+	h, _, err := tr.Search("helloworld", "Call")
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+	if h.(*handler).name != "helloworld.Call" {
+		t.Fatalf("invalid handler %v", h)
+	}
+	h, _, err = tr.Search("helloworld", "Stream")
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+	if h.(*handler).name != "helloworld.Stream" {
+		t.Fatalf("invalid handler %v", h)
+	}
 }
 
 func TestTrieWildcardPathPrefix(t *testing.T) {

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")
+	}
+}

util/test/test.go

@@ -0,0 +1,505 @@
+package test
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"database/sql/driver"
+	"encoding/csv"
+	"fmt"
+	"io"
+	"os"
+	"path"
+	"path/filepath"
+	"reflect"
+	"strings"
+	"time"
+
+	sqlmock "github.com/DATA-DOG/go-sqlmock"
+	"go.unistack.org/micro/v3/client"
+	"go.unistack.org/micro/v3/codec"
+	"go.unistack.org/micro/v3/errors"
+	"go.unistack.org/micro/v3/metadata"
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+)
+
+var ErrUnknownContentType = fmt.Errorf("unknown content type")
+
+type Extension struct {
+	Ext []string
+}
+
+var (
+	// ExtToTypes map file extension to content type
+	ExtToTypes = map[string][]string{
+		"json":  {"application/json", "application/grpc+json"},
+		"yaml":  {"application/yaml", "application/yml", "text/yaml", "text/yml"},
+		"yml":   {"application/yaml", "application/yml", "text/yaml", "text/yml"},
+		"proto": {"application/grpc", "application/grpc+proto", "application/proto"},
+	}
+	// DefaultExts specifies default file extensions to load data
+	DefaultExts = []string{"csv", "json", "yaml", "yml", "proto"}
+	// Codecs map to detect codec for test file or request content type
+	Codecs map[string]codec.Codec
+
+	// ResponseCompareFunc used to compare actual response with test case data
+	ResponseCompareFunc = func(expectRsp []byte, testRsp interface{}, expectCodec codec.Codec, testCodec codec.Codec) error {
+		var err error
+
+		expectMap := make(map[string]interface{})
+		if err = expectCodec.Unmarshal(expectRsp, &expectMap); err != nil {
+			return fmt.Errorf("failed to unmarshal err: %w", err)
+		}
+
+		testMap := make(map[string]interface{})
+		switch v := testRsp.(type) {
+		case *codec.Frame:
+			if err = testCodec.Unmarshal(v.Data, &testMap); err != nil {
+				return fmt.Errorf("failed to unmarshal err: %w", err)
+			}
+		case *errors.Error:
+			if err = expectCodec.Unmarshal([]byte(v.Error()), &testMap); err != nil {
+				return fmt.Errorf("failed to unmarshal err: %w", err)
+			}
+		case error:
+			st, ok := status.FromError(v)
+			if !ok {
+				return v
+			}
+			me := errors.Parse(st.Message())
+			if me.Code != 0 {
+				if err = expectCodec.Unmarshal([]byte(me.Error()), &testMap); err != nil {
+					return fmt.Errorf("failed to unmarshal err: %w", err)
+				}
+				break
+			}
+			for _, se := range st.Details() {
+				switch ne := se.(type) {
+				case proto.Message:
+					buf, err := testCodec.Marshal(ne)
+					if err != nil {
+						return fmt.Errorf("failed to marshal err: %w", err)
+					}
+					if err = testCodec.Unmarshal(buf, &testMap); err != nil {
+						return fmt.Errorf("failed to unmarshal err: %w", err)
+					}
+				default:
+					return st.Err()
+				}
+			}
+		case interface{ GRPCStatus() *status.Status }:
+			st := v.GRPCStatus()
+			me := errors.Parse(st.Message())
+			if me.Code != 0 {
+				if err = expectCodec.Unmarshal([]byte(me.Error()), &testMap); err != nil {
+					return fmt.Errorf("failed to unmarshal err: %w", err)
+				}
+				break
+			}
+		case *status.Status:
+			me := errors.Parse(v.Message())
+			if me.Code != 0 {
+				if err = expectCodec.Unmarshal([]byte(me.Error()), &testMap); err != nil {
+					return fmt.Errorf("failed to unmarshal err: %w", err)
+				}
+				break
+			}
+			for _, se := range v.Details() {
+				switch ne := se.(type) {
+				case proto.Message:
+					buf, err := testCodec.Marshal(ne)
+					if err != nil {
+						return fmt.Errorf("failed to marshal err: %w", err)
+					}
+					if err = testCodec.Unmarshal(buf, &testMap); err != nil {
+						return fmt.Errorf("failed to unmarshal err: %w", err)
+					}
+				default:
+					return v.Err()
+				}
+			}
+		}
+
+		if !reflect.DeepEqual(expectMap, testMap) {
+			return fmt.Errorf("test: %s != rsp: %s", expectMap, testMap)
+		}
+
+		return nil
+	}
+)
+
+func FromCSVString(columns []*sqlmock.Column, rows *sqlmock.Rows, s string) *sqlmock.Rows {
+	res := strings.NewReader(strings.TrimSpace(s))
+	csvReader := csv.NewReader(res)
+
+	for {
+		res, err := csvReader.Read()
+		if err != nil || res == nil {
+			break
+		}
+
+		var row []driver.Value
+		for i, v := range res {
+			item := CSVColumnParser(strings.TrimSpace(v))
+			if null, nullOk := columns[i].IsNullable(); null && nullOk && item == nil {
+				row = append(row, nil)
+			} else {
+				row = append(row, item)
+			}
+
+		}
+		rows = rows.AddRow(row...)
+	}
+
+	return rows
+}
+
+func CSVColumnParser(s string) []byte {
+	switch {
+	case strings.ToLower(s) == "null":
+		return nil
+	case s == "":
+		return nil
+	}
+	return []byte(s)
+}
+
+func NewResponseFromFile(rspfile string) (*codec.Frame, error) {
+	rspbuf, err := os.ReadFile(rspfile)
+	if err != nil {
+		return nil, err
+	}
+	return &codec.Frame{Data: rspbuf}, nil
+}
+
+func getCodec(codecs map[string]codec.Codec, ext string) (codec.Codec, error) {
+	var c codec.Codec
+	if cts, ok := ExtToTypes[ext]; ok {
+		for _, t := range cts {
+			if c, ok = codecs[t]; ok {
+				return c, nil
+			}
+		}
+	}
+	return nil, ErrUnknownContentType
+}
+
+func getContentType(codecs map[string]codec.Codec, ext string) (string, error) {
+	if cts, ok := ExtToTypes[ext]; ok {
+		for _, t := range cts {
+			if _, ok = codecs[t]; ok {
+				return t, nil
+			}
+		}
+	}
+	return "", ErrUnknownContentType
+}
+
+func getExt(name string) string {
+	ext := filepath.Ext(name)
+	if len(ext) > 0 && ext[0] == '.' {
+		ext = ext[1:]
+	}
+	return ext
+}
+
+func getNameWithoutExt(name string) string {
+	return strings.TrimSuffix(name, filepath.Ext(name))
+}
+
+func NewRequestFromFile(c client.Client, reqfile string) (client.Request, error) {
+	reqbuf, err := os.ReadFile(reqfile)
+	if err != nil {
+		return nil, err
+	}
+
+	endpoint := path.Base(path.Dir(reqfile))
+	if idx := strings.Index(endpoint, "_"); idx > 0 {
+		endpoint = endpoint[idx+1:]
+	}
+	ext := getExt(reqfile)
+
+	ct, err := getContentType(c.Options().Codecs, ext)
+	if err != nil {
+		return nil, err
+	}
+
+	req := c.NewRequest("test", endpoint, &codec.Frame{Data: reqbuf}, client.RequestContentType(ct))
+
+	return req, nil
+}
+
+func SQLFromFile(m sqlmock.Sqlmock, name string) error {
+	fp, err := os.Open(name)
+	if err != nil {
+		return err
+	}
+	defer fp.Close()
+	return SQLFromReader(m, fp)
+}
+
+func SQLFromBytes(m sqlmock.Sqlmock, buf []byte) error {
+	return SQLFromReader(m, bytes.NewReader(buf))
+}
+
+func SQLFromString(m sqlmock.Sqlmock, buf string) error {
+	return SQLFromReader(m, strings.NewReader(buf))
+}
+
+func SQLFromReader(m sqlmock.Sqlmock, r io.Reader) error {
+	var rows *sqlmock.Rows
+	var exp *sqlmock.ExpectedQuery
+	var columns []*sqlmock.Column
+
+	br := bufio.NewReader(r)
+
+	for {
+		s, err := br.ReadString('\n')
+		if err != nil && err != io.EOF {
+			return err
+		} else if err == io.EOF && len(s) == 0 {
+			if rows != nil && exp != nil {
+				exp.WillReturnRows(rows)
+			}
+			return nil
+		}
+
+		if s[0] != '#' {
+			r := csv.NewReader(strings.NewReader(s))
+			r.Comma = ','
+			var records [][]string
+			records, err = r.ReadAll()
+			if err != nil {
+				return err
+			}
+			if rows == nil && len(columns) > 0 {
+				rows = m.NewRowsWithColumnDefinition(columns...)
+			} else {
+				for idx := 0; idx < len(records); idx++ {
+					if len(columns) == 0 {
+						return fmt.Errorf("csv file not valid, does not have %q line", "# columns ")
+					}
+					rows = FromCSVString(columns, rows, strings.Join(records[idx], ","))
+				}
+			}
+			continue
+		}
+
+		if rows != nil {
+			exp.WillReturnRows(rows)
+			rows = nil
+		}
+
+		switch {
+		case strings.HasPrefix(strings.ToLower(s[2:]), "columns"):
+			for _, field := range strings.Split(s[2+len("columns")+1:], ",") {
+				args := strings.Split(field, "|")
+
+				column := sqlmock.NewColumn(args[0]).Nullable(false)
+
+				if len(args) > 1 {
+					for _, arg := range args {
+						switch arg {
+						case "BOOLEAN", "BOOL":
+							column = column.OfType("BOOL", false)
+						case "NUMBER", "DECIMAL":
+							column = column.OfType("DECIMAL", float64(0.0)).WithPrecisionAndScale(10, 4)
+						case "VARCHAR":
+							column = column.OfType("VARCHAR", nil)
+						case "NULL":
+							column = column.Nullable(true)
+						}
+					}
+				}
+
+				columns = append(columns, column)
+			}
+		case strings.HasPrefix(strings.ToLower(s[2:]), "begin"):
+			m.ExpectBegin()
+		case strings.HasPrefix(strings.ToLower(s[2:]), "commit"):
+			m.ExpectCommit()
+		case strings.HasPrefix(strings.ToLower(s[2:]), "rollback"):
+			m.ExpectRollback()
+		case strings.HasPrefix(strings.ToLower(s[2:]), "exec "):
+			m.ExpectExec(s[2+len("exec "):])
+		case strings.HasPrefix(strings.ToLower(s[2:]), "query "):
+			exp = m.ExpectQuery(s[2+len("query "):])
+		}
+	}
+}
+
+func Run(ctx context.Context, c client.Client, m sqlmock.Sqlmock, dir string, exts []string) error {
+	tcases, err := GetCases(dir, exts)
+	if err != nil {
+		return err
+	}
+
+	g, gctx := errgroup.WithContext(ctx)
+	if !strings.Contains(dir, "parallel") {
+		g.SetLimit(1)
+	}
+
+	for _, tcase := range tcases {
+
+		for _, dbfile := range tcase.dbfiles {
+			if err = SQLFromFile(m, dbfile); err != nil {
+				return err
+			}
+		}
+
+		tc := tcase
+		g.Go(func() error {
+			var xrid string
+			var gerr error
+
+			treq, err := NewRequestFromFile(c, tc.reqfile)
+			if err != nil {
+				gerr = fmt.Errorf("failed to read request from file %s err: %w", tc.reqfile, err)
+				return gerr
+			}
+
+			xrid = fmt.Sprintf("%s-%d", treq.Endpoint(), time.Now().Unix())
+
+			defer func() {
+				if gerr == nil {
+					fmt.Printf("test %s xrid: %s status: success\n", filepath.Dir(tc.reqfile), xrid)
+				} else {
+					fmt.Printf("test %s xrid: %s status: failure error: %v\n", filepath.Dir(tc.reqfile), xrid, err)
+				}
+			}()
+
+			data := &codec.Frame{}
+			md := metadata.New(1)
+			md.Set("X-Request-Id", xrid)
+			cerr := c.Call(metadata.NewOutgoingContext(gctx, md), treq, data, client.WithContentType(treq.ContentType()))
+
+			var rspfile string
+
+			if tc.errfile != "" {
+				rspfile = tc.errfile
+			} else if tc.rspfile != "" {
+				rspfile = tc.rspfile
+			} else {
+				gerr = fmt.Errorf("errfile and rspfile is empty")
+				return gerr
+			}
+
+			expectRsp, err := NewResponseFromFile(rspfile)
+			if err != nil {
+				gerr = fmt.Errorf("failed to read response from file %s err: %w", rspfile, err)
+				return gerr
+			}
+
+			testCodec, err := getCodec(Codecs, getExt(tc.reqfile))
+			if err != nil {
+				gerr = fmt.Errorf("failed to get response file codec err: %w", err)
+				return gerr
+			}
+
+			expectCodec, err := getCodec(Codecs, getExt(rspfile))
+			if err != nil {
+				gerr = fmt.Errorf("failed to get response file codec err: %w", err)
+				return gerr
+			}
+
+			if cerr == nil && tc.errfile != "" {
+				gerr = fmt.Errorf("expected err %s not happened", expectRsp.Data)
+				return gerr
+			} else if cerr != nil && tc.errfile != "" {
+				if err = ResponseCompareFunc(expectRsp.Data, cerr, expectCodec, testCodec); err != nil {
+					gerr = err
+					return gerr
+				}
+			} else if cerr != nil && tc.errfile == "" {
+				gerr = cerr
+				return gerr
+			} else if cerr == nil && tc.errfile == "" {
+				if err = ResponseCompareFunc(expectRsp.Data, data, expectCodec, testCodec); err != nil {
+					gerr = err
+					return gerr
+				}
+			}
+
+			/*
+				cf, err := getCodec(c.Options().Codecs, getExt(tc.rspfile))
+				if err != nil {
+					return err
+				}
+			*/
+
+			return nil
+		})
+
+	}
+
+	return g.Wait()
+}
+
+type Case struct {
+	dbfiles []string
+	reqfile string
+	rspfile string
+	errfile string
+}
+
+func GetCases(dir string, exts []string) ([]Case, error) {
+	var tcases []Case
+	entries, err := os.ReadDir(dir)
+	if len(entries) == 0 && err != nil {
+		return tcases, err
+	}
+
+	if exts == nil {
+		exts = DefaultExts
+	}
+
+	var dirs []string
+	var dbfiles []string
+	var reqfile, rspfile, errfile string
+
+	for _, entry := range entries {
+		if entry.IsDir() {
+			dirs = append(dirs, filepath.Join(dir, entry.Name()))
+			continue
+		}
+		if info, err := entry.Info(); err != nil {
+			return tcases, err
+		} else if !info.Mode().IsRegular() {
+			continue
+		}
+
+		for _, ext := range exts {
+			if getExt(entry.Name()) == ext {
+				name := getNameWithoutExt(entry.Name())
+				switch {
+				case strings.HasSuffix(name, "_db"):
+					dbfiles = append(dbfiles, filepath.Join(dir, entry.Name()))
+				case strings.HasSuffix(name, "_req"):
+					reqfile = filepath.Join(dir, entry.Name())
+				case strings.HasSuffix(name, "_rsp"):
+					rspfile = filepath.Join(dir, entry.Name())
+				case strings.HasSuffix(name, "_err"):
+					errfile = filepath.Join(dir, entry.Name())
+				}
+			}
+		}
+	}
+
+	if reqfile != "" && (rspfile != "" || errfile != "") {
+		tcases = append(tcases, Case{dbfiles: dbfiles, reqfile: reqfile, rspfile: rspfile, errfile: errfile})
+	}
+
+	for _, dir = range dirs {
+		ntcases, err := GetCases(dir, exts)
+		if len(ntcases) == 0 && err != nil {
+			return tcases, err
+		} else if len(ntcases) == 0 {
+			continue
+		}
+		tcases = append(tcases, ntcases...)
+	}
+
+	return tcases, nil
+}

util/test/test_test.go

@@ -0,0 +1,72 @@
+package test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/DATA-DOG/go-sqlmock"
+)
+
+func Test_SQLFromFile(t *testing.T) {
+	ctx := context.TODO()
+	db, c, err := sqlmock.New()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+
+	if err = SQLFromFile(c, "testdata/result/01_firstcase/Call_db.csv"); err != nil {
+		t.Fatal(err)
+	}
+
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	rows, err := tx.QueryContext(ctx, "select * from test;")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for rows.Next() {
+		var id int64
+		var name string
+		err = rows.Scan(&id, &name)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if id != 1 || name != "test" {
+			t.Fatalf("invalid rows %v %v", id, name)
+		}
+	}
+
+	if err = rows.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	if err = rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+
+	if err = tx.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	if err = c.ExpectationsWereMet(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func Test_GetCases(t *testing.T) {
+	files, err := GetCases("testdata/", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(files) == 0 {
+		t.Fatalf("no files matching")
+	}
+
+	if n := len(files); n != 1 {
+		t.Fatalf("invalid number of test cases %d", n)
+	}
+}

util/test/testdata/result/01_firstcase/Call_db.csv

@@ -0,0 +1,6 @@
+# begin
+# query select \* from test;
+# columns id|VARCHAR,name|VARCHAR
+id,name
+1,test
+# commit

util/test/testdata/result/01_firstcase/Call_req.json

@@ -0,0 +1 @@
+{}

util/test/testdata/result/01_firstcase/Call_rsp.json

@@ -0,0 +1 @@
+{}