dmctl, scheduler: support manually transfer-source

_utils/terror_gen/errors_release.txt

@@ -491,6 +491,7 @@ ErrSchedulerSubTaskStageInvalidUpdate,[code=46015:class=dm-master:scope=internal
 ErrSchedulerSubTaskOpTaskNotExist,[code=46016:class=dm-master:scope=internal:level=medium], "Message: subtasks with name %s need to be operate not exist, Workaround: Please use `query-status` command to see tasks."
 ErrSchedulerSubTaskOpSourceNotExist,[code=46017:class=dm-master:scope=internal:level=medium], "Message: sources %v need to be operate not exist"
 ErrSchedulerTaskNotExist,[code=46018:class=scheduler:scope=internal:level=medium], "Message: task with name %s not exist, Workaround: Please use `query-status` command to see tasks."
+ErrSchedulerRequireNotRunning,[code=46019:class=scheduler:scope=internal:level=high], "Message: tasks %v on source %s should not be running, Workaround: Please use `pause-task [-s source ...] task` to pause them first"
 ErrCtlGRPCCreateConn,[code=48001:class=dmctl:scope=internal:level=high], "Message: can not create grpc connection, Workaround: Please check your network connection."
 ErrCtlInvalidTLSCfg,[code=48002:class=dmctl:scope=internal:level=medium], "Message: invalid TLS config, Workaround: Please check the `ssl-ca`, `ssl-cert` and `ssl-key` config in command line."
 ErrNotSet,[code=50000:class=not-set:scope=not-set:level=high]

dm/ctl/ctl.go

@@ -74,6 +74,7 @@ func NewRootCmd() *cobra.Command {
 		master.NewOperateSchemaCmd(),
 		master.NewGetCfgCmd(),
 		master.NewHandleErrorCmd(),
+		master.NewTransferSourceCmd(),
 	)
 	// copied from (*cobra.Command).InitDefaultHelpCmd
 	helpCmd := &cobra.Command{

dm/ctl/master/transfer_source.go

@@ -0,0 +1,68 @@
+// Copyright 2021 PingCAP, Inc.
+//
+// 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package master
+
+import (
+	"context"
+	"errors"
+	"os"
+
+	"github.com/pingcap/dm/dm/ctl/common"
+	"github.com/pingcap/dm/dm/pb"
+
+	"github.com/spf13/cobra"
+)
+
+// NewTransferSourceCmd creates a TransferSource command
+func NewTransferSourceCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "transfer-source <source-id> <worker-id>",
+		Short: "Transfers a upstream MySQL/MariaDB source to a free worker.",
+		RunE:  transferSourceFunc,
+	}
+	return cmd
+}
+
+func transferSourceFunc(cmd *cobra.Command, _ []string) (err error) {
+	if len(cmd.Flags().Args()) != 2 {
+		cmd.SetOut(os.Stdout)
+		common.PrintCmdUsage(cmd)
+		err = errors.New("please check output to see error")
+		return
+	}
+
+	sourceID := cmd.Flags().Arg(0)
+	workerID := cmd.Flags().Arg(1)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	resp := &pb.TransferSourceResponse{}
+	err = common.SendRequest(
+		ctx,
+		"TransferSource",
+		&pb.TransferSourceRequest{
+			Source: sourceID,
+			Worker: workerID,
+		},
+		&resp,
+	)
+
+	if err != nil {
+		return
+	}
+
+	common.PrettyPrintResponse(resp)
+	return
+}

dm/master/scheduler/scheduler.go

