Repository: go-kratos/kratos
Pull request #1650 (2021-11-24 09:43:08)

feat(registry/nacos): add default kind option
<!--
🎉 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 mark it as a WIP(Work In Progress) PR or draft PR
4. Please use a conventional commits format title: `<type>[optional scope]: <description>`
    
    Some suggestion on <type>:

    refactor: A code change that neither fixes a bug nor adds a feature   
-->

#### Description (what this PR does / why we need it):
<!--
* The description should include the motivation for this PR or contrast this with previous behavior
-->
与java的dubbo框架写的grpc服务集成，元数据缺失kind字段信息也能正常访问

#### Which issue(s) this PR fixes (resolves / be part of):
<!--
* Automatically closes linked issue when PR is merged.
* If you PR is not fully resolved issue, please use `part of #<issue number>` instead.

Usage: `fixes/resolves #<issue number>`, or `fixes/resolves (paste link of issue)`.
-->
fixes #

#### Other special notes for reviewer:
<!--
* Somethings that need extra attention for reviewer
* Some additional notes, TODO list, etc.
-->


shenqidebaozi: 感觉这么改不是很合理呀，应该在元数据上加上 kind=xxx，举个例子，我司用的是 spring cloud + kratos，spring cloud 默认 HTTP。如果你这样改了。其他用户怎么办呢？
codecov-commenter: # [Codecov](https://codecov.io/gh/go-kratos/kratos/pull/1650?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos) Report
> Merging [#1650](https://codecov.io/gh/go-kratos/kratos/pull/1650?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos) (a0d4343) into [main](https://codecov.io/gh/go-kratos/kratos/commit/ef3322ec076462e192756f90b96e87aaa2c03271?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos) (ef3322e) will **increase** coverage by `0.08%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/go-kratos/kratos/pull/1650/graphs/tree.svg?width=650&height=150&src=pr&token=XwnmBTz78x&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos)](https://codecov.io/gh/go-kratos/kratos/pull/1650?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos)

```diff
@@            Coverage Diff             @@
##             main    #1650      +/-   ##
==========================================
+ Coverage   81.18%   81.27%   +0.08%     
==========================================
  Files          78       78              
  Lines        3429     3429              
==========================================
+ Hits         2784     2787       +3     
+ Misses        443      441       -2     
+ Partials      202      201       -1     
```


| [Impacted Files](https://codecov.io/gh/go-kratos/kratos/pull/1650?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos) | Coverage Δ | |
|---|---|---|
| [selector/node/ewma/node.go](https://codecov.io/gh/go-kratos/kratos/pull/1650/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos#diff-c2VsZWN0b3Ivbm9kZS9ld21hL25vZGUuZ28=) | `87.09% <0.00%> (+3.22%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/go-kratos/kratos/pull/1650?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/go-kratos/kratos/pull/1650?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos). Last update [ef3322e...a0d4343](https://codecov.io/gh/go-kratos/kratos/pull/1650?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos).


Patch:
diff --git a/contrib/registry/nacos/registry.go b/contrib/registry/nacos/registry.go
index 56414ef7..327b5109 100644
--- a/contrib/registry/nacos/registry.go
+++ b/contrib/registry/nacos/registry.go
@@ -23,6 +23,7 @@ type options struct {
 	weight  float64
 	cluster string
 	group   string
+	kind    string
 }
 
 // Option is nacos option.
@@ -48,6 +49,11 @@ func WithGroup(group string) Option {
 	return func(o *options) { o.group = group }
 }
 
+// WithDefaultKind with default kind option.
+func WithDefaultKind(kind string) Option {
+	return func(o *options) { o.kind = kind }
+}
+
 // Registry is nacos registry.
 type Registry struct {
 	opts options
@@ -61,6 +67,7 @@ func New(cli naming_client.INamingClient, opts ...Option) (r *Registry) {
 		cluster: "DEFAULT",
 		group:   "DEFAULT_GROUP",
 		weight:  100,
+		kind:    "grpc",
 	}
 	for _, option := range opts {
 		option(&op)
@@ -144,7 +151,7 @@ func (r *Registry) Deregister(ctx context.Context, service *registry.ServiceInst
 
 // Watch creates a watcher according to the service name.
 func (r *Registry) Watch(ctx context.Context, serviceName string) (registry.Watcher, error) {
-	return newWatcher(ctx, r.cli, serviceName, r.opts.group, []string{r.opts.cluster})
+	return newWatcher(ctx, r.cli, serviceName, r.opts.group, r.opts.kind, []string{r.opts.cluster})
 }
 
 // GetService return the service instances in memory according to the service name.
@@ -158,12 +165,16 @@ func (r *Registry) GetService(ctx context.Context, serviceName string) ([]*regis
 	}
 	items := make([]*registry.ServiceInstance, 0, len(res))
 	for _, in := range res {
+		kind := r.opts.kind
+		if k, ok := in.Metadata["kind"]; ok {
+			kind = k
+		}
 		items = append(items, &registry.ServiceInstance{
 			ID:        in.InstanceId,
 			Name:      in.ServiceName,
 			Version:   in.Metadata["version"],
 			Metadata:  in.Metadata,
-			Endpoints: []string{fmt.Sprintf("%s://%s:%d", in.Metadata["kind"], in.Ip, in.Port)},
+			Endpoints: []string{fmt.Sprintf("%s://%s:%d", kind, in.Ip, in.Port)},
 		})
 	}
 	return items, nil
diff --git a/contrib/registry/nacos/watcher.go b/contrib/registry/nacos/watcher.go
index 22961b69..eac90b12 100644
--- a/contrib/registry/nacos/watcher.go
+++ b/contrib/registry/nacos/watcher.go
@@ -20,14 +20,16 @@ type watcher struct {
 	cancel      context.CancelFunc
 	watchChan   chan struct{}
 	cli         naming_client.INamingClient
+	kind        string
 }
 
-func newWatcher(ctx context.Context, cli naming_client.INamingClient, serviceName string, groupName string, clusters []string) (*watcher, error) {
+func newWatcher(ctx context.Context, cli naming_client.INamingClient, serviceName, groupName, kind string, clusters []string) (*watcher, error) {
 	w := &watcher{
 		serviceName: serviceName,
 		clusters:    clusters,
 		groupName:   groupName,
 		cli:         cli,
+		kind:        kind,
 		watchChan:   make(chan struct{}, 1),
 	}
 	w.ctx, w.cancel = context.WithCancel(ctx)
@@ -59,12 +61,16 @@ func (w *watcher) Next() ([]*registry.ServiceInstance, error) {
 	}
 	items := make([]*registry.ServiceInstance, 0, len(res.Hosts))
 	for _, in := range res.Hosts {
+		kind := w.kind
+		if k, ok := in.Metadata["kind"]; ok {
+			kind = k
+		}
 		items = append(items, &registry.ServiceInstance{
 			ID:        in.InstanceId,
 			Name:      res.Name,
 			Version:   in.Metadata["version"],
 			Metadata:  in.Metadata,
-			Endpoints: []string{fmt.Sprintf("%s://%s:%d", in.Metadata["kind"], in.Ip, in.Port)},
+			Endpoints: []string{fmt.Sprintf("%s://%s:%d", kind, in.Ip, in.Port)},
 		})
 	}
 	return items, nil
diff --git a/examples/selector/client/main.go b/examples/selector/client/main.go
index 30517e48..f490646a 100644
--- a/examples/selector/client/main.go
+++ b/examples/selector/client/main.go
@@ -8,7 +8,6 @@ import (
 	"github.com/go-kratos/kratos/contrib/registry/consul/v2"
 	"github.com/go-kratos/kratos/examples/helloworld/helloworld"
 	"github.com/go-kratos/kratos/v2/middleware/recovery"
-	"github.com/go-kratos/kratos/v2/selector"
 	"github.com/go-kratos/kratos/v2/selector/filter"
 	"github.com/go-kratos/kratos/v2/selector/p2c"
 	"github.com/go-kratos/kratos/v2/selector/wrr"
@@ -32,10 +31,8 @@ func main() {
 		// 由于gRPC框架的限制只能使用全局balancer+filter的方式来实现selector
 		// 这里使用weighted round robin算法的balancer+静态version=1.0.0的Filter
 		grpc.WithBalancerName(wrr.Name),
-		grpc.WithNodeFilter(
-			func(node selector.Node) bool {
-				return node.Version() == "1.0.0"
-			},
+		grpc.WithFilter(
+			filter.Version("1.0.0"),
 		),
 	)
 	if err != nil {
diff --git a/selector/default_selector.go b/selector/default_selector.go
index ae5e4596..7550befd 100644
--- a/selector/default_selector.go
+++ b/selector/default_selector.go
@@ -27,7 +27,7 @@ func (d *Default) Select(ctx context.Context, opts ...SelectOption) (selected No
 	for _, o := range opts {
 		o(&options)
 	}
-	if len(d.Filters) > 0 {
+	if len(d.Filters) > 0 || len(options.Filters) > 0 {
 		newNodes := make([]Node, len(nodes))
 		for i, wc := range nodes {
 			newNodes[i] = wc
@@ -35,6 +35,9 @@ func (d *Default) Select(ctx context.Context, opts ...SelectOption) (selected No
 		for _, f := range d.Filters {
 			newNodes = f(ctx, newNodes)
 		}
+		for _, f := range options.Filters {
+			newNodes = f(ctx, newNodes)
+		}
 		candidates = make([]WeightedNode, len(newNodes))
 		for i, n := range newNodes {
 			candidates[i] = n.(WeightedNode)
@@ -43,9 +46,6 @@ func (d *Default) Select(ctx context.Context, opts ...SelectOption) (selected No
 		candidates = nodes
 	}
 
-	if len(options.Filters) > 0 {
-		candidates = d.nodeFilter(options.Filters, candidates)
-	}
 	if len(candidates) == 0 {
 		return nil, nil, ErrNoAvailable
 	}
@@ -56,23 +56,6 @@ func (d *Default) Select(ctx context.Context, opts ...SelectOption) (selected No
 	return wn.Raw(), done, nil
 }
 
-func (d *Default) nodeFilter(filters []NodeFilter, nodes []WeightedNode) []WeightedNode {
-	newNodes := make([]WeightedNode, 0, len(nodes))
-	for _, n := range nodes {
-		var remove bool
-		for _, f := range filters {
-			if !f(n) {
-				remove = true
-				break
-			}
-		}
-		if !remove {
-			newNodes = append(newNodes, n)
-		}
-	}
-	return newNodes
-}
-
 // Apply update nodes info.
 func (d *Default) Apply(nodes []Node) {
 	weightedNodes := make([]WeightedNode, 0, len(nodes))
diff --git a/selector/filter.go b/selector/filter.go
index 008be658..e88a52da 100644
--- a/selector/filter.go
+++ b/selector/filter.go
@@ -4,7 +4,3 @@ import "context"
 
 // Filter is select filter.
 type Filter func(context.Context, []Node) []Node
-
-// NodeFilter is node filter.
-// If it returns false, the node will be removed out from the balancer pick list
-type NodeFilter func(node Node) bool
diff --git a/selector/filter/version.go b/selector/filter/version.go
index fdcfcc7e..815d3fb7 100644
--- a/selector/filter/version.go
+++ b/selector/filter/version.go
@@ -9,12 +9,12 @@ import (
 // Version is version filter.
 func Version(version string) selector.Filter {
 	return func(_ context.Context, nodes []selector.Node) []selector.Node {
-		filters := make([]selector.Node, 0, len(nodes))
+		newNodes := nodes[:0]
 		for _, n := range nodes {
 			if n.Version() == version {
-				filters = append(filters, n)
+				newNodes = append(newNodes, n)
 			}
 		}
-		return filters
+		return newNodes
 	}
 }
diff --git a/selector/options.go b/selector/options.go
index 7e1cca4b..c9129fb3 100644
--- a/selector/options.go
+++ b/selector/options.go
@@ -2,14 +2,14 @@ package selector
 
 // SelectOptions is Select Options.
 type SelectOptions struct {
-	Filters []NodeFilter
+	Filters []Filter
 }
 
 // SelectOption is Selector option.
 type SelectOption func(*SelectOptions)
 
-// WithNodeFilter with filter options
-func WithNodeFilter(fn ...NodeFilter) SelectOption {
+// WithFilter with filter options
+func WithFilter(fn ...Filter) SelectOption {
 	return func(opts *SelectOptions) {
 		opts.Filters = fn
 	}
diff --git a/transport/grpc/balancer.go b/transport/grpc/balancer.go
index d44c96bf..5731bb9e 100644
--- a/transport/grpc/balancer.go
+++ b/transport/grpc/balancer.go
@@ -71,14 +71,14 @@ type Picker struct {
 
 // Pick pick instances.
 func (p *Picker) Pick(info gBalancer.PickInfo) (gBalancer.PickResult, error) {
-	var filters []selector.NodeFilter
+	var filters []selector.Filter
 	if tr, ok := transport.FromClientContext(info.Ctx); ok {
 		if gtr, ok := tr.(*Transport); ok {
-			filters = gtr.NodeFilters()
+			filters = gtr.SelectFilters()
 		}
 	}
 
-	n, done, err := p.selector.Select(info.Ctx, selector.WithNodeFilter(filters...))
+	n, done, err := p.selector.Select(info.Ctx, selector.WithFilter(filters...))
 	if err != nil {
 		return gBalancer.PickResult{}, err
 	}
diff --git a/transport/grpc/client.go b/transport/grpc/client.go
index a84a09a7..9d7ede93 100644
--- a/transport/grpc/client.go
+++ b/transport/grpc/client.go
@@ -80,8 +80,8 @@ func WithBalancerName(name string) ClientOption {
 	}
 }
 
-// WithNodeFilter with select filters
-func WithNodeFilter(filters ...selector.NodeFilter) ClientOption {
+// WithFilter with select filters
+func WithFilter(filters ...selector.Filter) ClientOption {
 	return func(o *clientOptions) {
 		o.filters = filters
 	}
@@ -97,7 +97,7 @@ type clientOptions struct {
 	ints         []grpc.UnaryClientInterceptor
 	grpcOpts     []grpc.DialOption
 	balancerName string
-	filters      []selector.NodeFilter
+	filters      []selector.Filter
 }
 
 // Dial returns a GRPC connection.
@@ -143,7 +143,7 @@ func dial(ctx context.Context, insecure bool, opts ...ClientOption) (*grpc.Clien
 	return grpc.DialContext(ctx, options.endpoint, grpcOpts...)
 }
 
-func unaryClientInterceptor(ms []middleware.Middleware, timeout time.Duration, filters []selector.NodeFilter) grpc.UnaryClientInterceptor {
+func unaryClientInterceptor(ms []middleware.Middleware, timeout time.Duration, filters []selector.Filter) grpc.UnaryClientInterceptor {
 	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
 		ctx = transport.NewClientContext(ctx, &Transport{
 			endpoint:  cc.Target(),
diff --git a/transport/grpc/transport.go b/transport/grpc/transport.go
index b7df2631..efa70e74 100644
--- a/transport/grpc/transport.go
+++ b/transport/grpc/transport.go
@@ -14,7 +14,7 @@ type Transport struct {
 	operation   string
 	reqHeader   headerCarrier
 	replyHeader headerCarrier
-	filters     []selector.NodeFilter
+	filters     []selector.Filter
 }
 
 // Kind returns the transport kind.
@@ -42,8 +42,8 @@ func (tr *Transport) ReplyHeader() transport.Header {
 	return tr.replyHeader
 }
 
-// Filters returns the client select filters.
-func (tr *Transport) NodeFilters() []selector.NodeFilter {
+// SelectFilters returns the client select filters.
+func (tr *Transport) SelectFilters() []selector.Filter {
 	return tr.filters
 }
 


Test patch:
diff --git a/selector/selector_test.go b/selector/selector_test.go
index f6f3838d..7c735a33 100644
--- a/selector/selector_test.go
+++ b/selector/selector_test.go
@@ -3,7 +3,6 @@ package selector
 import (
 	"context"
 	"math/rand"
-	"strconv"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -51,13 +50,13 @@ func (b *mockWeightedNodeBuilder) Build(n Node) WeightedNode {
 
 func mockFilter(version string) Filter {
 	return func(_ context.Context, nodes []Node) []Node {
-		filters := make([]Node, 0, len(nodes))
+		newNodes := nodes[:0]
 		for _, n := range nodes {
 			if n.Version() == version {
-				filters = append(filters, n)
+				newNodes = append(newNodes, n)
 			}
 		}
-		return filters
+		return newNodes
 	}
 }
 
@@ -107,9 +106,7 @@ func TestDefault(t *testing.T) {
 			Metadata:  map[string]string{"weight": "10"},
 		}))
 	selector.Apply(nodes)
-	n, done, err := selector.Select(context.Background(), WithNodeFilter(func(node Node) bool {
-		return (node.Version() == "v2.0.0")
-	}))
+	n, done, err := selector.Select(context.Background(), WithFilter(mockFilter("v2.0.0")))
 	assert.Nil(t, err)
 	assert.NotNil(t, n)
 	assert.NotNil(t, done)
@@ -121,74 +118,22 @@ func TestDefault(t *testing.T) {
 	done(context.Background(), DoneInfo{})
 
 	// no v3.0.0 instance
-	n, done, err = selector.Select(context.Background(), WithNodeFilter(func(node Node) bool {
-		return (node.Version() == "v3.0.0")
-	}))
+	n, done, err = selector.Select(context.Background(), WithFilter(mockFilter("v3.0.0")))
 	assert.Equal(t, ErrNoAvailable, err)
 	assert.Nil(t, done)
 	assert.Nil(t, n)
 
 	// apply zero instance
 	selector.Apply([]Node{})
-	n, done, err = selector.Select(context.Background(), WithNodeFilter(func(node Node) bool {
-		return (node.Version() == "v2.0.0")
-	}))
+	n, done, err = selector.Select(context.Background(), WithFilter(mockFilter("v2.0.0")))
 	assert.Equal(t, ErrNoAvailable, err)
 	assert.Nil(t, done)
 	assert.Nil(t, n)
 
 	// apply zero instance
 	selector.Apply(nil)
-	n, done, err = selector.Select(context.Background(), WithNodeFilter(func(node Node) bool {
-		return (node.Version() == "v2.0.0")
-	}))
+	n, done, err = selector.Select(context.Background(), WithFilter(mockFilter("v2.0.0")))
 	assert.Equal(t, ErrNoAvailable, err)
 	assert.Nil(t, done)
 	assert.Nil(t, n)
 }
-
-func TestNodeFilterWithRandom(t *testing.T) {
-	for i := 0; i < 100; i++ {
-		testBaseFilter(t, 1000, rand.Intn(1000))
-	}
-
-	testBaseFilter(t, 0, rand.Intn(1000))
-	testBaseFilter(t, 1, 1000)
-	testBaseFilter(t, 2, 1000)
-	testBaseFilter(t, 3, 1000)
-	testBaseFilter(t, 1, 0)
-	testBaseFilter(t, 2, 0)
-	testBaseFilter(t, 3, 0)
-}
-
-func testBaseFilter(t *testing.T, length int, reservedRatio int) {
-	var raw []WeightedNode
-	var targets map[string]WeightedNode = make(map[string]WeightedNode)
-	for i := 0; i < length; i++ {
-		addr := strconv.FormatInt(int64(i), 10)
-		raw = append(raw, &mockWeightedNode{Node: NewNode(
-			addr,
-			&registry.ServiceInstance{
-				ID:        addr,
-				Name:      "helloworld",
-				Endpoints: []string{addr},
-			})})
-		if reservedRatio > rand.Intn(length) {
-			targets[addr] = raw[i]
-		}
-	}
-
-	f := func(node Node) bool {
-		if _, ok := targets[node.Address()]; ok {
-			return true
-		}
-		return false
-	}
-	d := Default{}
-	raw = d.nodeFilter([]NodeFilter{f}, raw)
-	assert.Equal(t, len(targets), len(raw))
-	for _, n := range raw {
-		_, ok := targets[n.Address()]
-		assert.True(t, ok)
-	}
-}
diff --git a/transport/grpc/balancer_test.go b/transport/grpc/balancer_test.go
index da99aebe..c0ea6aa9 100644
--- a/transport/grpc/balancer_test.go
+++ b/transport/grpc/balancer_test.go
@@ -1,6 +1,7 @@
 package grpc
 
 import (
+	"context"
 	"testing"
 
 	"github.com/go-kratos/kratos/v2/selector"
@@ -24,8 +25,8 @@ func TestBalancerName(t *testing.T) {
 func TestFilters(t *testing.T) {
 	o := &clientOptions{}
 
-	WithNodeFilter(func(selector.Node) bool {
-		return true
+	WithFilter(func(_ context.Context, nodes []selector.Node) []selector.Node {
+		return nodes
 	})(o)
 	assert.Equal(t, 1, len(o.filters))
 }