feat: support mutli manager addrs (#846)

* feat: support mutli manager addrs (#844)

Signed-off-by: Gaius <gaius.qi@gmail.com>

client/config/dynconfig.go

@@ -75,10 +75,10 @@ type dynconfig struct {
 	done      chan bool
 }
 
-func NewDynconfig(managerClient internaldynconfig.ManagerClient, expire time.Duration) (Dynconfig, error) {
+func NewDynconfig(rawManagerClient managerclient.Client, hostOption HostOption, expire time.Duration) (Dynconfig, error) {
 	client, err := internaldynconfig.New(
 		internaldynconfig.ManagerSourceType,
-		internaldynconfig.WithManagerClient(managerClient),
+		internaldynconfig.WithManagerClient(newManagerClient(rawManagerClient, hostOption)),
 		internaldynconfig.WithExpireTime(expire),
 		internaldynconfig.WithCachePath(cachePath),
 	)
@@ -172,7 +172,7 @@ type managerClient struct {
 }
 
 // New the manager client used by dynconfig
-func NewManagerClient(client managerclient.Client, hostOption HostOption) internaldynconfig.ManagerClient {
+func newManagerClient(client managerclient.Client, hostOption HostOption) internaldynconfig.ManagerClient {
 	return &managerClient{
 		Client:     client,
 		hostOption: hostOption,

client/config/dynconfig_test.go

@@ -109,7 +109,7 @@ func TestDynconfigNewDynconfig(t *testing.T) {
 
 			mockManagerClient := mocks.NewMockClient(ctl)
 			tc.mock(mockManagerClient.EXPECT())
-			_, err := NewDynconfig(NewManagerClient(mockManagerClient, tc.hostOption), tc.expire)
+			_, err := NewDynconfig(mockManagerClient, tc.hostOption, tc.expire)
 			tc.expect(t, err)
 			tc.cleanFileCache(t)
 		})
@@ -283,7 +283,7 @@ func TestDynconfigGet(t *testing.T) {
 
 			mockManagerClient := mocks.NewMockClient(ctl)
 			tc.mock(mockManagerClient.EXPECT(), tc.data)
-			dynconfig, err := NewDynconfig(NewManagerClient(mockManagerClient, tc.hostOption), tc.expire)
+			dynconfig, err := NewDynconfig(mockManagerClient, tc.hostOption, tc.expire)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -462,7 +462,7 @@ func TestDynconfigGetSchedulers(t *testing.T) {
 
 			mockManagerClient := mocks.NewMockClient(ctl)
 			tc.mock(mockManagerClient.EXPECT(), tc.data)
-			dynconfig, err := NewDynconfig(NewManagerClient(mockManagerClient, tc.hostOption), tc.expire)
+			dynconfig, err := NewDynconfig(mockManagerClient, tc.hostOption, tc.expire)
 			if err != nil {
 				t.Fatal(err)
 			}

client/config/peerhost.go

@@ -114,7 +114,7 @@ func (p *DaemonOption) Validate() error {
 	}
 
 	if p.Scheduler.Manager.Enable {
-		if p.Scheduler.Manager.Addr == "" {
+		if len(p.Scheduler.NetAddrs) == 0 {
 			return errors.New("manager addr is not specified")
 		}
 
@@ -140,8 +140,8 @@ type SchedulerOption struct {
 type ManagerOption struct {
 	// Enable get configuration from manager
 	Enable bool `mapstructure:"enable" yaml:"enable"`
-	// Addr is manager addresse
+	// NetAddrs is manager addresses.
-	Addr string `mapstructure:"addr" yaml:"addr"`
+	NetAddrs []dfnet.NetAddr `mapstructure:"netAddrs" yaml:"netAddrs"`
 	// RefreshInterval is the refresh interval
 	RefreshInterval time.Duration `mapstructure:"refreshInterval" yaml:"refreshInterval"`
 }

client/config/peerhost_test.go

@@ -229,8 +229,13 @@ func TestPeerHostOption_Load(t *testing.T) {
 		KeepStorage: false,
 		Scheduler: SchedulerOption{
 			Manager: ManagerOption{
-				Enable:          false,
+				Enable: false,
-				Addr:            "127.0.0.1:65003",
+				NetAddrs: []dfnet.NetAddr{
+					{
+						Type: dfnet.TCP,
+						Addr: "127.0.0.1:65003",
+					},
+				},
 				RefreshInterval: 5 * time.Minute,
 			},
 			NetAddrs: []dfnet.NetAddr{

client/config/testdata/config/daemon.yaml

@@ -6,7 +6,9 @@ keepStorage: false
 scheduler:
   manager:
     enable: false
-    addr: "127.0.0.1:65003"
+    netAddrs:
+      - type: tcp
+        addr: 127.0.0.1:65003
     refreshInterval: 5m
   netAddrs:
     - type: tcp

client/daemon/daemon.go

@@ -107,16 +107,13 @@ func New(opt *config.DaemonOption) (Daemon, error) {
 	var dynconfig config.Dynconfig
 	if opt.Scheduler.Manager.Enable == true {
 		// New manager client
-		managerClient, err := managerclient.New(opt.Scheduler.Manager.Addr)
+		managerClient, err := managerclient.NewWithAddrs(opt.Scheduler.Manager.NetAddrs)
 		if err != nil {
 			return nil, err
 		}
 
 		// New dynconfig client
-		if dynconfig, err = config.NewDynconfig(
+		if dynconfig, err = config.NewDynconfig(managerClient, opt.Host, opt.Scheduler.Manager.RefreshInterval); err != nil {
-			config.NewManagerClient(managerClient, opt.Host),
-			opt.Scheduler.Manager.RefreshInterval,
-		); err != nil {
 			return nil, err
 		}
 

deploy/helm-charts

@@ -1 +1 @@
-Subproject commit 7245a3ca839403e6f5fd61279e7f54b0cce59f60
+Subproject commit 88270e37feec344381e63a26275dceb8296c12ac

docs/en/deployment/configuration/dfget.yaml

@@ -28,8 +28,10 @@ scheduler:
   manager:
     # get scheduler list dynamically from manager
     enable: false
-    # manager service address
+    # manager service addresses
-    addr: 127.0.0.1:65003
+    netAddrs:
+      - type: tcp
+        addr: 127.0.0.1:65003
     # scheduler list refresh interval
     refreshInterval: 5m
   # schedule timeout

docs/zh-CN/deployment/configuration/dfget.yaml

@@ -28,7 +28,9 @@ scheduler:
     # 通过 manager 接口动态获取 scheduler 列表
     enable: false
     # manager 服务地址
-    addr: 127.0.0.1:65003
+    netAddrs:
+      - type: tcp
+        addr: 127.0.0.1:65003
     # scheduler 列表刷新时间
     refreshInterval: 5m
   # 调度超时

pkg/rpc/manager/client/client.go

@@ -18,6 +18,7 @@ package client
 
 import (
 	"context"
+	"errors"
 	"time"
 
 	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
@@ -27,6 +28,8 @@ import (
 	"google.golang.org/grpc/backoff"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
+	"d7y.io/dragonfly/v2/pkg/basic/dfnet"
+	"d7y.io/dragonfly/v2/pkg/reachable"
 	"d7y.io/dragonfly/v2/pkg/rpc/manager"
 )
 
@@ -91,6 +94,19 @@ func New(target string) (Client, error) {
 	}, nil
 }
 
+func NewWithAddrs(netAddrs []dfnet.NetAddr) (Client, error) {
+	for _, netAddr := range netAddrs {
+		ipReachable := reachable.New(&reachable.Config{Address: netAddr.Addr})
+		if err := ipReachable.Check(); err == nil {
+			logger.Infof("use %s address for manager grpc client", netAddr.Addr)
+			return New(netAddr.Addr)
+		}
+		logger.Warnf("%s address can not reachable", netAddr.Addr)
+	}
+
+	return nil, errors.New("can not find available addresses")
+}
+
 func (c *client) GetScheduler(req *manager.GetSchedulerRequest) (*manager.Scheduler, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), contextTimeout)
 	defer cancel()