dm-ha/: add remove metadata feature

dm/config/task.go

@@ -510,7 +510,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/master/metrics.go

@@ -0,0 +1,39 @@
+// Copyright 2020 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 (
+	"github.com/pingcap/dm/pkg/metricsproxy"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	stmtHistogram = metricsproxy.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "dm",
+			Subsystem: "master",
+			Name:      "stmt_duration_time",
+			Help:      "Bucketed histogram of every statement query time (s).",
+			Buckets:   prometheus.ExponentialBuckets(0.000005, 2, 25),
+		}, []string{"type", "task"})
+)
+
+// RegisterMetrics registers metrics
+func RegisterMetrics() {
+	registry := prometheus.NewRegistry()
+	registry.MustRegister(prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}))
+	registry.MustRegister(prometheus.NewGoCollector())
+
+	registry.MustRegister(stmtHistogram)
+}

dm/master/server.go

@@ -38,12 +38,14 @@ 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/election"
 	"github.com/pingcap/dm/pkg/etcdutil"
 	"github.com/pingcap/dm/pkg/log"
 	"github.com/pingcap/dm/pkg/terror"
 	"github.com/pingcap/dm/pkg/tracing"
 	"github.com/pingcap/dm/pkg/utils"
+	shardmeta "github.com/pingcap/dm/syncer/sharding-meta"
 )
 
 const (
@@ -165,6 +167,9 @@ func (s *Server) Start(ctx context.Context) (err error) {
 	// gRPC API server
 	gRPCSvr := func(gs *grpc.Server) { pb.RegisterMasterServer(gs, s) }
 
+	// register metrics before serving
+	RegisterMetrics()
+
 	// start embed etcd server, gRPC API server and HTTP (API, status and debug) server.
 	s.etcd, err = startEtcd(etcdCfg, gRPCSvr, userHandles, etcdStartTimeout)
 	if err != nil {
@@ -360,15 +365,27 @@ 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)
+		}
+		if cfg.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()
+				return resp, nil
+			}
+			err = s.removeMetaData(ctx, cfg)
+			if err != nil {
+				resp.Msg = terror.Annotate(err, "while removing metadata").Error()
+				return resp, nil
+			}
+		}
 		err = s.scheduler.AddSubTasks(subtaskCfgPointersToInstances(stCfgs...)...)
 		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 +1349,62 @@ 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
+	}
+
+	sqls := make([]string, 0, 3)
+	sqls = append(sqls, fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS `%s`", cfg.MetaSchema))
+	// clear loader and syncer checkpoints
+	sqls = append(sqls, fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`",
+		cfg.MetaSchema, cfg.Name+"_loader_checkpoint"))
+	sqls = append(sqls, fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`",
+		cfg.MetaSchema, cfg.Name+"_syncer_checkpoint"))
+
+	// clear etcd data
+	if cfg.ShardMode == config.ShardPessimistic {
+		err := s.pessimist.RemoveMetaData(cfg.Name)
+		if err != nil {
+			return err
+		}
+		// clear shard meta data for pessimistic
+		sqls = append(sqls, fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`",
+			cfg.MetaSchema, fmt.Sprintf(shardmeta.MetaTableFormat, cfg.Name)))
+	} else if cfg.ShardMode == config.ShardOptimistic {
+		err := s.optimist.RemoveMetaData(cfg.Name)
+		if err != nil {
+			return err
+		}
+	}
+
+	// clear online ddl schema
+	if cfg.OnlineDDLScheme != "" {
+		sqls = append(sqls, fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`",
+			cfg.MetaSchema, fmt.Sprintf("%s_onlineddl", cfg.Name)))
+	}
+
+	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("unit", "remove metadata")))
+	_, err = dbConn.ExecuteSQL(ctctx, stmtHistogram, 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))

dm/master/shardddl/optimist.go

@@ -154,6 +154,31 @@ func (o *Optimist) ShowLocks(task string, sources []string) []*pb.DDLLock {
 	return ret
 }
 
