checker(dm): make start-task/resume-task/check-task return pre-check result only warning

dm/.golangci.yml

@@ -81,6 +81,9 @@ linters:
   # - maligned
 
 linters-settings:
+  dupl:
+    # tokens count to trigger issue, 150 by default
+    threshold: 200
   govet:
     # report about shadowed variables
     check-shadowing: true

dm/checker/checker.go

@@ -263,59 +263,59 @@ func (c *Checker) Process(ctx context.Context, pr chan pb.ProcessResult) {
 		errs = append(errs, unit.NewProcessError(err))
 	} else if !result.Summary.Passed {
 		errs = append(errs, unit.NewProcessError(errors.New("check was failed, please see detail")))
-		warnLeft, errLeft := c.warnCnt, c.errCnt
+	}
+	warnLeft, errLeft := c.warnCnt, c.errCnt
 
-		// remove success result if not pass
-		results := result.Results[:0]
-		for _, r := range result.Results {
-			if r.State == checker.StateSuccess {
-				continue
-			}
+	// remove success result if not pass
+	results := result.Results[:0]
+	for _, r := range result.Results {
+		if r.State == checker.StateSuccess {
+			continue
+		}
 
-			// handle results without r.Errors
-			if len(r.Errors) == 0 {
-				switch r.State {
-				case checker.StateWarning:
-					if warnLeft == 0 {
-						continue
-					}
-					warnLeft--
-					results = append(results, r)
-				case checker.StateFailure:
-					if errLeft == 0 {
-						continue
-					}
-					errLeft--
-					results = append(results, r)
+		// handle results without r.Errors
+		if len(r.Errors) == 0 {
+			switch r.State {
+			case checker.StateWarning:
+				if warnLeft == 0 {
+					continue
 				}
-				continue
+				warnLeft--
+				results = append(results, r)
+			case checker.StateFailure:
+				if errLeft == 0 {
+					continue
+				}
+				errLeft--
+				results = append(results, r)
 			}
+			continue
+		}
 
-			subErrors := make([]*checker.Error, 0, len(r.Errors))
-			for _, e := range r.Errors {
-				switch e.Severity {
-				case checker.StateWarning:
-					if warnLeft == 0 {
-						continue
-					}
-					warnLeft--
-					subErrors = append(subErrors, e)
-				case checker.StateFailure:
-					if errLeft == 0 {
-						continue
-					}
-					errLeft--
-					subErrors = append(subErrors, e)
+		subErrors := make([]*checker.Error, 0, len(r.Errors))
+		for _, e := range r.Errors {
+			switch e.Severity {
+			case checker.StateWarning:
+				if warnLeft == 0 {
+					continue
 				}
+				warnLeft--
+				subErrors = append(subErrors, e)
+			case checker.StateFailure:
+				if errLeft == 0 {
+					continue
+				}
+				errLeft--
+				subErrors = append(subErrors, e)
 			}
-			// skip display an empty Result
-			if len(subErrors) > 0 {
-				r.Errors = subErrors
-				results = append(results, r)
-			}
 		}
-		result.Results = results
+		// skip display an empty Result
+		if len(subErrors) > 0 {
+			r.Errors = subErrors
+			results = append(results, r)
+		}
 	}
+	result.Results = results
 
 	c.updateInstruction(result)
 
@@ -325,9 +325,12 @@ func (c *Checker) Process(ctx context.Context, pr chan pb.ProcessResult) {
 	default:
 	}
 
-	rawResult, err := json.MarshalIndent(result, "\t", "\t")
-	if err != nil {
-		rawResult = []byte(fmt.Sprintf("marshal error %v", err))
+	var rawResult []byte
+	if result.Summary.Successful != result.Summary.Total {
+		rawResult, err = json.MarshalIndent(result, "\t", "\t")
+		if err != nil {
+			rawResult = []byte(fmt.Sprintf("marshal error %v", err))
+		}
 	}
 
 	c.result.Lock()

dm/checker/cmd.go

@@ -15,28 +15,31 @@ package checker
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/pingcap/tiflow/dm/dm/config"
 	"github.com/pingcap/tiflow/dm/dm/pb"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
 
 var (
-	// ErrorMsgHeader used as the header of the error message when checking config failed.
-	ErrorMsgHeader = "fail to check synchronization configuration with type"
+	// CheckTaskMsgHeader used as the header of the error/warning message when checking config failed.
+	CheckTaskMsgHeader = "fail to check synchronization configuration with type"
+
+	CheckTaskSuccess = "check pass!!!"
 
 	// CheckSyncConfigFunc holds the CheckSyncConfig function.
-	CheckSyncConfigFunc func(ctx context.Context, cfgs []*config.SubTaskConfig, errCnt, warnCnt int64) error
+	CheckSyncConfigFunc func(ctx context.Context, cfgs []*config.SubTaskConfig, errCnt, warnCnt int64) (string, error)
 )
 
 func init() {
 	CheckSyncConfigFunc = CheckSyncConfig
 }
 
 // CheckSyncConfig checks synchronization configuration.
-func CheckSyncConfig(ctx context.Context, cfgs []*config.SubTaskConfig, errCnt, warnCnt int64) error {
+func CheckSyncConfig(ctx context.Context, cfgs []*config.SubTaskConfig, errCnt, warnCnt int64) (string, error) {
 	if len(cfgs) == 0 {
-		return nil
+		return "", nil
 	}
 
 	// all `IgnoreCheckingItems` and `Mode` of sub-task are same, so we take first one
@@ -53,25 +56,29 @@ func CheckSyncConfig(ctx context.Context, cfgs []*config.SubTaskConfig, errCnt,
 	}
 	checkingItems := config.FilterCheckingItems(ignoreCheckingItems)
 	if len(checkingItems) == 0 {
-		return nil
+		return "", nil
 	}
 
 	c := NewChecker(cfgs, checkingItems, errCnt, warnCnt)
 
 	if err := c.Init(ctx); err != nil {
-		return terror.Annotate(err, "fail to initial checker")
+		return "", terror.Annotate(err, "fail to initial checker")
 	}
 	defer c.Close()
 
 	pr := make(chan pb.ProcessResult, 1)
 	c.Process(ctx, pr)
-	for len(pr) > 0 {
+	if len(pr) > 0 {
 		r := <-pr
 		// we only want first error
 		if len(r.Errors) > 0 {
-			return terror.ErrTaskCheckSyncConfigError.Generate(ErrorMsgHeader, r.Errors[0].Message, string(r.Detail))
+			return "", terror.ErrTaskCheckSyncConfigError.Generate(CheckTaskMsgHeader, r.Errors[0].Message, string(r.Detail))
+		}
+		if len(r.Detail) == 0 {
+			return CheckTaskSuccess, nil
 		}
+		return fmt.Sprintf("%s: no errors but some warnings\n detail: %v", CheckTaskMsgHeader, string(r.Detail)), nil
 	}
 
-	return nil
+	return "", nil
 }

dm/dm/ctl/master/check_task.go

@@ -78,7 +78,7 @@ func checkTaskFunc(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 
-	if !common.PrettyPrintResponseWithCheckTask(resp, checker.ErrorMsgHeader) {
+	if !common.PrettyPrintResponseWithCheckTask(resp, checker.CheckTaskMsgHeader) {
 		common.PrettyPrintResponse(resp)
 	}
 	return nil

dm/dm/ctl/master/start_task.go

@@ -97,7 +97,7 @@ func startTaskFunc(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 
-	if !common.PrettyPrintResponseWithCheckTask(resp, checker.ErrorMsgHeader) {
+	if !common.PrettyPrintResponseWithCheckTask(resp, checker.CheckTaskMsgHeader) {
 		common.PrettyPrintResponse(resp)
 	}
 	return nil

dm/dm/master/openapi.go

@@ -445,7 +445,7 @@ func (s *Server) DMAPIStartTask(c *gin.Context) {
 	for i := range subTaskConfigList {
 		subTaskConfigPList[i] = &subTaskConfigList[i]
 	}
-	if err = checker.CheckSyncConfigFunc(newCtx, subTaskConfigPList,
+	if _, err = checker.CheckSyncConfigFunc(newCtx, subTaskConfigPList,
 		common.DefaultErrorCnt, common.DefaultWarnCnt); err != nil {
 		_ = c.Error(terror.WithClass(err, terror.ClassDMMaster))
 		return

dm/dm/master/openapi_test.go

@@ -849,6 +849,6 @@ func mockTaskQueryStatus(
 	).Return(queryResp, nil).MaxTimes(maxRetryNum)
 }
 
-func mockCheckSyncConfig(ctx context.Context, cfgs []*config.SubTaskConfig, errCnt, warnCnt int64) error {
-	return nil
+func mockCheckSyncConfig(ctx context.Context, cfgs []*config.SubTaskConfig, errCnt, warnCnt int64) (string, error) {
+	return "", nil
 }

dm/dm/master/server.go

@@ -427,12 +427,19 @@ func (s *Server) StartTask(ctx context.Context, req *pb.StartTaskRequest) (*pb.S
 	}
 
 	resp := &pb.StartTaskResponse{}
-	cfg, stCfgs, err := s.generateSubTask(ctx, req.Task, ctlcommon.DefaultErrorCnt, ctlcommon.DefaultWarnCnt)
+	cfg, stCfgs, err := s.generateSubTask(ctx, req.Task)
 	if err != nil {
 		resp.Msg = err.Error()
 		// nolint:nilerr
 		return resp, nil
 	}
+	msg, err := checker.CheckSyncConfigFunc(ctx, stCfgs, ctlcommon.DefaultErrorCnt, ctlcommon.DefaultWarnCnt)
+	if err != nil {
+		resp.Msg = terror.WithClass(err, terror.ClassDMMaster).Error()
+		return resp, nil
+	}
+	resp.Msg = msg
+
 	log.L().Info("", zap.String("task name", cfg.Name), zap.String("task", cfg.JSON()), zap.String("request", "StartTask"))
 
 	sourceRespCh := make(chan *pb.CommonWorkerResponse, len(stCfgs))
@@ -473,27 +480,27 @@ func (s *Server) StartTask(ctx context.Context, req *pb.StartTaskRequest) (*pb.S
 			// use same latch for remove-meta and start-task
 			release, err3 = s.scheduler.AcquireSubtaskLatch(cfg.Name)
 			if err3 != nil {
-				resp.Msg = terror.ErrSchedulerLatchInUse.Generate("RemoveMeta", cfg.Name).Error()
+				resp.Msg += terror.ErrSchedulerLatchInUse.Generate("RemoveMeta", cfg.Name).Error()
 				// nolint:nilerr
 				return resp, nil
 			}
 			defer release()
 			latched = true
 
 			if scm := s.scheduler.GetSubTaskCfgsByTask(cfg.Name); len(scm) > 0 {
-				resp.Msg = terror.Annotate(terror.ErrSchedulerSubTaskExist.Generate(cfg.Name, sources),
+				resp.Msg += terror.Annotate(terror.ErrSchedulerSubTaskExist.Generate(cfg.Name, sources),
 					"while remove-meta is true").Error()
 				return resp, nil
 			}
 			err = s.removeMetaData(ctx, cfg.Name, cfg.MetaSchema, cfg.TargetDB)
 			if err != nil {
-				resp.Msg = terror.Annotate(err, "while removing metadata").Error()
+				resp.Msg += terror.Annotate(err, "while removing metadata").Error()
 				return resp, nil
 			}
 		}
 		err = s.scheduler.AddSubTasks(latched, subtaskCfgPointersToInstances(stCfgs...)...)
 		if err != nil {
-			resp.Msg = err.Error()
+			resp.Msg += err.Error()
 			// nolint:nilerr
 			return resp, nil
 		}
@@ -504,7 +511,7 @@ func (s *Server) StartTask(ctx context.Context, req *pb.StartTaskRequest) (*pb.S
 
 		resp.Result = true
 		if cfg.RemoveMeta {
-			resp.Msg = "`remove-meta` in task config is deprecated, please use `start-task ... --remove-meta` instead"
+			resp.Msg += "`remove-meta` in task config is deprecated, please use `start-task ... --remove-meta` instead"
 		}
 		sourceResps = s.getSourceRespsAfterOperation(ctx, cfg.Name, sources, []string{}, req)
 	}
@@ -617,14 +624,20 @@ func (s *Server) UpdateTask(ctx context.Context, req *pb.UpdateTaskRequest) (*pb
 		return resp2, err2
 	}
 
-	cfg, stCfgs, err := s.generateSubTask(ctx, req.Task, ctlcommon.DefaultErrorCnt, ctlcommon.DefaultWarnCnt)
+	resp := &pb.UpdateTaskResponse{}
+	cfg, stCfgs, err := s.generateSubTask(ctx, req.Task)
 	if err != nil {
+		resp.Msg = err.Error()
 		// nolint:nilerr
-		return &pb.UpdateTaskResponse{
-			Result: false,
-			Msg:    err.Error(),
-		}, nil
+		return resp, nil
+	}
+
+	msg, err := checker.CheckSyncConfigFunc(ctx, stCfgs, ctlcommon.DefaultErrorCnt, ctlcommon.DefaultWarnCnt)
+	if err != nil {
+		resp.Msg = terror.WithClass(err, terror.ClassDMMaster).Error()
+		return resp, nil
 	}
+	resp.Msg = msg
 	log.L().Info("update task", zap.String("task name", cfg.Name), zap.Stringer("task", cfg))
 
 	workerRespCh := make(chan *pb.CommonWorkerResponse, len(stCfgs)+len(req.Sources))
@@ -663,10 +676,9 @@ func (s *Server) UpdateTask(ctx context.Context, req *pb.UpdateTaskRequest) (*pb
 		workerResps = append(workerResps, workerRespMap[worker])
 	}
 
-	return &pb.UpdateTaskResponse{
-		Result:  true,
-		Sources: workerResps,
-	}, nil
+	resp.Result = true
+	resp.Sources = workerResps
+	return resp, nil
 }
 
 type hasWokers interface {
@@ -1175,19 +1187,23 @@ func (s *Server) CheckTask(ctx context.Context, req *pb.CheckTaskRequest) (*pb.C
 		return resp2, err2
 	}
 
-	_, _, err := s.generateSubTask(ctx, req.Task, req.ErrCnt, req.WarnCnt)
+	resp := &pb.CheckTaskResponse{}
+	_, stCfgs, err := s.generateSubTask(ctx, req.Task)
 	if err != nil {
+		resp.Msg = err.Error()
 		// nolint:nilerr
-		return &pb.CheckTaskResponse{
-			Result: false,
-			Msg:    err.Error(),
-		}, nil
+		return resp, nil
 	}
 
-	return &pb.CheckTaskResponse{
-		Result: true,
-		Msg:    "check pass!!!",
-	}, nil
+	msg, err := checker.CheckSyncConfigFunc(ctx, stCfgs, req.ErrCnt, req.WarnCnt)
+	if err != nil {
+		resp.Msg = terror.WithClass(err, terror.ClassDMMaster).Error()
+		return resp, nil
+	}
+	resp.Msg = msg
+	resp.Result = true
+
+	return resp, nil
 }
 
 func parseAndAdjustSourceConfig(ctx context.Context, contents []string) ([]*config.SourceConfig, error) {
@@ -1434,7 +1450,7 @@ func (s *Server) OperateLeader(ctx context.Context, req *pb.OperateLeaderRequest
 	}, nil
 }
 
-func (s *Server) generateSubTask(ctx context.Context, task string, errCnt, warnCnt int64) (*config.TaskConfig, []*config.SubTaskConfig, error) {
+func (s *Server) generateSubTask(ctx context.Context, task string) (*config.TaskConfig, []*config.SubTaskConfig, error) {
 	cfg := config.NewTaskConfig()
 	err := cfg.Decode(task)
 	if err != nil {
@@ -1453,11 +1469,6 @@ func (s *Server) generateSubTask(ctx context.Context, task string, errCnt, warnC
 		return nil, nil, terror.WithClass(err, terror.ClassDMMaster)
 	}
 
-	err = checker.CheckSyncConfigFunc(ctx, stCfgs, errCnt, warnCnt)
-	if err != nil {
-		return nil, nil, terror.WithClass(err, terror.ClassDMMaster)
-	}
-
 	return cfg, stCfgs, nil
 }