feat: add handle-error command

_utils/terror_gen/errors_release.txt

@@ -120,6 +120,7 @@ ErrConnRegistryTLSConfig,[code=11113:class=functional:scope=internal:level=mediu
 ErrUpgradeVersionEtcdFail,[code=11114:class=functional:scope=internal:level=high], "Message: fail to operate DM cluster version in etcd, Workaround: Please use `list-member --master` to confirm whether the DM-master cluster is healthy"
 ErrInvalidV1WorkerMetaPath,[code=11115:class=functional:scope=internal:level=medium], "Message: %s is an invalid v1.0.x DM-worker meta path, Workaround: Please check no `meta-dir` set for v1.0.x DM-worker."
 ErrFailUpdateV1DBSchema,[code=11116:class=functional:scope=internal:level=medium], "Message: fail to upgrade v1.0.x DB schema, Workaround: Please confirm that you have not violated any restrictions in the upgrade documentation."
+ErrVerifyHandleErrorArgs,[code=11117:class=functional:scope=internal:level=low], "Workaround: Please make sure the args are correct."
 ErrConfigCheckItemNotSupport,[code=20001:class=config:scope=internal:level=medium], "Message: checking item %s is not supported\n%s, Workaround: Please check `ignore-checking-items` config in task configuration file, which can be set including `all`/`dump_privilege`/`replication_privilege`/`version`/`binlog_enable`/`binlog_format`/`binlog_row_image`/`table_schema`/`schema_of_shard_tables`/`auto_increment_ID`."
 ErrConfigTomlTransform,[code=20002:class=config:scope=internal:level=medium], "Message: %s, Workaround: Please check the configuration file has correct TOML format."
 ErrConfigYamlTransform,[code=20003:class=config:scope=internal:level=medium], "Message: %s, Workaround: Please check the configuration file has correct YAML format."
@@ -302,6 +303,9 @@ ErrSyncerUnitGenBAList,[code=36060:class=sync-unit:scope=internal:level=high], "
 ErrSyncerUnitHandleDDLFailed,[code=36061:class=sync-unit:scope=internal:level=high], "Message: fail to handle ddl job for %s"
 ErrSyncerShardDDLConflict,[code=36062:class=sync-unit:scope=internal:level=high], "Message: fail to handle shard ddl %v in optimistic mode, because schema conflict detected, Workaround: Please use show-ddl-locks command for more details."
 ErrSyncerFailpoint,[code=36063:class=sync-unit:scope=internal:level=low], "Message: failpoint specified error"
+ErrSyncerReplaceEvent,[code=36064:class=sync-unit:scope=internal:level=high]
+ErrSyncerOperatorNotExist,[code=36065:class=sync-unit:scope=internal:level=low], "Message: error operator not exist, position: %s"
+ErrSyncerReplaceEventNotExist,[code=36066:class=sync-unit:scope=internal:level=high], "Message: replace event not exist, location: %s"
 ErrMasterSQLOpNilRequest,[code=38001:class=dm-master:scope=internal:level=medium], "Message: nil request not valid"
 ErrMasterSQLOpNotSupport,[code=38002:class=dm-master:scope=internal:level=medium], "Message: op %s not supported"
 ErrMasterSQLOpWithoutSharding,[code=38003:class=dm-master:scope=internal:level=medium], "Message: operate request without --sharding specified not valid"

dm/command/argument.go

@@ -14,9 +14,6 @@
 package command
 
 import (
-	"regexp"
-	"strings"
-
 	"github.com/siddontang/go-mysql/mysql"
 
 	"github.com/pingcap/dm/pkg/binlog"
@@ -32,47 +29,12 @@ func TrimQuoteMark(s string) string {
 	return s
 }
 
-// VerifySQLOperateArgs verify args for sql operation, including sql-skip, sql-replace.
-// NOTE: only part of args verified in here and others need to be verified from outer.
-func VerifySQLOperateArgs(binlogPosStr, sqlPattern string, sharding bool) (*mysql.Position, *regexp.Regexp, error) {
-	binlogPosStr = TrimQuoteMark(binlogPosStr)
-	sqlPattern = TrimQuoteMark(sqlPattern)
-
-	var (
-		pos *mysql.Position
-		reg *regexp.Regexp
-	)
-
-	if len(binlogPosStr) > 0 && len(sqlPattern) > 0 {
-		return nil, nil, terror.ErrVerifySQLOperateArgs.New("cannot specify both --binlog-pos and --sql-pattern in sql operation")
+// VerifyBinlogPos verify binlog pos string
+func VerifyBinlogPos(pos string) (*mysql.Position, error) {
+	binlogPosStr := TrimQuoteMark(pos)
+	pos2, err := binlog.PositionFromStr(binlogPosStr)
+	if err != nil {
+		return nil, terror.ErrVerifyHandleErrorArgs.Generatef("invalid --binlog-pos %s in handle-error operation: %s", binlogPosStr, terror.Message(err))
 	}
-
-	if len(binlogPosStr) > 0 {
-		pos2, err := binlog.PositionFromStr(binlogPosStr)
-		if err != nil {
-			return nil, nil, terror.ErrVerifySQLOperateArgs.Generatef("invalid --binlog-pos %s in sql operation: %s", binlogPosStr, terror.Message(err))
-		}
-		pos = &pos2
-	} else if len(sqlPattern) > 0 {
-		var pattern string
-		if strings.HasPrefix(sqlPattern, "~") {
-			pattern = sqlPattern[1:]
-		} else {
-			pattern = "^" + regexp.QuoteMeta(sqlPattern) + "$"
-		}
-
-		var err error
-		reg, err = regexp.Compile(pattern)
-		if err != nil {
-			return nil, nil, terror.ErrVerifySQLOperateArgs.AnnotateDelegate(err, "invalid --sql-pattern %s in sql operation", sqlPattern)
-		}
-	} else {
-		return nil, nil, terror.ErrVerifySQLOperateArgs.New("must specify one of --binlog-pos and --sql-pattern in sql operation")
-	}
-
-	if sharding && len(binlogPosStr) > 0 {
-		return nil, nil, terror.ErrVerifySQLOperateArgs.New("cannot specify --binlog-pos with --sharding in sql operation")
-	}
-
-	return pos, reg, nil
+	return &pos2, nil
 }

dm/ctl/common/util.go

@@ -20,6 +20,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/pingcap/dm/dm/command"
 	"github.com/pingcap/dm/dm/config"
 	"github.com/pingcap/dm/dm/pb"
 	parserpkg "github.com/pingcap/dm/pkg/parser"
@@ -189,6 +190,7 @@ func ExtractSQLsFromArgs(args []string) ([]string, error) {
 	}
 
 	concat := strings.TrimSpace(strings.Join(args, " "))
+	concat = command.TrimQuoteMark(concat)
 
 	parser := parser.New()
 	nodes, err := parserpkg.Parse(parser, concat, "", "")

dm/ctl/ctl.go

@@ -63,9 +63,6 @@ func NewRootCmd() *cobra.Command {
 		master.NewUpdateTaskCmd(),
 		master.NewQueryStatusCmd(),
 		master.NewQueryErrorCmd(),
-		master.NewSQLReplaceCmd(),
-		master.NewSQLSkipCmd(),
-		master.NewSQLInjectCmd(),
 		master.NewShowDDLLocksCmd(),
 		master.NewUnlockDDLLockCmd(),
 		master.NewSwitchRelayMasterCmd(),
@@ -81,6 +78,7 @@ func NewRootCmd() *cobra.Command {
 		master.NewListMemberCmd(),
 		master.NewOperateSchemaCmd(),
 		master.NewGetTaskCfgCmd(),
+		master.NewHandleErrorCmd(),
 	)
 	return cmd
 }

dm/ctl/master/handle_error.go

@@ -0,0 +1,123 @@
+// 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 (
+	"context"
+	"errors"
+	"os"
+
+	"github.com/spf13/cobra"
+
+	"github.com/pingcap/dm/dm/command"
+	"github.com/pingcap/dm/dm/ctl/common"
+	"github.com/pingcap/dm/dm/pb"
+)
+
+// NewHandleErrorCmd creates a HandleError command
+func NewHandleErrorCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "handle-error <task-name | task-file> [-s source ...] [-b binlog-pos] <skip/replace/revert> [replace-sql1;replace-sql2;]",
+		Short: "skip/replace/revert the current error event or a specific binlog position (binlog-pos) event",
+		RunE:  handleErrorFunc,
+	}
+	cmd.Flags().StringP("binlog-pos", "b", "", "position used to match binlog event if matched the handler-error operation will be applied. The format like \"mysql-bin|000001.000003:3270\"")
+	return cmd
+}
+
+func convertOp(t string) pb.ErrorOp {
+	switch t {
+	case "skip":
+		return pb.ErrorOp_Skip
+	case "replace":
+		return pb.ErrorOp_Replace
+	case "revert":
+		return pb.ErrorOp_Revert
+	default:
+		return pb.ErrorOp_InvalidErrorOp
+	}
+}
+
+// handleErrorFunc does handle error request
+func handleErrorFunc(cmd *cobra.Command, _ []string) (err error) {
+	if len(cmd.Flags().Args()) < 2 {
+		cmd.SetOut(os.Stdout)
+		cmd.Usage()
+		err = errors.New("please check output to see error")
+		return
+	}
+
+	taskName := common.GetTaskNameFromArgOrFile(cmd.Flags().Arg(0))
+	operation := cmd.Flags().Arg(1)
+	var sqls []string
+
+	op := convertOp(operation)
+	switch op {
+	case pb.ErrorOp_Skip, pb.ErrorOp_Revert:
+		if len(cmd.Flags().Args()) > 2 {
+			common.PrintLines("replace-sqls can not be used for 'skip/revert' operation")
+			err = errors.New("please check output to see error")
+			return
+		}
+	case pb.ErrorOp_Replace:
+		if len(cmd.Flags().Args()) <= 2 {
+			common.PrintLines("must specify the replace-sqls for replace operation")
+			err = errors.New("please check output to see error")
+			return
+		}
+
+		sqls, err = common.ExtractSQLsFromArgs(cmd.Flags().Args()[2:])
+		if err != nil {
+			return
+		}
+	default:
+		common.PrintLines("invalid operation '%s', please use `skip`, `replace` or `revert`", operation)
+		err = errors.New("please check output to see error")
+		return
+	}
+
+	binlogPos, err := cmd.Flags().GetString("binlog-pos")
+	if err != nil {
+		return
+	}
+	if len(binlogPos) != 0 {
+		_, err = command.VerifyBinlogPos(binlogPos)
+		if err != nil {
+			return
+		}
+	}
+
+	sources, err := common.GetSourceArgs(cmd)
+	if err != nil {
+		return
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	cli := common.MasterClient()
+
+	resp, err := cli.HandleError(ctx, &pb.HandleErrorRequest{
+		Op:        op,
+		Task:      taskName,
+		BinlogPos: binlogPos,
+		Sqls:      sqls,
+		Sources:   sources,
+	})
+	if err != nil {
+		return
+	}
+
+	common.PrettyPrintResponse(resp)
+	return
+}