test: add registry tests (#1260)

* add registry tests

* add registrar timeout

Co-authored-by: 包子 <baozhecheng@foxmail.com>

.github/workflows/go.yml

@@ -8,9 +8,21 @@ on:
   workflow_dispatch:
 
 jobs:
-
   build:
     runs-on: ubuntu-latest
+    services:
+      etcd:
+        image: gcr.io/etcd-development/etcd:v3.5.0
+        ports:
+          - 2379:2379
+        env:
+          ETCD_LISTEN_CLIENT_URLS: http://0.0.0.0:2379
+          ETCD_ADVERTISE_CLIENT_URLS: http://0.0.0.0:2379
+      consul:
+        image: consul:latest
+        ports:
+          - 8500:8500
+
     steps:
     - uses: actions/checkout@v2
 

app.go

@@ -7,6 +7,7 @@ import (
 	"os/signal"
 	"sync"
 	"syscall"
+	"time"
 
 	"github.com/go-kratos/kratos/v2/log"
 	"github.com/go-kratos/kratos/v2/registry"
@@ -39,6 +40,7 @@ func New(opts ...Option) *App {
 		ctx:              context.Background(),
 		logger:           log.DefaultLogger,
 		sigs:             []os.Signal{syscall.SIGTERM, syscall.SIGQUIT, syscall.SIGINT},
+		registrarTimeout: 10 * time.Second,
 	}
 	if id, err := uuid.NewUUID(); err == nil {
 		options.id = id.String()
@@ -92,7 +94,9 @@ func (a *App) Run() error {
 	}
 	wg.Wait()
 	if a.opts.registrar != nil {
-		if err := a.opts.registrar.Register(a.opts.ctx, instance); err != nil {
+		ctx, cancel := context.WithTimeout(a.opts.ctx, a.opts.registrarTimeout)
+		defer cancel()
+		if err := a.opts.registrar.Register(ctx, instance); err != nil {
 			return err
 		}
 		a.instance = instance
@@ -118,7 +122,9 @@ func (a *App) Run() error {
 // Stop gracefully stops the application.
 func (a *App) Stop() error {
 	if a.opts.registrar != nil && a.instance != nil {
-		if err := a.opts.registrar.Deregister(a.opts.ctx, a.instance); err != nil {
+		ctx, cancel := context.WithTimeout(a.opts.ctx, a.opts.registrarTimeout)
+		defer cancel()
+		if err := a.opts.registrar.Deregister(ctx, a.instance); err != nil {
 			return err
 		}
 	}

examples/registry/registry_test.go

@@ -0,0 +1,124 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"testing"
+	"time"
+
+	"github.com/go-kratos/kratos/examples/helloworld/helloworld"
+	pb "github.com/go-kratos/kratos/examples/helloworld/helloworld"
+
+	consulregistry "github.com/go-kratos/consul/registry"
+	etcdregistry "github.com/go-kratos/etcd/registry"
+	"github.com/go-kratos/kratos/v2"
+	"github.com/go-kratos/kratos/v2/registry"
+	"github.com/go-kratos/kratos/v2/transport/grpc"
+	"github.com/go-kratos/kratos/v2/transport/http"
+	consul "github.com/hashicorp/consul/api"
+	etcd "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 startServer(r registry.Registrar) (app *kratos.App, err error) {
+	httpSrv := http.NewServer()
+	grpcSrv := grpc.NewServer()
+
+	s := &server{}
+	pb.RegisterGreeterServer(grpcSrv, s)
+	pb.RegisterGreeterHTTPServer(httpSrv, s)
+
+	app = kratos.New(
+		kratos.Name("helloworld"),
+		kratos.Server(
+			httpSrv,
+			grpcSrv,
+		),
+		kratos.Registrar(r),
+		kratos.RegistrarTimeout(5*time.Second),
+	)
+	go func() {
+		err = app.Run()
+	}()
+	time.Sleep(time.Second)
+	return
+}
+
+func callGRPC(t *testing.T, r registry.Discovery) {
+	conn, err := grpc.DialInsecure(
+		context.Background(),
+		grpc.WithEndpoint("discovery:///helloworld"),
+		grpc.WithDiscovery(r),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	client := helloworld.NewGreeterClient(conn)
+	reply, err := client.SayHello(context.Background(), &helloworld.HelloRequest{Name: "kratos"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Logf("[grpc] SayHello %+v\n", reply)
+}
+
+func callHTTP(t *testing.T, r registry.Discovery) {
+	conn, err := http.NewClient(
+		context.Background(),
+		http.WithEndpoint("discovery:///helloworld"),
+		http.WithDiscovery(r),
+		http.WithBlock(),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	client := helloworld.NewGreeterHTTPClient(conn)
+	reply, err := client.SayHello(context.Background(), &helloworld.HelloRequest{Name: "kratos"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Logf("[http] SayHello %+v\n", reply)
+}
+
+func TestETCD(t *testing.T) {
+	client, err := etcd.New(etcd.Config{
+		Endpoints: []string{"127.0.0.1:2379"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	r := etcdregistry.New(client)
+	srv, err := startServer(r)
+	if err != nil {
+		t.Fatal(err)
+	}
+	callHTTP(t, r)
+	callGRPC(t, r)
+	srv.Stop()
+}
+
+func TestConsul(t *testing.T) {
+	client, err := consul.NewClient(consul.DefaultConfig())
+	if err != nil {
+		log.Fatal(err)
+	}
+	r := consulregistry.New(client)
+	srv, err := startServer(r)
+	if err != nil {
+		t.Fatal(err)
+	}
+	callHTTP(t, r)
+	callGRPC(t, r)
+	srv.Stop()
+}

options.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"net/url"
 	"os"
+	"time"
 
 	"github.com/go-kratos/kratos/v2/log"
 	"github.com/go-kratos/kratos/v2/registry"
@@ -26,7 +27,7 @@ type options struct {
 
 	logger           log.Logger
 	registrar        registry.Registrar
-
+	registrarTimeout time.Duration
 	servers          []transport.Server
 }
 
@@ -79,3 +80,8 @@ func Signal(sigs ...os.Signal) Option {
 func Registrar(r registry.Registrar) Option {
 	return func(o *options) { o.registrar = r }
 }
+
+// RegistrarTimeout with registrar timeout.
+func RegistrarTimeout(t time.Duration) Option {
+	return func(o *options) { o.registrarTimeout = t }
+}