Repository: go-kratos/kratos
Pull request #2460 (2022-10-18 08:36:31)

fix(selector/ewma): judge net error
dial tcp 127.0.0.1:9000: connect: connection refused
dial tcp 127.0.0.1:9000: i/o timeout


Patch:
diff --git a/api/metadata/server.go b/api/metadata/server.go
index 3600fc14..b1c4c17b 100644
--- a/api/metadata/server.go
+++ b/api/metadata/server.go
@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"sort"
 	"sync"
 
 	"google.golang.org/grpc"
@@ -115,6 +116,8 @@ func (s *Server) ListServices(ctx context.Context, in *ListServicesRequest) (*Li
 			reply.Methods = append(reply.Methods, fmt.Sprintf("/%s/%s", name, method))
 		}
 	}
+	sort.Strings(reply.Services)
+	sort.Strings(reply.Methods)
 	return reply, nil
 }
 
diff --git a/contrib/config/etcd/README.md b/contrib/config/etcd/README.md
index d7668745..d69edd82 100644
--- a/contrib/config/etcd/README.md
+++ b/contrib/config/etcd/README.md
@@ -30,12 +30,16 @@ if err != nil {
 c := config.New(config.WithSource(source))
 defer c.Close()
 
+// load sources before get
+if err := c.Load(); err != nil {
+    log.Fatalln(err)
+}
+
 // acquire config value
 foo, err := c.Value("/app-config").String()
 if err != nil {
-    log.Println(err)
+    log.Fatalln(err)
 }
-println(foo)
 
+log.Println(foo)
 ```
-
diff --git a/contrib/registry/eureka/client.go b/contrib/registry/eureka/client.go
index 46b83416..020a9a90 100644
--- a/contrib/registry/eureka/client.go
+++ b/contrib/registry/eureka/client.go
@@ -307,39 +307,50 @@ func (e *Client) buildAPI(currentTimes int, params ...string) string {
 	return strings.Join(params, "/")
 }
 
-func (e *Client) do(ctx context.Context, method string, params []string, input io.Reader, output interface{}) error {
-	for i := 0; i < e.maxRetry; i++ {
-		request, err := http.NewRequest(method, e.buildAPI(i, params...), input)
+func (e *Client) request(ctx context.Context, method string, params []string, input io.Reader, output interface{}, i int) (bool, error) {
+	request, err := http.NewRequestWithContext(ctx, method, e.buildAPI(i, params...), input)
+	if err != nil {
+		return false, err
+	}
+	request.Header.Add("User-Agent", "go-eureka-client")
+	request.Header.Add("Accept", "application/json;charset=UTF-8")
+	request.Header.Add("Content-Type", "application/json;charset=UTF-8")
+	resp, err := e.client.Do(request)
+	if err != nil {
+		return true, err
+	}
+	defer func() {
+		_, _ = io.Copy(io.Discard, resp.Body)
+		_ = resp.Body.Close()
+	}()
+
+	if output != nil && resp.StatusCode/100 == 2 {
+		data, err := io.ReadAll(resp.Body)
 		if err != nil {
-			return err
+			return false, err
 		}
-		request = request.WithContext(ctx)
-		request.Header.Add("User-Agent", "go-eureka-client")
-		request.Header.Add("Accept", "application/json;charset=UTF-8")
-		request.Header.Add("Content-Type", "application/json;charset=UTF-8")
-		resp, err := e.client.Do(request)
+		err = json.Unmarshal(data, output)
 		if err != nil {
-			continue
-		}
-		defer func() {
-			_, _ = io.Copy(io.Discard, resp.Body)
-			resp.Body.Close()
-		}()
-
-		if output != nil && resp.StatusCode/100 == 2 {
-			data, err := io.ReadAll(resp.Body)
-			if err != nil {
-				return err
-			}
-			if err = json.Unmarshal(data, output); err != nil {
-				return err
-			}
+			return false, err
 		}
+	}
 
-		if resp.StatusCode >= http.StatusBadRequest {
-			return fmt.Errorf("response Error %d", resp.StatusCode)
-		}
+	if resp.StatusCode >= http.StatusBadRequest {
+		return false, fmt.Errorf("response Error %d", resp.StatusCode)
+	}
 
+	return false, nil
+}
+
+func (e *Client) do(ctx context.Context, method string, params []string, input io.Reader, output interface{}) error {
+	for i := 0; i < e.maxRetry; i++ {
+		retry, err := e.request(ctx, method, params, input, output, i)
+		if retry {
+			continue
+		}
+		if err != nil {
+			return err
+		}
 		return nil
 	}
 	return fmt.Errorf("retry after %d times", e.maxRetry)
diff --git a/middleware/metadata/metadata.go b/middleware/metadata/metadata.go
index 5c29189f..da9e43a9 100644
--- a/middleware/metadata/metadata.go
+++ b/middleware/metadata/metadata.go
@@ -51,16 +51,19 @@ func Server(opts ...Option) middleware.Middleware {
 	}
 	return func(handler middleware.Handler) middleware.Handler {
 		return func(ctx context.Context, req interface{}) (reply interface{}, err error) {
-			if tr, ok := transport.FromServerContext(ctx); ok {
-				md := options.md.Clone()
-				header := tr.RequestHeader()
-				for _, k := range header.Keys() {
-					if options.hasPrefix(k) {
-						md.Set(k, header.Get(k))
-					}
+			tr, ok := transport.FromServerContext(ctx)
+			if !ok {
+				return handler(ctx, req)
+			}
+
+			md := options.md.Clone()
+			header := tr.RequestHeader()
+			for _, k := range header.Keys() {
+				if options.hasPrefix(k) {
+					md.Set(k, header.Get(k))
 				}
-				ctx = metadata.NewServerContext(ctx, md)
 			}
+			ctx = metadata.NewServerContext(ctx, md)
 			return handler(ctx, req)
 		}
 	}
@@ -76,25 +79,28 @@ func Client(opts ...Option) middleware.Middleware {
 	}
 	return func(handler middleware.Handler) middleware.Handler {
 		return func(ctx context.Context, req interface{}) (reply interface{}, err error) {
-			if tr, ok := transport.FromClientContext(ctx); ok {
-				header := tr.RequestHeader()
-				// x-md-local-
-				for k, v := range options.md {
+			tr, ok := transport.FromClientContext(ctx)
+			if !ok {
+				return handler(ctx, req)
+			}
+
+			header := tr.RequestHeader()
+			// x-md-local-
+			for k, v := range options.md {
+				header.Set(k, v)
+			}
+			if md, ok := metadata.FromClientContext(ctx); ok {
+				for k, v := range md {
 					header.Set(k, v)
 				}
-				if md, ok := metadata.FromClientContext(ctx); ok {
-					for k, v := range md {
+			}
+			// x-md-global-
+			if md, ok := metadata.FromServerContext(ctx); ok {
+				for k, v := range md {
+					if options.hasPrefix(k) {
 						header.Set(k, v)
 					}
 				}
-				// x-md-global-
-				if md, ok := metadata.FromServerContext(ctx); ok {
-					for k, v := range md {
-						if options.hasPrefix(k) {
-							header.Set(k, v)
-						}
-					}
-				}
 			}
 			return handler(ctx, req)
 		}
diff --git a/selector/node/ewma/node.go b/selector/node/ewma/node.go
index 4fed4aaa..46d5c41c 100644
--- a/selector/node/ewma/node.go
+++ b/selector/node/ewma/node.go
@@ -4,6 +4,7 @@ import (
 	"container/list"
 	"context"
 	"math"
+	"net"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -158,8 +159,9 @@ func (n *Node) Pick() selector.DoneFunc {
 			if n.errHandler != nil && n.errHandler(di.Err) {
 				success = 0
 			}
+			var netErr net.Error
 			if errors.Is(context.DeadlineExceeded, di.Err) || errors.Is(context.Canceled, di.Err) ||
-				errors.IsServiceUnavailable(di.Err) || errors.IsGatewayTimeout(di.Err) {
+				errors.IsServiceUnavailable(di.Err) || errors.IsGatewayTimeout(di.Err) || errors.As(di.Err, netErr) {
 				success = 0
 			}
 		}


Test patch:
diff --git a/contrib/log/fluent/fluent_test.go b/contrib/log/fluent/fluent_test.go
index 5b7b3e17..41f7080d 100644
--- a/contrib/log/fluent/fluent_test.go
+++ b/contrib/log/fluent/fluent_test.go
@@ -11,18 +11,23 @@ import (
 )
 
 func TestMain(m *testing.M) {
+	listener := func(ln net.Listener) {
+		conn, err := ln.Accept()
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		_, err = io.ReadAll(conn)
+		if err != nil {
+			return
+		}
+	}
+
 	if ln, err := net.Listen("tcp", ":24224"); err == nil {
 		defer ln.Close()
 		go func() {
 			for {
-				conn, err := ln.Accept()
-				if err != nil {
-					return
-				}
-				defer conn.Close()
-				if _, err = io.ReadAll(conn); err != nil {
-					continue
-				}
+				listener(ln)
 			}
 		}()
 	}
diff --git a/middleware/auth/jwt/jwt_test.go b/middleware/auth/jwt/jwt_test.go
index 188744b0..711757bc 100644
--- a/middleware/auth/jwt/jwt_test.go
+++ b/middleware/auth/jwt/jwt_test.go
@@ -254,7 +254,7 @@ func TestServer(t *testing.T) {
 			}
 			if test.exceptErr == nil {
 				if testToken == nil {
-					t.Errorf("except testToken not nil, but got nil")
+					t.Fatal("except testToken not nil, but got nil")
 				}
 				_, ok := testToken.(jwt.MapClaims)
 				if !ok {
@@ -493,3 +493,26 @@ func TestClientMissKey(t *testing.T) {
 		}
 	})
 }
+
+func TestNewContextAndFromContext(t *testing.T) {
+	tests := []struct {
+		name   string
+		claims jwt.MapClaims
+	}{
+		{"val not nil", jwt.MapClaims{"name": "kratos"}},
+		{"val nil", nil},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			ctx := NewContext(context.Background(), test.claims)
+
+			claims, ok := FromContext(ctx)
+			if !ok {
+				t.Fatal("ctx not found authKey{}")
+			}
+			if !reflect.DeepEqual(test.claims, claims) {
+				t.Errorf(`want: %s, got: %v`, test.claims, claims)
+			}
+		})
+	}
+}
diff --git a/middleware/metadata/metadata_test.go b/middleware/metadata/metadata_test.go
index e46a94f5..b083f9c1 100644
--- a/middleware/metadata/metadata_test.go
+++ b/middleware/metadata/metadata_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"net/http"
+	"reflect"
 	"testing"
 
 	"github.com/go-kratos/kratos/v2/metadata"
@@ -33,15 +34,18 @@ func (tr *testTransport) Operation() string               { return "" }
 func (tr *testTransport) RequestHeader() transport.Header { return tr.header }
 func (tr *testTransport) ReplyHeader() transport.Header   { return tr.header }
 
+var (
+	globalKey   = "x-md-global-key"
+	globalValue = "global-value"
+	localKey    = "x-md-local-key"
+	localValue  = "local-value"
+	customKey   = "x-md-local-custom"
+	customValue = "custom-value"
+	constKey    = "x-md-local-const"
+	constValue  = "x-md-local-const"
+)
+
 func TestSever(t *testing.T) {
-	var (
-		globalKey   = "x-md-global-key"
-		globalValue = "global-value"
-		localKey    = "x-md-local-key"
-		localValue  = "local-value"
-		constKey    = "x-md-local-const"
-		constValue  = "x-md-local-const"
-	)
 	hs := func(ctx context.Context, in interface{}) (interface{}, error) {
 		md, ok := metadata.FromServerContext(ctx)
 		if !ok {
@@ -75,16 +79,6 @@ func TestSever(t *testing.T) {
 }
 
 func TestClient(t *testing.T) {
-	var (
-		globalKey   = "x-md-global-key"
-		globalValue = "global-value"
-		localKey    = "x-md-local-key"
-		localValue  = "local-value"
-		customKey   = "x-md-local-custom"
-		customValue = "custom-value"
-		constKey    = "x-md-local-const"
-		constValue  = "x-md-local-const"
-	)
 	hs := func(ctx context.Context, in interface{}) (interface{}, error) {
 		tr, ok := transport.FromClientContext(ctx)
 		if !ok {
@@ -127,13 +121,52 @@ func TestClient(t *testing.T) {
 	}
 }
 
-func Test_WithPropagatedPrefix(t *testing.T) {
-	o := &options{
+func TestWithConstants(t *testing.T) {
+	md := metadata.Metadata{
+		constKey: constValue,
+	}
+	options := &options{
+		md: metadata.Metadata{
+			"override": "override",
+		},
+	}
+
+	WithConstants(md)(options)
+	if !reflect.DeepEqual(md, options.md) {
+		t.Errorf("want: %v, got: %v", md, options.md)
+	}
+}
+
+func TestOptions_WithPropagatedPrefix(t *testing.T) {
+	options := &options{
 		prefix: []string{"override"},
 	}
-	WithPropagatedPrefix("something", "another")(o)
+	prefixes := []string{"something", "another"}
 
-	if len(o.prefix) != 2 {
+	WithPropagatedPrefix(prefixes...)(options)
+	if !reflect.DeepEqual(prefixes, options.prefix) {
 		t.Error("The prefix must be overridden.")
 	}
 }
+
+func TestOptions_hasPrefix(t *testing.T) {
+	tests := []struct {
+		name    string
+		options *options
+		key     string
+		exists  bool
+	}{
+		{"exists key upper", &options{prefix: []string{"prefix"}}, "PREFIX_true", true},
+		{"exists key lower", &options{prefix: []string{"prefix"}}, "prefix_true", true},
+		{"not exists key upper", &options{prefix: []string{"prefix"}}, "false_PREFIX", false},
+		{"not exists key lower", &options{prefix: []string{"prefix"}}, "false_prefix", false},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			exists := test.options.hasPrefix(test.key)
+			if test.exists != exists {
+				t.Errorf("key: '%sr', not exists prefixs: %v", test.key, test.options.prefix)
+			}
+		})
+	}
+}