dm-ha/: add remove metadata feature

_utils/terror_gen/errors_release.txt

@@ -339,6 +339,7 @@ ErrMasterRequestIsNotForwardToLeader,[code=38043:class=dm-master:scope=internal:
 ErrMasterIsNotAsyncRequest,[code=38044:class=dm-master:scope=internal:level=medium],"request %s is not an async one, needn't wait for ok"
 ErrMasterFailToGetExpectResult,[code=38045:class=dm-master:scope=internal:level=medium],"fail to get expected result"
 ErrMasterPessimistNotStarted,[code=38046:class=dm-master:scope=internal:level=medium],"the shardddl pessimist has not started"
+ErrMasterOptimistNotStarted,[code=38047:class=dm-master:scope=internal:level=medium],"the shardddl optimist has not started"
 ErrWorkerParseFlagSet,[code=40001:class=dm-worker:scope=internal:level=medium],"parse dm-worker config flag set"
 ErrWorkerInvalidFlag,[code=40002:class=dm-worker:scope=internal:level=medium],"'%s' is an invalid flag"
 ErrWorkerDecodeConfigFromFile,[code=40003:class=dm-worker:scope=internal:level=medium],"toml decode file"

dm/config/subtask.go

@@ -149,7 +149,6 @@ type SubTaskConfig struct {
 	ServerID                uint32 `toml:"server-id" json:"server-id"`
 	Flavor                  string `toml:"flavor" json:"flavor"`
 	MetaSchema              string `toml:"meta-schema" json:"meta-schema"`
-	RemoveMeta              bool   `toml:"remove-meta" json:"remove-meta"`
 	HeartbeatUpdateInterval int    `toml:"heartbeat-update-interval" json:"heartbeat-update-interval"`
 	HeartbeatReportInterval int    `toml:"heartbeat-report-interval" json:"heartbeat-report-interval"`
 	EnableHeartbeat         bool   `toml:"enable-heartbeat" json:"enable-heartbeat"`

dm/config/task.go

@@ -256,9 +256,7 @@ type TaskConfig struct {
 	// we store detail status in meta
 	// don't save configuration into it
 	MetaSchema string `yaml:"meta-schema"`
-	// remove meta from downstreaming database
-	// now we delete checkpoint and online ddl information
-	RemoveMeta              bool   `yaml:"remove-meta"`
+
 	EnableHeartbeat         bool   `yaml:"enable-heartbeat"`
 	HeartbeatUpdateInterval int    `yaml:"heartbeat-update-interval"`
 	HeartbeatReportInterval int    `yaml:"heartbeat-report-interval"`
@@ -510,7 +508,6 @@ func (c *TaskConfig) SubTaskConfigs(sources map[string]DBConfig) ([]*SubTaskConf
 		cfg.Mode = c.TaskMode
 		cfg.CaseSensitive = c.CaseSensitive
 		cfg.MetaSchema = c.MetaSchema
-		cfg.RemoveMeta = c.RemoveMeta
 		cfg.EnableHeartbeat = c.EnableHeartbeat
 		cfg.HeartbeatUpdateInterval = c.HeartbeatUpdateInterval
 		cfg.HeartbeatReportInterval = c.HeartbeatReportInterval

dm/config/task_test.go

@@ -29,7 +29,6 @@ name: test
 task-mode: all
 is-sharding: true
 meta-schema: "dm_meta"
-remove-meta: false
 enable-heartbeat: true
 timezone: "Asia/Shanghai"
 ignore-checking-items: ["all"]
@@ -56,7 +55,6 @@ name: test1
 task-mode: all
 is-sharding: true
 meta-schema: "dm_meta"
-remove-meta: false
 enable-heartbeat: true
 timezone: "Asia/Shanghai"
 ignore-checking-items: ["all"]
@@ -80,7 +78,7 @@ mysql-instances:
 	err := taskConfig.Decode(errorTaskConfig1)
 	// field server-id is not a member of TaskConfig
 	c.Check(err, NotNil)
-	c.Assert(err, ErrorMatches, "*line 19: field server-id not found in type config.MySQLInstance*")
+	c.Assert(err, ErrorMatches, "*line 18: field server-id not found in type config.MySQLInstance*")
 
 	err = taskConfig.Decode(errorTaskConfig2)
 	// field name duplicate
@@ -94,7 +92,6 @@ name: test
 task-mode: all
 is-sharding: true
 meta-schema: "dm_meta"
-remove-meta: false
 enable-heartbeat: true
 timezone: "Asia/Shanghai"
 ignore-checking-items: ["all"]
@@ -113,7 +110,6 @@ task-mode: all
 task-mode: all
 is-sharding: true
 meta-schema: "dm_meta"
-remove-meta: false
 enable-heartbeat: true
 timezone: "Asia/Shanghai"
 ignore-checking-items: ["all"]
@@ -131,7 +127,6 @@ name: test
 task-mode: all
 is-sharding: true
 meta-schema: "dm_meta"
-remove-meta: false
 enable-heartbeat: true
 heartbeat-update-interval: 1
 heartbeat-report-interval: 1
@@ -210,7 +205,6 @@ task-mode: all
 is-sharding: true
 shard-mode: "optimistic"
 meta-schema: "dm_meta"
-remove-meta: false
 enable-heartbeat: true
 heartbeat-update-interval: 1
 heartbeat-report-interval: 1

dm/ctl/master/start_task.go

@@ -23,15 +23,17 @@ import (
 	"github.com/pingcap/dm/checker"
 	"github.com/pingcap/dm/dm/ctl/common"
 	"github.com/pingcap/dm/dm/pb"
+	"github.com/pingcap/dm/pkg/terror"
 )
 
 // NewStartTaskCmd creates a StartTask command
 func NewStartTaskCmd() *cobra.Command {
 	cmd := &cobra.Command{
-		Use:   "start-task [-s source ...] <config-file>",
+		Use:   "start-task [-s source ...] [--remove-meta] <config-file>",
 		Short: "start a task as defined in the config file",
 		Run:   startTaskFunc,
 	}
+	cmd.Flags().BoolP("remove-meta", "", false, "whether to remove task's meta data")
 	return cmd
 }
 
@@ -54,14 +56,21 @@ func startTaskFunc(cmd *cobra.Command, _ []string) {
 		return
 	}
 
+	removeMeta, err := cmd.Flags().GetBool("remove-meta")
+	if err != nil {
+		common.PrintLines("%s", terror.Message(err))
+		return
+	}
+
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
 	// start task
 	cli := common.MasterClient()
 	resp, err := cli.StartTask(ctx, &pb.StartTaskRequest{
-		Task:    string(content),
-		Sources: sources,
+		Task:       string(content),
+		Sources:    sources,
+		RemoveMeta: removeMeta,
 	})
 	if err != nil {
 		common.PrintLines("can not start task:\n%v", errors.ErrorStack(err))

dm/dm-ansible/conf/task_advanced.yaml.example

@@ -3,7 +3,6 @@ name: test # global unique
 task-mode: all  # full/incremental/all
 is-sharding: true  # whether multi dm-worker do one sharding job
 meta-schema: "dm_meta"  # meta schema in downstreaming database to store meta informaton of dm
-remove-meta: false  # remove meta from downstreaming database, now we delete checkpoint and online ddl information
 enable-heartbeat: false  # whether to enable heartbeat for calculating lag between master and syncer
 # heartbeat-update-interval: 1  # interval to do heartbeat and save timestamp, default 1s
 # heartbeat-report-interval: 10 # interval to report time lap to prometheus, default 10s
@@ -23,8 +22,7 @@ mysql-instances:             # one or more source database, config more source d
     # `full` / `all`:
     #    never be used
     # `incremental`:
-    #    if `remove-meta` is true, this will be used
-    #    else if checkpoints already exists in `meta-schema`, this will not be used
+    #    if checkpoints already exists in `meta-schema`, this will not be used
     #    otherwise, this will be used
     meta:
       binlog-name: mysql-bin.000001

dm/master/server.go

@@ -38,6 +38,8 @@ import (
 	"github.com/pingcap/dm/dm/master/workerrpc"
 	"github.com/pingcap/dm/dm/pb"
 	"github.com/pingcap/dm/pkg/conn"
+	tcontext "github.com/pingcap/dm/pkg/context"
+	"github.com/pingcap/dm/pkg/cputil"
 	"github.com/pingcap/dm/pkg/election"
 	"github.com/pingcap/dm/pkg/etcdutil"
 	"github.com/pingcap/dm/pkg/log"
@@ -81,6 +83,9 @@ type Server struct {
 	leaderClient   pb.MasterClient
 	leaderGrpcConn *grpc.ClientConn
 
+	// removeMetaLock locks start task when removing meta
+	removeMetaLock sync.RWMutex
+
 	// WaitGroup for background functions.
 	bgFunWg sync.WaitGroup
 
@@ -360,15 +365,31 @@ func (s *Server) StartTask(ctx context.Context, req *pb.StartTaskRequest) (*pb.S
 	if len(sourceRespCh) > 0 {
 		sourceResps = sortCommonWorkerResults(sourceRespCh)
 	} else {
+		sources := make([]string, 0, len(stCfgs))
+		for _, stCfg := range stCfgs {
+			sources = append(sources, stCfg.SourceID)
+		}
+		s.removeMetaLock.Lock()
+		if req.RemoveMeta {
+			if scm := s.scheduler.GetSubTaskCfgsByTask(cfg.Name); len(scm) > 0 {
+				resp.Msg = terror.Annotate(terror.ErrSchedulerSubTaskExist.Generate(cfg.Name, sources),
+					"while remove-meta is true").Error()
+				s.removeMetaLock.Unlock()
+				return resp, nil
+			}
+			err = s.removeMetaData(ctx, cfg)
+			if err != nil {
+				resp.Msg = terror.Annotate(err, "while removing metadata").Error()
+				s.removeMetaLock.Unlock()
+				return resp, nil
+			}
+		}
 		err = s.scheduler.AddSubTasks(subtaskCfgPointersToInstances(stCfgs...)...)
+		s.removeMetaLock.Unlock()
 		if err != nil {
 			resp.Msg = errors.ErrorStack(err)
 			return resp, nil
 		}
-		sources := make([]string, 0, len(stCfgs))
-		for _, stCfg := range stCfgs {
-			sources = append(sources, stCfg.SourceID)
-		}
 		resp.Result = true
 		sourceResps = s.getSourceRespsAfterOperation(ctx, cfg.Name, sources, []string{}, req)
 	}
@@ -1332,6 +1353,55 @@ func (s *Server) generateSubTask(ctx context.Context, task string) (*config.Task
 	return cfg, stCfgs, nil
 }
 
+func (s *Server) removeMetaData(ctx context.Context, cfg *config.TaskConfig) error {
+	toDB := *cfg.TargetDB
+	toDB.Adjust()
+	if len(toDB.Password) > 0 {
+		pswdTo, err := utils.Decrypt(toDB.Password)
+		if err != nil {
+			return err
+		}
+		toDB.Password = pswdTo
+	}
+
+	// clear shard meta data for pessimistic/optimist
+	err := s.pessimist.RemoveMetaData(cfg.Name)
+	if err != nil {
+		return err
+	}
+	err = s.optimist.RemoveMetaData(cfg.Name)
+	if err != nil {
+		return err
+	}
+
+	// set up db and clear meta data in downstream db
+	baseDB, err := conn.DefaultDBProvider.Apply(toDB)
+	if err != nil {
+		return terror.WithScope(err, terror.ScopeDownstream)
+	}
+	defer baseDB.Close()
+	dbConn, err := baseDB.GetBaseConn(ctx)
+	if err != nil {
+		return terror.WithScope(err, terror.ScopeDownstream)
+	}
+	defer baseDB.CloseBaseConn(dbConn)
+	ctctx := tcontext.Background().WithContext(ctx).WithLogger(log.With(zap.String("job", "remove metadata")))
+
+	sqls := make([]string, 0, 4)
+	// clear loader and syncer checkpoints
+	sqls = append(sqls, fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`",
+		cfg.MetaSchema, cputil.LoaderCheckpoint(cfg.Name)))
+	sqls = append(sqls, fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`",
+		cfg.MetaSchema, cputil.SyncerCheckpoint(cfg.Name)))
+	sqls = append(sqls, fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`",
+		cfg.MetaSchema, cputil.SyncerShardMeta(cfg.Name)))
+	sqls = append(sqls, fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`",
+		cfg.MetaSchema, cputil.SyncerOnlineDDL(cfg.Name)))
+
+	_, err = dbConn.ExecuteSQL(ctctx, nil, cfg.Name, sqls)
+	return err
+}
+
 func extractWorkerError(result *pb.ProcessResult) error {
 	if result != nil && len(result.Errors) > 0 {
 		return terror.ErrMasterOperRespNotSuccess.Generate(utils.JoinProcessErrors(result.Errors))