Repository: go-kratos/kratos (go)

Issue:
consul服务注册和发现两个问题
问题1：服务A已起来，但其他服务在20秒内无法发现服务A，原因是默认健康检查间隔 20 秒
问题2：服务A启动时去发现其他服务，结果无法找到节点，原因是consul的服务发现底层目前是异步的，需要先同步请求一次

Pull request description:
fix(registry/consul): fix can not find service in 20s
<!--
🎉 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>:

    fix: A bug fix
    feat: A new feature
    test: Adding missing tests or correcting existing tests
    refactor: A code change that neither fixes a bug nor adds a feature
    break: Changes has break change

    docs: Documentation only changes
    deps: Changes external dependencies
    style: Changes that do not affect the meaning of the code (white-space, formatting, etc)
    chore Daily work, examples, etc.
    ci: Changes to our CI configuration files and scripts
-->

#### 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 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 #1727 problem 1

#### Other special notes for reviewer:
<!--
* Somethings that need extra attention for reviewer
* Some additional notes, TODO list, etc.
-->


Patch:
diff --git a/contrib/registry/consul/client.go b/contrib/registry/consul/client.go
index b711a5bf..62654db0 100644
--- a/contrib/registry/consul/client.go
+++ b/contrib/registry/consul/client.go
@@ -21,11 +21,17 @@ type Client struct {
 
 	// resolve service entry endpoints
 	resolver ServiceResolver
+	// healthcheck time interval in seconds
+	healthcheckInterval int
 }
 
 // NewClient creates consul client
 func NewClient(cli *api.Client) *Client {
-	c := &Client{cli: cli, resolver: defaultResolver}
+	c := &Client{
+		cli:                 cli,
+		resolver:            defaultResolver,
+		healthcheckInterval: 10,
+	}
 	c.ctx, c.cancel = context.WithCancel(context.Background())
 	return c
 }
