feat: add retry feature for etcd (#91)

* feat: add retry feature for etcd

* use sonic

* update

* add unit test

* add unit test

etcd/etcd_test.go

@@ -24,11 +24,11 @@ import (
 	"testing"
 	"time"
 
-	"github.com/cloudwego/hertz/pkg/app/client/discovery"
 	"github.com/stretchr/testify/require"
 
 	"github.com/cloudwego/hertz/pkg/app"
 	"github.com/cloudwego/hertz/pkg/app/client"
+	"github.com/cloudwego/hertz/pkg/app/client/discovery"
 	"github.com/cloudwego/hertz/pkg/app/middlewares/client/sd"
 	"github.com/cloudwego/hertz/pkg/app/server"
 	"github.com/cloudwego/hertz/pkg/app/server/registry"
@@ -397,6 +397,26 @@ func TestEtcdRegistryWithEnvironmentVariable(t *testing.T) {
 	teardownEmbedEtcd(s)
 }
 
+func TestRetryOption(t *testing.T) {
+	o := newOptionForServer([]string{"127.0.0.1:2345"})
+	assert.Equal(t, o.etcdCfg.Endpoints, []string{"127.0.0.1:2345"})
+	assert.Equal(t, uint(5), o.retryCfg.maxAttemptTimes)
+	assert.Equal(t, 30*time.Second, o.retryCfg.observeDelay)
+	assert.Equal(t, 10*time.Second, o.retryCfg.retryDelay)
+}
+
+func TestRetryCustomConfig(t *testing.T) {
+	o := newOptionForServer(
+		[]string{"127.0.0.1:2345"},
+		WithMaxAttemptTimes(10),
+		WithObserveDelay(20*time.Second),
+		WithRetryDelay(5*time.Second),
+	)
+	assert.Equal(t, uint(10), o.retryCfg.maxAttemptTimes)
+	assert.Equal(t, 20*time.Second, o.retryCfg.observeDelay)
+	assert.Equal(t, 5*time.Second, o.retryCfg.retryDelay)
+}
+
 func setupEmbedEtcd(t *testing.T) (*embed.Etcd, string) {
 	endpoint := fmt.Sprintf("unix://localhost:%06d", os.Getpid())
 	u, err := url.Parse(endpoint)

etcd/example/server/retry/main.go

@@ -0,0 +1,51 @@
+// Copyright 2021 CloudWeGo Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"context"
+	"time"
+
+	"github.com/cloudwego/hertz/pkg/app"
+	"github.com/cloudwego/hertz/pkg/app/server"
+	"github.com/cloudwego/hertz/pkg/app/server/registry"
+	"github.com/cloudwego/hertz/pkg/common/utils"
+	"github.com/cloudwego/hertz/pkg/protocol/consts"
+	"github.com/hertz-contrib/registry/etcd"
+)
+
+func main() {
+	r, _ := etcd.NewEtcdRegistry(
+		[]string{"127.0.0.1:2379"},
+		etcd.WithMaxAttemptTimes(10),
+		etcd.WithObserveDelay(20*time.Second),
+		etcd.WithRetryDelay(5*time.Second),
+	)
+
+	addr := "127.0.0.1:8888"
+	h := server.Default(
+		server.WithHostPorts(addr),
+		server.WithRegistry(r, &registry.Info{
+			ServiceName: "hertz.test.demo",
+			Addr:        utils.NewNetAddr("tcp", addr),
+			Weight:      10,
+			Tags:        nil,
+		}),
+	)
+	h.GET("/ping", func(_ context.Context, ctx *app.RequestContext) {
+		ctx.JSON(consts.StatusOK, utils.H{"ping": "pong2"})
+	})
+	h.Spin()
+}

etcd/common.go → etcd/option.go

@@ -22,6 +22,7 @@ import (
 	"io/ioutil"
 	"os"
 	"strconv"
+	"time"
 
 	"github.com/cloudwego/hertz/pkg/app/server/registry"
 	"github.com/cloudwego/hertz/pkg/common/hlog"
@@ -32,6 +33,50 @@ const (
 	defaultTTL = 60
 )
 
+type option struct {
+	// etcd client config
+	etcdCfg  clientv3.Config
+	retryCfg *retryCfg
+}
+
+type retryCfg struct {
+	// The maximum number of call attempt times, including the initial call
+	maxAttemptTimes uint
+	// observeDelay is the delay time for checking the service status under normal conditions
+	observeDelay time.Duration
+	// retryDelay is the delay time for attempting to register the service after disconnecting
+	retryDelay time.Duration
+}
+
+type Option func(o *option)
+
+// WithMaxAttemptTimes sets the maximum number of call attempt times, including the initial call
+func WithMaxAttemptTimes(maxAttemptTimes uint) Option {
+	return func(o *option) {
+		o.retryCfg.maxAttemptTimes = maxAttemptTimes
+	}
+}
+
+// WithObserveDelay sets the delay time for checking the service status under normal conditions
+func WithObserveDelay(observeDelay time.Duration) Option {
+	return func(o *option) {
+		o.retryCfg.observeDelay = observeDelay
+	}
+}
+
+// WithRetryDelay sets the delay time of retry
+func WithRetryDelay(t time.Duration) Option {
+	return func(o *option) {
+		o.retryCfg.retryDelay = t
+	}
+}
+
+func (o *option) apply(opts ...Option) {
+	for _, opt := range opts {
+		opt(o)
+	}
+}
+
 // instanceInfo used to stored service basic info in etcd.
 type instanceInfo struct {
 	Network string            `json:"network"`
@@ -74,25 +119,22 @@ func getTTL() int64 {
 	return ttl
 }
 
-// Option sets options such as username, tls etc.
-type Option func(cfg *clientv3.Config)
-
 // WithTLSOpt returns a option that authentication by tls/ssl.
 func WithTLSOpt(certFile, keyFile, caFile string) Option {
-	return func(cfg *clientv3.Config) {
+	return func(o *option) {
 		tlsCfg, err := newTLSConfig(certFile, keyFile, caFile, "")
 		if err != nil {
 			hlog.Errorf("HERTZ: tls failed with err: %v , skipping tls.", err)
 		}
-		cfg.TLS = tlsCfg
+		o.etcdCfg.TLS = tlsCfg
 	}
 }
 
 // WithAuthOpt returns an option that authentication by username and password.
 func WithAuthOpt(username, password string) Option {
-	return func(cfg *clientv3.Config) {
-		cfg.Username = username
-		cfg.Password = password
+	return func(o *option) {
+		o.etcdCfg.Username = username
+		o.etcdCfg.Password = password
 	}
 }
 

etcd/readme.md

@@ -191,6 +191,61 @@ func main() {
 	}
 }
 ```
+## Retry
+
+After the service is registered to `ETCD`, it will regularly check the status of the service. If any abnormal status is found, it will try to register the service again. `observeDelay` is the delay time for checking the service status under normal conditions, and `retryDelay` is the delay time for attempting to register the service after disconnecting.
+
+### Default Retry Config
+
+| Config Name         | Default Value    | Description                                                                               |
+|:--------------------|:-----------------|:------------------------------------------------------------------------------------------|
+| WithMaxAttemptTimes | 5                | Used to set the maximum number of attempts, if 0, it means infinite attempts              |
+| WithObserveDelay    | 30 * time.Second | Used to set the delay time for checking service status under normal connection conditions |
+| WithRetryDelay      | 10 * time.Second | Used to set the retry delay time after disconnecting                                      |
+
+### Example
+
+```go
+package main
+
+import (
+	"context"
+	"time"
+
+	"github.com/cloudwego/hertz/pkg/app"
+	"github.com/cloudwego/hertz/pkg/app/server"
+	"github.com/cloudwego/hertz/pkg/app/server/registry"
+	"github.com/cloudwego/hertz/pkg/common/utils"
+	"github.com/cloudwego/hertz/pkg/protocol/consts"
+	"github.com/hertz-contrib/registry/etcd"
+)
+
+func main() {
+	r, _ := etcd.NewEtcdRegistry(
+		[]string{"127.0.0.1:2379"},
+		etcd.WithMaxAttemptTimes(10),
+		etcd.WithObserveDelay(20*time.Second),
+		etcd.WithRetryDelay(5*time.Second),
+	)
+
+	addr := "127.0.0.1:8888"
+	h := server.Default(
+		server.WithHostPorts(addr),
+		server.WithRegistry(r, &registry.Info{
+			ServiceName: "hertz.test.demo",
+			Addr:        utils.NewNetAddr("tcp", addr),
+			Weight:      10,
+			Tags:        nil,
+		}),
+	)
+	h.GET("/ping", func(_ context.Context, ctx *app.RequestContext) {
+		ctx.JSON(consts.StatusOK, utils.H{"ping": "pong2"})
+	})
+	h.Spin()
+}
+
+```
+
 ## How to Dynamically specify ip and port
 
 To dynamically specify an IP and port, one should first set the environment variables `HERTZ_IP_TO_REGISTRY` and `HERTZ_PORT_TO_REGISTRY`. If these variables are not set, the system defaults to using the service's listening IP and port. Notably, if the service's listening IP is either not set or set to "::", the system will automatically retrieve and use the machine's IPV4 address.