test: refine and improve pkg/utils unit test coverage to 82.3% (#1054)

Makefile

@@ -203,11 +203,11 @@ coverage_fix_cover_mode:
 	sed -i "s/mode: count/mode: atomic/g" $(TEST_DIR)/cov.*.dmctl.*.out
 
 coverage: coverage_fix_cover_mode retool_setup
-	retool do gocovmerge "$(TEST_DIR)"/cov.* | grep -vE ".*.pb.go|.*.__failpoint_binding__.go" > "$(TEST_DIR)/all_cov.out"
-	retool do gocovmerge "$(TEST_DIR)"/cov.unit_test*.out | grep -vE ".*.pb.go|.*.__failpoint_binding__.go" > $(TEST_DIR)/unit_test.out
+	retool do gocovmerge "$(TEST_DIR)"/cov.* | grep -vE ".*.pb.go|.*.__failpoint_binding__.go|.*debug-tools.*|.*portal.*|.*tracer.*|.*tracing.*" > "$(TEST_DIR)/all_cov.out"
+	retool do gocovmerge "$(TEST_DIR)"/cov.unit_test*.out | grep -vE ".*.pb.go|.*.__failpoint_binding__.go|.*debug-tools.*|.*portal.*|.*tracer.*|.*tracing.*" > $(TEST_DIR)/unit_test.out
 ifeq ("$(JenkinsCI)", "1")
-	@retool do goveralls -coverprofile=$(TEST_DIR)/all_cov.out -service=jenkins-ci -repotoken $(COVERALLS_TOKEN)
 	@bash <(curl -s https://codecov.io/bash) -f $(TEST_DIR)/unit_test.out -t $(CODECOV_TOKEN)
+	@retool do goveralls -coverprofile=$(TEST_DIR)/all_cov.out -service=jenkins-ci -repotoken $(COVERALLS_TOKEN)
 else
 	go tool cover -html "$(TEST_DIR)/all_cov.out" -o "$(TEST_DIR)/all_cov.html"
 	go tool cover -html "$(TEST_DIR)/unit_test.out" -o "$(TEST_DIR)/unit_test_cov.html"

dm/master/server.go

@@ -44,6 +44,7 @@ import (
 	"github.com/pingcap/dm/dm/master/shardddl"
 	"github.com/pingcap/dm/dm/master/workerrpc"
 	"github.com/pingcap/dm/dm/pb"
+	"github.com/pingcap/dm/dm/unit"
 	"github.com/pingcap/dm/pkg/conn"
 	tcontext "github.com/pingcap/dm/pkg/context"
 	"github.com/pingcap/dm/pkg/cputil"
@@ -1597,7 +1598,7 @@ func (s *Server) removeMetaData(ctx context.Context, cfg *config.TaskConfig) err
 
 func extractWorkerError(result *pb.ProcessResult) error {
 	if result != nil && len(result.Errors) > 0 {
-		return terror.ErrMasterOperRespNotSuccess.Generate(utils.JoinProcessErrors(result.Errors))
+		return terror.ErrMasterOperRespNotSuccess.Generate(unit.JoinProcessErrors(result.Errors))
 	}
 	return nil
 }

dm/unit/unit.go

@@ -93,3 +93,12 @@ func IsCtxCanceledProcessErr(err *pb.ProcessError) bool {
 	}
 	return false
 }
+
+// JoinProcessErrors return the string of pb.ProcessErrors joined by ", "
+func JoinProcessErrors(errors []*pb.ProcessError) string {
+	serrs := make([]string, 0, len(errors))
+	for _, serr := range errors {
+		serrs = append(serrs, serr.String())
+	}
+	return strings.Join(serrs, ", ")
+}

dm/unit/unit_test.go

@@ -18,8 +18,10 @@ import (
 	"testing"
 
 	"github.com/pingcap/check"
-	"github.com/pingcap/dm/pkg/terror"
 	"github.com/pingcap/errors"
+
+	"github.com/pingcap/dm/dm/pb"
+	"github.com/pingcap/dm/pkg/terror"
 )
 
 func TestSuite(t *testing.T) {
@@ -45,3 +47,13 @@ func (t *testUnitSuite) TestIsCtxCanceledProcessErr(c *check.C) {
 	c.Assert(err.GetMessage(), check.Equals, terr.Message())
 	c.Assert(err.GetRawCause(), check.Equals, "")
 }
+
+func (t *testUnitSuite) TestJoinProcessErrors(c *check.C) {
+	errs := []*pb.ProcessError{
+		NewProcessError(terror.ErrDBDriverError.Generate()),
+		NewProcessError(terror.ErrSyncUnitDMLStatementFound.Generate()),
+	}
+
+	c.Assert(JoinProcessErrors(errs), check.Equals,
+		`ErrCode:10001 ErrClass:"database" ErrScope:"not-set" ErrLevel:"high" Message:"database driver error" Workaround:"Please check the database connection and the database config in configuration file." , ErrCode:36014 ErrClass:"sync-unit" ErrScope:"internal" ErrLevel:"high" Message:"only support ROW format binlog, unexpected DML statement found in query event" `)
+}

dm/worker/server.go

@@ -548,7 +548,7 @@ func (s *Server) QueryError(ctx context.Context, req *pb.QueryErrorRequest) (*pb
 		Worker: s.cfg.Name,
 	}
 	if sourceStatus.Result != nil && len(sourceStatus.Result.Errors) > 0 {
-		sourceError.SourceError = utils.JoinProcessErrors(sourceStatus.Result.Errors)
+		sourceError.SourceError = unit.JoinProcessErrors(sourceStatus.Result.Errors)
 	}
 	resp := &pb.QueryErrorResponse{
 		Result:      true,

dm/worker/status.go

@@ -19,7 +19,7 @@ import (
 	"sort"
 
 	"github.com/pingcap/dm/dm/pb"
-	"github.com/pingcap/dm/pkg/utils"
+	"github.com/pingcap/dm/dm/unit"
 
 	"github.com/golang/protobuf/jsonpb"
 	"go.uber.org/zap"
@@ -181,7 +181,7 @@ func (w *Worker) Error(stName string) []*pb.SubTaskError {
 					stError.Error = &pb.SubTaskError_Sync{Sync: us.(*pb.SyncError)}
 				}
 			} else if result := st.Result(); result != nil {
-				processErrorMsg := utils.JoinProcessErrors(result.Errors)
+				processErrorMsg := unit.JoinProcessErrors(result.Errors)
 				if len(processErrorMsg) > 0 {
 					stError.Error = &pb.SubTaskError_Msg{Msg: processErrorMsg}
 				}

dumpling/dumpling.go

@@ -125,7 +125,7 @@ func (m *Dumpling) Process(ctx context.Context, pr chan pb.ProcessResult) {
 		m.logger.Info("dump data finished", zap.Duration("cost time", time.Since(begin)))
 	} else {
 		m.logger.Error("dump data exits with error", zap.Duration("cost time", time.Since(begin)),
-			zap.String("error", utils.JoinProcessErrors(errs)))
+			zap.String("error", unit.JoinProcessErrors(errs)))
 	}
 
 	pr <- pb.ProcessResult{

loader/db.go

@@ -20,7 +20,6 @@ import (
 	"time"
 
 	"github.com/go-sql-driver/mysql"
-	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	tmysql "github.com/pingcap/parser/mysql"
 	"github.com/pingcap/tidb-tools/pkg/dbutil"
@@ -220,19 +219,13 @@ func createConns(tctx *tcontext.Context, cfg *config.SubTaskConfig, workerCount
 }
 
 func isErrDBExists(err error) bool {
-	return isMySQLError(err, tmysql.ErrDBCreateExists)
+	return utils.IsMySQLError(err, tmysql.ErrDBCreateExists)
 }
 
 func isErrTableExists(err error) bool {
-	return isMySQLError(err, tmysql.ErrTableExists)
+	return utils.IsMySQLError(err, tmysql.ErrTableExists)
 }
 
 func isErrDupEntry(err error) bool {
-	return isMySQLError(err, tmysql.ErrDupEntry)
-}
-
-func isMySQLError(err error, code uint16) bool {
-	err = errors.Cause(err)
-	e, ok := err.(*mysql.MySQLError)
-	return ok && e.Number == code
+	return utils.IsMySQLError(err, tmysql.ErrDupEntry)
 }

pkg/retry/errors.go

@@ -16,8 +16,11 @@ package retry
 import (
 	"database/sql/driver"
 
-	"github.com/pingcap/dm/pkg/terror"
 	"github.com/pingcap/errors"
+	tmysql "github.com/pingcap/parser/mysql"
+	gmysql "github.com/siddontang/go-mysql/mysql"
+
+	"github.com/pingcap/dm/pkg/terror"
 )
 
 var (
@@ -61,7 +64,7 @@ var (
 func IsConnectionError(err error) bool {
 	err = errors.Cause(err)
 	switch err {
-	case driver.ErrBadConn:
+	case driver.ErrBadConn, tmysql.ErrBadConn, gmysql.ErrBadConn:
 		return true
 	}
 	return false

pkg/utils/common.go

@@ -54,7 +54,7 @@ func TrimCtrlChars(s string) string {
 
 // FetchAllDoTables returns all need to do tables after filtered (fetches from upstream MySQL)
 func FetchAllDoTables(db *sql.DB, bw *filter.Filter) (map[string][]string, error) {
-	schemas, err := getSchemas(db, maxRetryCount)
+	schemas, err := dbutil.GetSchemas(context.Background(), db)
 
 	failpoint.Inject("FetchAllDoTablesFailed", func(val failpoint.Value) {
 		err = tmysql.NewErr(uint16(val.(int)))
@@ -144,66 +144,6 @@ func FetchTargetDoTables(db *sql.DB, bw *filter.Filter, router *router.Table) (m
 	return mapper, nil
 }
 
-func getSchemas(db *sql.DB, maxRetry int) ([]string, error) {
-	query := "SHOW DATABASES"
-	rows, err := querySQL(db, query, maxRetry)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-
-	// show an example.
-	/*
-		mysql> SHOW DATABASES;
-		+--------------------+
-		| Database           |
-		+--------------------+
-		| information_schema |
-		| mysql              |
-		| performance_schema |
-		| sys                |
-		| test_db            |
-		+--------------------+
-	*/
-	schemas := make([]string, 0, 10)
-	for rows.Next() {
-		var schema string
-		err = rows.Scan(&schema)
-		if err != nil {
-			return nil, terror.DBErrorAdapt(err, terror.ErrDBDriverError)
-		}
-		schemas = append(schemas, schema)
-	}
-	return schemas, terror.DBErrorAdapt(rows.Err(), terror.ErrDBDriverError)
-}
-
-func querySQL(db *sql.DB, query string, maxRetry int) (*sql.Rows, error) {
-	var (
-		err  error
-		rows *sql.Rows
-	)
-
-	for i := 0; i < maxRetry; i++ {
-		if i > 0 {
-			log.L().Warn("query retry", zap.Int("retry number", i), zap.String("query", TruncateString(query, -1)))
-			time.Sleep(retryTimeout)
-		} else {
-			log.L().Debug("query sql", zap.String("query", TruncateString(query, -1)))
-		}
-
-		rows, err = db.Query(query)
-		if err != nil {
-			log.L().Warn("query retry", zap.String("query", TruncateString(query, -1)), log.ShortError(err))
-			continue
-		}
-
-		return rows, nil
-	}
-
-	log.L().Error("query failed", zap.String("query", TruncateString(query, -1)), zap.Error(err))
-	return nil, terror.DBErrorAdapt(err, terror.ErrDBDriverError)
-}
-
 // CompareShardingDDLs compares s and t ddls
 // only concern in content, ignore order of ddl
 func CompareShardingDDLs(s, t []string) bool {