Repository: go-kratos/kratos
Pull request #2323 (2022-08-28 11:54:51)

fix: init global selector to wrr
<!--
🎉 Thanks for sending a pull request to Kratos! Here are some tips for you:

1. If this is your first time contributing to Kratos, please read our contribution guide: https://go-kratos.dev/en/docs/community/contribution/
2. Ensure you have added or ran the appropriate tests and lint for your PR, please use `make lint` and `make test` before filing your PR, use `make clean` to tidy your go mod.
3. If the PR is unfinished, you may need to mark it as a WIP(Work In Progress) PR or draft PR
4. Please use a semantic commits format title, such as `<type>[optional scope]: <description>`, see: https://go-kratos.dev/docs/community/contribution#type
5. at the same time, please note that similar work should be submitted in one PR as far as possible to reduce the workload of reviewers. Do not split a work into multiple PR unless it should.
-->

<!--
🎉 感谢您向 Kratos 发送 PR 请求！以下是一些提示：
如果这是你第一次为 Kratos 贡献，请阅读我们的贡献指南：https://go-kratos.dev/en/docs/community/contribution/
2、确保您已经为您的 PR 添加或运行了适当的测试和lint，请在提交PR之前使用“make lint”和“make test”，使用“make clean”整理您的 go.mod。
3、如果请购单未完成，您可能需要将其标记为 WIP（在制品）PR 或 draft PR
4、请使用语义提交格式标题，如“<类型>[可选范围]：<说明>`，请参阅：https://go-kratos.dev/docs/community/contribution#type
5. 同时请注意，同类的工作请尽量在一个PR中提交，以减轻 review 者的工作负担，不要把一项工作拆分成很多个PR，除非它应该这样做。
-->

<!--
-->

#### Description (what this PR does / why we need it):
<!--
* The description should include the motivation for this PR or contrast this with previous behavior
-->


#### Which issue(s) this PR fixes (resolves / be part of):
<!--
* Automatically closes linked issue when PR is merged.
* If your PR is not fully resolved the issue, please use `part of #<issue number>` instead.

Usage: `fixes/resolves #<issue number>`, or `fixes/resolves (paste link of issue)`.
-->


#### Other special notes for the reviewers:
<!--
* Somethings that need extra attention for the reviewers
* Some additional notes, TODO list, etc.
-->


Wang-TaoTao: 官方文档写的默认负载均衡选择器是wrr，但底层实现上为p2c。与包子老师沟通后，将底层初始化选择器修改为wrr，符合文档描述。

