worker, ha: increase keepalive TTL to 1 minute, and 30 minutes if rel…

…ay enabled

dm/worker/config.go

@@ -33,8 +33,11 @@ import (
 // SampleConfigFile is sample config file of dm-worker
 // later we can read it from dm/worker/dm-worker.toml
 // and assign it to SampleConfigFile while we build dm-worker
-var SampleConfigFile string
-var defaultKeepAliveTTL = int64(10)
+var (
+	SampleConfigFile         string
+	defaultKeepAliveTTL      = int64(60)      // 1 minute
+	relayEnabledKeepAliveTTL = int64(60 * 30) // 30 minutes
+)
 
 // NewConfig creates a new base config for worker.
 func NewConfig() *Config {

dm/worker/server.go

@@ -588,6 +588,10 @@ func (s *Server) startWorker(cfg *config.SourceConfig) error {
 			return err
 		}
 		startRelay = !relayStage.IsDeleted && relayStage.Expect == pb.Stage_Running
+		// change keepalive TTL to 30 minutes if it's default value
+		if s.cfg.KeepAliveTTL == defaultKeepAliveTTL {
+			ha.NotifyKeepAliveChange(relayEnabledKeepAliveTTL)
+		}
 	}
 	go func() {
 		w.Start(startRelay)

pkg/ha/keepalive.go

@@ -27,6 +27,15 @@ import (
 	"github.com/pingcap/dm/pkg/log"
 )
 
+var (
+	keepAliveUpdateCh = make(chan int64, 10)
+)
+
+// NotifyKeepAliveChange is used to dynamically change keepalive TTL and don't let watcher observe a DELETE of old key
+func NotifyKeepAliveChange(newTTL int64) {
+	keepAliveUpdateCh <- newTTL
+}
+
 // WorkerEvent represents the PUT/DELETE keepalive event of DM-worker.
 type WorkerEvent struct {
 	WorkerName string    `json:"worker-name"` // the worker name of the worker.
@@ -98,7 +107,12 @@ func KeepAlive(ctx context.Context, cli *clientv3.Client, workerName string, kee
 		}
 	}()
 
-	ch, err := cli.KeepAlive(ctx, lease.ID)
+	keepAliveCtx, keepAliveCancel := context.WithCancel(ctx)
+	defer func() {
+		keepAliveCancel()
+	}()
+
+	ch, err := cli.KeepAlive(keepAliveCtx, lease.ID)
 	if err != nil {
 		return err
 	}
@@ -112,6 +126,30 @@ func KeepAlive(ctx context.Context, cli *clientv3.Client, workerName string, kee
 		case <-ctx.Done():
 			log.L().Info("ctx is canceled, keepalive will exit now")
 			return nil
+		case newTTL := <-keepAliveUpdateCh:
+			// create a new lease with new TTL, and overwrite original KV
+			cliCtx, cancel := context.WithTimeout(ctx, etcdutil.DefaultRequestTimeout)
+			defer cancel()
+			lease, err = cli.Grant(cliCtx, newTTL)
+			if err != nil {
+				log.L().Error("meet error when change keepalive TTL", zap.Error(err))
+				return err
+			}
+			_, err = cli.Put(cliCtx, k, workerEventJSON, clientv3.WithLease(lease.ID))
+			if err != nil {
+				log.L().Error("meet error when change keepalive TTL", zap.Error(err))
+				return err
+			}
+
+			oldCancel := keepAliveCancel
+			keepAliveCtx, keepAliveCancel = context.WithCancel(ctx)
+			ch, err = cli.KeepAlive(keepAliveCtx, lease.ID)
+			if err != nil {
+				log.L().Error("meet error when change keepalive TTL", zap.Error(err))
+				return err
+			}
+			// after new keepalive is succeed, we cancel the old keepalive
+			oldCancel()
 		}
 	}
 }