Skip to content

Commit

Permalink
feat: add retry feature for etcd (#91)
Browse files Browse the repository at this point in the history
* feat: add retry feature for etcd

* use sonic

* update

* add unit test

* add unit test
  • Loading branch information
Skyenought authored Oct 5, 2023
1 parent cab5b29 commit eb24a62
Show file tree
Hide file tree
Showing 7 changed files with 307 additions and 35 deletions.
22 changes: 21 additions & 1 deletion etcd/etcd_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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)
Expand Down
51 changes: 51 additions & 0 deletions etcd/example/server/retry/main.go
Original file line number Diff line number Diff line change
@@ -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()
}
File renamed without changes.
58 changes: 50 additions & 8 deletions etcd/common.go → etcd/option.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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"`
Expand Down Expand Up @@ -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
}
}

Expand Down
55 changes: 55 additions & 0 deletions etcd/readme.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Loading

0 comments on commit eb24a62

Please sign in to comment.