Patch:
diff --git a/README.md b/README.md
index f4023812..815be2fc 100644
--- a/README.md
+++ b/README.md
@@ -90,7 +90,6 @@ If you discover a security vulnerability within Kratos, please send an e-mail to
 - [Wechat Group](https://github.com/go-kratos/kratos/issues/682)
 - [Discord Group](https://discord.gg/BWzJsUJ)
 - [go-kratos.dev](https://go-kratos.dev/en)
-- QQ Group: 716486124
 
 ## Contributors
 
diff --git a/cmd/kratos/internal/base/repo.go b/cmd/kratos/internal/base/repo.go
index 55b12ac0..bdcb3f01 100644
--- a/cmd/kratos/internal/base/repo.go
+++ b/cmd/kratos/internal/base/repo.go
@@ -3,13 +3,15 @@ package base
 import (
 	"context"
 	"fmt"
-	stdurl "net/url"
+	"net"
 	"os"
 	"os/exec"
 	"path"
 	"strings"
 )
 
+var unExpandVarPath = []string{"~", ".", ".."}
+
 // Repo is git repository manager.
 type Repo struct {
 	url    string
@@ -18,27 +20,21 @@ type Repo struct {
 }
 
 func repoDir(url string) string {
-	if !strings.Contains(url, "//") {
-		url = "//" + url
-	}
-	if strings.HasPrefix(url, "//git@") {
-		url = "ssh:" + url
-	} else if strings.HasPrefix(url, "//") {
-		url = "https:" + url
+	vcsURL, err := ParseVCSUrl(url)
+	if err != nil {
+		return url
 	}
-	u, err := stdurl.Parse(url)
-	if err == nil {
-		url = fmt.Sprintf("%s://%s%s", u.Scheme, u.Hostname(), u.Path)
+	// check host contains port
+	host, _, err := net.SplitHostPort(vcsURL.Host)
+	if err != nil {
+		host = vcsURL.Host
 	}
-	var start int
-	start = strings.Index(url, "//")
-	if start == -1 {
-		start = strings.Index(url, ":") + 1
-	} else {
-		start += 2
+	for _, p := range unExpandVarPath {
+		host = strings.TrimLeft(host, p)
 	}
-	end := strings.LastIndex(url, "/")
-	return url[start:end]
+	dir := path.Base(path.Dir(vcsURL.Path))
+	url = fmt.Sprintf("%s/%s", host, dir)
+	return url
 }
 
 // NewRepo new a repository manager.
diff --git a/cmd/kratos/internal/base/vcs_url.go b/cmd/kratos/internal/base/vcs_url.go
new file mode 100644
index 00000000..ef76db30
--- /dev/null
+++ b/cmd/kratos/internal/base/vcs_url.go
@@ -0,0 +1,58 @@
+package base
+
+import (
+	"errors"
+	"net/url"
+	"regexp"
+	"strings"
+)
+
+var (
+	scpSyntaxRe = regexp.MustCompile(`^([a-zA-Z0-9_]+)@([a-zA-Z0-9._-]+):(.*)$`)
+	scheme      = []string{"git", "https", "http", "git+ssh", "ssh", "file", "ftp", "ftps"}
+)
+
+// ParseVCSUrl ref https://github.com/golang/go/blob/master/src/cmd/go/internal/vcs/vcs.go
+// see https://go-review.googlesource.com/c/go/+/12226/
+// git url define https://git-scm.com/docs/git-clone#_git_urls
+func ParseVCSUrl(repo string) (*url.URL, error) {
+	var (
+		repoURL *url.URL
+		err     error
+	)
+
+	if m := scpSyntaxRe.FindStringSubmatch(repo); m != nil {
+		// Match SCP-like syntax and convert it to a URL.
+		// Eg, "git@github.com:user/repo" becomes
+		// "ssh://git@github.com/user/repo".
+		repoURL = &url.URL{
+			Scheme: "ssh",
+			User:   url.User(m[1]),
+			Host:   m[2],
+			Path:   m[3],
+		}
+	} else {
+		if !strings.Contains(repo, "//") {
+			repo = "//" + repo
+		}
+		if strings.HasPrefix(repo, "//git@") {
+			repo = "ssh:" + repo
+		} else if strings.HasPrefix(repo, "//") {
+			repo = "https:" + repo
+		}
+		repoURL, err = url.Parse(repo)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// Iterate over insecure schemes too, because this function simply
+	// reports the state of the repo. If we can't see insecure schemes then
+	// we can't report the actual repo URL.
+	for _, s := range scheme {
+		if repoURL.Scheme == s {
+			return repoURL, nil
+		}
+	}
+	return nil, errors.New("unable to parse repo url")
+}
diff --git a/internal/host/host.go b/internal/host/host.go
index d5cc7ffa..bf204eed 100644
--- a/internal/host/host.go
+++ b/internal/host/host.go
@@ -50,16 +50,13 @@ func Extract(hostPort string, lis net.Listener) (string, error) {
 	if err != nil {
 		return "", err
 	}
-	lowest := int(^uint(0) >> 1)
+	minIndex := int(^uint(0) >> 1)
 	var result net.IP
 	for _, iface := range ifaces {
 		if (iface.Flags & net.FlagUp) == 0 {
 			continue
 		}
-		if iface.Index < lowest || result == nil {
-			lowest = iface.Index
-		}
-		if result != nil {
+		if iface.Index >= minIndex && result != nil {
 			continue
 		}
 		addrs, err := iface.Addrs()
@@ -77,7 +74,9 @@ func Extract(hostPort string, lis net.Listener) (string, error) {
 				continue
 			}
 			if isValidIP(ip.String()) {
+				minIndex = iface.Index
 				result = ip
+				break
 			}
 		}
 	}
diff --git a/transport/grpc/client.go b/transport/grpc/client.go
index a1424d20..da6ceb02 100644
--- a/transport/grpc/client.go
+++ b/transport/grpc/client.go
@@ -10,7 +10,7 @@ import (
 	"github.com/go-kratos/kratos/v2/middleware"
 	"github.com/go-kratos/kratos/v2/registry"
 	"github.com/go-kratos/kratos/v2/selector"
-	"github.com/go-kratos/kratos/v2/selector/p2c"
+	"github.com/go-kratos/kratos/v2/selector/wrr"
 	"github.com/go-kratos/kratos/v2/transport"
 	"github.com/go-kratos/kratos/v2/transport/grpc/resolver/discovery"
 
@@ -25,7 +25,7 @@ import (
 
 func init() {
 	if selector.GlobalSelector() == nil {
-		selector.SetGlobalSelector(p2c.NewBuilder())
+		selector.SetGlobalSelector(wrr.NewBuilder())
 	}
 }
 
diff --git a/transport/http/client.go b/transport/http/client.go
index 79ac4f91..c0ad5ee3 100644
--- a/transport/http/client.go
+++ b/transport/http/client.go
@@ -16,13 +16,13 @@ import (
 	"github.com/go-kratos/kratos/v2/middleware"
 	"github.com/go-kratos/kratos/v2/registry"
 	"github.com/go-kratos/kratos/v2/selector"
-	"github.com/go-kratos/kratos/v2/selector/p2c"
+	"github.com/go-kratos/kratos/v2/selector/wrr"
 	"github.com/go-kratos/kratos/v2/transport"
 )
 
 func init() {
 	if selector.GlobalSelector() == nil {
-		selector.SetGlobalSelector(p2c.NewBuilder())
+		selector.SetGlobalSelector(wrr.NewBuilder())
 	}
 }
 


Test patch:
diff --git a/cmd/kratos/internal/base/repo_test.go b/cmd/kratos/internal/base/repo_test.go
index 15ed33c8..dfcdbb93 100644
--- a/cmd/kratos/internal/base/repo_test.go
+++ b/cmd/kratos/internal/base/repo_test.go
@@ -7,7 +7,37 @@ import (
 )
 
 func TestRepo(t *testing.T) {
-	os.RemoveAll("/tmp/test_repo")
+	urls := []string{
+		// ssh://[user@]host.xz[:port]/path/to/repo.git/
+		"ssh://git@github.com:7875/go-kratos/kratos.git",
+		// git://host.xz[:port]/path/to/repo.git/
+		"git://github.com:7875/go-kratos/kratos.git",
+		// http[s]://host.xz[:port]/path/to/repo.git/
+		"https://github.com:7875/go-kratos/kratos.git",
+		// ftp[s]://host.xz[:port]/path/to/repo.git/
+		"ftps://github.com:7875/go-kratos/kratos.git",
+		//[user@]host.xz:path/to/repo.git/
+		"git@github.com:go-kratos/kratos.git",
+		// ssh://[user@]host.xz[:port]/~[user]/path/to/repo.git/
+		"ssh://git@github.com:7875/go-kratos/kratos.git",
+		// git://host.xz[:port]/~[user]/path/to/repo.git/
+		"git://github.com:7875/go-kratos/kratos.git",
+		//[user@]host.xz:/~[user]/path/to/repo.git/
+		"git@github.com:go-kratos/kratos.git",
+		///path/to/repo.git/
+		"//github.com/go-kratos/kratos.git",
+		// file:///path/to/repo.git/
+		"file://./github.com/go-kratos/kratos.git",
+	}
+	for _, url := range urls {
+		dir := repoDir(url)
+		if dir != "github.com/go-kratos" && dir != "/go-kratos" {
+			t.Fatal(url, "repoDir test failed", dir)
+		}
+	}
+}
+
+func TestRepoClone(t *testing.T) {
 	r := NewRepo("https://github.com/go-kratos/service-layout.git", "")
 	if err := r.Clone(context.Background()); err != nil {
 		t.Fatal(err)
@@ -15,19 +45,7 @@ func TestRepo(t *testing.T) {
 	if err := r.CopyTo(context.Background(), "/tmp/test_repo", "github.com/go-kratos/kratos-layout", nil); err != nil {
 		t.Fatal(err)
 	}
-	urls := []string{
-		"ssh://git@gitlab.xxx.com:1234/foo/bar.git",
-		"ssh://gitlab.xxx.com:1234/foo/bar.git",
-		"//git@gitlab.xxx.com:1234/foo/bar.git",
-		"git@gitlab.xxx.com:1234/foo/bar.git",
-		"gitlab.xxx.com:1234/foo/bar.git",
-		"gitlab.xxx.com/foo/bar.git",
-		"gitlab.xxx.com/foo/bar",
-	}
-	for _, url := range urls {
-		dir := repoDir(url)
-		if dir != "gitlab.xxx.com/foo" {
-			t.Fatal("repoDir test failed", dir)
-		}
-	}
+	t.Cleanup(func() {
+		os.RemoveAll("/tmp/test_repo")
+	})
 }
diff --git a/cmd/kratos/internal/base/vcs_url_test.go b/cmd/kratos/internal/base/vcs_url_test.go
new file mode 100644
index 00000000..09b8d568
--- /dev/null
+++ b/cmd/kratos/internal/base/vcs_url_test.go
@@ -0,0 +1,55 @@
+package base
+
+import (
+	"net"
+	"strings"
+	"testing"
+)
+
+func TestParseVCSUrl(t *testing.T) {
+	repos := []string{
+		// ssh://[user@]host.xz[:port]/path/to/repo.git/
+		"ssh://git@github.com:7875/go-kratos/kratos.git",
+		// git://host.xz[:port]/path/to/repo.git/
+		"git://github.com:7875/go-kratos/kratos.git",
+		// http[s]://host.xz[:port]/path/to/repo.git/
+		"https://github.com:7875/go-kratos/kratos.git",
+		// ftp[s]://host.xz[:port]/path/to/repo.git/
+		"ftps://github.com:7875/go-kratos/kratos.git",
+		//[user@]host.xz:path/to/repo.git/
+		"git@github.com:go-kratos/kratos.git",
+		// ssh://[user@]host.xz[:port]/~[user]/path/to/repo.git/
+		"ssh://git@github.com:7875/go-kratos/kratos.git",
+		// git://host.xz[:port]/~[user]/path/to/repo.git/
+		"git://github.com:7875/go-kratos/kratos.git",
+		//[user@]host.xz:/~[user]/path/to/repo.git/
+		"git@github.com:go-kratos/kratos.git",
+		///path/to/repo.git/
+		"~/go-kratos/kratos.git",
+		// file:///path/to/repo.git/
+		"file://~/go-kratos/kratos.git",
+	}
+	for _, repo := range repos {
+		url, err := ParseVCSUrl(repo)
+		if err != nil {
+			t.Fatal(repo, err)
+		}
+		urlPath := strings.TrimLeft(url.Path, "/")
+		if urlPath != "go-kratos/kratos.git" {
+			t.Fatal(repo, "parse url failed", urlPath)
+		}
+	}
+}
+
+func TestParseSsh(t *testing.T) {
+	repo := "ssh://git@github.com:7875/go-kratos/kratos.git"
+	url, err := ParseVCSUrl(repo)
+	if err != nil {
+		t.Fatal(err)
+	}
+	host, _, err := net.SplitHostPort(url.Host)
+	if err != nil {
+		host = url.Host
+	}
+	t.Log(host, url.Path)
+}
diff --git a/selector/selector_test.go b/selector/selector_test.go
index f7c1f1bc..bcc08264 100644
--- a/selector/selector_test.go
+++ b/selector/selector_test.go
@@ -12,6 +12,8 @@ import (
 	"github.com/go-kratos/kratos/v2/registry"
 )
 
+var errNodeNotMatch = errors.New("node is not match")
+
 type mockWeightedNode struct {
 	Node
 
@@ -80,6 +82,18 @@ func (b *mockBalancer) Pick(ctx context.Context, nodes []WeightedNode) (selected
 	return
 }
 
+type mockMustErrorBalancerBuilder struct{}
+
+func (b *mockMustErrorBalancerBuilder) Build() Balancer {
+	return &mockMustErrorBalancer{}
+}
+
+type mockMustErrorBalancer struct{}
+
+func (b *mockMustErrorBalancer) Pick(ctx context.Context, nodes []WeightedNode) (selected WeightedNode, done DoneFunc, err error) {
+	return nil, nil, errNodeNotMatch
+}
+
 func TestDefault(t *testing.T) {
 	builder := DefaultBuilder{
 		Node:     &mockWeightedNodeBuilder{},
@@ -107,6 +121,7 @@ func TestDefault(t *testing.T) {
 			Endpoints: []string{"http://127.0.0.1:9090"},
 			Metadata:  map[string]string{"weight": "10"},
 		}))
+
 	selector.Apply(nodes)
 	n, done, err := selector.Select(context.Background(), WithNodeFilter(mockFilter("v2.0.0")))
 	if err != nil {
@@ -121,6 +136,9 @@ func TestDefault(t *testing.T) {
 	if !reflect.DeepEqual("v2.0.0", n.Version()) {
 		t.Errorf("expect %v, got %v", "v2.0.0", n.Version())
 	}
+	if n.Scheme() == "" {
+		t.Errorf("expect %v, got %v", "", n.Scheme())
+	}
 	if n.Address() == "" {
 		t.Errorf("expect %v, got %v", "", n.Address())
 	}
@@ -135,6 +153,21 @@ func TestDefault(t *testing.T) {
 	}
 	done(context.Background(), DoneInfo{})
 
+	// peer in ctx
+	ctx := NewPeerContext(context.Background(), &Peer{
+		Node: mockWeightedNode{},
+	})
+	n, done, err = selector.Select(ctx)
+	if err != nil {
+		t.Errorf("expect %v, got %v", ErrNoAvailable, err)
+	}
+	if done == nil {
+		t.Errorf("expect %v, got %v", nil, done)
+	}
+	if n == nil {
+		t.Errorf("expect %v, got %v", nil, n)
+	}
+
 	// no v3.0.0 instance
 	n, done, err = selector.Select(context.Background(), WithNodeFilter(mockFilter("v3.0.0")))
 	if !errors.Is(ErrNoAvailable, err) {
@@ -172,4 +205,87 @@ func TestDefault(t *testing.T) {
 	if n != nil {
 		t.Errorf("expect %v, got %v", nil, n)
 	}
+
+	// without node_filters
+	n, done, err = selector.Select(context.Background())
+	if !errors.Is(ErrNoAvailable, err) {
+		t.Errorf("expect %v, got %v", ErrNoAvailable, err)
+	}
+	if done != nil {
+		t.Errorf("expect %v, got %v", nil, done)
+	}
+	if n != nil {
+		t.Errorf("expect %v, got %v", nil, n)
+	}
+}
+
+func TestWithoutApply(t *testing.T) {
+	builder := DefaultBuilder{
+		Node:     &mockWeightedNodeBuilder{},
+		Balancer: &mockBalancerBuilder{},
+	}
+	selector := builder.Build()
+	n, done, err := selector.Select(context.Background())
+	if !errors.Is(ErrNoAvailable, err) {
+		t.Errorf("expect %v, got %v", ErrNoAvailable, err)
+	}
+	if done != nil {
+		t.Errorf("expect %v, got %v", nil, done)
+	}
+	if n != nil {
+		t.Errorf("expect %v, got %v", nil, n)
+	}
+}
+
+func TestNoPick(t *testing.T) {
+	builder := DefaultBuilder{
+		Node:     &mockWeightedNodeBuilder{},
+		Balancer: &mockMustErrorBalancerBuilder{},
+	}
+	var nodes []Node
+	nodes = append(nodes, NewNode(
+		"http",
+		"127.0.0.1:8080",
+		&registry.ServiceInstance{
+			ID:        "127.0.0.1:8080",
+			Name:      "helloworld",
+			Version:   "v2.0.0",
+			Endpoints: []string{"http://127.0.0.1:8080"},
+			Metadata:  map[string]string{"weight": "10"},
+		}))
+	nodes = append(nodes, NewNode(
+		"http",
+		"127.0.0.1:9090",
+		&registry.ServiceInstance{
+			ID:        "127.0.0.1:9090",
+			Name:      "helloworld",
+			Version:   "v1.0.0",
+			Endpoints: []string{"http://127.0.0.1:9090"},
+			Metadata:  map[string]string{"weight": "10"},
+		}))
+	selector := builder.Build()
+	selector.Apply(nodes)
+	n, done, err := selector.Select(context.Background())
+	if !errors.Is(errNodeNotMatch, err) {
+		t.Errorf("expect %v, got %v", errNodeNotMatch, err)
+	}
+	if done != nil {
+		t.Errorf("expect %v, got %v", nil, done)
+	}
+	if n != nil {
+		t.Errorf("expect %v, got %v", nil, n)
+	}
+}
+
+func TestGolobal(t *testing.T) {
+	builder := DefaultBuilder{
+		Node:     &mockWeightedNodeBuilder{},
+		Balancer: &mockBalancerBuilder{},
+	}
+	SetGlobalSelector(&builder)
+
+	gBuilder := GlobalSelector()
+	if gBuilder == nil {
+		t.Errorf("expect %v, got %v", nil, gBuilder)
+	}
 }