@@ -15,10 +15,12 @@ package scheduler
 
 import (
 	"context"
+	"errors"
 	"sort"
 	"sync"
 	"time"
 
+	"github.com/pingcap/failpoint"
 	"go.etcd.io/etcd/clientv3"
 	"go.uber.org/zap"
 
@@ -351,6 +353,81 @@ func (s *Scheduler) GetSourceCfgByID(source string) *config.SourceConfig {
 	return &clone
 }
 
+// TransferSource unbinds the source and binds it to a free worker. If fails halfway, the old worker should try recover
+func (s *Scheduler) TransferSource(source, worker string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if !s.started {
+		return terror.ErrSchedulerNotStarted.Generate()
+	}
+
+	// 1. check existence or no need
+	if _, ok := s.sourceCfgs[source]; !ok {
+		return terror.ErrSchedulerSourceCfgNotExist.Generate(source)
+	}
+	w, ok := s.workers[worker]
+	if !ok {
+		return terror.ErrSchedulerWorkerNotExist.Generate(worker)
+	}
+	oldWorker, hasOldWorker := s.bounds[source]
+	if hasOldWorker && oldWorker.BaseInfo().Name == worker {
+		return nil
+	}
+
+	// 2. check new worker is free
+	stage := w.Stage()
+	if stage != WorkerFree {
+		return terror.ErrSchedulerWorkerInvalidTrans.Generate(worker, stage, WorkerBound)
+	}
+
+	// 3. if no old worker, bound it directly
+	if !hasOldWorker {
+		s.logger.Warn("in transfer source, found a free worker and not bound source, which should not happened",
+			zap.String("source", source),
+			zap.String("worker", worker))
+		err := s.boundSourceToWorker(source, w)
+		if err == nil {
+			delete(s.unbounds, source)
+		}
+		return err
+	}
+
+	// 4. if there's old worker, make sure it's not running
+	var runningTasks []string
+	for task, subtaskM := range s.expectSubTaskStages {
+		subtaskStage, ok2 := subtaskM[source]
+		if !ok2 {
+			continue
+		}
+		if subtaskStage.Expect == pb.Stage_Running {
+			runningTasks = append(runningTasks, task)
+		}
+	}
+	if len(runningTasks) > 0 {
+		return terror.ErrSchedulerRequireNotRunning.Generate(runningTasks, source)
+	}
+
+	// 5. replace the source bound
+	failpoint.Inject("failToReplaceSourceBound", func(_ failpoint.Value) {
+		failpoint.Return(errors.New("failToPutSourceBound"))
+	})
+	_, err := ha.ReplaceSourceBound(s.etcdCli, source, oldWorker.BaseInfo().Name, worker)
+	if err != nil {
+		return err
+	}
+	oldWorker.ToFree()
+	// we have checked w.stage is free, so there should not be an error
+	_ = s.updateStatusForBound(w, ha.NewSourceBound(source, worker))
+
+	// 6. try bound the old worker
+	_, err = s.tryBoundForWorker(oldWorker)
+	if err != nil {
+		s.logger.Warn("in transfer source, error when try bound the old worker", zap.Error(err))
+	}
+	return nil
+}
+
 // AddSubTasks adds the information of one or more subtasks for one task.
 func (s *Scheduler) AddSubTasks(cfgs ...config.SubTaskConfig) error {
 	s.mu.Lock()

dm/master/scheduler/scheduler_test.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	. "github.com/pingcap/check"
+	"github.com/pingcap/failpoint"
 	"go.etcd.io/etcd/clientv3"
 	v3rpc "go.etcd.io/etcd/etcdserver/api/v3rpc/rpctypes"
 	"go.etcd.io/etcd/integration"
@@ -1008,7 +1009,7 @@ func (t *testScheduler) TestLastBound(c *C) {
 	c.Assert(err, IsNil)
 	c.Assert(bounded, IsFalse)
 
-	// after worker3 become offline, worker2 should be bounded to worker2
+	// after worker3 become offline, source2 should be bounded to worker2
 	s.updateStatusForUnbound(sourceID2)
 	_, ok := s.bounds[sourceID2]
 	c.Assert(ok, IsFalse)
@@ -1018,3 +1019,98 @@ func (t *testScheduler) TestLastBound(c *C) {
 	c.Assert(bounded, IsTrue)
 	c.Assert(s.bounds[sourceID2], DeepEquals, worker2)
 }
+
+func (t *testScheduler) TestTransferSource(c *C) {
+	defer clearTestInfoOperation(c)
+
+	var (
+		logger      = log.L()
+		s           = NewScheduler(&logger, config.Security{})
+		sourceID1   = "mysql-replica-1"
+		sourceID2   = "mysql-replica-2"
+		sourceID3   = "mysql-replica-3"
+		sourceID4   = "mysql-replica-4"
+		workerName1 = "dm-worker-1"
+		workerName2 = "dm-worker-2"
+		workerName3 = "dm-worker-3"
+		workerName4 = "dm-worker-4"
+	)
+
+	worker1 := &Worker{baseInfo: ha.WorkerInfo{Name: workerName1}}
+	worker2 := &Worker{baseInfo: ha.WorkerInfo{Name: workerName2}}
+	worker3 := &Worker{baseInfo: ha.WorkerInfo{Name: workerName3}}
+	worker4 := &Worker{baseInfo: ha.WorkerInfo{Name: workerName4}}
+
+	// step 1: start an empty scheduler
+	s.started = true
+	s.etcdCli = etcdTestCli
+	s.workers[workerName1] = worker1
+	s.workers[workerName2] = worker2
+	s.workers[workerName3] = worker3
+	s.workers[workerName4] = worker4
+	s.sourceCfgs[sourceID1] = config.SourceConfig{}
+	s.sourceCfgs[sourceID2] = config.SourceConfig{}
+
+	worker1.ToFree()
+	c.Assert(s.boundSourceToWorker(sourceID1, worker1), IsNil)
+	worker2.ToFree()
+	c.Assert(s.boundSourceToWorker(sourceID2, worker2), IsNil)
+
+	c.Assert(s.bounds[sourceID1], DeepEquals, worker1)
+	c.Assert(s.bounds[sourceID2], DeepEquals, worker2)
+
+	worker3.ToFree()
+	worker4.ToFree()
+
+	// test invalid transfer: source not exists
+	c.Assert(s.TransferSource("not-exist", workerName3), NotNil)
+
+	// test valid transfer: source -> worker = bound -> free
+	c.Assert(s.TransferSource(sourceID1, workerName4), IsNil)
+	c.Assert(s.bounds[sourceID1], DeepEquals, worker4)
+	c.Assert(worker1.Stage(), Equals, WorkerFree)
+
+	// test valid transfer: source -> worker = unbound -> free
+	s.sourceCfgs[sourceID3] = config.SourceConfig{}
+	s.unbounds[sourceID3] = struct{}{}
+	c.Assert(s.TransferSource(sourceID3, workerName3), IsNil)
+	c.Assert(s.bounds[sourceID3], DeepEquals, worker3)
+
+	// test valid transfer: self
+	c.Assert(s.TransferSource(sourceID3, workerName3), IsNil)
+	c.Assert(s.bounds[sourceID3], DeepEquals, worker3)
+
+	// test invalid transfer: source -> worker = bound -> bound
+	c.Assert(s.TransferSource(sourceID1, workerName3), NotNil)
+	c.Assert(s.bounds[sourceID1], DeepEquals, worker4)
+	c.Assert(s.bounds[sourceID3], DeepEquals, worker3)
+
+	// test invalid transfer: source -> worker = bound -> offline
+	worker1.ToOffline()
+	c.Assert(s.TransferSource(sourceID1, workerName1), NotNil)
+	c.Assert(s.bounds[sourceID1], DeepEquals, worker4)
+
+	// test invalid transfer: source -> worker = unbound -> bound
+	s.sourceCfgs[sourceID4] = config.SourceConfig{}
+	s.unbounds[sourceID4] = struct{}{}
+	c.Assert(s.TransferSource(sourceID4, workerName3), NotNil)
+	c.Assert(s.bounds[sourceID3], DeepEquals, worker3)
+	delete(s.unbounds, sourceID4)
+	delete(s.sourceCfgs, sourceID4)
+
+	worker1.ToFree()
+	// now we have (worker1, nil) (worker2, source2) (worker3, source3) (worker4, source1)
+
+	// test fail halfway won't left old worker unbound
+	c.Assert(failpoint.Enable("github.com/pingcap/dm/dm/master/scheduler/failToReplaceSourceBound", `return()`), IsNil)
+	c.Assert(s.TransferSource(sourceID1, workerName1), NotNil)
+	c.Assert(s.bounds[sourceID1], DeepEquals, worker4)
+	c.Assert(worker1.Stage(), Equals, WorkerFree)
+	c.Assert(failpoint.Disable("github.com/pingcap/dm/dm/master/scheduler/failToReplaceSourceBound"), IsNil)
+
+	// test can't transfer when there's any running task on the source
+	s.expectSubTaskStages["test"] = map[string]ha.Stage{sourceID1: {Expect: pb.Stage_Running}}
+	c.Assert(s.TransferSource(sourceID1, workerName1), NotNil)
+	c.Assert(s.bounds[sourceID1], DeepEquals, worker4)
+	c.Assert(worker1.Stage(), Equals, WorkerFree)
+}

dm/master/server.go

@@ -1982,6 +1982,26 @@ func (s *Server) HandleError(ctx context.Context, req *pb.HandleErrorRequest) (*
 	}, nil
 }
 
+// TransferSource implements MasterServer.TransferSource
+func (s *Server) TransferSource(ctx context.Context, req *pb.TransferSourceRequest) (*pb.TransferSourceResponse, error) {
+	var (
+		resp2 = &pb.TransferSourceResponse{}
+		err2  error
+	)
+	shouldRet := s.sharedLogic(ctx, req, &resp2, &err2)
+	if shouldRet {
+		return resp2, err2
+	}
+
+	err := s.scheduler.TransferSource(req.Source, req.Worker)
+	if err != nil {
+		resp2.Msg = err.Error()
+		return resp2, nil
+	}
+	resp2.Result = true
+	return resp2, nil
+}
+
 // sharedLogic does some shared logic for each RPC implementation
 // arguments with `Pointer` suffix should be pointer to that variable its name indicated
 // return `true` means caller should return with variable that `xxPointer` modified