worker: sync master's endpoint peroidcally

dm/worker/server.go

@@ -48,6 +48,7 @@ var (
 	keepaliveTimeout        = 3 * time.Second
 	keepaliveTime           = 3 * time.Second
 	retryConnectSleepTime   = time.Second
+	syncMasterEndpointsTime = 3 * time.Second // enough?
 	getMinPosForSubTaskFunc = getMinPosForSubTask
 )
 
@@ -108,6 +109,12 @@ func (s *Server) Start() error {
 		return err
 	}
 
+	s.wg.Add(1)
+	go func() {
+		s.syncMasterEndpoints(s.ctx)
+		s.wg.Done()
+	}()
+
 	bound, sourceCfg, revBound, err := ha.GetSourceBoundConfig(s.etcdClient, s.cfg.Name)
 	if err != nil {
 		// TODO: need retry
@@ -177,6 +184,28 @@ func (s *Server) Start() error {
 	return terror.ErrWorkerStartService.Delegate(err)
 }
 
+func (s *Server) syncMasterEndpoints(ctx context.Context) {
+	clientURLs := []string{}
+	for {
+		clientURLs = clientURLs[:0]
+		resp, err := s.etcdClient.MemberList(ctx)
+		if err != nil {
+			log.L().Error("can't get etcd member list", zap.Error(err))
+		} else {
+			for _, m := range resp.Members {
+				clientURLs = append(clientURLs, m.GetClientURLs()...)
+			}
+			log.L().Debug("will sync endpoints to", zap.Strings("client URLs", clientURLs))
+			s.etcdClient.SetEndpoints(clientURLs...)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(syncMasterEndpointsTime):
+		}
+	}
+}
+
 func (s *Server) observeSourceBound(ctx context.Context, etcdCli *clientv3.Client, rev int64) error {
 	var wg sync.WaitGroup
 	for {

dm/worker/server_test.go

@@ -101,7 +101,7 @@ func (t *testServer) TestServer(c *C) {
 		keepAliveTTL = int64(1)
 	)
 	etcdDir := c.MkDir()
-	ETCD, err := createMockETCD(etcdDir, "host://"+masterAddr)
+	ETCD, err := createMockETCD(etcdDir, "http://"+masterAddr)
 	c.Assert(err, IsNil)
 	defer ETCD.Close()
 	cfg := NewConfig()
@@ -230,7 +230,7 @@ func (t *testServer) TestWatchSourceBoundEtcdCompact(c *C) {
 		startRev     = int64(1)
 	)
 	etcdDir := c.MkDir()
-	ETCD, err := createMockETCD(etcdDir, "host://"+masterAddr)
+	ETCD, err := createMockETCD(etcdDir, "http://"+masterAddr)
 	c.Assert(err, IsNil)
 	defer ETCD.Close()
 	cfg := NewConfig()
@@ -376,7 +376,7 @@ func (t *testServer) testRetryConnectMaster(c *C, s *Server, ETCD *embed.Etcd, d
 	// When worker server fail to keepalive with etcd, sever should close its worker
 	c.Assert(s.getWorker(true), IsNil)
 	c.Assert(s.getSourceStatus(true).Result, IsNil)
-	ETCD, err := createMockETCD(dir, "host://"+hostName)
+	ETCD, err := createMockETCD(dir, "http://"+hostName)
 	c.Assert(err, IsNil)
 	time.Sleep(3 * time.Second)
 	return ETCD

dm/worker/worker_test.go

@@ -106,7 +106,7 @@ func (t *testServer2) TestTaskAutoResume(c *C) {
 	)
 	hostName := "127.0.0.1:8261"
 	etcdDir := c.MkDir()
-	ETCD, err := createMockETCD(etcdDir, "host://"+hostName)
+	ETCD, err := createMockETCD(etcdDir, "http://"+hostName)
 	c.Assert(err, IsNil)
 	defer ETCD.Close()
 
@@ -226,7 +226,7 @@ func (t *testWorkerEtcdCompact) TestWatchSubtaskStageEtcdCompact(c *C) {
 		startRev     = int64(1)
 	)
 	etcdDir := c.MkDir()
-	ETCD, err := createMockETCD(etcdDir, "host://"+masterAddr)
+	ETCD, err := createMockETCD(etcdDir, "http://"+masterAddr)
 	c.Assert(err, IsNil)
 	defer ETCD.Close()
 	cfg := NewConfig()
@@ -337,7 +337,7 @@ func (t *testWorkerEtcdCompact) TestWatchRelayStageEtcdCompact(c *C) {
 		startRev     = int64(1)
 	)
 	etcdDir := c.MkDir()
-	ETCD, err := createMockETCD(etcdDir, "host://"+masterAddr)
+	ETCD, err := createMockETCD(etcdDir, "http://"+masterAddr)
 	c.Assert(err, IsNil)
 	defer ETCD.Close()
 	cfg := NewConfig()

pkg/ha/keepalive.go

@@ -71,7 +71,6 @@ func workerEventFromKey(key string) (WorkerEvent, error) {
 // KeepAlive puts the join time of the workerName into etcd.
 // this key will be kept in etcd until the worker is blocked or failed
 // k/v: workerName -> join time.
-// TODO: fetch the actual master endpoints, the master member maybe changed.
 func KeepAlive(ctx context.Context, cli *clientv3.Client, workerName string, keepAliveTTL int64) error {
 	cliCtx, cancel := context.WithTimeout(ctx, etcdutil.DefaultRequestTimeout)
 	defer cancel()

tests/ha/conf/dm-worker1.toml

@@ -1,2 +1,2 @@
 name = "worker1"
-join = "127.0.0.1:8261,127.0.0.1:8361,127.0.0.1:8461,127.0.0.1:8561,127.0.0.1:8661"
+join = "127.0.0.1:8261"

tests/ha/conf/dm-worker2.toml

@@ -1,2 +1,2 @@
 name = "worker2"
-join = "127.0.0.1:8261,127.0.0.1:8361,127.0.0.1:8461,127.0.0.1:8561,127.0.0.1:8661"
+join = "127.0.0.1:8261"