Repository: pingcap/tiflow
Pull request #6053 (2022-06-27 07:58:46)

*(dm): move some SQL utils under BaseDB/BaseConn
<!--
Thank you for contributing to TiFlow! 
Please read MD's [CONTRIBUTING](https://github.com/pingcap/tiflow/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve?
<!--
Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.
 -->

Issue Number: ref #4287 

### What is changed and how it works?

BaseDB/BaseConn can retry on errors, this PR let some SQL utils use BaseDB/BaseConn

### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - pass existing test

#### Questions <!-- Authors should answer these questions and reviewers should consider these questions. -->

##### Will it cause performance regression or break compatibility?

##### Do you need to update user documentation, design documentation or monitoring documentation?

### Release note <!-- bugfixes or new feature need a release note -->

```release-note
`None`.
```


ti-chi-bot: [REVIEW NOTIFICATION]

This pull request has not been approved.


To complete the [pull request process](https://book.prow.tidb.io/#/en/workflows/pr), please ask the reviewers in the [list](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/6053/owners) to review by filling `/cc @reviewer` in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the [list](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/6053/owners) by filling  `/assign @committer` in the comment to help you merge this pull request.

The full list of commands accepted by this bot can be found [here](https://prow.tidb.io/command-help?repo=pingcap%2Ftiflow).

<details>

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.
</details>

<!--Review Notification Identifier-->
ti-chi-bot: @lance6716: PR needs rebase.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
lance6716: > rest lgtm. It seems some of the functions in the `utils` pkg (e.g. GetFlavor, GetSlaveServerID) have not been utilizing `baseDB`. Will it be done in future pr or any further consideration around that?

I'll open a tracking issue for them, they should also be done.
lance6716: /merge
ti-chi-bot: This pull request has been accepted and is ready to merge. <details>Commit hash: 0802cf42eeeff3d166f27879fe153a32fa587eaa</details>
ti-chi-bot: @lance6716: Your PR was out of date, I have automatically updated it for you.

At the same time I will also trigger all tests for you:

/run-all-tests

If the CI test fails, you just re-trigger the test that failed and the bot will merge the PR for you after the CI passes.


<details>

Instructions for interacting with me using PR comments are available [here](https://prow.tidb.io/command-help).  If you have questions or suggestions related to my behavior, please file an issue against the [ti-community-infra/tichi](https://github.com/ti-community-infra/tichi/issues/new?title=Prow%20issue:) repository.
</details>
lance6716: /run-leak-test

Patch:
diff --git a/dm/checker/checker.go b/dm/checker/checker.go
index 6ce515241..3b5bee703 100644
--- a/dm/checker/checker.go
+++ b/dm/checker/checker.go
@@ -216,7 +216,7 @@ func (c *Checker) Init(ctx context.Context) (err error) {
 				c.checkList = append(c.checkList, checker.NewOnlineDDLChecker(instance.sourceDB.DB, checkSchemas, c.onlineDDL, instance.baList))
 			}
 			if _, ok := c.checkingItems[config.BinlogDBChecking]; ok {
-				c.checkList = append(c.checkList, checker.NewBinlogDBChecker(instance.sourceDB.DB, instance.sourceDBinfo, checkSchemas, instance.cfg.CaseSensitive))
+				c.checkList = append(c.checkList, checker.NewBinlogDBChecker(instance.sourceDB, instance.sourceDBinfo, checkSchemas, instance.cfg.CaseSensitive))
 			}
 		}
 	}
diff --git a/dm/dm/config/subtask.go b/dm/dm/config/subtask.go
index 54390d1ea..6c8e37145 100644
--- a/dm/dm/config/subtask.go
+++ b/dm/dm/config/subtask.go
@@ -15,6 +15,8 @@ package config
 
 import (
 	"bytes"
+	"context"
+	"database/sql"
 	_ "embed"
 	"encoding/json"
 	"flag"
@@ -26,6 +28,7 @@ import (
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb-tools/pkg/column-mapping"
 	extstorage "github.com/pingcap/tidb/br/pkg/storage"
+	"github.com/pingcap/tidb/util/dbutil"
 	"github.com/pingcap/tidb/util/filter"
 	regexprrouter "github.com/pingcap/tidb/util/regexpr-router"
 	router "github.com/pingcap/tidb/util/table-router"
@@ -143,6 +146,16 @@ func (db *DBConfig) AdjustWithTimeZone(timeZone string) {
 	db.Adjust()
 }
 
+// FetchTimeZoneSetting fetch target db global time_zone setting.
+// TODO: move GetTimeZoneOffset and FormatTimeZoneOffset from TiDB to tiflow.
+func FetchTimeZoneSetting(ctx context.Context, db *sql.DB) (string, error) {
+	dur, err := dbutil.GetTimeZoneOffset(ctx, db)
+	if err != nil {
+		return "", err
+	}
+	return dbutil.FormatTimeZoneOffset(dur), nil
+}
+
 // Clone returns a deep copy of DBConfig. This function only fixes data race when adjusting Session.
 func (db *DBConfig) Clone() *DBConfig {
 	if db == nil {
diff --git a/dm/dm/worker/source_worker.go b/dm/dm/worker/source_worker.go
index 40838ede3..556dd0cbd 100644
--- a/dm/dm/worker/source_worker.go
+++ b/dm/dm/worker/source_worker.go
@@ -32,6 +32,7 @@ import (
 	"github.com/pingcap/tiflow/dm/dm/pb"
 	"github.com/pingcap/tiflow/dm/pkg/binlog"
 	"github.com/pingcap/tiflow/dm/pkg/conn"
+	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
 	"github.com/pingcap/tiflow/dm/pkg/etcdutil"
 	"github.com/pingcap/tiflow/dm/pkg/ha"
 	"github.com/pingcap/tiflow/dm/pkg/log"
@@ -268,10 +269,11 @@ func (w *SourceWorker) updateSourceStatus(ctx context.Context, needLock bool) er
 	}
 	w.sourceDBMu.Unlock()
 
+	// TODO: write NewSourceStatusFromDB, so engine can easily to use
 	var status binlog.SourceStatus
 	ctx, cancel := context.WithTimeout(ctx, utils.DefaultDBTimeout)
 	defer cancel()
-	pos, gtidSet, err := utils.GetPosAndGs(ctx, w.sourceDB.DB, cfg.Flavor)
+	pos, gtidSet, err := conn.GetPosAndGs(tcontext.NewContext(ctx, log.L()), w.sourceDB, cfg.Flavor)
 	if err != nil {
 		return err
 	}
diff --git a/dm/dumpling/dumpling.go b/dm/dumpling/dumpling.go
index 1bf2e9693..7ea069ea8 100644
--- a/dm/dumpling/dumpling.go
+++ b/dm/dumpling/dumpling.go
@@ -329,8 +329,13 @@ func (m *Dumpling) constructArgs(ctx context.Context) (*export.Config, error) {
 	tz := m.cfg.Timezone
 	if len(tz) == 0 {
 		// use target db time_zone as default
+		baseDB, err2 := conn.DefaultDBProvider.Apply(&m.cfg.To)
+		if err2 != nil {
+			return nil, err2
+		}
+		defer baseDB.Close()
 		var err1 error
-		tz, err1 = conn.FetchTimeZoneSetting(ctx, &m.cfg.To)
+		tz, err1 = config.FetchTimeZoneSetting(ctx, baseDB.DB)
 		if err1 != nil {
 			return nil, err1
 		}
diff --git a/dm/loader/lightning.go b/dm/loader/lightning.go
index f32a7278e..6922f23c1 100644
--- a/dm/loader/lightning.go
+++ b/dm/loader/lightning.go
@@ -149,8 +149,13 @@ func (l *LightningLoader) Init(ctx context.Context) (err error) {
 
 	timeZone := l.cfg.Timezone
 	if len(timeZone) == 0 {
+		baseDB, err2 := conn.DefaultDBProvider.Apply(&l.cfg.To)
+		if err2 != nil {
+			return err2
+		}
+		defer baseDB.Close()
 		var err1 error
-		timeZone, err1 = conn.FetchTimeZoneSetting(ctx, &l.cfg.To)
+		timeZone, err1 = config.FetchTimeZoneSetting(ctx, baseDB.DB)
 		if err1 != nil {
 			return err1
 		}
diff --git a/dm/loader/loader.go b/dm/loader/loader.go
index 671e2f04b..da37b8a42 100644
--- a/dm/loader/loader.go
+++ b/dm/loader/loader.go
@@ -540,8 +540,13 @@ func (l *Loader) Init(ctx context.Context) (err error) {
 	}
 	timeZone := l.cfg.Timezone
 	if len(timeZone) == 0 {
+		baseDB, err2 := conn.DefaultDBProvider.Apply(&l.cfg.To)
+		if err2 != nil {
+			return err2
+		}
+		defer baseDB.Close()
 		var err1 error
-		timeZone, err1 = conn.FetchTimeZoneSetting(ctx, &lcfg.To)
+		timeZone, err1 = config.FetchTimeZoneSetting(ctx, baseDB.DB)
 		if err1 != nil {
 			return err1
 		}
diff --git a/dm/pkg/checker/binlog.go b/dm/pkg/checker/binlog.go
index 23f505e80..ef9797e3b 100644
--- a/dm/pkg/checker/binlog.go
+++ b/dm/pkg/checker/binlog.go
@@ -20,6 +20,10 @@ import (
 	"strings"
 
 	"github.com/pingcap/tidb/util/dbutil"
+
+	"github.com/pingcap/tiflow/dm/pkg/conn"
+	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
+	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 )
 
@@ -183,14 +187,14 @@ func (pc *MySQLBinlogRowImageChecker) Name() string {
 
 // BinlogDBChecker checks if migrated dbs are in binlog_do_db or binlog_ignore_db.
 type BinlogDBChecker struct {
-	db            *sql.DB
+	db            *conn.BaseDB
 	dbinfo        *dbutil.DBConfig
 	schemas       map[string]struct{}
 	caseSensitive bool
 }
 
 // NewBinlogDBChecker returns a RealChecker.
-func NewBinlogDBChecker(db *sql.DB, dbinfo *dbutil.DBConfig, schemas map[string]struct{}, caseSensitive bool) RealChecker {
+func NewBinlogDBChecker(db *conn.BaseDB, dbinfo *dbutil.DBConfig, schemas map[string]struct{}, caseSensitive bool) RealChecker {
 	newSchemas := make(map[string]struct{}, len(schemas))
 	for schema := range schemas {
 		newSchemas[schema] = struct{}{}
@@ -207,12 +211,13 @@ func (c *BinlogDBChecker) Check(ctx context.Context) *Result {
 		Extra: fmt.Sprintf("address of db instance - %s:%d", c.dbinfo.Host, c.dbinfo.Port),
 	}
 
-	flavor, err := utils.GetFlavor(ctx, c.db)
+	flavor, err := utils.GetFlavor(ctx, c.db.DB)
 	if err != nil {
 		markCheckError(result, err)
 		return result
 	}
-	binlogDoDB, binlogIgnoreDB, err := utils.GetBinlogDB(ctx, c.db, flavor)
+	tctx := tcontext.NewContext(ctx, log.L())
+	binlogDoDB, binlogIgnoreDB, err := conn.GetBinlogDB(tctx, c.db, flavor)
 	if err != nil {
 		markCheckError(result, err)
 		return result
diff --git a/dm/pkg/conn/basedb.go b/dm/pkg/conn/basedb.go
index 97e108703..07e47e695 100644
--- a/dm/pkg/conn/basedb.go
+++ b/dm/pkg/conn/basedb.go
@@ -170,6 +170,7 @@ func (d *BaseDB) GetBaseConn(ctx context.Context) (*BaseConn, error) {
 	return baseConn, nil
 }
 
+// TODO: retry can be done inside the BaseDB.
 func (d *BaseDB) ExecContext(tctx *tcontext.Context, query string, args ...interface{}) (sql.Result, error) {
 	if tctx.L().Core().Enabled(zap.DebugLevel) {
 		tctx.L().Debug("exec context",
@@ -179,6 +180,7 @@ func (d *BaseDB) ExecContext(tctx *tcontext.Context, query string, args ...inter
 	return d.DB.ExecContext(tctx.Ctx, query, args...)
 }
 
+// TODO: retry can be done inside the BaseDB.
 func (d *BaseDB) QueryContext(tctx *tcontext.Context, query string, args ...interface{}) (*sql.Rows, error) {
 	if tctx.L().Core().Enabled(zap.DebugLevel) {
 		tctx.L().Debug("query context",
diff --git a/dm/pkg/conn/utils.go b/dm/pkg/conn/utils.go
index 63e865367..7a1d3f877 100644
--- a/dm/pkg/conn/utils.go
+++ b/dm/pkg/conn/utils.go
@@ -14,24 +14,189 @@
 package conn
 
 import (
-	"context"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
 
-	"github.com/pingcap/errors"
-	"github.com/pingcap/tidb/util/dbutil"
+	gmysql "github.com/go-mysql-org/go-mysql/mysql"
+	"github.com/pingcap/failpoint"
+	tmysql "github.com/pingcap/tidb/parser/mysql"
+	"go.uber.org/zap"
 
-	"github.com/pingcap/tiflow/dm/dm/config"
+	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
+	"github.com/pingcap/tiflow/dm/pkg/gtid"
+	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
 
-// FetchTimeZoneSetting fetch target db global time_zone setting.
-func FetchTimeZoneSetting(ctx context.Context, cfg *config.DBConfig) (string, error) {
-	db, err := DefaultDBProvider.Apply(cfg)
+const (
+	// DefaultDBTimeout represents a DB operation timeout for common usages.
+	DefaultDBTimeout = 30 * time.Second
+)
+
+// GetGlobalVariable gets server's global variable.
+func GetGlobalVariable(ctx *tcontext.Context, db *BaseDB, variable string) (value string, err error) {
+	failpoint.Inject("GetGlobalVariableFailed", func(val failpoint.Value) {
+		items := strings.Split(val.(string), ",")
+		if len(items) != 2 {
+			ctx.L().Fatal("failpoint GetGlobalVariableFailed's value is invalid", zap.String("val", val.(string)))
+		}
+		variableName := items[0]
+		errCode, err1 := strconv.ParseUint(items[1], 10, 16)
+		if err1 != nil {
+			ctx.L().Fatal("failpoint GetGlobalVariableFailed's value is invalid", zap.String("val", val.(string)))
+		}
+		if variable == variableName {
+			err = tmysql.NewErr(uint16(errCode))
+			ctx.L().Warn("GetGlobalVariable failed", zap.String("variable", variable), zap.String("failpoint", "GetGlobalVariableFailed"), zap.Error(err))
+			failpoint.Return("", terror.DBErrorAdapt(err, terror.ErrDBDriverError))
+		}
+	})
+
+	conn, err := db.GetBaseConn(ctx.Context())
+	if err != nil {
+		return "", terror.DBErrorAdapt(err, terror.ErrDBDriverError)
+	}
+	defer conn.close()
+	return getVariable(ctx, conn, variable, true)
+}
+
+func getVariable(ctx *tcontext.Context, conn *BaseConn, variable string, isGlobal bool) (value string, err error) {
+	var template string
+	if isGlobal {
+		template = "SHOW GLOBAL VARIABLES LIKE '%s'"
+	} else {
+		template = "SHOW VARIABLES LIKE '%s'"
+	}
+	query := fmt.Sprintf(template, variable)
+	row, err := conn.QuerySQL(ctx, query)
 	if err != nil {
 		return "", err
 	}
-	defer db.Close()
-	dur, err := dbutil.GetTimeZoneOffset(ctx, db.DB)
+	defer row.Close()
+
+	// Show an example.
+	/*
+		mysql> SHOW GLOBAL VARIABLES LIKE "binlog_format";
+		+---------------+-------+
+		| Variable_name | Value |
+		+---------------+-------+
+		| binlog_format | ROW   |
+		+---------------+-------+
+	*/
+
+	if !row.Next() {
+		return "", terror.ErrDBDriverError.Generatef("variable %s not found", variable)
+	}
+
+	err = row.Scan(&variable, &value)
 	if err != nil {
-		return "", errors.Trace(err)
+		return "", terror.DBErrorAdapt(err, terror.ErrDBDriverError)
+	}
+	return value, nil
+}
+
+// GetMasterStatus gets status from master.
+// When the returned error is nil, the gtid must be not nil.
+func GetMasterStatus(ctx *tcontext.Context, db *BaseDB, flavor string) (
+	string, uint32, string, string, string, error,
+) {
+	var (
+		binlogName     string
+		pos            uint32
+		binlogDoDB     string
+		binlogIgnoreDB string
+		gtidStr        string
+		err            error
+	)
+	// need REPLICATION SLAVE privilege
+	rows, err := db.QueryContext(ctx, `SHOW MASTER STATUS`)
+	if err != nil {
+		err = terror.DBErrorAdapt(err, terror.ErrDBDriverError)
+		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
+	}
+	defer rows.Close()
+
+	if !rows.Next() {
+		err = terror.ErrNoMasterStatus.Generate()
+		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
 	}
-	return dbutil.FormatTimeZoneOffset(dur), nil
+	// Show an example.
+	/*
+		MySQL [test]> SHOW MASTER STATUS;
+		+-----------+----------+--------------+------------------+--------------------------------------------+
+		| File      | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set                          |
+		+-----------+----------+--------------+------------------+--------------------------------------------+
+		| ON.000001 |     4822 |              |                  | 85ab69d1-b21f-11e6-9c5e-64006a8978d2:1-46
+		+-----------+----------+--------------+------------------+--------------------------------------------+
+	*/
+	/*
+		For MariaDB,SHOW MASTER STATUS:
+		+--------------------+----------+--------------+------------------+
+		| File               | Position | Binlog_Do_DB | Binlog_Ignore_DB |
+		+--------------------+----------+--------------+------------------+
+		| mariadb-bin.000016 |      475 |              |                  |
+		+--------------------+----------+--------------+------------------+
+		SELECT @@global.gtid_binlog_pos;
+		+--------------------------+
+		| @@global.gtid_binlog_pos |
+		+--------------------------+
+		| 0-1-2                    |
+		+--------------------------+
+	*/
+	if flavor == gmysql.MySQLFlavor {
+		err = rows.Scan(&binlogName, &pos, &binlogDoDB, &binlogIgnoreDB, &gtidStr)
+	} else {
+		err = rows.Scan(&binlogName, &pos, &binlogDoDB, &binlogIgnoreDB)
+	}
+	if err != nil {
+		err = terror.DBErrorAdapt(err, terror.ErrDBDriverError)
+		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
+	}
+	if flavor == gmysql.MariaDBFlavor {
+		gtidStr, err = GetGlobalVariable(ctx, db, "gtid_binlog_pos")
+		if err != nil {
+			return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
+		}
+	}
+
+	if rows.Next() {
+		ctx.L().Warn("SHOW MASTER STATUS returns more than one row, will only use first row")
+	}
+	if rows.Close() != nil {
+		err = terror.DBErrorAdapt(rows.Err(), terror.ErrDBDriverError)
+		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
+	}
+	if rows.Err() != nil {
+		err = terror.DBErrorAdapt(rows.Err(), terror.ErrDBDriverError)
+		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
+	}
+
+	return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
+}
+
+// GetPosAndGs get binlog position and gmysql.GTIDSet from `show master status`.
+func GetPosAndGs(ctx *tcontext.Context, db *BaseDB, flavor string) (
+	binlogPos gmysql.Position,
+	gs gmysql.GTIDSet,
+	err error,
+) {
+	binlogName, pos, _, _, gtidStr, err := GetMasterStatus(ctx, db, flavor)
+	if err != nil {
+		return
+	}
+	binlogPos = gmysql.Position{
+		Name: binlogName,
+		Pos:  pos,
+	}
+
+	gs, err = gtid.ParserGTID(flavor, gtidStr)
+	return
+}
+
+// GetBinlogDB get binlog_do_db and binlog_ignore_db from `show master status`.
+func GetBinlogDB(ctx *tcontext.Context, db *BaseDB, flavor string) (string, string, error) {
+	// nolint:dogsled
+	_, _, binlogDoDB, binlogIgnoreDB, _, err := GetMasterStatus(ctx, db, flavor)
+	return binlogDoDB, binlogIgnoreDB, err
 }
diff --git a/dm/pkg/utils/db.go b/dm/pkg/utils/db.go
index 5a89af11b..9ac9af862 100644
--- a/dm/pkg/utils/db.go
+++ b/dm/pkg/utils/db.go
@@ -33,7 +33,6 @@ import (
 	"github.com/pingcap/tidb/util/dbutil"
 	"go.uber.org/zap"
 
-	"github.com/pingcap/tiflow/dm/pkg/gtid"
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
@@ -168,125 +167,6 @@ func GetSlaveServerID(ctx context.Context, db *sql.DB) (map[uint32]struct{}, err
 	return serverIDs, nil
 }
 
-// GetPosAndGs get binlog position and gmysql.GTIDSet from `show master status`.
-func GetPosAndGs(ctx context.Context, db *sql.DB, flavor string) (
-	binlogPos gmysql.Position,
-	gs gmysql.GTIDSet,
-	err error,
-) {
-	binlogName, pos, _, _, gtidStr, err := GetMasterStatus(ctx, db, flavor)
-	if err != nil {
-		return
-	}
-	binlogPos = gmysql.Position{
-		Name: binlogName,
-		Pos:  pos,
-	}
-
-	gs, err = gtid.ParserGTID(flavor, gtidStr)
-	return
-}
-
-// GetBinlogDB get binlog_do_db and binlog_ignore_db from `show master status`.
-func GetBinlogDB(ctx context.Context, db *sql.DB, flavor string) (string, string, error) {
-	// nolint:dogsled
-	_, _, binlogDoDB, binlogIgnoreDB, _, err := GetMasterStatus(ctx, db, flavor)
-	return binlogDoDB, binlogIgnoreDB, err
-}
-
-// GetMasterStatus gets status from master.
-// When the returned error is nil, the gtid must be not nil.
-func GetMasterStatus(ctx context.Context, db *sql.DB, flavor string) (
-	string, uint32, string, string, string, error,
-) {
-	var (
-		binlogName     string
-		pos            uint32
-		binlogDoDB     string
-		binlogIgnoreDB string
-		gtidStr        string
-		err            error
-	)
-	// need REPLICATION SLAVE privilege
-	rows, err := db.QueryContext(ctx, `SHOW MASTER STATUS`)
-	if err != nil {
-		err = terror.DBErrorAdapt(err, terror.ErrDBDriverError)
-		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-	}
-	defer rows.Close()
-
-	if !rows.Next() {
-		err = terror.ErrNoMasterStatus.Generate()
-		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-	}
-	// Show an example.
-	/*
-		MySQL [test]> SHOW MASTER STATUS;
-		+-----------+----------+--------------+------------------+--------------------------------------------+
-		| File      | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set                          |
-		+-----------+----------+--------------+------------------+--------------------------------------------+
-		| ON.000001 |     4822 |              |                  | 85ab69d1-b21f-11e6-9c5e-64006a8978d2:1-46
-		+-----------+----------+--------------+------------------+--------------------------------------------+
-	*/
-	/*
-		For MariaDB,SHOW MASTER STATUS:
-		+--------------------+----------+--------------+------------------+
-		| File               | Position | Binlog_Do_DB | Binlog_Ignore_DB |
-		+--------------------+----------+--------------+------------------+
-		| mariadb-bin.000016 |      475 |              |                  |
-		+--------------------+----------+--------------+------------------+
-		SELECT @@global.gtid_binlog_pos;
-		+--------------------------+
-		| @@global.gtid_binlog_pos |
-		+--------------------------+
-		| 0-1-2                    |
-		+--------------------------+
-	*/
-	if flavor == gmysql.MySQLFlavor {
-		err = rows.Scan(&binlogName, &pos, &binlogDoDB, &binlogIgnoreDB, &gtidStr)
-	} else {
-		err = rows.Scan(&binlogName, &pos, &binlogDoDB, &binlogIgnoreDB)
-	}
-	if err != nil {
-		err = terror.DBErrorAdapt(err, terror.ErrDBDriverError)
-		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-	}
-	if flavor == gmysql.MariaDBFlavor {
-		gtidStr, err = GetGlobalVariable(ctx, db, "gtid_binlog_pos")
-		if err != nil {
-			return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-		}
-	}
-
-	if rows.Next() {
-		log.L().Warn("SHOW MASTER STATUS returns more than one row, will only use first row")
-	}
-	if rows.Close() != nil {
-		err = terror.DBErrorAdapt(rows.Err(), terror.ErrDBDriverError)
-		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-	}
-	if rows.Err() != nil {
-		err = terror.DBErrorAdapt(rows.Err(), terror.ErrDBDriverError)
-		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-	}
-
-	return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-}
-
-// GetMariaDBGTID gets MariaDB's `gtid_binlog_pos`
-// it can not get by `SHOW MASTER STATUS`.
-func GetMariaDBGTID(ctx context.Context, db *sql.DB) (gmysql.GTIDSet, error) {
-	gtidStr, err := GetGlobalVariable(ctx, db, "gtid_binlog_pos")
-	if err != nil {
-		return nil, err
-	}
-	gs, err := gtid.ParserGTID(gmysql.MariaDBFlavor, gtidStr)
-	if err != nil {
-		return nil, err
-	}
-	return gs, nil
-}
-
 // GetGlobalVariable gets server's global variable.
 func GetGlobalVariable(ctx context.Context, db *sql.DB, variable string) (value string, err error) {
 	failpoint.Inject("GetGlobalVariableFailed", func(val failpoint.Value) {
diff --git a/dm/relay/relay.go b/dm/relay/relay.go
index d88c35a0a..2a15bc683 100644
--- a/dm/relay/relay.go
+++ b/dm/relay/relay.go
@@ -39,6 +39,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/binlog/common"
 	binlogReader "github.com/pingcap/tiflow/dm/pkg/binlog/reader"
 	"github.com/pingcap/tiflow/dm/pkg/conn"
+	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
 	"github.com/pingcap/tiflow/dm/pkg/gtid"
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	parserpkg "github.com/pingcap/tiflow/dm/pkg/parser"
@@ -596,7 +597,8 @@ func (r *Relay) handleEvents(
 					r.logger.Error("the requested binlog files have purged in the master server or the master server have switched, currently DM do no support to handle this error",
 						zap.String("db host", cfg.Host), zap.Int("db port", cfg.Port), zap.Stringer("last pos", lastPos), log.ShortError(err))
 					// log the status for debug
-					pos, gs, err2 := utils.GetPosAndGs(ctx, r.db.DB, r.cfg.Flavor)
+					tctx := tcontext.NewContext(ctx, r.logger)
+					pos, gs, err2 := conn.GetPosAndGs(tctx, r.db, r.cfg.Flavor)
 					if err2 == nil {
 						r.logger.Info("current master status", zap.Stringer("position", pos), log.WrapStringerField("GTID sets", gs))
 					}
@@ -782,7 +784,8 @@ func (r *Relay) reSetupMeta(ctx context.Context) error {
 
 	var latestPosName, latestGTIDStr string
 	if (r.cfg.EnableGTID && len(r.cfg.BinlogGTID) == 0) || (!r.cfg.EnableGTID && len(r.cfg.BinLogName) == 0) {
-		latestPos, latestGTID, err2 := utils.GetPosAndGs(ctx, r.db.DB, r.cfg.Flavor)
+		tctx := tcontext.NewContext(ctx, r.logger)
+		latestPos, latestGTID, err2 := conn.GetPosAndGs(tctx, r.db, r.cfg.Flavor)
 		if err2 != nil {
 			return err2
 		}
@@ -862,7 +865,8 @@ func (r *Relay) doIntervalOps(ctx context.Context) {
 				return
 			}
 			ctx2, cancel2 := context.WithTimeout(ctx, utils.DefaultDBTimeout)
-			pos, _, err := utils.GetPosAndGs(ctx2, r.db.DB, r.cfg.Flavor)
+			tctx := tcontext.NewContext(ctx2, r.logger)
+			pos, _, err := conn.GetPosAndGs(tctx, r.db, r.cfg.Flavor)
 			cancel2()
 			if err != nil {
 				r.logger.Warn("get master status", zap.Error(err))
diff --git a/dm/syncer/dbconn/db.go b/dm/syncer/dbconn/db.go
index c2d847f0e..674ed9529 100644
--- a/dm/syncer/dbconn/db.go
+++ b/dm/syncer/dbconn/db.go
@@ -20,17 +20,17 @@ import (
 
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/util/dbutil"
-	"github.com/pingcap/tiflow/dm/pkg/metricsproxy"
-	"github.com/pingcap/tiflow/dm/syncer/metrics"
 	"go.uber.org/zap"
 
 	"github.com/pingcap/tiflow/dm/dm/config"
 	"github.com/pingcap/tiflow/dm/pkg/conn"
 	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
 	"github.com/pingcap/tiflow/dm/pkg/log"
+	"github.com/pingcap/tiflow/dm/pkg/metricsproxy"
 	"github.com/pingcap/tiflow/dm/pkg/retry"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
+	"github.com/pingcap/tiflow/dm/syncer/metrics"
 )
 
 var retryTimeout = 3 * time.Second
diff --git a/dm/syncer/dbconn/upstream_db.go b/dm/syncer/dbconn/upstream_db.go
index cdb48dc8f..cd61c022f 100644
--- a/dm/syncer/dbconn/upstream_db.go
+++ b/dm/syncer/dbconn/upstream_db.go
@@ -50,8 +50,8 @@ func NewUpStreamConn(dbCfg *config.DBConfig) (*UpStreamConn, error) {
 }
 
 // GetMasterStatus returns binlog location that extracted from SHOW MASTER STATUS.
-func (conn *UpStreamConn) GetMasterStatus(ctx context.Context, flavor string) (mysql.Position, mysql.GTIDSet, error) {
-	pos, gtidSet, err := utils.GetPosAndGs(ctx, conn.BaseDB.DB, flavor)
+func (c *UpStreamConn) GetMasterStatus(ctx *tcontext.Context, flavor string) (mysql.Position, mysql.GTIDSet, error) {
+	pos, gtidSet, err := conn.GetPosAndGs(ctx, c.BaseDB, flavor)
 
 	failpoint.Inject("GetMasterStatusFailed", func(val failpoint.Value) {
 		err = tmysql.NewErr(uint16(val.(int)))
@@ -62,13 +62,13 @@ func (conn *UpStreamConn) GetMasterStatus(ctx context.Context, flavor string) (m
 }
 
 // GetServerUUID returns upstream server UUID.
-func (conn *UpStreamConn) GetServerUUID(ctx context.Context, flavor string) (string, error) {
-	return utils.GetServerUUID(ctx, conn.BaseDB.DB, flavor)
+func (c *UpStreamConn) GetServerUUID(ctx context.Context, flavor string) (string, error) {
+	return utils.GetServerUUID(ctx, c.BaseDB.DB, flavor)
 }
 
 // GetServerUnixTS returns the result of current timestamp in upstream.
-func (conn *UpStreamConn) GetServerUnixTS(ctx context.Context) (int64, error) {
-	return utils.GetServerUnixTS(ctx, conn.BaseDB.DB)
+func (c *UpStreamConn) GetServerUnixTS(ctx context.Context) (int64, error) {
+	return utils.GetServerUnixTS(ctx, c.BaseDB.DB)
 }
 
 // GetCharsetAndCollationInfo returns charset and collation info.
@@ -117,18 +117,18 @@ func GetCharsetAndCollationInfo(tctx *tcontext.Context, conn *DBConn) (map[strin
 }
 
 // GetParser returns the parser with correct flag for upstream.
-func (conn *UpStreamConn) GetParser(ctx context.Context) (*parser.Parser, error) {
-	return utils.GetParser(ctx, conn.BaseDB.DB)
+func (c *UpStreamConn) GetParser(ctx context.Context) (*parser.Parser, error) {
+	return utils.GetParser(ctx, c.BaseDB.DB)
 }
 
 // KillConn kills a connection in upstream.
-func (conn *UpStreamConn) KillConn(ctx context.Context, connID uint32) error {
-	return utils.KillConn(ctx, conn.BaseDB.DB, connID)
+func (c *UpStreamConn) KillConn(ctx context.Context, connID uint32) error {
+	return utils.KillConn(ctx, c.BaseDB.DB, connID)
 }
 
 // FetchAllDoTables returns tables matches allow-list.
-func (conn *UpStreamConn) FetchAllDoTables(ctx context.Context, bw *filter.Filter) (map[string][]string, error) {
-	return utils.FetchAllDoTables(ctx, conn.BaseDB.DB, bw)
+func (c *UpStreamConn) FetchAllDoTables(ctx context.Context, bw *filter.Filter) (map[string][]string, error) {
+	return utils.FetchAllDoTables(ctx, c.BaseDB.DB, bw)
 }
 
 // CloseUpstreamConn closes the UpStreamConn.
diff --git a/dm/syncer/relay.go b/dm/syncer/relay.go
index 5325dd628..a96f845c5 100644
--- a/dm/syncer/relay.go
+++ b/dm/syncer/relay.go
@@ -63,7 +63,7 @@ func (s *Syncer) setInitActiveRelayLog(ctx context.Context) error {
 		}
 	default:
 		// start from dumper or loader, get current pos from master
-		pos, _, err = s.fromDB.GetMasterStatus(ctx, s.cfg.Flavor)
+		pos, _, err = s.fromDB.GetMasterStatus(s.tctx.WithContext(ctx), s.cfg.Flavor)
 		if err != nil {
 			return terror.Annotatef(err, "get master status")
 		}
diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go
index 2eae19f51..2d55c3410 100644
--- a/dm/syncer/syncer.go
+++ b/dm/syncer/syncer.go
@@ -1939,7 +1939,7 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 		failpoint.Inject("FakeRedirect", func(val failpoint.Value) {
 			if len(shardingReSyncCh) == 0 && shardingReSync == nil {
 				if strVal, ok := val.(string); ok {
-					pos, gtidSet, err2 := s.fromDB.GetMasterStatus(ctx, s.cfg.Flavor)
+					pos, gtidSet, err2 := s.fromDB.GetMasterStatus(s.tctx.WithContext(ctx), s.cfg.Flavor)
 					if err2 != nil {
 						s.tctx.L().Error("fail to get master status in failpoint FakeRedirect", zap.Error(err2))
 						os.Exit(1)
diff --git a/dm/syncer/util.go b/dm/syncer/util.go
index 0871aa772..279196696 100644
--- a/dm/syncer/util.go
+++ b/dm/syncer/util.go
@@ -109,7 +109,12 @@ func printServerVersion(tctx *tcontext.Context, db *conn.BaseDB, scope string) {
 func str2TimezoneOrFromDB(tctx *tcontext.Context, tzStr string, dbCfg *config.DBConfig) (*time.Location, error) {
 	var err error
 	if len(tzStr) == 0 {
-		tzStr, err = conn.FetchTimeZoneSetting(tctx.Ctx, dbCfg)
+		baseDB, err2 := conn.DefaultDBProvider.Apply(dbCfg)
+		if err2 != nil {
+			return nil, err2
+		}
+		defer baseDB.Close()
+		tzStr, err = config.FetchTimeZoneSetting(tctx.Ctx, baseDB.DB)
 		if err != nil {
 			return nil, err
 		}


Test patch:
diff --git a/dm/dm/config/subtask_test.go b/dm/dm/config/subtask_test.go
index a18cd4206..fc51edcad 100644
--- a/dm/dm/config/subtask_test.go
+++ b/dm/dm/config/subtask_test.go
@@ -14,11 +14,12 @@
 package config
 
 import (
+	"context"
 	"reflect"
 
+	"github.com/DATA-DOG/go-sqlmock"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/tidb/util/filter"
-
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
 
@@ -316,3 +317,14 @@ func (t *testConfig) TestDBConfigClone(c *C) {
 	c.Assert(a, DeepEquals, b)
 	c.Assert(a.Security, Not(Equals), b.Security)
 }
+
+func (t *testConfig) TestFetchTZSetting(c *C) {
+	db, mock, err := sqlmock.New()
+	c.Assert(err, IsNil)
+
+	mock.ExpectQuery("SELECT cast\\(TIMEDIFF\\(NOW\\(6\\), UTC_TIMESTAMP\\(6\\)\\) as time\\);").
+		WillReturnRows(mock.NewRows([]string{""}).AddRow("01:00:00"))
+	tz, err := FetchTimeZoneSetting(context.Background(), db)
+	c.Assert(err, IsNil)
+	c.Assert(tz, Equals, "+01:00")
+}
diff --git a/dm/dm/worker/subtask_test.go b/dm/dm/worker/subtask_test.go
index ccaaf8b72..9b2b4b1ab 100644
--- a/dm/dm/worker/subtask_test.go
+++ b/dm/dm/worker/subtask_test.go
@@ -20,6 +20,11 @@ import (
 	"time"
 
 	"github.com/go-mysql-org/go-mysql/mysql"
+	. "github.com/pingcap/check"
+	"github.com/pingcap/errors"
+	"github.com/stretchr/testify/require"
+	clientv3 "go.etcd.io/etcd/client/v3"
+
 	"github.com/pingcap/tiflow/dm/dm/config"
 	"github.com/pingcap/tiflow/dm/dm/pb"
 	"github.com/pingcap/tiflow/dm/dm/unit"
@@ -30,11 +35,6 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 	"github.com/pingcap/tiflow/dm/relay"
 	"github.com/pingcap/tiflow/dm/syncer"
-	"github.com/stretchr/testify/require"
-
-	. "github.com/pingcap/check"
-	"github.com/pingcap/errors"
-	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 const (
diff --git a/dm/pkg/checker/binlog_test.go b/dm/pkg/checker/binlog_test.go
index 6a7475237..35186e4d0 100644
--- a/dm/pkg/checker/binlog_test.go
+++ b/dm/pkg/checker/binlog_test.go
@@ -19,6 +19,8 @@ import (
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/pingcap/tidb/util/dbutil"
 	"github.com/stretchr/testify/require"
+
+	"github.com/pingcap/tiflow/dm/pkg/conn"
 )
 
 func TestBinlogDB(t *testing.T) {
@@ -115,7 +117,7 @@ func TestBinlogDB(t *testing.T) {
 	}
 
 	for _, cs := range cases {
-		binlogDBChecker := NewBinlogDBChecker(db, &dbutil.DBConfig{}, cs.schemas, cs.caseSensitive)
+		binlogDBChecker := NewBinlogDBChecker(conn.NewBaseDB(db), &dbutil.DBConfig{}, cs.schemas, cs.caseSensitive)
 		versionRow := sqlmock.NewRows([]string{"Variable_name", "Value"}).AddRow("version", "mysql")
 		masterStatusRow := sqlmock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB", "Executed_Gtid_Set"}).
 			AddRow("", 0, cs.doDB, cs.ignoreDB, "")
diff --git a/dm/pkg/conn/utils_test.go b/dm/pkg/conn/utils_test.go
index b366fb251..e031deda1 100644
--- a/dm/pkg/conn/utils_test.go
+++ b/dm/pkg/conn/utils_test.go
@@ -15,22 +15,48 @@ package conn
 
 import (
 	"context"
+	"testing"
 
-	. "github.com/pingcap/check"
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/stretchr/testify/require"
 
-	"github.com/pingcap/tiflow/dm/dm/config"
+	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
+	"github.com/pingcap/tiflow/dm/pkg/log"
 )
 
-var _ = Suite(testUtilSuite{})
-
-type testUtilSuite struct{}
-
-func (s testUtilSuite) TestFetchTZSetting(c *C) {
-	m := InitMockDB(c)
-
-	m.ExpectQuery("SELECT cast\\(TIMEDIFF\\(NOW\\(6\\), UTC_TIMESTAMP\\(6\\)\\) as time\\);").
-		WillReturnRows(m.NewRows([]string{""}).AddRow("01:00:00"))
-	tz, err := FetchTimeZoneSetting(context.Background(), &config.DBConfig{})
-	c.Assert(err, IsNil)
-	c.Assert(tz, Equals, "+01:00")
+func TestGetBinlogDB(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), DefaultDBTimeout)
+	defer cancel()
+
+	tctx := tcontext.NewContext(ctx, log.L())
+
+	db, mock, err := sqlmock.New()
+	require.NoError(t, err)
+	baseDB := NewBaseDB(db)
+
+	// 5 columns for MySQL
+	rows := mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB", "Executed_Gtid_Set"}).AddRow(
+		"mysql-bin.000009", 11232, "do_db", "ignore_db", "074be7f4-f0f1-11ea-95bd-0242ac120002:1-699",
+	)
+	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
+
+	binlogDoDB, binlogIgnoreDB, err := GetBinlogDB(tctx, baseDB, "mysql")
+	require.Nil(t, err)
+	require.Equal(t, binlogDoDB, "do_db")
+	require.Equal(t, binlogIgnoreDB, "ignore_db")
+	require.Nil(t, mock.ExpectationsWereMet())
+
+	// 4 columns for MariaDB
+	rows = mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB"}).AddRow(
+		"mysql-bin.000009", 11232, "do_db", "ignore_db",
+	)
+	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
+	rows = mock.NewRows([]string{"Variable_name", "Value"}).AddRow("gtid_binlog_pos", "1-2-100")
+	mock.ExpectQuery(`SHOW GLOBAL VARIABLES LIKE 'gtid_binlog_pos'`).WillReturnRows(rows)
+
+	binlogDoDB, binlogIgnoreDB, err = GetBinlogDB(tctx, baseDB, "mariadb")
+	require.Nil(t, err)
+	require.Equal(t, binlogDoDB, "do_db")
+	require.Equal(t, binlogIgnoreDB, "ignore_db")
+	require.Nil(t, mock.ExpectationsWereMet())
 }
diff --git a/dm/pkg/utils/db_test.go b/dm/pkg/utils/db_test.go
index a5a1aaf06..1bfce9f32 100644
--- a/dm/pkg/utils/db_test.go
+++ b/dm/pkg/utils/db_test.go
@@ -75,88 +75,6 @@ func (t *testDBSuite) TestGetRandomServerID(c *C) {
 	c.Assert(serverID, Not(Equals), 101)
 }
 
-func TestGetPosAndGs(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), DefaultDBTimeout)
-	defer cancel()
-
-	db, mock, err := sqlmock.New()
-	require.Nil(t, err)
-
-	// 5 columns for MySQL
-	rows := mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB", "Executed_Gtid_Set"}).AddRow(
-		"mysql-bin.000009", 11232, "do_db", "ignore_db", "074be7f4-f0f1-11ea-95bd-0242ac120002:1-699",
-	)
-	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
-
-	pos, gs, err := GetPosAndGs(ctx, db, "mysql")
-	require.Nil(t, err)
-	require.Equal(t, pos, gmysql.Position{
-		Name: "mysql-bin.000009",
-		Pos:  11232,
-	})
-	require.Equal(t, gs.String(), "074be7f4-f0f1-11ea-95bd-0242ac120002:1-699")
-	require.Nil(t, mock.ExpectationsWereMet())
-
-	// 4 columns for MariaDB
-	rows = mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB"}).AddRow(
-		"mysql-bin.000009", 11232, "do_db", "ignore_db",
-	)
-	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
-	rows = mock.NewRows([]string{"Variable_name", "Value"}).AddRow("gtid_binlog_pos", "1-2-100")
-	mock.ExpectQuery(`SHOW GLOBAL VARIABLES LIKE 'gtid_binlog_pos'`).WillReturnRows(rows)
-
-	pos, gs, err = GetPosAndGs(ctx, db, "mariadb")
-	require.Nil(t, err)
-	require.Equal(t, pos, gmysql.Position{
-		Name: "mysql-bin.000009",
-		Pos:  11232,
-	})
-	require.Equal(t, gs.String(), "1-2-100")
-	require.Nil(t, mock.ExpectationsWereMet())
-
-	// some upstream (maybe a polarDB secondary node)
-	rows = mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB"})
-	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
-
-	_, gs, err = GetPosAndGs(ctx, db, "mysql")
-	require.Nil(t, gs)
-	require.NotNil(t, err)
-}
-
-func TestGetBinlogDB(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), DefaultDBTimeout)
-	defer cancel()
-
-	db, mock, err := sqlmock.New()
-	require.Nil(t, err)
-
-	// 5 columns for MySQL
-	rows := mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB", "Executed_Gtid_Set"}).AddRow(
-		"mysql-bin.000009", 11232, "do_db", "ignore_db", "074be7f4-f0f1-11ea-95bd-0242ac120002:1-699",
-	)
-	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
-
-	binlogDoDB, binlogIgnoreDB, err := GetBinlogDB(ctx, db, "mysql")
-	require.Nil(t, err)
-	require.Equal(t, binlogDoDB, "do_db")
-	require.Equal(t, binlogIgnoreDB, "ignore_db")
-	require.Nil(t, mock.ExpectationsWereMet())
-
-	// 4 columns for MariaDB
-	rows = mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB"}).AddRow(
-		"mysql-bin.000009", 11232, "do_db", "ignore_db",
-	)
-	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
-	rows = mock.NewRows([]string{"Variable_name", "Value"}).AddRow("gtid_binlog_pos", "1-2-100")
-	mock.ExpectQuery(`SHOW GLOBAL VARIABLES LIKE 'gtid_binlog_pos'`).WillReturnRows(rows)
-
-	binlogDoDB, binlogIgnoreDB, err = GetBinlogDB(ctx, db, "mariadb")
-	require.Nil(t, err)
-	require.Equal(t, binlogDoDB, "do_db")
-	require.Equal(t, binlogIgnoreDB, "ignore_db")
-	require.Nil(t, mock.ExpectationsWereMet())
-}
-
 func (t *testDBSuite) TestGetMariaDBGtidDomainID(c *C) {
 	ctx, cancel := context.WithTimeout(context.Background(), DefaultDBTimeout)
 	defer cancel()
diff --git a/dm/pkg/utils/string_test.go b/dm/pkg/utils/string_test.go
index 12c946cde..df60aba4d 100644
--- a/dm/pkg/utils/string_test.go
+++ b/dm/pkg/utils/string_test.go
@@ -14,14 +14,14 @@
 package utils
 
 import (
-	. "github.com/pingcap/check"
-)
+	"testing"
 
-var _ = Suite(&testStringSuite{})
+	"github.com/stretchr/testify/require"
+)
 
-type testStringSuite struct{}
+func TestTruncateString(t *testing.T) {
+	t.Parallel()
 
-func (t *testStringSuite) TestTruncateString(c *C) {
 	cases := []struct {
 		s      string
 		n      int
@@ -38,11 +38,13 @@ func (t *testStringSuite) TestTruncateString(c *C) {
 	}
 
 	for _, cs := range cases {
-		c.Assert(TruncateString(cs.s, cs.n), Equals, cs.expect)
+		require.Equal(t, cs.expect, TruncateString(cs.s, cs.n))
 	}
 }
 
-func (t *testStringSuite) TestTruncateInterface(c *C) {
+func TestTruncateInterface(t *testing.T) {
+	t.Parallel()
+
 	type compose struct {
 		b bool
 		i int
@@ -82,6 +84,6 @@ func (t *testStringSuite) TestTruncateInterface(c *C) {
 	}
 
 	for _, cs := range cases {
-		c.Assert(TruncateInterface(cs.v, cs.n), Equals, cs.expect)
+		require.Equal(t, cs.expect, TruncateInterface(cs.v, cs.n))
 	}
 }
diff --git a/dm/syncer/dbconn/upstream_db_test.go b/dm/syncer/dbconn/upstream_db_test.go
index 6389ddd0e..b66a0a2dd 100644
--- a/dm/syncer/dbconn/upstream_db_test.go
+++ b/dm/syncer/dbconn/upstream_db_test.go
@@ -24,6 +24,8 @@ import (
 	. "github.com/pingcap/check"
 
 	"github.com/pingcap/tiflow/dm/dm/config"
+	"github.com/pingcap/tiflow/dm/pkg/conn"
+	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 )
 
@@ -79,7 +81,7 @@ func (s *testDBSuite) resetBinlogSyncer(c *C) {
 		s.syncer.Close()
 	}
 
-	pos, _, err := utils.GetPosAndGs(context.Background(), s.db, "mysql")
+	pos, _, err := conn.GetPosAndGs(tcontext.Background(), conn.NewBaseDB(s.db), "mysql")
 	c.Assert(err, IsNil)
 
 	s.syncer = replication.NewBinlogSyncer(cfg)