worker(dm): add retry for watch when network is weak

dm/dm/worker/server.go

@@ -239,8 +239,8 @@ func (s *Server) Start() error {
 // worker keepalive with master
 // If worker loses connect from master, it would stop all task and try to connect master again.
 func (s *Server) startKeepAlive() {
-	s.kaWg.Add(1)
 	s.kaCtx, s.kaCancel = context.WithCancel(s.ctx)
+	s.kaWg.Add(1)
 	go s.doStartKeepAlive()
 }
 
@@ -431,7 +431,7 @@ func (s *Server) observeSourceBound(ctx context.Context, rev int64) error {
 			}
 		} else {
 			if err != nil {
-				log.L().Error("observeSourceBound is failed and will quit now", zap.Error(err))
+				log.L().Error("observeSourceBound is failed and will quit now outer logic will trigger a retry", zap.Error(err))
 			} else {
 				log.L().Info("observeSourceBound will quit now")
 			}
@@ -564,7 +564,7 @@ OUTER:
 			}
 			// TODO: Deal with err
 			log.L().Error("WatchSourceBound received an error", zap.Error(err))
-			if etcdutil.IsRetryableError(err) {
+			if etcdutil.IsRetryableError(err) || errors.Cause(err) == etcdutil.ErrEtcdWatchChannelClose {
 				return err
 			}
 		}

dm/dm/worker/source_worker.go

@@ -521,7 +521,7 @@ func (w *SourceWorker) DisableHandleSubtasks() {
 
 	// close all sub tasks
 	w.subTaskHolder.closeAllSubTasks()
-	w.l.Info("handling subtask enabled")
+	w.l.Info("handling subtask disabled")
 }
 
 // fetchSubTasksAndAdjust gets source's subtask stages and configs, adjust some values by worker's config and status

dm/pkg/etcdutil/etcdutil.go

@@ -55,6 +55,8 @@ var etcdDefaultTxnRetryParam = retry.Params{
 
 var etcdDefaultTxnStrategy = retry.FiniteRetryStrategy{}
 
+var ErrEtcdWatchChannelClose = errors.New("etcd watch channel closed")
+
 // CreateClient creates an etcd client with some default config items.
 func CreateClient(endpoints []string, tlsCfg *tls.Config) (*clientv3.Client, error) {
 	return clientv3.New(clientv3.Config{

dm/pkg/ha/bound.go

@@ -269,7 +269,20 @@ func WatchSourceBound(ctx context.Context, cli *clientv3.Client, worker string,
 		case <-ctx.Done():
 			return
 		case resp, ok := <-ch:
+			failpoint.Inject("WatchSourceBoundChanClosed", func() {
+				ok = false
+			})
 			if !ok {
+				log.L().Info("WatchSourceBound chan closed! observeSourceBound will exit！")
+				select {
+				case errCh <- etcdutil.ErrEtcdWatchChannelClose:
+					// etcd client will return a closed channel if underlying grpc connection is closed
+					// see https://github.com/etcd-io/etcd/blob/20c89df5e5e2d738efb9c276d954d754eb86918b/client/v3/watch.go#L325
+					// and https://github.com/etcd-io/etcd/issues/8980#issuecomment-350070282
+					// we return an error to let outer logic do the whole retry for `observeSourceBound`
+				case <-ctx.Done():
+					return
+				}
 				return
 			}
 			if resp.Canceled {

dm/tests/_utils/ha_cases_lib.sh

@@ -155,6 +155,8 @@ function cleanup() {
 		$(mysql -h127.0.0.1 -p123456 -P${i} -uroot -e "drop database if exists ha_test2;")
 		sleep 1
 	done
+	export GO_FAILPOINTS=""
+	cleanup_process $*
 }
 
 function isolate_master() {

dm/tests/ha_cases3_1/run.sh

@@ -48,8 +48,9 @@ function test_multi_task_running() {
 
 function test_isolate_master_and_worker() {
 	echo "[$(date)] <<<<<< start test_isolate_master_and_worker >>>>>>"
-
-	test_multi_task_running
+	cleanup
+	prepare_sql_multi_task
+	start_multi_tasks_cluster
 
 	# join master4 and master5
 	run_dm_master $WORK_DIR/master-join4 $MASTER_PORT4 $cur/conf/dm-master-join4.toml
@@ -129,7 +130,48 @@ function test_isolate_master_and_worker() {
 	echo "[$(date)] <<<<<< finish test_isolate_master_and_worker >>>>>>"
 }
 
+function test_watch_source_bound_exit() {
+	echo "[$(date)] <<<<<< start test_watch_source_bound_exit >>>>>>"
+	cleanup
+	export GO_FAILPOINTS="github.com/pingcap/tiflow/dm/pkg/ha/WatchSourceBoundChanClosed=return()"
+
+	echo "start DM worker and master cluster"
+	run_dm_master $WORK_DIR/master1 $MASTER_PORT1 $cur/conf/dm-master1.toml
+	run_dm_master $WORK_DIR/master2 $MASTER_PORT2 $cur/conf/dm-master2.toml
+	run_dm_master $WORK_DIR/master3 $MASTER_PORT3 $cur/conf/dm-master3.toml
+	check_rpc_alive $cur/../bin/check_master_online 127.0.0.1:$MASTER_PORT1
+	check_rpc_alive $cur/../bin/check_master_online 127.0.0.1:$MASTER_PORT2
+	check_rpc_alive $cur/../bin/check_master_online 127.0.0.1:$MASTER_PORT3
+
+	echo "start worker and operate mysql config to worker"
+	run_dm_worker $WORK_DIR/worker1 $WORKER1_PORT $cur/conf/dm-worker1.toml
+	check_rpc_alive $cur/../bin/check_worker_online 127.0.0.1:$WORKER1_PORT
+	cp $cur/conf/source1.yaml $WORK_DIR/source1.yaml
+	sed -i "/relay-binlog-name/i\relay-dir: $WORK_DIR/worker1/relay_log" $WORK_DIR/source1.yaml
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT1" \
+		"operate-source create $WORK_DIR/source1.yaml" \
+		"\"result\": true" 1 \
+		"fail to get expected result" 1
+
+	# master think worker1 is online and bound source1
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT1" \
+		"list-member --name worker1" \
+		"\"source\": \"mysql-replica-01\"" 1
+
+	# but worker1 is not handling this source because worker watch source bound thread is exited
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT1" \
+		"query-status" \
+		"no mysql source is being handled in the worker" 1
+
+	# dm-worker will retry watch source bound again and agin
+	check_log_contain_with_retry 'observeSourceBound is failed and will quit now outer logic will trigger a retry' $WORK_DIR/worker1/log/dm-worker.log
+
+	echo "[$(date)] <<<<<< finish test_watch_source_bound_exit >>>>>>"
+}
+
 function run() {
+	test_multi_task_running
+	test_watch_source_bound_exit
 	test_isolate_master_and_worker # TICASE-934, 935, 936, 987, 992, 998, 999
 }