+// RemoveMetaData removes meta data for a specified task
+// NOTE: this function can only be used when the specified task is not running
+func (o *Optimist) RemoveMetaData(task string) error {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	if o.closed {
+		return nil
+	}
+
+	infos, ops, _, err := optimism.GetInfosOperationsThroughTask(o.cli, task)
+	if err != nil {
+		return err
+	}
+	for _, info := range infos {
+		o.lk.RemoveLockThroughInfo(info)
+	}
+	for _, op := range ops {
+		o.lk.RemoveLock(op.ID)
+	}
+
+	o.tk.RemoveTableThroughTask(task)
+	_, err = optimism.DeleteInfosOperationsTablesThroughTask(o.cli, task)
+	return err
+}
+
 // run runs jobs in the background.
 func (o *Optimist) run(ctx context.Context, revSource, revInfo, revOperation int64) error {
 	for {

dm/master/shardddl/pessimist.go

@@ -369,6 +369,27 @@ func (p *Pessimist) UnlockLock(ctx context.Context, ID, replaceOwner string, for
 	return nil
 }
 
+// RemoveMetaData removes meta data for a specified task
+// NOTE: this function can only be used when the specified task is not running
+func (p *Pessimist) RemoveMetaData(task string) error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.closed {
+		return terror.ErrMasterPessimistNotStarted.Generate()
+	}
+
+	infos, ops, _, err := pessimism.GetInfosOperationsThroughTask(p.cli, task)
+	for _, info := range infos {
+		p.lk.RemoveLockThroughInfo(info)
+	}
+	for _, op := range ops {
+		p.lk.RemoveLock(op.ID)
+	}
+
+	_, err = pessimism.DeleteInfosOperationsThroughTask(p.cli, task)
+	return err
+}
+
 // recoverLocks recovers shard DDL locks based on shard DDL info and shard DDL lock operation.
 func (p *Pessimist) recoverLocks(ifm map[string]map[string]pessimism.Info, opm map[string]map[string]pessimism.Operation) error {
 	// construct locks based on the shard DDL info.

dm/worker/join.go

@@ -56,6 +56,7 @@ func (s *Server) JoinMaster(endpoints []string) error {
 		ctx1, cancel1 = context.WithTimeout(ctx, 3*time.Second)
 		resp, err := client.RegisterWorker(ctx1, req)
 		cancel1()
+		conn.Close()
 		if err != nil {
 			log.L().Error("fail to register worker", zap.Error(err))
 			continue

pkg/shardddl/optimism/keeper.go

@@ -65,6 +65,12 @@ func (lk *LockKeeper) RemoveLock(lockID string) bool {
 	return ok
 }
 
+// RemoveLockThroughInfo removes a lock.
+func (lk *LockKeeper) RemoveLockThroughInfo(info Info) bool {
+	lockID := genDDLLockID(info)
+	return lk.RemoveLock(lockID)
+}
+
 // FindLock finds a lock.
 func (lk *LockKeeper) FindLock(lockID string) *Lock {
 	lk.mu.RLock()
@@ -193,6 +199,19 @@ func (tk *TableKeeper) RemoveTable(task, source, upSchema, upTable, downSchema,
 	return removed
 }
 
+// RemoveTableThroughTask removes tables from the source tables through task name.
+// it returns whether removed (exit before).
+func (tk *TableKeeper) RemoveTableThroughTask(task string) bool {
+	tk.mu.Lock()
+	defer tk.mu.Unlock()
+
+	if _, ok := tk.tables[task]; !ok {
+		return false
+	}
+	delete(tk.tables, task)
+	return true
+}
+
 // FindTables finds source tables by task name and downstream table name.
 func (tk *TableKeeper) FindTables(task, downSchema, downTable string) []TargetTable {
 	tk.mu.RLock()

pkg/shardddl/optimism/operation.go

@@ -168,6 +168,39 @@ func GetAllOperations(cli *clientv3.Client) (map[string]map[string]map[string]ma
 	return opm, resp.Header.Revision, nil
 }
 
+// GetInfosOperationsThroughTask gets all shard DDL info and operation in etcd currently.
+// This function should often be called by DM-master.
+func GetInfosOperationsThroughTask(cli *clientv3.Client, task string) ([]Info, []Operation, int64, error) {
+	respTxn, _, err := etcdutil.DoOpsInOneTxnWithRetry(cli,
+		clientv3.OpGet(common.ShardDDLOptimismInfoKeyAdapter.Encode(task), clientv3.WithPrefix()),
+		clientv3.OpGet(common.ShardDDLOptimismOperationKeyAdapter.Encode(task), clientv3.WithPrefix()))
+
+	if err != nil {
+		return nil, nil, 0, err
+	}
+	infoResp := respTxn.Responses[0].GetResponseRange()
+	opsResp := respTxn.Responses[1].GetResponseRange()
+	var (
+		infos = make([]Info, 0, len(infoResp.Kvs))
+		ops   = make([]Operation, 0, len(opsResp.Kvs))
+	)
+	for _, kv := range infoResp.Kvs {
+		info, err2 := infoFromJSON(string(kv.Value))
+		if err2 != nil {
+			return nil, nil, 0, err2
+		}
+		infos = append(infos, info)
+	}
+	for _, kv := range opsResp.Kvs {
+		op, err2 := operationFromJSON(string(kv.Value))
+		if err2 != nil {
+			return nil, nil, 0, err2
+		}
+		ops = append(ops, op)
+	}
+	return infos, ops, respTxn.Header.Revision, nil
+}
+
 // WatchOperationPut watches PUT operations for DDL lock operation.
 // If want to watch all operations matching, pass empty string for `task`, `source`, `upSchema` and `upTable`.
 // This function can be called by DM-worker and DM-master.

pkg/shardddl/optimism/ops.go

@@ -16,6 +16,7 @@ package optimism
 import (
 	"go.etcd.io/etcd/clientv3"
 
+	"github.com/pingcap/dm/dm/common"
 	"github.com/pingcap/dm/pkg/etcdutil"
 )
 
@@ -59,3 +60,13 @@ func DeleteInfosOperations(cli *clientv3.Client, infos []Info, ops []Operation)
 	_, rev, err := etcdutil.DoOpsInOneTxnWithRetry(cli, opsDel...)
 	return rev, err
 }
+
+// DeleteInfosOperationsTablesThroughTask deletes the shard DDL infos and operations in etcd.
+func DeleteInfosOperationsTablesThroughTask(cli *clientv3.Client, task string) (int64, error) {
+	opsDel := make([]clientv3.Op, 0, 3)
+	opsDel = append(opsDel, clientv3.OpDelete(common.ShardDDLOptimismInfoKeyAdapter.Encode(task), clientv3.WithPrefix()))
+	opsDel = append(opsDel, clientv3.OpDelete(common.ShardDDLOptimismOperationKeyAdapter.Encode(task), clientv3.WithPrefix()))
+	opsDel = append(opsDel, clientv3.OpDelete(common.ShardDDLOptimismSourceTablesKeyAdapter.Encode(task), clientv3.WithPrefix()))
+	_, rev, err := etcdutil.DoOpsInOneTxnWithRetry(cli, opsDel...)
+	return rev, err
+}

pkg/shardddl/pessimism/keeper.go

@@ -64,6 +64,12 @@ func (lk *LockKeeper) RemoveLock(lockID string) bool {
 	return ok
 }
 
+// RemoveLockThroughInfo removes a lock through given info.
+func (lk *LockKeeper) RemoveLockThroughInfo(info Info) bool {
+	lockID := genDDLLockID(info)
+	return lk.RemoveLock(lockID)
+}
+
 // FindLock finds a lock.
 func (lk *LockKeeper) FindLock(lockID string) *Lock {
 	lk.mu.RLock()

pkg/shardddl/pessimism/operation.go

@@ -165,6 +165,38 @@ func GetAllOperations(cli *clientv3.Client) (map[string]map[string]Operation, in
 	return opm, resp.Header.Revision, nil
 }
 
+// GetInfosOperationsThroughTask gets all DDL lock infos and operations in etcd currently.
+func GetInfosOperationsThroughTask(cli *clientv3.Client, task string) ([]Info, []Operation, int64, error) {
+	respTxn, _, err := etcdutil.DoOpsInOneTxnWithRetry(cli,
+		clientv3.OpGet(common.ShardDDLPessimismInfoKeyAdapter.Encode(task), clientv3.WithPrefix()),
+		clientv3.OpGet(common.ShardDDLPessimismOperationKeyAdapter.Encode(task), clientv3.WithPrefix()))
+
+	if err != nil {
+		return nil, nil, 0, err
+	}
+	infoResp := respTxn.Responses[0].GetResponseRange()
+	opsResp := respTxn.Responses[1].GetResponseRange()
+	var (
+		infos = make([]Info, 0, len(infoResp.Kvs))
+		ops   = make([]Operation, 0, len(opsResp.Kvs))
+	)
+	for _, kv := range infoResp.Kvs {
+		info, err2 := infoFromJSON(string(kv.Value))
+		if err2 != nil {
+			return nil, nil, 0, err2
+		}
+		infos = append(infos, info)
+	}
+	for _, kv := range opsResp.Kvs {
+		op, err2 := operationFromJSON(string(kv.Value))
+		if err2 != nil {
+			return nil, nil, 0, err2
+		}
+		ops = append(ops, op)
+	}
+	return infos, ops, respTxn.Header.Revision, nil
+}
+
 // WatchOperationPut watches PUT operations for DDL lock operation.
 // If want to watch all operations, pass empty string for `task` and `source`.
 // This function can be called by DM-worker and DM-master.