Repository: go-kratos/kratos
Pull request #1652 (2021-11-26 02:07:40)

feat: middleware seletor add ctx match
Fix context not being passed correctly(BREAKCHANGE):
```
MatchFunc   func(operation string) bool
-> 
MatchFunc   func(ctx context.Context, operation string) bool
```

codecov-commenter: # [Codecov](https://codecov.io/gh/go-kratos/kratos/pull/1652?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos) Report
> Merging [#1652](https://codecov.io/gh/go-kratos/kratos/pull/1652?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos) (b062279) into [main](https://codecov.io/gh/go-kratos/kratos/commit/2acede53f3e821cef7d3e167dc1cbd6dac22609b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos) (2acede5) will **decrease** coverage by `0.08%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/go-kratos/kratos/pull/1652/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/1652?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    #1652      +/-   ##
==========================================
- Coverage   81.27%   81.19%   -0.09%     
==========================================
  Files          78       78              
  Lines        3429     3440      +11     
==========================================
+ Hits         2787     2793       +6     
- Misses        441      445       +4     
- Partials      201      202       +1     
```


| [Impacted Files](https://codecov.io/gh/go-kratos/kratos/pull/1652?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos) | Coverage Δ | |
|---|---|---|
| [middleware/selector/selector.go](https://codecov.io/gh/go-kratos/kratos/pull/1652/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos#diff-bWlkZGxld2FyZS9zZWxlY3Rvci9zZWxlY3Rvci5nbw==) | `94.20% <100.00%> (+1.09%)` | :arrow_up: |
| [selector/node/ewma/node.go](https://codecov.io/gh/go-kratos/kratos/pull/1652/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos#diff-c2VsZWN0b3Ivbm9kZS9ld21hL25vZGUuZ28=) | `83.87% <0.00%> (-3.23%)` | :arrow_down: |
| [internal/context/context.go](https://codecov.io/gh/go-kratos/kratos/pull/1652/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos#diff-aW50ZXJuYWwvY29udGV4dC9jb250ZXh0Lmdv) | `97.01% <0.00%> (-2.99%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/go-kratos/kratos/pull/1652?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/1652?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos). Last update [2acede5...b062279](https://codecov.io/gh/go-kratos/kratos/pull/1652?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/cmd/kratos/internal/proto/add/add.go b/cmd/kratos/internal/proto/add/add.go
index 2d0d6572..5173e6d1 100644
--- a/cmd/kratos/internal/proto/add/add.go
+++ b/cmd/kratos/internal/proto/add/add.go
@@ -18,7 +18,7 @@ var CmdAdd = &cobra.Command{
 }
 
 func run(cmd *cobra.Command, args []string) {
-	// kratos add helloworld/v1/helloworld.proto
+	// kratos proto add helloworld/v1/helloworld.proto
 	input := args[0]
 	n := strings.LastIndex(input, "/")
 	if n == -1 {
diff --git a/contrib/config/apollo/watcher.go b/contrib/config/apollo/watcher.go
index b1eea6a3..c36d3004 100644
--- a/contrib/config/apollo/watcher.go
+++ b/contrib/config/apollo/watcher.go
@@ -1,6 +1,7 @@
 package apollo
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/go-kratos/kratos/v2/config"
@@ -66,11 +67,13 @@ func newWatcher(a *apollo, logger log.Logger) (config.Watcher, error) {
 	}
 
 	changeCh := make(chan []*config.KeyValue)
-	a.client.AddChangeListener(&customChangeListener{in: changeCh, logger: logger})
+	listener := &customChangeListener{in: changeCh, logger: logger}
+	a.client.AddChangeListener(listener)
 
 	return &watcher{
 		out: changeCh,
 		cancelFn: func() {
+			a.client.RemoveChangeListener(listener)
 			close(changeCh)
 		},
 	}, nil
@@ -78,7 +81,11 @@ func newWatcher(a *apollo, logger log.Logger) (config.Watcher, error) {
 
 // Next will be blocked until the Stop method is called
 func (w *watcher) Next() ([]*config.KeyValue, error) {
-	return <-w.out, nil
+	kv, ok := <-w.out
+	if !ok {
+		return nil, context.Canceled
+	}
+	return kv, nil
 }
 
 func (w *watcher) Stop() error {
diff --git a/contrib/registry/nacos/README.md b/contrib/registry/nacos/README.md
index 05c30547..1a3e09cb 100644
--- a/contrib/registry/nacos/README.md
+++ b/contrib/registry/nacos/README.md
@@ -1,50 +1,90 @@
 # Nacos Registry
 
+## example
+### server
 ```go
+package main
+
 import (
-		"github.com/go-kratos/kratos/v2"
-		"github.com/go-kratos/kratos/v2/transport/grpc"
-		
-		"github.com/nacos-group/nacos-sdk-go/clients"
-		"github.com/nacos-group/nacos-sdk-go/common/constant"
-    	"github.com/nacos-group/nacos-sdk-go/vo"
+	"log"
+
+	"github.com/nacos-group/nacos-sdk-go/clients"
+	"github.com/nacos-group/nacos-sdk-go/common/constant"
+	"github.com/nacos-group/nacos-sdk-go/vo"
+
+	"github.com/go-kratos/kratos/contrib/registry/nacos/v2"
+	"github.com/go-kratos/kratos/v2"
 )
 
-sc := []constant.ServerConfig{
-	*constant.NewServerConfig("127.0.0.1", 8848),
-}
+func main() {
+	sc := []constant.ServerConfig{
+		*constant.NewServerConfig("127.0.0.1", 8848),
+	}
+	
 
-cc := constant.ClientConfig{
-	NamespaceId:         "public",
-	TimeoutMs:           5000,
-}
+	client, err := clients.NewNamingClient(
+		vo.NacosClientParam{
+			ServerConfigs: sc,
+		},
+	)
 
-client, err := clients.NewNamingClient(
-	vo.NacosClientParam{
-		ClientConfig:  &cc,
-		ServerConfigs: sc,
-	},
-)
+	if err != nil {
+		log.Panic(err)
+	}
+
+	r := nacos.New(client)
 
-if err != nil {
-	log.Panic(err)
+	// server
+	app := kratos.New(
+		kratos.Name("helloworld"),
+		kratos.Registrar(r),
+	)
+	if err := app.Run(); err != nil {
+		log.Fatal(err)
+	}
 }
+```
+### client
+```go
+package main
+
+import (
+	"context"
+	"log"
 
-r := nacos.New(client)
+	"github.com/nacos-group/nacos-sdk-go/clients"
+	"github.com/nacos-group/nacos-sdk-go/common/constant"
+	"github.com/nacos-group/nacos-sdk-go/vo"
 
-// server
-app := kratos.New(
-	kratos.Name("helloworld"),
-	kratos.Registrar(r),
+	"github.com/go-kratos/kratos/contrib/registry/nacos/v2"
+	"github.com/go-kratos/kratos/v2/transport/grpc"
 )
-if err := app.Run(); err != nil {
-	log.Fatal(err)
-}
 
-// client
-conn, err := grpc.DialInsecure(
-	context.Background(),
-	grpc.WithEndpoint("discovery:///helloworld"),
-	grpc.WithDiscovery(r),
-)
+func main() {
+
+	cc := constant.ClientConfig{
+		NamespaceId: "public",
+		TimeoutMs:   5000,
+	}
+
+	client, err := clients.NewNamingClient(
+		vo.NacosClientParam{
+			ClientConfig: &cc,
+		},
+	)
+
+	if err != nil {
+		log.Panic(err)
+	}
+
+	r := nacos.New(client)
+
+	// client
+	conn, err := grpc.DialInsecure(
+		context.Background(),
+		grpc.WithEndpoint("discovery:///helloworld"),
+		grpc.WithDiscovery(r),
+	)
+	defer conn.Close()
+}
 ```
\ No newline at end of file
diff --git a/contrib/registry/nacos/registry.go b/contrib/registry/nacos/registry.go
index 56414ef7..5a5d9744 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)
@@ -72,7 +79,7 @@ func New(cli naming_client.INamingClient, opts ...Option) (r *Registry) {
 }
 
 // Register the registration.
-func (r *Registry) Register(ctx context.Context, si *registry.ServiceInstance) error {
+func (r *Registry) Register(_ context.Context, si *registry.ServiceInstance) error {
 	if si.Name == "" {
 		return fmt.Errorf("kratos/nacos: serviceInstance.name cannot is empty")
 	}
@@ -89,11 +96,20 @@ func (r *Registry) Register(ctx context.Context, si *registry.ServiceInstance) e
 		if err != nil {
 			return err
 		}
+		var rmd map[string]string
 		if si.Metadata == nil {
-			si.Metadata = make(map[string]string)
+			rmd = map[string]string{
+				"kind":    u.Scheme,
+				"version": si.Version,
+			}
+		} else {
+			rmd = make(map[string]string, len(si.Metadata)+2)
+			for k, v := range si.Metadata {
+				rmd[k] = v
+			}
+			rmd["kind"] = u.Scheme
+			rmd["version"] = si.Version
 		}
-		si.Metadata["kind"] = u.Scheme
-		si.Metadata["version"] = si.Version
 		_, e := r.cli.RegisterInstance(vo.RegisterInstanceParam{
 			Ip:          host,
 			Port:        uint64(p),
@@ -102,7 +118,7 @@ func (r *Registry) Register(ctx context.Context, si *registry.ServiceInstance) e
 			Enable:      true,
 			Healthy:     true,
 			Ephemeral:   true,
-			Metadata:    si.Metadata,
+			Metadata:    rmd,
 			ClusterName: r.opts.cluster,
 			GroupName:   r.opts.group,
 		})
@@ -114,7 +130,7 @@ func (r *Registry) Register(ctx context.Context, si *registry.ServiceInstance) e
 }
 
 // Deregister the registration.
-func (r *Registry) Deregister(ctx context.Context, service *registry.ServiceInstance) error {
+func (r *Registry) Deregister(_ context.Context, service *registry.ServiceInstance) error {
 	for _, endpoint := range service.Endpoints {
 		u, err := url.Parse(endpoint)
 		if err != nil {
@@ -144,11 +160,11 @@ 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.
-func (r *Registry) GetService(ctx context.Context, serviceName string) ([]*registry.ServiceInstance, error) {
+func (r *Registry) GetService(_ context.Context, serviceName string) ([]*registry.ServiceInstance, error) {
 	res, err := r.cli.SelectInstances(vo.SelectInstancesParam{
 		ServiceName: serviceName,
 		HealthyOnly: true,
@@ -158,12 +174,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/blog/cmd/blog/wire.go b/examples/blog/cmd/blog/wire.go
index 2d3db884..30e5ebc4 100644
--- a/examples/blog/cmd/blog/wire.go
+++ b/examples/blog/cmd/blog/wire.go
@@ -1,3 +1,4 @@
+//go:build wireinject
 // +build wireinject
 
 // The build tag makes sure the stub is not built in the final build.
diff --git a/examples/blog/cmd/blog/wire_gen.go b/examples/blog/cmd/blog/wire_gen.go
index 118c1296..844b7c8c 100644
--- a/examples/blog/cmd/blog/wire_gen.go
+++ b/examples/blog/cmd/blog/wire_gen.go
@@ -1,7 +1,8 @@
 // Code generated by Wire. DO NOT EDIT.
 
 //go:generate go run github.com/google/wire/cmd/wire
-//+build !wireinject
+//go:build !wireinject
+// +build !wireinject
 
 package main
 
diff --git a/examples/i18n/cmd/i18n/wire.go b/examples/i18n/cmd/i18n/wire.go
index 7713a0fa..a06ff957 100644
--- a/examples/i18n/cmd/i18n/wire.go
+++ b/examples/i18n/cmd/i18n/wire.go
@@ -1,3 +1,4 @@
+//go:build wireinject
 // +build wireinject
 
 // The build tag makes sure the stub is not built in the final build.
diff --git a/examples/i18n/cmd/i18n/wire_gen.go b/examples/i18n/cmd/i18n/wire_gen.go
index 7e635e66..4983e47e 100644
--- a/examples/i18n/cmd/i18n/wire_gen.go
+++ b/examples/i18n/cmd/i18n/wire_gen.go
@@ -1,7 +1,8 @@
 // Code generated by Wire. DO NOT EDIT.
 
 //go:generate go run github.com/google/wire/cmd/wire
-//+build !wireinject
+//go:build !wireinject
+// +build !wireinject
 
 package main
 
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/middleware/circuitbreaker/circuitbreaker.go b/middleware/circuitbreaker/circuitbreaker.go
index 5317e5d2..f6eb5f41 100644
--- a/middleware/circuitbreaker/circuitbreaker.go
+++ b/middleware/circuitbreaker/circuitbreaker.go
@@ -42,7 +42,7 @@ func Client(opts ...Option) middleware.Middleware {
 	}
 	return func(handler middleware.Handler) middleware.Handler {
 		return func(ctx context.Context, req interface{}) (interface{}, error) {
-			info, _ := transport.FromServerContext(ctx)
+			info, _ := transport.FromClientContext(ctx)
 			breaker := opt.group.Get(info.Operation()).(circuitbreaker.CircuitBreaker)
 			if err := breaker.Allow(); err != nil {
 				// rejected
diff --git a/middleware/selector/selector.go b/middleware/selector/selector.go
index 53eaf4b6..06b4d2f9 100644
--- a/middleware/selector/selector.go
+++ b/middleware/selector/selector.go
@@ -11,7 +11,7 @@ import (
 
 type (
 	transporter func(ctx context.Context) (transport.Transporter, bool)
-	MatchFunc   func(operation string) bool
+	MatchFunc   func(ctx context.Context, operation string) bool
 )
 
 var (
@@ -83,7 +83,13 @@ func (b *Builder) Build() middleware.Middleware {
 }
 
 // matchs is match operation compliance Builder
-func (b *Builder) matchs(operation string) bool {
+func (b *Builder) matchs(ctx context.Context, transporter transporter) bool {
+	info, ok := transporter(ctx)
+	if !ok {
+		return false
+	}
+
+	operation := info.Operation()
 	for _, prefix := range b.prefix {
 		if prefixMatch(prefix, operation) {
 			return true
@@ -101,21 +107,19 @@ func (b *Builder) matchs(operation string) bool {
 	}
 
 	if b.match != nil {
-		return b.match(operation)
+		if b.match(ctx, operation) {
+			return true
+		}
 	}
+
 	return false
 }
 
 // selector middleware
-func selector(transporter transporter, match MatchFunc, ms ...middleware.Middleware) middleware.Middleware {
+func selector(transporter transporter, match func(context.Context, transporter) bool, ms ...middleware.Middleware) middleware.Middleware {
 	return func(handler middleware.Handler) middleware.Handler {
 		return func(ctx context.Context, req interface{}) (reply interface{}, err error) {
-			info, ok := transporter(ctx)
-			if !ok {
-				return handler(ctx, req)
-			}
-
-			if !match(info.Operation()) {
+			if !match(ctx, transporter) {
 				return handler(ctx, req)
 			}
 			return middleware.Chain(ms...)(handler)(ctx, req)
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..0181792d 100644
--- a/transport/grpc/balancer.go
+++ b/transport/grpc/balancer.go
@@ -49,6 +49,11 @@ type Builder struct {
 
 // Build creates a grpc Picker.
 func (b *Builder) Build(info base.PickerBuildInfo) gBalancer.Picker {
+	if len(info.ReadySCs) == 0 {
+		// Block the RPC until a new picker is available via UpdateState().
+		return base.NewErrPicker(gBalancer.ErrNoSubConnAvailable)
+	}
+
 	nodes := make([]selector.Node, 0)
 	for conn, info := range info.ReadySCs {
 		ins, _ := info.Address.Attributes.Value("rawServiceInstance").(*registry.ServiceInstance)
@@ -71,14 +76,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
 }
 
diff --git a/transport/http/client.go b/transport/http/client.go
index c40c43bb..17cd34e3 100644
--- a/transport/http/client.go
+++ b/transport/http/client.go
@@ -231,7 +231,7 @@ func (client *Client) Invoke(ctx context.Context, method, path string, args inte
 
 func (client *Client) invoke(ctx context.Context, req *http.Request, args interface{}, reply interface{}, c callInfo, opts ...CallOption) error {
 	h := func(ctx context.Context, in interface{}) (interface{}, error) {
-		res, err := client.do(ctx, req, c)
+		res, err := client.do(req.WithContext(ctx))
 		if res != nil {
 			cs := csAttempt{res: res}
 			for _, o := range opts {
@@ -263,19 +263,18 @@ func (client *Client) Do(req *http.Request, opts ...CallOption) (*http.Response,
 			return nil, err
 		}
 	}
-	ctx := req.Context()
 
-	return client.do(ctx, req, c)
+	return client.do(req)
 }
 
-func (client *Client) do(ctx context.Context, req *http.Request, c callInfo) (*http.Response, error) {
+func (client *Client) do(req *http.Request) (*http.Response, error) {
 	var done func(context.Context, selector.DoneInfo)
 	if client.r != nil {
 		var (
 			err  error
 			node selector.Node
 		)
-		if node, done, err = client.opts.selector.Select(ctx); err != nil {
+		if node, done, err = client.opts.selector.Select(req.Context()); err != nil {
 			return nil, errors.ServiceUnavailable("NODE_NOT_FOUND", err.Error())
 		}
 		if client.insecure {
@@ -288,14 +287,14 @@ func (client *Client) do(ctx context.Context, req *http.Request, c callInfo) (*h
 	}
 	resp, err := client.cc.Do(req)
 	if err == nil {
-		err = client.opts.errorDecoder(ctx, resp)
+		err = client.opts.errorDecoder(req.Context(), resp)
 	}
 
 	if err != nil {
 		return nil, err
 	}
 	if done != nil {
-		done(ctx, selector.DoneInfo{Err: err})
+		done(req.Context(), selector.DoneInfo{Err: err})
 	}
 	return resp, nil
 }


Test patch:
diff --git a/middleware/selector/selector_test.go b/middleware/selector/selector_test.go
index aec1a45a..b66c3064 100644
--- a/middleware/selector/selector_test.go
+++ b/middleware/selector/selector_test.go
@@ -17,6 +17,7 @@ type Transport struct {
 	kind      transport.Kind
 	endpoint  string
 	operation string
+	headers   *mockHeader
 }
 
 func (tr *Transport) Kind() transport.Kind {
@@ -32,13 +33,33 @@ func (tr *Transport) Operation() string {
 }
 
 func (tr *Transport) RequestHeader() transport.Header {
-	return nil
+	return tr.headers
 }
 
 func (tr *Transport) ReplyHeader() transport.Header {
 	return nil
 }
 
+type mockHeader struct {
+	m map[string]string
+}
+
+func (m *mockHeader) Get(key string) string {
+	return m.m[key]
+}
+
+func (m *mockHeader) Set(key, value string) {
+	m.m[key] = value
+}
+
+func (m *mockHeader) Keys() []string {
+	keys := make([]string, 0, len(m.m))
+	for k := range m.m {
+		keys = append(keys, k)
+	}
+	return keys
+}
+
 func TestMatch(t *testing.T) {
 	tests := []struct {
 		name string
@@ -139,7 +160,7 @@ func TestFunc(t *testing.T) {
 				t.Log(req)
 				return "reply", nil
 			}
-			next = Server(testMiddleware).Match(func(operation string) bool {
+			next = Server(testMiddleware).Match(func(ctx context.Context, operation string) bool {
 				if strings.HasPrefix(operation, "/go-kratos.dev") || strings.HasSuffix(operation, "world") {
 					return true
 				}
@@ -152,6 +173,66 @@ func TestFunc(t *testing.T) {
 	}
 }
 
+func TestHeaderFunc(t *testing.T) {
+	tests := []struct {
+		name string
+		ctx  context.Context
+	}{
+		{
+			name: "/hello.Update/world",
+			ctx: transport.NewServerContext(context.Background(), &Transport{
+				operation: "/hello.Update/world",
+				headers:   &mockHeader{map[string]string{"X-Test": "test"}},
+			}),
+		},
+		{
+			name: "/hi.Create/world",
+			ctx: transport.NewServerContext(context.Background(), &Transport{
+				operation: "/hi.Create/world",
+				headers:   &mockHeader{map[string]string{"X-Test": "test2", "go-kratos": "kratos"}},
+			}),
+		},
+		{
+			name: "/test.Name/1234",
+			ctx: transport.NewServerContext(context.Background(), &Transport{
+				operation: "/test.Name/1234",
+				headers:   &mockHeader{map[string]string{"X-Test": "test3"}},
+			}),
+		},
+		{
+			name: "/go-kratos.dev/kratos",
+			ctx: transport.NewServerContext(context.Background(), &Transport{
+				operation: "/go-kratos.dev/kratos",
+				headers:   &mockHeader{map[string]string{"X-Test": "test"}},
+			}),
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			next := func(ctx context.Context, req interface{}) (interface{}, error) {
+				t.Log(req)
+				return "reply", nil
+			}
+			next = Server(testMiddleware).Match(func(ctx context.Context, operation string) bool {
+				tr, ok := transport.FromServerContext(ctx)
+				if !ok {
+					return false
+				}
+				if tr.RequestHeader().Get("X-Test") == "test" {
+					return true
+				}
+				if tr.RequestHeader().Get("go-kratos") == "kratos" {
+					return true
+				}
+				return false
+			}).Build()(next)
+			reply, err := next(test.ctx, test.name)
+			assert.Equal(t, reply, "reply")
+			assert.Nil(t, err)
+		})
+	}
+}
+
 func testMiddleware(handler middleware.Handler) middleware.Handler {
 	return func(ctx context.Context, req interface{}) (reply interface{}, err error) {
 		fmt.Println("before")
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))
 }