@@ -107,7 +113,7 @@ func (c *Client) Register(_ context.Context, svc *registry.ServiceInstance, enab
 		for _, address := range checkAddresses {
 			asr.Checks = append(asr.Checks, &api.AgentServiceCheck{
 				TCP:                            address,
-				Interval:                       "20s",
+				Interval:                       fmt.Sprintf("%ds", c.healthcheckInterval),
 				DeregisterCriticalServiceAfter: "70s",
 			})
 		}
@@ -116,8 +122,9 @@ func (c *Client) Register(_ context.Context, svc *registry.ServiceInstance, enab
 	if err != nil {
 		return err
 	}
+	_ = c.cli.Agent().UpdateTTL("service:"+svc.ID, "pass", "pass")
 	go func() {
-		ticker := time.NewTicker(time.Second * 20)
+		ticker := time.NewTicker(time.Second * time.Duration(c.healthcheckInterval))
 		defer ticker.Stop()
 		for {
 			select {
diff --git a/contrib/registry/consul/registry.go b/contrib/registry/consul/registry.go
index 41cdd8c7..e2dd7a6e 100644
--- a/contrib/registry/consul/registry.go
+++ b/contrib/registry/consul/registry.go
@@ -35,6 +35,15 @@ func WithServiceResolver(fn ServiceResolver) Option {
 	}
 }
 
+// WithHealthCheckInterval with healthcheck interval in seconds.
+func WithHealthCheckInterval(interval int) Option {
+	return func(o *Registry) {
+		if o.cli != nil {
+			o.cli.healthcheckInterval = interval
+		}
+	}
+}
+
 // Config is consul registry config
 type Config struct {
 	*api.Config
diff --git a/examples/go.mod b/examples/go.mod
index 83e41611..3f2f1342 100644
--- a/examples/go.mod
+++ b/examples/go.mod
@@ -33,6 +33,7 @@ require (
 	github.com/prometheus/client_golang v1.11.0
 	github.com/segmentio/kafka-go v0.4.17
 	github.com/sirupsen/logrus v1.8.1
+	github.com/soheilhy/cmux v0.1.4
 	github.com/stretchr/testify v1.7.0
 	go.etcd.io/etcd/client/v3 v3.5.0
 	go.opentelemetry.io/otel v1.0.0
@@ -43,6 +44,7 @@ require (
 	golang.org/x/text v0.3.7
 	google.golang.org/genproto v0.0.0-20210831024726-fe130286e0e2
 	google.golang.org/grpc v1.42.0
+	google.golang.org/grpc/examples v0.0.0-20211228230712-fbaf7c558210 // indirect
 	google.golang.org/protobuf v1.27.1
 )
 
diff --git a/examples/go.sum b/examples/go.sum
index e028f1f3..d86393d8 100644
--- a/examples/go.sum
+++ b/examples/go.sum
@@ -588,6 +588,7 @@ github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1
 github.com/smartystreets/goconvey v0.0.0-20190330032615-68dc04aab96a/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
 github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIKYqbNC9s=
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
+github.com/soheilhy/cmux v0.1.4 h1:0HKaf1o97UwFjHH9o5XsHUOF+tqmdA7KEzXLpiyaw0E=
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
 github.com/sony/gobreaker v0.4.1/go.mod h1:ZKptC7FHNvhBz7dN2LGjPVBz2sZJmc0/PkyDJOjmxWY=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
@@ -1013,6 +1014,7 @@ google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEY
 google.golang.org/genproto v0.0.0-20200618031413-b414f8b61790/go.mod h1:jDfRM7FcilCzHH/e9qn6dsT145K34l5v+OpcnNgKAAA=
 google.golang.org/genproto v0.0.0-20200729003335-053ba62fc06f/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20200804131852-c06518451d9c/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
+google.golang.org/genproto v0.0.0-20200806141610-86f49bd18e98/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20200825200019-8632dd797987/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c/go.mod h1:UODoCrxHCcBojKKwX1terBiRUaqAsFqJiF615XL43r0=
 google.golang.org/genproto v0.0.0-20210617175327-b9e0b3197ced/go.mod h1:SzzZ/N+nwJDaO1kznhnlzqS8ocJICar6hYhVyhi++24=
@@ -1045,6 +1047,8 @@ google.golang.org/grpc v1.39.0/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnD
 google.golang.org/grpc v1.40.0/go.mod h1:ogyxbiOoUXAkP+4+xa6PZSE9DZgIHtSpzjDTB9KAK34=
 google.golang.org/grpc v1.42.0 h1:XT2/MFpuPFsEX2fWh3YQtHkZ+WYZFQRfaUgLZYj/p6A=
 google.golang.org/grpc v1.42.0/go.mod h1:k+4IHHFw41K8+bbowsex27ge2rCb65oeWqe4jJ590SU=
+google.golang.org/grpc/examples v0.0.0-20211228230712-fbaf7c558210 h1:xTGItiLKOFqOpgkCD5DK1dWd8RzqFz+2+S2H785PnRg=
+google.golang.org/grpc/examples v0.0.0-20211228230712-fbaf7c558210/go.mod h1:gID3PKrg7pWKntu9Ss6zTLJ0ttC0X9IHgREOCZwbCVU=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
diff --git a/examples/registry/etcd/cmuxServer/main.go b/examples/registry/etcd/cmuxServer/main.go
new file mode 100644
index 00000000..7237c3ae
--- /dev/null
+++ b/examples/registry/etcd/cmuxServer/main.go
@@ -0,0 +1,80 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net"
+	"strings"
+
+	"github.com/go-kratos/kratos/contrib/registry/etcd/v2"
+	pb "github.com/go-kratos/kratos/examples/helloworld/helloworld"
+	"github.com/go-kratos/kratos/v2"
+	"github.com/go-kratos/kratos/v2/middleware/recovery"
+	"github.com/go-kratos/kratos/v2/transport/grpc"
+	"github.com/go-kratos/kratos/v2/transport/http"
+	"github.com/soheilhy/cmux"
+	etcdclient "go.etcd.io/etcd/client/v3"
+)
+
+// server is used to implement helloworld.GreeterServer.
+type server struct {
+	pb.UnimplementedGreeterServer
+}
+
+// SayHello implements helloworld.GreeterServer
+func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
+	return &pb.HelloReply{Message: fmt.Sprintf("Welcome %+v!", in.Name)}, nil
+}
+
+func main() {
+	l, err := net.Listen("tcp", ":9091")
+	if err != nil {
+		log.Panic(err)
+	}
+	m := cmux.New(l)
+
+	client, err := etcdclient.New(etcdclient.Config{
+		Endpoints: []string{"127.0.0.1:2379"},
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	grpcSrv := grpc.NewServer(
+		grpc.Listener(m.MatchWithWriters(cmux.HTTP2MatchHeaderFieldPrefixSendSettings("content-type", "application/grpc"))),
+		grpc.Middleware(
+			recovery.Recovery(),
+		),
+	)
+	httpSrv := http.NewServer(
+		http.Listener(m.Match(cmux.Any())),
+		http.Middleware(
+			recovery.Recovery(),
+		),
+	)
+
+	s := &server{}
+	pb.RegisterGreeterServer(grpcSrv, s)
+	pb.RegisterGreeterHTTPServer(httpSrv, s)
+
+	r := etcd.New(client)
+	app := kratos.New(
+		kratos.Name("helloworld"),
+		kratos.Server(
+			httpSrv,
+			grpcSrv,
+		),
+		kratos.Registrar(r),
+	)
+
+	go func() {
+		if err := m.Serve(); !strings.Contains(err.Error(), "use of closed network connection") {
+			panic(err)
+		}
+	}()
+
+	if err := app.Run(); err != nil {
+		log.Fatal(err)
+	}
+}
diff --git a/transport/grpc/server.go b/transport/grpc/server.go
index 748302c2..0ae458d6 100644
--- a/transport/grpc/server.go
+++ b/transport/grpc/server.go
@@ -5,7 +5,6 @@ import (
 	"crypto/tls"
 	"net"
 	"net/url"
-	"sync"
 	"time"
 
 	"github.com/go-kratos/kratos/v2/internal/endpoint"
@@ -75,6 +74,13 @@ func TLSConfig(c *tls.Config) ServerOption {
 	}
 }
 
+// Listener with server lis
+func Listener(lis net.Listener) ServerOption {
+	return func(s *Server) {
+		s.lis = lis
+	}
+}
+
 // UnaryInterceptor returns a ServerOption that sets the UnaryServerInterceptor for the server.
 func UnaryInterceptor(in ...grpc.UnaryServerInterceptor) ServerOption {
 	return func(s *Server) {
@@ -95,7 +101,6 @@ type Server struct {
 	baseCtx    context.Context
 	tlsConf    *tls.Config
 	lis        net.Listener
-	once       sync.Once
 	err        error
 	network    string
 	address    string
@@ -139,6 +144,8 @@ func NewServer(opts ...ServerOption) *Server {
 	}
 	srv.Server = grpc.NewServer(grpcOpts...)
 	srv.metadata = apimd.NewServer(srv.Server)
+	// listen and endpoint
+	srv.err = srv.listenAndEndpoint()
 	// internal register
 	grpc_health_v1.RegisterHealthServer(srv.Server, srv.health)
 	apimd.RegisterMetadataServer(srv.Server, srv.metadata)
@@ -150,21 +157,6 @@ func NewServer(opts ...ServerOption) *Server {
 // examples:
 //   grpc://127.0.0.1:9000?isSecure=false
 func (s *Server) Endpoint() (*url.URL, error) {
-	s.once.Do(func() {
-		lis, err := net.Listen(s.network, s.address)
-		if err != nil {
-			s.err = err
-			return
-		}
-		addr, err := host.Extract(s.address, lis)
-		if err != nil {
-			_ = lis.Close()
-			s.err = err
-			return
-		}
-		s.lis = lis
-		s.endpoint = endpoint.NewEndpoint("grpc", addr, s.tlsConf != nil)
-	})
 	if s.err != nil {
 		return nil, s.err
 	}
@@ -173,8 +165,8 @@ func (s *Server) Endpoint() (*url.URL, error) {
 
 // Start start the gRPC server.
 func (s *Server) Start(ctx context.Context) error {
-	if _, err := s.Endpoint(); err != nil {
-		return err
+	if s.err != nil {
+		return s.err
 	}
 	s.baseCtx = ctx
 	s.log.Infof("[gRPC] server listening on: %s", s.lis.Addr().String())
@@ -190,6 +182,23 @@ func (s *Server) Stop(ctx context.Context) error {
 	return nil
 }
 
+func (s *Server) listenAndEndpoint() error {
+	if s.lis == nil {
+		lis, err := net.Listen(s.network, s.address)
+		if err != nil {
+			return err
+		}
+		s.lis = lis
+	}
+	addr, err := host.Extract(s.address, s.lis)
+	if err != nil {
+		_ = s.lis.Close()
+		return err
+	}
+	s.endpoint = endpoint.NewEndpoint("grpc", addr, s.tlsConf != nil)
+	return nil
+}
+
 func (s *Server) unaryServerInterceptor() grpc.UnaryServerInterceptor {
 	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
 		ctx, cancel := ic.Merge(ctx, s.baseCtx)
diff --git a/transport/http/server.go b/transport/http/server.go
index da8a29fe..f30268cf 100644
--- a/transport/http/server.go
+++ b/transport/http/server.go
@@ -7,7 +7,6 @@ import (
 	"net"
 	"net/http"
 	"net/url"
-	"sync"
 	"time"
 
 	"github.com/go-kratos/kratos/v2/internal/endpoint"
@@ -107,12 +106,18 @@ func StrictSlash(strictSlash bool) ServerOption {
 	}
 }
 
+// Listener with server lis
+func Listener(lis net.Listener) ServerOption {
+	return func(s *Server) {
+		s.lis = lis
+	}
+}
+
 // Server is an HTTP server wrapper.
 type Server struct {
 	*http.Server
 	lis         net.Listener
 	tlsConf     *tls.Config
-	once        sync.Once
 	endpoint    *url.URL
 	err         error
 	network     string
@@ -149,6 +154,7 @@ func NewServer(opts ...ServerOption) *Server {
 		Handler:   FilterChain(srv.filters...)(srv.router),
 		TLSConfig: srv.tlsConf,
 	}
+	srv.err = srv.listenAndEndpoint()
 	return srv
 }
 
@@ -219,22 +225,6 @@ func (s *Server) filter() mux.MiddlewareFunc {
 // examples:
 //   http://127.0.0.1:8000?isSecure=false
 func (s *Server) Endpoint() (*url.URL, error) {
-	s.once.Do(func() {
-		lis, err := net.Listen(s.network, s.address)
-		if err != nil {
-			s.err = err
-			return
-		}
-		addr, err := host.Extract(s.address, lis)
-		if err != nil {
-			lis.Close()
-			s.err = err
-			return
-		}
-		s.lis = lis
-
-		s.endpoint = endpoint.NewEndpoint("http", addr, s.tlsConf != nil)
-	})
 	if s.err != nil {
 		return nil, s.err
 	}
@@ -243,8 +233,8 @@ func (s *Server) Endpoint() (*url.URL, error) {
 
 // Start start the HTTP server.
 func (s *Server) Start(ctx context.Context) error {
-	if _, err := s.Endpoint(); err != nil {
-		return err
+	if s.err != nil {
+		return s.err
 	}
 	s.BaseContext = func(net.Listener) context.Context {
 		return ctx
@@ -267,3 +257,20 @@ func (s *Server) Stop(ctx context.Context) error {
 	s.log.Info("[HTTP] server stopping")
 	return s.Shutdown(ctx)
 }
+
+func (s *Server) listenAndEndpoint() error {
+	if s.lis == nil {
+		lis, err := net.Listen(s.network, s.address)
+		if err != nil {
+			return err
+		}
+		s.lis = lis
+	}
+	addr, err := host.Extract(s.address, s.lis)
+	if err != nil {
+		_ = s.lis.Close()
+		return err
+	}
+	s.endpoint = endpoint.NewEndpoint("http", addr, s.tlsConf != nil)
+	return nil
+}


Test patch:
diff --git a/transport/grpc/server_test.go b/transport/grpc/server_test.go
index 87c3483c..29f0b224 100644
--- a/transport/grpc/server_test.go
+++ b/transport/grpc/server_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"net"
 	"net/url"
 	"strings"
 	"testing"
@@ -117,11 +118,9 @@ func TestNetwork(t *testing.T) {
 }
 
 func TestAddress(t *testing.T) {
-	o := &Server{}
 	v := "abc"
-	Address(v)(o)
+	o := NewServer(Address(v))
 	assert.Equal(t, v, o.address)
-
 	u, err := o.Endpoint()
 	assert.NotNil(t, err)
 	assert.Nil(t, u)
@@ -216,3 +215,10 @@ func TestServer_unaryServerInterceptor(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, "hi", rv.(*testResp).Data)
 }
+
+func TestListener(t *testing.T) {
+	lis := &net.TCPListener{}
+	s := &Server{}
+	Listener(lis)(s)
+	assert.Equal(t, s.lis, lis)
+}
diff --git a/transport/http/client_test.go b/transport/http/client_test.go
index dd5af239..2da84f75 100644
--- a/transport/http/client_test.go
+++ b/transport/http/client_test.go
@@ -5,12 +5,14 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
+	"fmt"
 	"io"
 	nethttp "net/http"
+	"strconv"
 	"testing"
 	"time"
 
-	"github.com/go-kratos/kratos/v2/errors"
+	kratosErrors "github.com/go-kratos/kratos/v2/errors"
 	"github.com/go-kratos/kratos/v2/middleware"
 	"github.com/go-kratos/kratos/v2/registry"
 	"github.com/stretchr/testify/assert"
@@ -116,8 +118,16 @@ func (*mockDiscovery) Watch(ctx context.Context, serviceName string) (registry.W
 
 type mockWatcher struct{}
 
-func (*mockWatcher) Next() ([]*registry.ServiceInstance, error) {
-	return nil, nil
+func (m *mockWatcher) Next() ([]*registry.ServiceInstance, error) {
+	instance := &registry.ServiceInstance{
+		ID:        "1",
+		Name:      "kratos",
+		Version:   "v1",
+		Metadata:  map[string]string{},
+		Endpoints: []string{fmt.Sprintf("http://127.0.0.1:9001?isSecure=%s", strconv.FormatBool(false))},
+	}
+	time.Sleep(time.Millisecond * 500)
+	return []*registry.ServiceInstance{instance}, nil
 }
 
 func (*mockWatcher) Stop() error {
@@ -202,9 +212,9 @@ func TestDefaultErrorDecoder(t *testing.T) {
 	}
 	err2 := DefaultErrorDecoder(context.TODO(), resp2)
 	assert.Error(t, err2)
-	assert.Equal(t, int32(500), err2.(*errors.Error).GetCode())
-	assert.Equal(t, "hi", err2.(*errors.Error).GetMessage())
-	assert.Equal(t, "FOO", err2.(*errors.Error).GetReason())
+	assert.Equal(t, int32(500), err2.(*kratosErrors.Error).GetCode())
+	assert.Equal(t, "hi", err2.(*kratosErrors.Error).GetMessage())
+	assert.Equal(t, "FOO", err2.(*kratosErrors.Error).GetReason())
 }
 
 func TestCodecForResponse(t *testing.T) {
diff --git a/transport/http/server_test.go b/transport/http/server_test.go
index b5238bb0..f41dc042 100644
--- a/transport/http/server_test.go
+++ b/transport/http/server_test.go
@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"strings"
 	"testing"
@@ -236,3 +237,10 @@ func TestTLSConfig(t *testing.T) {
 	TLSConfig(v)(o)
 	assert.Equal(t, v, o.tlsConf)
 }
+
+func TestListener(t *testing.T) {
+	lis := &net.TCPListener{}
+	s := &Server{}
+	Listener(lis)(s)
+	assert.Equal(t, s.lis, lis)
+}