Repository: pingcap/dumpling
Pull request #242 (2021-02-25 04:20:16)

log, metrics: refactor metrics and log system for dm and other repositories to use
<!--
Thank you for contributing to Dumpling! Please read the [CONTRIBUTING](https://github.com/pingcap/dumpling/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.

If you want to open the **Challenge Program** pull request, please use the following template:
https://raw.githubusercontent.com/pingcap/.github/master/.github/PULL_REQUEST_TEMPLATE/challenge-program.md

You can use it with query parameters: https://github.com/pingcap/dumpling/compare/master...${you branch}?template=challenge-program.md
-->

### What problem does this PR solve? <!--add issue link with summary if exists-->
1. Now dumpling only has only one global logger. DM may init several dumpling loggers, which may cause a data race in dumpling. Besides, each logger has a different task name, which may be confusing if we only have one global logger.
2. DM can't set customized `conf.Labels` because we have preset their labels in `metrics.go`.

### What is changed and how it works?
1. Use a customized logger for each dumper instance.
2. Don't set prometheus vector at first.

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test

Related changes

 - Need to cherry-pick to the release branch

### Release note

<!-- bugfixes or new feature need a release note, must in the form of a list, such as

- support -T/--tables-list argument

or if no need to be included in the release note, just add the following line

- No release note
-->
- No release note


lance6716: /lgtm

need remove some call of `Context()` in test files

Patch:
diff --git a/cmd/dumpling/main.go b/cmd/dumpling/main.go
index fb045a2..e3f48db 100644
--- a/cmd/dumpling/main.go
+++ b/cmd/dumpling/main.go
@@ -25,7 +25,6 @@ import (
 
 	"github.com/pingcap/dumpling/v4/cli"
 	"github.com/pingcap/dumpling/v4/export"
-	"github.com/pingcap/dumpling/v4/log"
 )
 
 func main() {
@@ -64,20 +63,20 @@ func main() {
 	registry := prometheus.NewRegistry()
 	registry.MustRegister(prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}))
 	registry.MustRegister(prometheus.NewGoCollector())
+	export.InitMetricsVector(conf.Labels)
 	export.RegisterMetrics(registry)
 	prometheus.DefaultGatherer = registry
 	dumper, err := export.NewDumper(context.Background(), conf)
 	if err != nil {
-		log.Error("dump failed error stack info", zap.Error(err))
 		fmt.Printf("\ncreate dumper failed: %s\n", err.Error())
 		os.Exit(1)
 	}
 	err = dumper.Dump()
 	dumper.Close()
 	if err != nil {
-		log.Error("dump failed error stack info", zap.Error(err))
+		dumper.L().Error("dump failed error stack info", zap.Error(err))
 		fmt.Printf("\ndump failed: %s\n", err.Error())
 		os.Exit(1)
 	}
-	log.Info("dump data successfully, dumpling will exit now")
+	dumper.L().Info("dump data successfully, dumpling will exit now")
 }
diff --git a/go.sum b/go.sum
index 51be893..3538c38 100644
--- a/go.sum
+++ b/go.sum
@@ -637,6 +637,7 @@ github.com/pingcap/check v0.0.0-20191107115940-caf2b9e6ccf4/go.mod h1:PYMCGwN0JH
 github.com/pingcap/check v0.0.0-20191216031241-8a5a85928f12/go.mod h1:PYMCGwN0JHjoqGr3HrZoD+b8Tgx8bKnArhSq8YVzUMc=
 github.com/pingcap/check v0.0.0-20200212061837-5e12011dc712 h1:R8gStypOBmpnHEx1qi//SaqxJVI4inOqljg/Aj5/390=
 github.com/pingcap/check v0.0.0-20200212061837-5e12011dc712/go.mod h1:PYMCGwN0JHjoqGr3HrZoD+b8Tgx8bKnArhSq8YVzUMc=
+github.com/pingcap/dm v1.1.0-alpha.0.20200521025928-83063141c5fd h1:j1xRUlwfdEgmTwBDGw7DB3wyE+9nQj99BtrQfhz+97M=
 github.com/pingcap/dm v1.1.0-alpha.0.20200521025928-83063141c5fd/go.mod h1:I5AAhwb0JPfLZINukML5VU9rB6mCcVA/Jq5OFFtMuEk=
 github.com/pingcap/dumpling v0.0.0-20200423082233-887d037b5b5c/go.mod h1:VJTcnA0MLL9tzDceTDoRh3k5UnOq9Hk6wh/ATo+B8I8=
 github.com/pingcap/errcode v0.0.0-20180921232412-a1a7271709d9/go.mod h1:4b2X8xSqxIroj/IZ9MX/VGZhAwc11wB9wRIzHvz6SeM=
diff --git a/v4/context/context.go b/v4/context/context.go
new file mode 100644
index 0000000..d38eb25
--- /dev/null
+++ b/v4/context/context.go
@@ -0,0 +1,64 @@
+// Copyright 2021 PingCAP, Inc. Licensed under Apache-2.0.
+// This code is copied from https://github.com/pingcap/dm/blob/master/pkg/context/context.go
+
+package context
+
+import (
+	gcontext "context"
+
+	"github.com/pingcap/dumpling/v4/log"
+)
+
+// Context is used to in dm to record some context field like
+// * go context
+// * logger
+type Context struct {
+	gcontext.Context
+	logger log.Logger
+}
+
+// Background return a nop context
+func Background() *Context {
+	return &Context{
+		Context: gcontext.Background(),
+		logger:  log.Zap(),
+	}
+}
+
+// NewContext return a new Context
+func NewContext(ctx gcontext.Context, logger log.Logger) *Context {
+	return &Context{
+		Context: ctx,
+		logger:  logger,
+	}
+}
+
+// WithContext set go context
+func (c *Context) WithContext(ctx gcontext.Context) *Context {
+	return &Context{
+		Context: ctx,
+		logger:  c.logger,
+	}
+}
+
+// WithCancel sets a cancel context.
+func (c *Context) WithCancel() (*Context, gcontext.CancelFunc) {
+	ctx, cancel := gcontext.WithCancel(c.Context)
+	return &Context{
+		Context: ctx,
+		logger:  c.logger,
+	}, cancel
+}
+
+// WithLogger set logger
+func (c *Context) WithLogger(logger log.Logger) *Context {
+	return &Context{
+		Context: c.Context,
+		logger:  logger,
+	}
+}
+
+// L returns real logger
+func (c *Context) L() log.Logger {
+	return c.logger
+}
diff --git a/v4/export/block_allow_list.go b/v4/export/block_allow_list.go
index 70b9f5c..2ce57eb 100644
--- a/v4/export/block_allow_list.go
+++ b/v4/export/block_allow_list.go
@@ -5,11 +5,11 @@ package export
 import (
 	"go.uber.org/zap"
 
-	"github.com/pingcap/dumpling/v4/log"
+	tcontext "github.com/pingcap/dumpling/v4/context"
 )
 
-func filterTables(conf *Config) {
-	log.Debug("filter tables")
+func filterTables(tctx *tcontext.Context, conf *Config) {
+	tctx.L().Debug("start to filter tables")
 	dbTables := DatabaseTables{}
 	ignoredDBTable := DatabaseTables{}
 
@@ -31,7 +31,7 @@ func filterTables(conf *Config) {
 	}
 
 	if len(ignoredDBTable) > 0 {
-		log.Debug("ignore table", zap.String("tables", ignoredDBTable.Literal()))
+		tctx.L().Debug("ignore table", zap.String("tables", ignoredDBTable.Literal()))
 	}
 
 	conf.Tables = dbTables
diff --git a/v4/export/config.go b/v4/export/config.go
index 26d044e..cc85b85 100644
--- a/v4/export/config.go
+++ b/v4/export/config.go
@@ -13,6 +13,8 @@ import (
 	"text/template"
 	"time"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/coreos/go-semver/semver"
 	"github.com/docker/go-units"
 	"github.com/go-sql-driver/mysql"
@@ -23,8 +25,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/spf13/pflag"
 	"go.uber.org/zap"
-
-	"github.com/pingcap/dumpling/v4/log"
 )
 
 const (
@@ -172,8 +172,8 @@ func DefaultConfig() *Config {
 // String returns dumpling's config in json format
 func (conf *Config) String() string {
 	cfg, err := json.Marshal(conf)
-	if err != nil {
-		log.Error("fail to marshal config to json", zap.Error(err))
+	if err != nil && conf.Logger != nil {
+		conf.Logger.Error("fail to marshal config to json", zap.Error(err))
 	}
 	return string(cfg)
 }
@@ -579,8 +579,8 @@ var (
 )
 
 // ParseServerInfo parses exported server type and version info from version string
-func ParseServerInfo(src string) ServerInfo {
-	log.Debug("parse server info", zap.String("server info string", src))
+func ParseServerInfo(tctx *tcontext.Context, src string) ServerInfo {
+	tctx.L().Debug("parse server info", zap.String("server info string", src))
 	lowerCase := strings.ToLower(src)
 	serverInfo := ServerInfo{}
 	switch {
@@ -594,7 +594,7 @@ func ParseServerInfo(src string) ServerInfo {
 		serverInfo.ServerType = ServerTypeUnknown
 	}
 
-	log.Info("detect server type",
+	tctx.L().Info("detect server type",
 		zap.String("type", serverInfo.ServerType.String()))
 
 	var versionStr string
@@ -608,12 +608,12 @@ func ParseServerInfo(src string) ServerInfo {
 	var err error
 	serverInfo.ServerVersion, err = semver.NewVersion(versionStr)
 	if err != nil {
-		log.Warn("fail to parse version",
+		tctx.L().Warn("fail to parse version",
 			zap.String("version", versionStr))
 		return serverInfo
 	}
 
-	log.Info("detect server version",
+	tctx.L().Info("detect server version",
 		zap.String("version", serverInfo.ServerVersion.String()))
 	return serverInfo
 }
@@ -660,22 +660,6 @@ func adjustConfig(conf *Config, fns ...func(*Config) error) error {
 	return nil
 }
 
-func initLogger(conf *Config) error {
-	if conf.Logger != nil {
-		log.SetAppLogger(conf.Logger)
-	} else {
-		err := log.InitAppLogger(&log.Config{
-			Level:  conf.LogLevel,
-			File:   conf.LogFile,
-			Format: conf.LogFormat,
-		})
-		if err != nil {
-			return errors.Trace(err)
-		}
-	}
-	return nil
-}
-
 func registerTLSConfig(conf *Config) error {
 	if len(conf.Security.CAPath) > 0 {
 		tlsConfig, err := utils.ToTLSConfig(conf.Security.CAPath, conf.Security.CertPath, conf.Security.KeyPath)
diff --git a/v4/export/consistency.go b/v4/export/consistency.go
index bfcde5a..0c418ff 100644
--- a/v4/export/consistency.go
+++ b/v4/export/consistency.go
@@ -6,6 +6,8 @@ import (
 	"context"
 	"database/sql"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/pingcap/br/pkg/utils"
 	"github.com/pingcap/errors"
 )
@@ -49,7 +51,7 @@ func NewConsistencyController(ctx context.Context, conf *Config, session *sql.DB
 
 // ConsistencyController is the interface that controls the consistency of exporting progress
 type ConsistencyController interface {
-	Setup(context.Context) error
+	Setup(*tcontext.Context) error
 	TearDown(context.Context) error
 	PingContext(context.Context) error
 }
@@ -58,7 +60,7 @@ type ConsistencyController interface {
 type ConsistencyNone struct{}
 
 // Setup implements ConsistencyController.Setup
-func (c *ConsistencyNone) Setup(_ context.Context) error {
+func (c *ConsistencyNone) Setup(_ *tcontext.Context) error {
 	return nil
 }
 
@@ -79,11 +81,11 @@ type ConsistencyFlushTableWithReadLock struct {
 }
 
 // Setup implements ConsistencyController.Setup
-func (c *ConsistencyFlushTableWithReadLock) Setup(ctx context.Context) error {
+func (c *ConsistencyFlushTableWithReadLock) Setup(tctx *tcontext.Context) error {
 	if c.serverType == ServerTypeTiDB {
 		return errors.New("'flush table with read lock' cannot be used to ensure the consistency in TiDB")
 	}
-	return FlushTableWithReadLock(ctx, c.conn)
+	return FlushTableWithReadLock(tctx, c.conn)
 }
 
 // TearDown implements ConsistencyController.TearDown
@@ -113,13 +115,13 @@ type ConsistencyLockDumpingTables struct {
 }
 
 // Setup implements ConsistencyController.Setup
-func (c *ConsistencyLockDumpingTables) Setup(ctx context.Context) error {
+func (c *ConsistencyLockDumpingTables) Setup(tctx *tcontext.Context) error {
 	blockList := make(map[string]map[string]interface{})
-	return utils.WithRetry(ctx, func() error {
+	return utils.WithRetry(tctx, func() error {
 		lockTablesSQL := buildLockTablesSQL(c.allTables, blockList)
-		_, err := c.conn.ExecContext(ctx, lockTablesSQL)
+		_, err := c.conn.ExecContext(tctx, lockTablesSQL)
 		return errors.Trace(err)
-	}, newLockTablesBackoffer(blockList))
+	}, newLockTablesBackoffer(tctx, blockList))
 }
 
 // TearDown implements ConsistencyController.TearDown
diff --git a/v4/export/dump.go b/v4/export/dump.go
index 007a863..50da732 100755
--- a/v4/export/dump.go
+++ b/v4/export/dump.go
@@ -11,6 +11,7 @@ import (
 	"strings"
 	"time"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
 	"github.com/pingcap/dumpling/v4/log"
 
 	// import mysql driver
@@ -26,7 +27,7 @@ import (
 
 // Dumper is the dump progress structure
 type Dumper struct {
-	ctx       context.Context
+	tctx      *tcontext.Context
 	conf      *Config
 	cancelCtx context.CancelFunc
 
@@ -38,14 +39,13 @@ type Dumper struct {
 
 // NewDumper returns a new Dumper
 func NewDumper(ctx context.Context, conf *Config) (*Dumper, error) {
-	ctx, cancelFn := context.WithCancel(ctx)
+	tctx, cancelFn := tcontext.Background().WithContext(ctx).WithCancel()
 	d := &Dumper{
-		ctx:       ctx,
+		tctx:      tctx,
 		conf:      conf,
 		cancelCtx: cancelFn,
 	}
 	err := adjustConfig(conf,
-		initLogger,
 		registerTLSConfig,
 		validateSpecifiedSQL,
 		validateFileFormat)
@@ -53,6 +53,7 @@ func NewDumper(ctx context.Context, conf *Config) (*Dumper, error) {
 		return nil, err
 	}
 	err = runSteps(d,
+		initLogger,
 		createExternalStore,
 		startHTTPService,
 		openSQLDB,
@@ -76,43 +77,43 @@ func (d *Dumper) Dump() (dumpErr error) {
 		err     error
 		conCtrl ConsistencyController
 	)
-	ctx, conf, pool := d.ctx, d.conf, d.dbHandle
-	m := newGlobalMetadata(d.extStore, conf.Snapshot)
+	tctx, conf, pool := d.tctx, d.conf, d.dbHandle
+	m := newGlobalMetadata(tctx, d.extStore, conf.Snapshot)
 	defer func() {
 		if dumpErr == nil {
-			_ = m.writeGlobalMetaData(ctx)
+			_ = m.writeGlobalMetaData()
 		}
 	}()
 
 	// for consistency lock, we should get table list at first to generate the lock tables SQL
 	if conf.Consistency == consistencyTypeLock {
-		conn, err = createConnWithConsistency(ctx, pool)
+		conn, err = createConnWithConsistency(tctx, pool)
 		if err != nil {
 			return errors.Trace(err)
 		}
-		if err = prepareTableListToDump(conf, conn); err != nil {
+		if err = prepareTableListToDump(tctx, conf, conn); err != nil {
 			conn.Close()
 			return err
 		}
 		conn.Close()
 	}
 
-	conCtrl, err = NewConsistencyController(ctx, conf, pool)
+	conCtrl, err = NewConsistencyController(tctx, conf, pool)
 	if err != nil {
 		return err
 	}
-	if err = conCtrl.Setup(ctx); err != nil {
+	if err = conCtrl.Setup(tctx); err != nil {
 		return errors.Trace(err)
 	}
 	// To avoid lock is not released
 	defer func() {
-		err = conCtrl.TearDown(ctx)
+		err = conCtrl.TearDown(tctx)
 		if err != nil {
-			log.Error("fail to tear down consistency controller", zap.Error(err))
+			tctx.L().Error("fail to tear down consistency controller", zap.Error(err))
 		}
 	}()
 
-	metaConn, err := createConnWithConsistency(ctx, pool)
+	metaConn, err := createConnWithConsistency(tctx, pool)
 	if err != nil {
 		return err
 	}
@@ -126,25 +127,25 @@ func (d *Dumper) Dump() (dumpErr error) {
 	// for consistency none, the binlog pos in metadata might be earlier than dumped data. We need to enable safe-mode to assure data safety.
 	err = m.recordGlobalMetaData(metaConn, conf.ServerInfo.ServerType, false)
 	if err != nil {
-		log.Info("get global metadata failed", zap.Error(err))
+		tctx.L().Info("get global metadata failed", zap.Error(err))
 	}
 
 	// for other consistencies, we should get table list after consistency is set up and GlobalMetaData is cached
 	if conf.Consistency != consistencyTypeLock {
-		if err = prepareTableListToDump(conf, metaConn); err != nil {
+		if err = prepareTableListToDump(tctx, conf, metaConn); err != nil {
 			return err
 		}
 	}
 
 	rebuildConn := func(conn *sql.Conn) (*sql.Conn, error) {
 		// make sure that the lock connection is still alive
-		err1 := conCtrl.PingContext(ctx)
+		err1 := conCtrl.PingContext(tctx)
 		if err1 != nil {
 			return conn, errors.Trace(err1)
 		}
 		// give up the last broken connection
 		conn.Close()
-		newConn, err1 := createConnWithConsistency(ctx, pool)
+		newConn, err1 := createConnWithConsistency(tctx, pool)
 		if err1 != nil {
 			return conn, errors.Trace(err1)
 		}
@@ -160,9 +161,9 @@ func (d *Dumper) Dump() (dumpErr error) {
 	}
 
 	taskChan := make(chan Task, defaultDumpThreads)
-	taskChannelCapacity.With(conf.Labels).Add(defaultDumpThreads)
-	wg, writingCtx := errgroup.WithContext(ctx)
-	writers, tearDownWriters, err := d.startWriters(writingCtx, wg, taskChan, rebuildConn)
+	AddGauge(taskChannelCapacity, conf.Labels, defaultDumpThreads)
+	wg, writingCtx := errgroup.WithContext(tctx)
+	writers, tearDownWriters, err := d.startWriters(tctx.WithContext(writingCtx), wg, taskChan, rebuildConn)
 	if err != nil {
 		return err
 	}
@@ -170,9 +171,9 @@ func (d *Dumper) Dump() (dumpErr error) {
 
 	if conf.TransactionalConsistency {
 		if conf.Consistency == consistencyTypeFlush || conf.Consistency == consistencyTypeLock {
-			log.Info("All the dumping transactions have started. Start to unlock tables")
+			tctx.L().Info("All the dumping transactions have started. Start to unlock tables")
 		}
-		if err = conCtrl.TearDown(ctx); err != nil {
+		if err = conCtrl.TearDown(tctx); err != nil {
 			return errors.Trace(err)
 		}
 	}
@@ -183,22 +184,22 @@ func (d *Dumper) Dump() (dumpErr error) {
 		// record again, to provide a location to exit safe mode for DM
 		err = m.recordGlobalMetaData(metaConn, conf.ServerInfo.ServerType, true)
 		if err != nil {
-			log.Info("get global metadata (after connection pool established) failed", zap.Error(err))
+			tctx.L().Info("get global metadata (after connection pool established) failed", zap.Error(err))
 		}
 	}
 
-	summary.SetLogCollector(summary.NewLogCollector(log.Info))
+	summary.SetLogCollector(summary.NewLogCollector(tctx.L().Info))
 	summary.SetUnit(summary.BackupUnit)
 	defer summary.Summary(summary.BackupUnit)
 
-	logProgressCtx, logProgressCancel := context.WithCancel(ctx)
+	logProgressCtx, logProgressCancel := tctx.WithCancel()
 	go d.runLogProgress(logProgressCtx)
 	defer logProgressCancel()
 
 	tableDataStartTime := time.Now()
 
 	failpoint.Inject("PrintTiDBMemQuotaQuery", func(_ failpoint.Value) {
-		row := d.dbHandle.QueryRowContext(ctx, "select @@tidb_mem_quota_query;")
+		row := d.dbHandle.QueryRowContext(tctx, "select @@tidb_mem_quota_query;")
 		var s string
 		err = row.Scan(&s)
 		if err != nil {
@@ -227,27 +228,27 @@ func (d *Dumper) Dump() (dumpErr error) {
 	return nil
 }
 
-func (d *Dumper) startWriters(ctx context.Context, wg *errgroup.Group, taskChan <-chan Task,
+func (d *Dumper) startWriters(tctx *tcontext.Context, wg *errgroup.Group, taskChan <-chan Task,
 	rebuildConnFn func(*sql.Conn) (*sql.Conn, error)) ([]*Writer, func(), error) {
 	conf, pool := d.conf, d.dbHandle
 	writers := make([]*Writer, conf.Threads)
 	for i := 0; i < conf.Threads; i++ {
-		conn, err := createConnWithConsistency(ctx, pool)
+		conn, err := createConnWithConsistency(tctx, pool)
 		if err != nil {
 			return nil, func() {}, err
 		}
-		writer := NewWriter(ctx, int64(i), conf, conn, d.extStore)
+		writer := NewWriter(tctx, int64(i), conf, conn, d.extStore)
 		writer.rebuildConnFn = rebuildConnFn
 		writer.setFinishTableCallBack(func(task Task) {
 			if td, ok := task.(*TaskTableData); ok {
-				finishedTablesCounter.With(conf.Labels).Inc()
-				log.Debug("finished dumping table data",
+				IncCounter(finishedTablesCounter, conf.Labels)
+				tctx.L().Debug("finished dumping table data",
 					zap.String("database", td.Meta.DatabaseName()),
 					zap.String("table", td.Meta.TableName()))
 			}
 		})
 		writer.setFinishTaskCallBack(func(task Task) {
-			taskChannelCapacity.With(conf.Labels).Inc()
+			IncGauge(taskChannelCapacity, conf.Labels)
 		})
 		wg.Go(func() error {
 			return writer.run(taskChan)
@@ -274,7 +275,7 @@ func (d *Dumper) dumpDatabases(metaConn *sql.Conn, taskChan chan<- Task) error {
 		d.sendTaskToChan(task, taskChan)
 
 		for _, table := range tables {
-			log.Debug("start dumping table...", zap.String("database", dbName),
+			d.L().Debug("start dumping table...", zap.String("database", dbName),
 				zap.String("table", table.Name))
 			meta, err := dumpTableMeta(conf, metaConn, dbName, table)
 			if err != nil {
@@ -328,7 +329,7 @@ func (d *Dumper) concurrentDumpTable(conn *sql.Conn, meta TableMeta, taskChan ch
 	if conf.ServerInfo.ServerType == ServerTypeTiDB &&
 		conf.ServerInfo.ServerVersion != nil &&
 		conf.ServerInfo.ServerVersion.Compare(*tableSampleVersion) >= 0 {
-		log.Debug("dumping TiDB tables with TABLESAMPLE",
+		d.L().Debug("dumping TiDB tables with TABLESAMPLE",
 			zap.String("database", db), zap.String("table", tbl))
 		return d.concurrentDumpTiDBTables(conn, meta, taskChan)
 	}
@@ -338,7 +339,7 @@ func (d *Dumper) concurrentDumpTable(conn *sql.Conn, meta TableMeta, taskChan ch
 	}
 	if field == "" {
 		// skip split chunk logic if not found proper field
-		log.Warn("fallback to sequential dump due to no proper field",
+		d.L().Warn("fallback to sequential dump due to no proper field",
 			zap.String("database", db), zap.String("table", tbl))
 		return d.sequentialDumpTable(conn, meta, taskChan)
 	}
@@ -347,18 +348,18 @@ func (d *Dumper) concurrentDumpTable(conn *sql.Conn, meta TableMeta, taskChan ch
 	if err != nil {
 		return err
 	}
-	log.Debug("get int bounding values",
+	d.L().Debug("get int bounding values",
 		zap.String("lower", min.String()),
 		zap.String("upper", max.String()))
 
-	count := estimateCount(db, tbl, conn, field, conf)
-	log.Info("get estimated rows count",
+	count := estimateCount(d.tctx, db, tbl, conn, field, conf)
+	d.L().Info("get estimated rows count",
 		zap.String("database", db),
 		zap.String("table", tbl),
 		zap.Uint64("estimateCount", count))
 	if count < conf.Rows {
 		// skip chunk logic if estimates are low
-		log.Warn("skip concurrent dump due to estimate count < rows",
+		d.L().Warn("skip concurrent dump due to estimate count < rows",
 			zap.Uint64("estimate count", count),
 			zap.Uint64("conf.rows", conf.Rows),
 			zap.String("database", db),
@@ -407,38 +408,38 @@ func (d *Dumper) concurrentDumpTable(conn *sql.Conn, meta TableMeta, taskChan ch
 }
 
 func (d *Dumper) sendTaskToChan(task Task, taskChan chan<- Task) (ctxDone bool) {
-	ctx, conf := d.ctx, d.conf
+	tctx, conf := d.tctx, d.conf
 	select {
-	case <-ctx.Done():
+	case <-tctx.Done():
 		return true
 	case taskChan <- task:
-		log.Debug("send task to writer",
+		tctx.L().Debug("send task to writer",
 			zap.String("task", task.Brief()))
-		taskChannelCapacity.With(conf.Labels).Desc()
+		DecGauge(taskChannelCapacity, conf.Labels)
 		return false
 	}
 }
 
 func (d *Dumper) selectMinAndMaxIntValue(conn *sql.Conn, db, tbl, field string) (*big.Int, *big.Int, error) {
-	ctx, conf, zero := d.ctx, d.conf, &big.Int{}
+	tctx, conf, zero := d.tctx, d.conf, &big.Int{}
 	query := fmt.Sprintf("SELECT MIN(`%s`),MAX(`%s`) FROM `%s`.`%s`",
 		escapeString(field), escapeString(field), escapeString(db), escapeString(tbl))
 	if conf.Where != "" {
 		query = fmt.Sprintf("%s WHERE %s", query, conf.Where)
 	}
-	log.Debug("split chunks", zap.String("query", query))
+	tctx.L().Debug("split chunks", zap.String("query", query))
 
 	var smin sql.NullString
 	var smax sql.NullString
-	row := conn.QueryRowContext(ctx, query)
+	row := conn.QueryRowContext(tctx, query)
 	err := row.Scan(&smin, &smax)
 	if err != nil {
-		log.Error("split chunks - get max min failed", zap.String("query", query), zap.Error(err))
+		tctx.L().Error("split chunks - get max min failed", zap.String("query", query), zap.Error(err))
 		return zero, zero, errors.Trace(err)
 	}
 	if !smax.Valid || !smin.Valid {
 		// found no data
-		log.Warn("no data to dump", zap.String("database", db), zap.String("table", tbl))
+		tctx.L().Warn("no data to dump", zap.String("database", db), zap.String("table", tbl))
 		return zero, zero, nil
 	}
 
@@ -483,6 +484,11 @@ func (d *Dumper) concurrentDumpTiDBTables(conn *sql.Conn, meta TableMeta, taskCh
 	return nil
 }
 
+// L returns real logger
+func (d *Dumper) L() log.Logger {
+	return d.tctx.L()
+}
+
 func selectTiDBTableSample(conn *sql.Conn, dbName, tableName string) (pkFields []string, pkVals []string, err error) {
 	pkFields, pkColTypes, err := GetPrimaryKeyAndColumnTypes(conn, dbName, tableName)
 	if err != nil {
@@ -527,7 +533,7 @@ func buildTiDBTableSampleQuery(pkFields []string, dbName, tblName string) string
 	return fmt.Sprintf(template, pks, escapeString(dbName), escapeString(tblName), pks)
 }
 
-func prepareTableListToDump(conf *Config, db *sql.Conn) error {
+func prepareTableListToDump(tctx *tcontext.Context, conf *Config, db *sql.Conn) error {
 	databases, err := prepareDumpingDatabases(conf, db)
 	if err != nil {
 		return err
@@ -546,7 +552,7 @@ func prepareTableListToDump(conf *Config, db *sql.Conn) error {
 		conf.Tables.Merge(views)
 	}
 
-	filterTables(conf)
+	filterTables(tctx, conf)
 	return nil
 }
 
@@ -634,14 +640,34 @@ func runSteps(d *Dumper, steps ...func(*Dumper) error) error {
 	return nil
 }
 
+func initLogger(d *Dumper) error {
+	conf := d.conf
+	var logger log.Logger
+	if conf.Logger != nil {
+		logger = log.NewAppLogger(conf.Logger)
+	} else {
+		var err error
+		logger, err = log.InitAppLogger(&log.Config{
+			Level:  conf.LogLevel,
+			File:   conf.LogFile,
+			Format: conf.LogFormat,
+		})
+		if err != nil {
+			return errors.Trace(err)
+		}
+	}
+	d.tctx = d.tctx.WithLogger(logger)
+	return nil
+}
+
 // createExternalStore is an initialization step of Dumper.
 func createExternalStore(d *Dumper) error {
-	ctx, conf := d.ctx, d.conf
+	tctx, conf := d.tctx, d.conf
 	b, err := storage.ParseBackend(conf.OutputDirPath, &conf.BackendOptions)
 	if err != nil {
 		return errors.Trace(err)
 	}
-	extStore, err := storage.Create(ctx, b, false)
+	extStore, err := storage.Create(tctx, b, false)
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -654,9 +680,9 @@ func startHTTPService(d *Dumper) error {
 	conf := d.conf
 	if conf.StatusAddr != "" {
 		go func() {
-			err := startDumplingService(conf.StatusAddr)
+			err := startDumplingService(d.tctx, conf.StatusAddr)
 			if err != nil {
-				log.Warn("meet error when stopping dumpling http service", zap.Error(err))
+				d.L().Warn("meet error when stopping dumpling http service", zap.Error(err))
 			}
 		}()
 	}
@@ -682,7 +708,7 @@ func detectServerInfo(d *Dumper) error {
 		conf.ServerInfo = ServerInfoUnknown
 		return err
 	}
-	conf.ServerInfo = ParseServerInfo(versionStr)
+	conf.ServerInfo = ParseServerInfo(d.tctx, versionStr)
 	return nil
 }
 
@@ -705,24 +731,24 @@ func resolveAutoConsistency(d *Dumper) error {
 
 // tidbSetPDClientForGC is an initialization step of Dumper.
 func tidbSetPDClientForGC(d *Dumper) error {
-	ctx, si, pool := d.ctx, d.conf.ServerInfo, d.dbHandle
+	tctx, si, pool := d.tctx, d.conf.ServerInfo, d.dbHandle
 	if si.ServerType != ServerTypeTiDB ||
 		si.ServerVersion == nil ||
 		si.ServerVersion.Compare(*gcSafePointVersion) < 0 {
 		return nil
 	}
-	pdAddrs, err := GetPdAddrs(pool)
+	pdAddrs, err := GetPdAddrs(tctx, pool)
 	if err != nil {
 		return err
 	}
 	if len(pdAddrs) > 0 {
-		doPdGC, err := checkSameCluster(ctx, pool, pdAddrs)
+		doPdGC, err := checkSameCluster(tctx, pool, pdAddrs)
 		if err != nil {
-			log.Warn("meet error while check whether fetched pd addr and TiDB belong to one cluster", zap.Error(err), zap.Strings("pdAddrs", pdAddrs))
+			tctx.L().Warn("meet error while check whether fetched pd addr and TiDB belong to one cluster", zap.Error(err), zap.Strings("pdAddrs", pdAddrs))
 		} else if doPdGC {
-			pdClient, err := pd.NewClientWithContext(ctx, pdAddrs, pd.SecurityOption{})
+			pdClient, err := pd.NewClientWithContext(tctx, pdAddrs, pd.SecurityOption{})
 			if err != nil {
-				log.Warn("create pd client to control GC failed", zap.Error(err), zap.Strings("pdAddrs", pdAddrs))
+				tctx.L().Warn("create pd client to control GC failed", zap.Error(err), zap.Strings("pdAddrs", pdAddrs))
 			}
 			d.tidbPDClientForGC = pdClient
 		}
@@ -734,17 +760,17 @@ func tidbSetPDClientForGC(d *Dumper) error {
 func tidbGetSnapshot(d *Dumper) error {
 	conf, doPdGC := d.conf, d.tidbPDClientForGC != nil
 	consistency := conf.Consistency
-	pool, ctx := d.dbHandle, d.ctx
+	pool, tctx := d.dbHandle, d.tctx
 	if conf.Snapshot == "" && (doPdGC || consistency == "snapshot") {
-		conn, err := pool.Conn(ctx)
+		conn, err := pool.Conn(tctx)
 		if err != nil {
-			log.Warn("cannot get snapshot from TiDB", zap.Error(err))
+			tctx.L().Warn("cannot get snapshot from TiDB", zap.Error(err))
 			return nil
 		}
 		snapshot, err := getSnapshot(conn)
 		_ = conn.Close()
 		if err != nil {
-			log.Warn("cannot get snapshot from TiDB", zap.Error(err))
+			tctx.L().Warn("cannot get snapshot from TiDB", zap.Error(err))
 			return nil
 		}
 		conf.Snapshot = snapshot
@@ -755,16 +781,16 @@ func tidbGetSnapshot(d *Dumper) error {
 
 // tidbStartGCSavepointUpdateService is an initialization step of Dumper.
 func tidbStartGCSavepointUpdateService(d *Dumper) error {
-	ctx, pool, conf := d.ctx, d.dbHandle, d.conf
+	tctx, pool, conf := d.tctx, d.dbHandle, d.conf
 	snapshot, si := conf.Snapshot, conf.ServerInfo
 	if d.tidbPDClientForGC != nil {
 		snapshotTS, err := parseSnapshotToTSO(pool, snapshot)
 		if err != nil {
 			return err
 		}
-		go updateServiceSafePoint(ctx, d.tidbPDClientForGC, defaultDumpGCSafePointTTL, snapshotTS)
+		go updateServiceSafePoint(tctx, d.tidbPDClientForGC, defaultDumpGCSafePointTTL, snapshotTS)
 	} else if si.ServerType == ServerTypeTiDB {
-		log.Warn("If the amount of data to dump is large, criteria: (data more than 60GB or dumped time more than 10 minutes)\n" +
+		tctx.L().Warn("If the amount of data to dump is large, criteria: (data more than 60GB or dumped time more than 10 minutes)\n" +
 			"you'd better adjust the tikv_gc_life_time to avoid export failure due to TiDB GC during the dump process.\n" +
 			"Before dumping: run sql `update mysql.tidb set VARIABLE_VALUE = '720h' where VARIABLE_NAME = 'tikv_gc_life_time';` in tidb.\n" +
 			"After dumping: run sql `update mysql.tidb set VARIABLE_VALUE = '10m' where VARIABLE_NAME = 'tikv_gc_life_time';` in tidb.\n")
@@ -772,30 +798,30 @@ func tidbStartGCSavepointUpdateService(d *Dumper) error {
 	return nil
 }
 
-func updateServiceSafePoint(ctx context.Context, pdClient pd.Client, ttl int64, snapshotTS uint64) {
+func updateServiceSafePoint(tctx *tcontext.Context, pdClient pd.Client, ttl int64, snapshotTS uint64) {
 	updateInterval := time.Duration(ttl/2) * time.Second
 	tick := time.NewTicker(updateInterval)
 	dumplingServiceSafePointID := fmt.Sprintf("%s_%d", dumplingServiceSafePointPrefix, time.Now().UnixNano())
-	log.Info("generate dumpling gc safePoint id", zap.String("id", dumplingServiceSafePointID))
+	tctx.L().Info("generate dumpling gc safePoint id", zap.String("id", dumplingServiceSafePointID))
 
 	for {
-		log.Debug("update PD safePoint limit with ttl",
+		tctx.L().Debug("update PD safePoint limit with ttl",
 			zap.Uint64("safePoint", snapshotTS),
 			zap.Int64("ttl", ttl))
 		for retryCnt := 0; retryCnt <= 10; retryCnt++ {
-			_, err := pdClient.UpdateServiceGCSafePoint(ctx, dumplingServiceSafePointID, ttl, snapshotTS)
+			_, err := pdClient.UpdateServiceGCSafePoint(tctx, dumplingServiceSafePointID, ttl, snapshotTS)
 			if err == nil {
 				break
 			}
-			log.Debug("update PD safePoint failed", zap.Error(err), zap.Int("retryTime", retryCnt))
+			tctx.L().Debug("update PD safePoint failed", zap.Error(err), zap.Int("retryTime", retryCnt))
 			select {
-			case <-ctx.Done():
+			case <-tctx.Done():
 				return
 			case <-time.After(time.Second):
 			}
 		}
 		select {
-		case <-ctx.Done():
+		case <-tctx.Done():
 			return
 		case <-tick.C:
 		}
@@ -826,7 +852,7 @@ func setSessionParam(d *Dumper) error {
 		}
 	}
 	var err error
-	if d.dbHandle, err = resetDBWithSessionParams(pool, conf.GetDSN(""), conf.SessionParams); err != nil {
+	if d.dbHandle, err = resetDBWithSessionParams(d.tctx, pool, conf.GetDSN(""), conf.SessionParams); err != nil {
 		return errors.Trace(err)
 	}
 	return nil
diff --git a/v4/export/http_handler.go b/v4/export/http_handler.go
index 092c624..656e68e 100644
--- a/v4/export/http_handler.go
+++ b/v4/export/http_handler.go
@@ -9,7 +9,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/pingcap/dumpling/v4/log"
+	tcontext "github.com/pingcap/dumpling/v4/context"
 
 	"github.com/pingcap/errors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -19,7 +19,7 @@ import (
 
 var cmuxReadTimeout = 10 * time.Second
 
-func startHTTPServer(lis net.Listener) {
+func startHTTPServer(tctx *tcontext.Context, lis net.Listener) {
 	router := http.NewServeMux()
 	router.Handle("/metrics", promhttp.Handler())
 
@@ -35,11 +35,11 @@ func startHTTPServer(lis net.Listener) {
 	err := httpServer.Serve(lis)
 	err = errors.Cause(err)
 	if err != nil && !isErrNetClosing(err) && err != http.ErrServerClosed {
-		log.Warn("http server return with error", zap.Error(err))
+		tctx.L().Warn("http server return with error", zap.Error(err))
 	}
 }
 
-func startDumplingService(addr string) error {
+func startDumplingService(tctx *tcontext.Context, addr string) error {
 	rootLis, err := net.Listen("tcp", addr)
 	if err != nil {
 		return errors.Annotate(err, "start listening")
@@ -50,7 +50,7 @@ func startDumplingService(addr string) error {
 	m.SetReadTimeout(cmuxReadTimeout) // set a timeout, ref: https://github.com/pingcap/tidb-binlog/pull/352
 
 	httpL := m.Match(cmux.HTTP1Fast())
-	go startHTTPServer(httpL)
+	go startHTTPServer(tctx, httpL)
 
 	err = m.Serve() // start serving, block
 	if err != nil && isErrNetClosing(err) {
diff --git a/v4/export/ir.go b/v4/export/ir.go
index cb9734f..c776bcf 100644
--- a/v4/export/ir.go
+++ b/v4/export/ir.go
@@ -82,14 +82,6 @@ type MetaIR interface {
 	MetaSQL() string
 }
 
-// Logger used for logging when export.
-type Logger interface {
-	Debug(format string, args ...interface{})
-	Info(format string, args ...interface{})
-	Warn(format string, args ...interface{})
-	Error(format string, args ...interface{})
-}
-
 func setTableMetaFromRows(rows *sql.Rows) (TableMeta, error) {
 	tps, err := rows.ColumnTypes()
 	if err != nil {
diff --git a/v4/export/metadata.go b/v4/export/metadata.go
index da0cbeb..ac84e43 100644
--- a/v4/export/metadata.go
+++ b/v4/export/metadata.go
@@ -10,14 +10,15 @@ import (
 	"strings"
 	"time"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/pingcap/br/pkg/storage"
 	"github.com/pingcap/errors"
 	"go.uber.org/zap"
-
-	"github.com/pingcap/dumpling/v4/log"
 )
 
 type globalMetadata struct {
+	tctx            *tcontext.Context
 	buffer          bytes.Buffer
 	afterConnBuffer bytes.Buffer
 	snapshot        string
@@ -34,8 +35,9 @@ const (
 	gtidSetFieldIndex = 4
 )
 
-func newGlobalMetadata(s storage.ExternalStorage, snapshot string) *globalMetadata {
+func newGlobalMetadata(tctx *tcontext.Context, s storage.ExternalStorage, snapshot string) *globalMetadata {
 	return &globalMetadata{
+		tctx:     tctx,
 		storage:  s,
 		buffer:   bytes.Buffer{},
 		snapshot: snapshot,
@@ -58,12 +60,12 @@ func (m *globalMetadata) recordFinishTime(t time.Time) {
 func (m *globalMetadata) recordGlobalMetaData(db *sql.Conn, serverType ServerType, afterConn bool) error { // revive:disable-line:flag-parameter
 	if afterConn {
 		m.afterConnBuffer.Reset()
-		return recordGlobalMetaData(db, &m.afterConnBuffer, serverType, afterConn, m.snapshot)
+		return recordGlobalMetaData(m.tctx, db, &m.afterConnBuffer, serverType, afterConn, m.snapshot)
 	}
-	return recordGlobalMetaData(db, &m.buffer, serverType, afterConn, m.snapshot)
+	return recordGlobalMetaData(m.tctx, db, &m.buffer, serverType, afterConn, m.snapshot)
 }
 
-func recordGlobalMetaData(db *sql.Conn, buffer *bytes.Buffer, serverType ServerType, afterConn bool, snapshot string) error { // revive:disable-line:flag-parameter
+func recordGlobalMetaData(tctx *tcontext.Context, db *sql.Conn, buffer *bytes.Buffer, serverType ServerType, afterConn bool, snapshot string) error { // revive:disable-line:flag-parameter
 	writeMasterStatusHeader := func() {
 		buffer.WriteString("SHOW MASTER STATUS:")
 		if afterConn {
@@ -134,7 +136,7 @@ func recordGlobalMetaData(db *sql.Conn, buffer *bytes.Buffer, serverType ServerT
 		var gtidSet string
 		err = db.QueryRowContext(context.Background(), "SELECT @@global.gtid_binlog_pos").Scan(&gtidSet)
 		if err != nil {
-			log.Error("fail to get gtid for mariaDB", zap.Error(err))
+			tctx.L().Error("fail to get gtid for mariaDB", zap.Error(err))
 		}
 
 		if logFile != "" {
@@ -211,15 +213,15 @@ func recordGlobalMetaData(db *sql.Conn, buffer *bytes.Buffer, serverType ServerT
 	})
 }
 
-func (m *globalMetadata) writeGlobalMetaData(ctx context.Context) error {
+func (m *globalMetadata) writeGlobalMetaData() error {
 	// keep consistent with mydumper. Never compress metadata
-	fileWriter, tearDown, err := buildFileWriter(ctx, m.storage, metadataPath, storage.NoCompression)
+	fileWriter, tearDown, err := buildFileWriter(m.tctx, m.storage, metadataPath, storage.NoCompression)
 	if err != nil {
 		return err
 	}
-	defer tearDown(ctx)
+	defer tearDown(m.tctx)
 
-	return write(ctx, fileWriter, m.String())
+	return write(m.tctx, fileWriter, m.String())
 }
 
 func getValidStr(str []string, idx int) string {
diff --git a/v4/export/metrics.go b/v4/export/metrics.go
index 1d0ed38..b3713bf 100644
--- a/v4/export/metrics.go
+++ b/v4/export/metrics.go
@@ -10,27 +10,43 @@ import (
 )
 
 var (
+	finishedSizeCounter            *prometheus.CounterVec
+	finishedRowsCounter            *prometheus.CounterVec
+	finishedTablesCounter          *prometheus.CounterVec
+	writeTimeHistogram             *prometheus.HistogramVec
+	receiveWriteChunkTimeHistogram *prometheus.HistogramVec
+	errorCount                     *prometheus.CounterVec
+	taskChannelCapacity            *prometheus.GaugeVec
+)
+
+// InitMetricsVector inits metrics vectors.
+// This function must run before RegisterMetrics
+func InitMetricsVector(labels prometheus.Labels) {
+	labelNames := make([]string, 0, len(labels))
+	for name := range labels {
+		labelNames = append(labelNames, name)
+	}
 	finishedSizeCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "dumpling",
 			Subsystem: "dump",
 			Name:      "finished_size",
 			Help:      "counter for dumpling finished file size",
-		}, []string{})
+		}, labelNames)
 	finishedRowsCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "dumpling",
 			Subsystem: "dump",
 			Name:      "finished_rows",
 			Help:      "counter for dumpling finished rows",
-		}, []string{})
+		}, labelNames)
 	finishedTablesCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "dumpling",
 			Subsystem: "dump",
 			Name:      "finished_tables",
 			Help:      "counter for dumpling finished tables",
-		}, []string{})
+		}, labelNames)
 	writeTimeHistogram = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
 			Namespace: "dumpling",
@@ -38,7 +54,7 @@ var (
 			Name:      "write_duration_time",
 			Help:      "Bucketed histogram of write time (s) of files",
 			Buckets:   prometheus.ExponentialBuckets(0.00005, 2, 20),
-		}, []string{})
+		}, labelNames)
 	receiveWriteChunkTimeHistogram = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
 			Namespace: "dumpling",
@@ -46,25 +62,28 @@ var (
 			Name:      "receive_chunk_duration_time",
 			Help:      "Bucketed histogram of write time (s) of files",
 			Buckets:   prometheus.ExponentialBuckets(0.00005, 2, 20),
-		}, []string{})
+		}, labelNames)
 	errorCount = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "dumpling",
 			Subsystem: "dump",
 			Name:      "error_count",
 			Help:      "Total error count during dumping progress",
-		}, []string{})
-	taskChannelCapacity = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
+		}, labelNames)
+	taskChannelCapacity = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
 			Namespace: "dumpling",
 			Subsystem: "dump",
 			Name:      "channel_capacity",
 			Help:      "The task channel capacity during dumping progress",
-		}, []string{})
-)
+		}, labelNames)
+}
 
 // RegisterMetrics registers metrics.
 func RegisterMetrics(registry *prometheus.Registry) {
+	if finishedSizeCounter == nil {
+		return
+	}
 	registry.MustRegister(finishedSizeCounter)
 	registry.MustRegister(finishedRowsCounter)
 	registry.MustRegister(finishedTablesCounter)
@@ -76,6 +95,9 @@ func RegisterMetrics(registry *prometheus.Registry) {
 
 // RemoveLabelValuesWithTaskInMetrics removes metrics of specified labels.
 func RemoveLabelValuesWithTaskInMetrics(labels prometheus.Labels) {
+	if finishedSizeCounter == nil {
+		return
+	}
 	finishedSizeCounter.Delete(labels)
 	finishedRowsCounter.Delete(labels)
 	finishedTablesCounter.Delete(labels)
@@ -86,10 +108,62 @@ func RemoveLabelValuesWithTaskInMetrics(labels prometheus.Labels) {
 }
 
 // ReadCounter reports the current value of the counter.
-func ReadCounter(counter prometheus.Counter) float64 {
+func ReadCounter(counterVec *prometheus.CounterVec, labels prometheus.Labels) float64 {
+	if counterVec == nil {
+		return math.NaN()
+	}
+	counter := counterVec.With(labels)
 	var metric dto.Metric
 	if err := counter.Write(&metric); err != nil {
 		return math.NaN()
 	}
 	return metric.Counter.GetValue()
 }
+
+// AddCounter adds a counter.
+func AddCounter(counterVec *prometheus.CounterVec, labels prometheus.Labels, v float64) {
+	if counterVec == nil {
+		return
+	}
+	counterVec.With(labels).Add(v)
+}
+
+// IncCounter incs a counter.
+func IncCounter(counterVec *prometheus.CounterVec, labels prometheus.Labels) {
+	if counterVec == nil {
+		return
+	}
+	counterVec.With(labels).Inc()
+}
+
+// ObserveHistogram observes a histogram
+func ObserveHistogram(histogramVec *prometheus.HistogramVec, labels prometheus.Labels, v float64) {
+	if histogramVec == nil {
+		return
+	}
+	histogramVec.With(labels).Observe(v)
+}
+
+// AddGauge adds a gauge
+func AddGauge(gaugeVec *prometheus.GaugeVec, labels prometheus.Labels, v float64) {
+	if gaugeVec == nil {
+		return
+	}
+	gaugeVec.With(labels).Add(v)
+}
+
+// IncGauge incs a gauge
+func IncGauge(gaugeVec *prometheus.GaugeVec, labels prometheus.Labels) {
+	if gaugeVec == nil {
+		return
+	}
+	gaugeVec.With(labels).Inc()
+}
+
+// DecGauge decs a gauge
+func DecGauge(gaugeVec *prometheus.GaugeVec, labels prometheus.Labels) {
+	if gaugeVec == nil {
+		return
+	}
+	gaugeVec.With(labels).Dec()
+}
diff --git a/v4/export/prepare.go b/v4/export/prepare.go
index 00bbbf0..362ff7c 100644
--- a/v4/export/prepare.go
+++ b/v4/export/prepare.go
@@ -10,8 +10,6 @@ import (
 	"text/template"
 
 	"github.com/pingcap/errors"
-
-	"github.com/pingcap/dumpling/v4/log"
 )
 
 const (
@@ -99,12 +97,10 @@ func prepareDumpingDatabases(conf *Config, db *sql.Conn) ([]string, error) {
 }
 
 func listAllTables(db *sql.Conn, databaseNames []string) (DatabaseTables, error) {
-	log.Debug("list all the tables")
 	return ListAllDatabasesTables(db, databaseNames, TableTypeBase)
 }
 
 func listAllViews(db *sql.Conn, databaseNames []string) (DatabaseTables, error) {
-	log.Debug("list all the views")
 	return ListAllDatabasesTables(db, databaseNames, TableTypeView)
 }
 
diff --git a/v4/export/retry.go b/v4/export/retry.go
index 089d572..753f6f7 100644
--- a/v4/export/retry.go
+++ b/v4/export/retry.go
@@ -6,12 +6,12 @@ import (
 	"strings"
 	"time"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb-tools/pkg/dbutil"
 	"go.uber.org/zap"
-
-	"github.com/pingcap/dumpling/v4/log"
 )
 
 const (
@@ -64,14 +64,16 @@ func (b *dumpChunkBackoffer) Attempt() int {
 	return b.attempt
 }
 
-func newLockTablesBackoffer(blockList map[string]map[string]interface{}) *lockTablesBackoffer {
+func newLockTablesBackoffer(tctx *tcontext.Context, blockList map[string]map[string]interface{}) *lockTablesBackoffer {
 	return &lockTablesBackoffer{
+		tctx:      tctx,
 		attempt:   lockTablesRetryTime,
 		blockList: blockList,
 	}
 }
 
 type lockTablesBackoffer struct {
+	tctx      *tcontext.Context
 	attempt   int
 	blockList map[string]map[string]interface{}
 }
@@ -82,7 +84,7 @@ func (b *lockTablesBackoffer) NextBackoff(err error) time.Duration {
 		b.attempt--
 		db, table, err := getTableFromMySQLError(mysqlErr.Message)
 		if err != nil {
-			log.Error("retry lock tables meet error", zap.Error(err))
+			b.tctx.L().Error("retry lock tables meet error", zap.Error(err))
 			b.attempt = 0
 			return 0
 		}
diff --git a/v4/export/sql.go b/v4/export/sql.go
index d2f7d1f..398c492 100644
--- a/v4/export/sql.go
+++ b/v4/export/sql.go
@@ -12,10 +12,10 @@ import (
 	"strconv"
 	"strings"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/pingcap/errors"
 	"go.uber.org/zap"
-
-	"github.com/pingcap/dumpling/v4/log"
 )
 
 // ShowDatabases shows the databases of a database server.
@@ -204,41 +204,6 @@ func SelectAllFromTable(conf *Config, db *sql.Conn, database, table string) (Tab
 	}, nil
 }
 
-// SelectFromSQL dumps data serialized from a specified SQL
-func SelectFromSQL(conf *Config, conn *sql.Conn) (TableMeta, TableDataIR, error) {
-	log.Info("dump data from sql", zap.String("sql", conf.SQL))
-	rows, err := conn.QueryContext(context.Background(), conf.SQL)
-	if err != nil {
-		return nil, nil, errors.Annotatef(err, "sql: %s", conf.SQL)
-	}
-	defer rows.Close()
-	colTypes, err := rows.ColumnTypes()
-	if err != nil {
-		return nil, nil, errors.Annotatef(err, "sql: %s", conf.SQL)
-	}
-	cols, err := rows.Columns()
-	if err != nil {
-		return nil, nil, errors.Annotatef(err, "sql: %s", conf.SQL)
-	}
-	for i := range cols {
-		cols[i] = wrapBackTicks(cols[i])
-	}
-	meta := &tableMeta{
-		colTypes:      colTypes,
-		selectedField: strings.Join(cols, ","),
-		specCmts: []string{
-			"/*!40101 SET NAMES binary*/;",
-		},
-	}
-	data := &tableData{
-		query:  conf.SQL,
-		colLen: len(cols),
-	}
-	rows.Close()
-
-	return meta, data, rows.Err()
-}
-
 func buildSelectQuery(database, table string, fields string, where string, orderByClause string) string {
 	var query strings.Builder
 	query.WriteString("SELECT ")
@@ -465,11 +430,11 @@ func GetSpecifiedColumnValue(rows *sql.Rows, columnName string) ([]string, error
 }
 
 // GetPdAddrs gets PD address from TiDB
-func GetPdAddrs(db *sql.DB) ([]string, error) {
+func GetPdAddrs(tctx *tcontext.Context, db *sql.DB) ([]string, error) {
 	query := "SELECT * FROM information_schema.cluster_info where type = 'pd';"
-	rows, err := db.Query(query)
+	rows, err := db.QueryContext(tctx, query)
 	if err != nil {
-		log.Warn("can't execute query from db",
+		tctx.L().Warn("can't execute query from db",
 			zap.String("query", query), zap.Error(err))
 		return []string{}, errors.Annotatef(err, "sql: %s", query)
 	}
@@ -478,11 +443,11 @@ func GetPdAddrs(db *sql.DB) ([]string, error) {
 }
 
 // GetTiDBDDLIDs gets DDL IDs from TiDB
-func GetTiDBDDLIDs(db *sql.DB) ([]string, error) {
+func GetTiDBDDLIDs(tctx *tcontext.Context, db *sql.DB) ([]string, error) {
 	query := "SELECT * FROM information_schema.tidb_servers_info;"
-	rows, err := db.Query(query)
+	rows, err := db.QueryContext(tctx, query)
 	if err != nil {
-		log.Warn("can't execute query from db",
+		tctx.L().Warn("can't execute query from db",
 			zap.String("query", query), zap.Error(err))
 		return []string{}, errors.Annotatef(err, "sql: %s", query)
 	}
@@ -514,14 +479,14 @@ func isUnknownSystemVariableErr(err error) bool {
 	return strings.Contains(err.Error(), "Unknown system variable")
 }
 
-func resetDBWithSessionParams(db *sql.DB, dsn string, params map[string]interface{}) (*sql.DB, error) {
+func resetDBWithSessionParams(tctx *tcontext.Context, db *sql.DB, dsn string, params map[string]interface{}) (*sql.DB, error) {
 	support := make(map[string]interface{})
 	for k, v := range params {
 		s := fmt.Sprintf("SET SESSION %s = ?", k)
-		_, err := db.Exec(s, v)
+		_, err := db.ExecContext(tctx, s, v)
 		if err != nil {
 			if isUnknownSystemVariableErr(err) {
-				log.Info("session variable is not supported by db", zap.String("variable", k), zap.Reflect("value", v))
+				tctx.L().Info("session variable is not supported by db", zap.String("variable", k), zap.Reflect("value", v))
 				continue
 			}
 			return nil, errors.Trace(err)
@@ -727,7 +692,7 @@ func pickupPossibleField(dbName, tableName string, db *sql.Conn, conf *Config) (
 	return fieldName, nil
 }
 
-func estimateCount(dbName, tableName string, db *sql.Conn, field string, conf *Config) uint64 {
+func estimateCount(tctx *tcontext.Context, dbName, tableName string, db *sql.Conn, field string, conf *Config) uint64 {
 	query := fmt.Sprintf("EXPLAIN SELECT `%s` FROM `%s`.`%s`", escapeString(field), escapeString(dbName), escapeString(tableName))
 
 	if conf.Where != "" {
@@ -735,7 +700,7 @@ func estimateCount(dbName, tableName string, db *sql.Conn, field string, conf *C
 		query += conf.Where
 	}
 
-	estRows := detectEstimateRows(db, query, []string{"rows", "estRows", "count"})
+	estRows := detectEstimateRows(tctx, db, query, []string{"rows", "estRows", "count"})
 	/* tidb results field name is estRows (before 4.0.0-beta.2: count)
 		+-----------------------+----------+-----------+---------------------------------------------------------+
 		| id                    | estRows  | task      | access object | operator info                           |
@@ -764,10 +729,10 @@ func estimateCount(dbName, tableName string, db *sql.Conn, field string, conf *C
 	return 0
 }
 
-func detectEstimateRows(db *sql.Conn, query string, fieldNames []string) uint64 {
-	rows, err := db.QueryContext(context.Background(), query)
+func detectEstimateRows(tctx *tcontext.Context, db *sql.Conn, query string, fieldNames []string) uint64 {
+	rows, err := db.QueryContext(tctx, query)
 	if err != nil {
-		log.Warn("can't execute query from db",
+		tctx.L().Warn("can't execute query from db",
 			zap.String("query", query), zap.Error(err))
 		return 0
 	}
@@ -775,13 +740,13 @@ func detectEstimateRows(db *sql.Conn, query string, fieldNames []string) uint64
 	rows.Next()
 	columns, err := rows.Columns()
 	if err != nil {
-		log.Warn("can't get columns from db",
+		tctx.L().Warn("can't get columns from db",
 			zap.String("query", query), zap.Error(err))
 		return 0
 	}
 	err = rows.Err()
 	if err != nil {
-		log.Warn("rows meet some error during the query",
+		tctx.L().Warn("rows meet some error during the query",
 			zap.String("query", query), zap.Error(err))
 		return 0
 	}
@@ -802,14 +767,14 @@ found:
 	}
 	err = rows.Scan(addr...)
 	if err != nil || fieldIndex < 0 {
-		log.Warn("can't get estimate count from db",
+		tctx.L().Warn("can't get estimate count from db",
 			zap.String("query", query), zap.Error(err))
 		return 0
 	}
 
 	estRows, err := strconv.ParseFloat(oneRow[fieldIndex].String, 64)
 	if err != nil {
-		log.Warn("can't get parse rows from db",
+		tctx.L().Warn("can't get parse rows from db",
 			zap.String("query", query), zap.Error(err))
 		return 0
 	}
diff --git a/v4/export/status.go b/v4/export/status.go
index 5c8ad54..26048f4 100644
--- a/v4/export/status.go
+++ b/v4/export/status.go
@@ -3,19 +3,18 @@
 package export
 
 import (
-	"context"
 	"fmt"
 	"time"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/docker/go-units"
 	"go.uber.org/zap"
-
-	"github.com/pingcap/dumpling/v4/log"
 )
 
 const logProgressTick = 2 * time.Minute
 
-func (d *Dumper) runLogProgress(ctx context.Context) {
+func (d *Dumper) runLogProgress(tctx *tcontext.Context) {
 	conf := d.conf
 	totalTables := float64(calculateTableCount(conf.Tables))
 	logProgressTicker := time.NewTicker(logProgressTick)
@@ -24,17 +23,17 @@ func (d *Dumper) runLogProgress(ctx context.Context) {
 	defer logProgressTicker.Stop()
 	for {
 		select {
-		case <-ctx.Done():
-			log.Debug("stopping log progress")
+		case <-tctx.Done():
+			tctx.L().Debug("stopping log progress")
 			return
 		case <-logProgressTicker.C:
 			nanoseconds := float64(time.Since(lastCheckpoint).Nanoseconds())
 
-			completedTables := ReadCounter(finishedTablesCounter.With(conf.Labels))
-			finishedBytes := ReadCounter(finishedSizeCounter.With(conf.Labels))
-			finishedRows := ReadCounter(finishedRowsCounter.With(conf.Labels))
+			completedTables := ReadCounter(finishedTablesCounter, conf.Labels)
+			finishedBytes := ReadCounter(finishedSizeCounter, conf.Labels)
+			finishedRows := ReadCounter(finishedRowsCounter, conf.Labels)
 
-			log.Info("progress",
+			tctx.L().Info("progress",
 				zap.String("tables", fmt.Sprintf("%.0f/%.0f (%.1f%%)", completedTables, totalTables, completedTables/totalTables*100)),
 				zap.String("finished rows", fmt.Sprintf("%.0f", finishedRows)),
 				zap.String("finished size", units.HumanSize(finishedBytes)),
diff --git a/v4/export/util.go b/v4/export/util.go
index 7157a0d..6df177c 100644
--- a/v4/export/util.go
+++ b/v4/export/util.go
@@ -9,6 +9,8 @@ import (
 	"strings"
 	"time"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/pingcap/errors"
 	"go.etcd.io/etcd/clientv3"
 )
@@ -31,7 +33,7 @@ func getPdDDLIDs(pCtx context.Context, cli *clientv3.Client) ([]string, error) {
 	return pdDDLIds, nil
 }
 
-func checkSameCluster(ctx context.Context, db *sql.DB, pdAddrs []string) (bool, error) {
+func checkSameCluster(tctx *tcontext.Context, db *sql.DB, pdAddrs []string) (bool, error) {
 	cli, err := clientv3.New(clientv3.Config{
 		Endpoints:   pdAddrs,
 		DialTimeout: defaultEtcdDialTimeOut,
@@ -39,11 +41,11 @@ func checkSameCluster(ctx context.Context, db *sql.DB, pdAddrs []string) (bool,
 	if err != nil {
 		return false, errors.Trace(err)
 	}
-	tidbDDLIDs, err := GetTiDBDDLIDs(db)
+	tidbDDLIDs, err := GetTiDBDDLIDs(tctx, db)
 	if err != nil {
 		return false, err
 	}
-	pdDDLIDs, err := getPdDDLIDs(ctx, cli)
+	pdDDLIDs, err := getPdDDLIDs(tctx, cli)
 	if err != nil {
 		return false, err
 	}
diff --git a/v4/export/writer.go b/v4/export/writer.go
index 9f7766c..1387251 100644
--- a/v4/export/writer.go
+++ b/v4/export/writer.go
@@ -4,25 +4,24 @@ package export
 
 import (
 	"bytes"
-	"context"
 	"database/sql"
 	"fmt"
 	"strings"
 	"text/template"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/pingcap/br/pkg/storage"
 	"github.com/pingcap/br/pkg/utils"
 	"github.com/pingcap/errors"
 	"go.uber.org/zap"
-
-	"github.com/pingcap/dumpling/v4/log"
 )
 
 // Writer is the abstraction that keep pulling data from database and write to files.
 // Every writer owns a snapshot connection, and will try to get a task from task stream chan and work on it.
 type Writer struct {
 	id         int64
-	ctx        context.Context
+	tctx       *tcontext.Context
 	conf       *Config
 	conn       *sql.Conn
 	extStorage storage.ExternalStorage
@@ -36,10 +35,10 @@ type Writer struct {
 }
 
 // NewWriter returns a new Writer with given configurations
-func NewWriter(ctx context.Context, id int64, config *Config, conn *sql.Conn, externalStore storage.ExternalStorage) *Writer {
+func NewWriter(tctx *tcontext.Context, id int64, config *Config, conn *sql.Conn, externalStore storage.ExternalStorage) *Writer {
 	sw := &Writer{
 		id:                  id,
-		ctx:                 ctx,
+		tctx:                tctx,
 		conf:                config,
 		conn:                conn,
 		extStorage:          externalStore,
@@ -74,8 +73,8 @@ func countTotalTask(writers []*Writer) int {
 func (w *Writer) run(taskStream <-chan Task) error {
 	for {
 		select {
-		case <-w.ctx.Done():
-			log.Warn("context has been done, the writer will exit",
+		case <-w.tctx.Done():
+			w.tctx.L().Warn("context has been done, the writer will exit",
 				zap.Int64("writer ID", w.id))
 			return nil
 		case task, ok := <-taskStream:
@@ -110,34 +109,34 @@ func (w *Writer) handleTask(task Task) error {
 		}
 		return nil
 	default:
-		log.Warn("unsupported writer task type", zap.String("type", fmt.Sprintf("%T", t)))
+		w.tctx.L().Warn("unsupported writer task type", zap.String("type", fmt.Sprintf("%T", t)))
 		return nil
 	}
 }
 
 // WriteDatabaseMeta writes database meta to a file
 func (w *Writer) WriteDatabaseMeta(db, createSQL string) error {
-	ctx, conf := w.ctx, w.conf
+	tctx, conf := w.tctx, w.conf
 	fileName, err := (&outputFileNamer{DB: db}).render(conf.OutputFileTemplate, outputFileTemplateSchema)
 	if err != nil {
 		return err
 	}
-	return writeMetaToFile(ctx, db, createSQL, w.extStorage, fileName+".sql", conf.CompressType)
+	return writeMetaToFile(tctx, db, createSQL, w.extStorage, fileName+".sql", conf.CompressType)
 }
 
 // WriteTableMeta writes table meta to a file
 func (w *Writer) WriteTableMeta(db, table, createSQL string) error {
-	ctx, conf := w.ctx, w.conf
+	tctx, conf := w.tctx, w.conf
 	fileName, err := (&outputFileNamer{DB: db, Table: table}).render(conf.OutputFileTemplate, outputFileTemplateTable)
 	if err != nil {
 		return err
 	}
-	return writeMetaToFile(ctx, db, createSQL, w.extStorage, fileName+".sql", conf.CompressType)
+	return writeMetaToFile(tctx, db, createSQL, w.extStorage, fileName+".sql", conf.CompressType)
 }
 
 // WriteViewMeta writes view meta to a file
 func (w *Writer) WriteViewMeta(db, view, createTableSQL, createViewSQL string) error {
-	ctx, conf := w.ctx, w.conf
+	tctx, conf := w.tctx, w.conf
 	fileNameTable, err := (&outputFileNamer{DB: db, Table: view}).render(conf.OutputFileTemplate, outputFileTemplateTable)
 	if err != nil {
 		return err
@@ -146,27 +145,27 @@ func (w *Writer) WriteViewMeta(db, view, createTableSQL, createViewSQL string) e
 	if err != nil {
 		return err
 	}
-	err = writeMetaToFile(ctx, db, createTableSQL, w.extStorage, fileNameTable+".sql", conf.CompressType)
+	err = writeMetaToFile(tctx, db, createTableSQL, w.extStorage, fileNameTable+".sql", conf.CompressType)
 	if err != nil {
 		return err
 	}
-	return writeMetaToFile(ctx, db, createViewSQL, w.extStorage, fileNameView+".sql", conf.CompressType)
+	return writeMetaToFile(tctx, db, createViewSQL, w.extStorage, fileNameView+".sql", conf.CompressType)
 }
 
 // WriteTableData writes table data to a file with retry
 func (w *Writer) WriteTableData(meta TableMeta, ir TableDataIR, currentChunk int) error {
-	ctx, conf, conn := w.ctx, w.conf, w.conn
+	tctx, conf, conn := w.tctx, w.conf, w.conn
 	retryTime := 0
 	var lastErr error
-	return utils.WithRetry(ctx, func() (err error) {
+	return utils.WithRetry(tctx, func() (err error) {
 		defer func() {
 			lastErr = err
 			if err != nil {
-				errorCount.With(conf.Labels).Inc()
+				IncCounter(errorCount, conf.Labels)
 			}
 		}()
 		retryTime++
-		log.Debug("trying to dump table chunk", zap.Int("retryTime", retryTime), zap.String("db", meta.DatabaseName()),
+		tctx.L().Debug("trying to dump table chunk", zap.Int("retryTime", retryTime), zap.String("db", meta.DatabaseName()),
 			zap.String("table", meta.TableName()), zap.Int("chunkIndex", currentChunk), zap.NamedError("lastError", lastErr))
 		// don't rebuild connection when dump for the first time
 		if retryTime > 1 {
@@ -176,7 +175,7 @@ func (w *Writer) WriteTableData(meta TableMeta, ir TableDataIR, currentChunk int
 				return
 			}
 		}
-		err = ir.Start(ctx, conn)
+		err = ir.Start(tctx, conn)
 		if err != nil {
 			return
 		}
@@ -187,11 +186,11 @@ func (w *Writer) WriteTableData(meta TableMeta, ir TableDataIR, currentChunk int
 			}
 		}
 		defer ir.Close()
-		return w.tryToWriteTableData(ctx, meta, ir, currentChunk)
+		return w.tryToWriteTableData(tctx, meta, ir, currentChunk)
 	}, newDumpChunkBackoffer(canRebuildConn(conf.Consistency, conf.TransactionalConsistency)))
 }
 
-func (w *Writer) tryToWriteTableData(ctx context.Context, meta TableMeta, ir TableDataIR, curChkIdx int) error {
+func (w *Writer) tryToWriteTableData(tctx *tcontext.Context, meta TableMeta, ir TableDataIR, curChkIdx int) error {
 	conf, format := w.conf, w.fileFmt
 	namer := newOutputFileNamer(meta, curChkIdx, conf.Rows != UnspecifiedSize, conf.FileSize != UnspecifiedSize)
 	fileName, err := namer.NextName(conf.OutputFileTemplate, w.fileFmt.Extension())
@@ -200,9 +199,9 @@ func (w *Writer) tryToWriteTableData(ctx context.Context, meta TableMeta, ir Tab
 	}
 
 	for {
-		fileWriter, tearDown := buildInterceptFileWriter(ctx, w.extStorage, fileName, conf.CompressType)
-		err = format.WriteInsert(ctx, conf, meta, ir, fileWriter)
-		tearDown(ctx)
+		fileWriter, tearDown := buildInterceptFileWriter(tctx, w.extStorage, fileName, conf.CompressType)
+		err = format.WriteInsert(tctx, conf, meta, ir, fileWriter)
+		tearDown(tctx)
 		if err != nil {
 			return err
 		}
@@ -222,14 +221,14 @@ func (w *Writer) tryToWriteTableData(ctx context.Context, meta TableMeta, ir Tab
 	return nil
 }
 
-func writeMetaToFile(ctx context.Context, target, metaSQL string, s storage.ExternalStorage, path string, compressType storage.CompressType) error {
-	fileWriter, tearDown, err := buildFileWriter(ctx, s, path, compressType)
+func writeMetaToFile(tctx *tcontext.Context, target, metaSQL string, s storage.ExternalStorage, path string, compressType storage.CompressType) error {
+	fileWriter, tearDown, err := buildFileWriter(tctx, s, path, compressType)
 	if err != nil {
 		return errors.Trace(err)
 	}
-	defer tearDown(ctx)
+	defer tearDown(tctx)
 
-	return WriteMeta(ctx, &metaData{
+	return WriteMeta(tctx, &metaData{
 		target:  target,
 		metaSQL: metaSQL,
 		specCmts: []string{
diff --git a/v4/export/writer_util.go b/v4/export/writer_util.go
index 34e7b96..0697f57 100755
--- a/v4/export/writer_util.go
+++ b/v4/export/writer_util.go
@@ -12,14 +12,14 @@ import (
 	"sync"
 	"time"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/pingcap/br/pkg/storage"
 	"github.com/pingcap/br/pkg/summary"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
-
-	"github.com/pingcap/dumpling/v4/log"
 )
 
 const lengthLimit = 1048576
@@ -59,7 +59,7 @@ func newWriterPipe(w storage.ExternalFileWriter, fileSizeLimit, statementSizeLim
 	}
 }
 
-func (b *writerPipe) Run(ctx context.Context) {
+func (b *writerPipe) Run(tctx *tcontext.Context) {
 	defer close(b.closed)
 	var errOccurs bool
 	receiveChunkTime := time.Now()
@@ -72,11 +72,11 @@ func (b *writerPipe) Run(ctx context.Context) {
 			if errOccurs {
 				continue
 			}
-			receiveWriteChunkTimeHistogram.With(b.labels).Observe(time.Since(receiveChunkTime).Seconds())
+			ObserveHistogram(receiveWriteChunkTimeHistogram, b.labels, time.Since(receiveChunkTime).Seconds())
 			receiveChunkTime = time.Now()
-			err := writeBytes(ctx, b.w, s.Bytes())
-			writeTimeHistogram.With(b.labels).Observe(time.Since(receiveChunkTime).Seconds())
-			finishedSizeCounter.With(b.labels).Add(float64(s.Len()))
+			err := writeBytes(tctx, b.w, s.Bytes())
+			ObserveHistogram(writeTimeHistogram, b.labels, time.Since(receiveChunkTime).Seconds())
+			AddCounter(finishedSizeCounter, b.labels, float64(s.Len()))
 			b.finishedFileSize += uint64(s.Len())
 			s.Reset()
 			pool.Put(s)
@@ -85,7 +85,7 @@ func (b *writerPipe) Run(ctx context.Context) {
 				b.errCh <- err
 			}
 			receiveChunkTime = time.Now()
-		case <-ctx.Done():
+		case <-tctx.Done():
 			return
 		}
 	}
@@ -115,26 +115,26 @@ func (b *writerPipe) ShouldSwitchStatement() bool {
 }
 
 // WriteMeta writes MetaIR to a storage.ExternalFileWriter
-func WriteMeta(ctx context.Context, meta MetaIR, w storage.ExternalFileWriter) error {
-	log.Debug("start dumping meta data", zap.String("target", meta.TargetName()))
+func WriteMeta(tctx *tcontext.Context, meta MetaIR, w storage.ExternalFileWriter) error {
+	tctx.L().Debug("start dumping meta data", zap.String("target", meta.TargetName()))
 
 	specCmtIter := meta.SpecialComments()
 	for specCmtIter.HasNext() {
-		if err := write(ctx, w, fmt.Sprintf("%s\n", specCmtIter.Next())); err != nil {
+		if err := write(tctx, w, fmt.Sprintf("%s\n", specCmtIter.Next())); err != nil {
 			return err
 		}
 	}
 
-	if err := write(ctx, w, meta.MetaSQL()); err != nil {
+	if err := write(tctx, w, meta.MetaSQL()); err != nil {
 		return err
 	}
 
-	log.Debug("finish dumping meta data", zap.String("target", meta.TargetName()))
+	tctx.L().Debug("finish dumping meta data", zap.String("target", meta.TargetName()))
 	return nil
 }
 
 // WriteInsert writes TableDataIR to a storage.ExternalFileWriter in sql type
-func WriteInsert(pCtx context.Context, cfg *Config, meta TableMeta, tblIR TableDataIR, w storage.ExternalFileWriter) error {
+func WriteInsert(pCtx *tcontext.Context, cfg *Config, meta TableMeta, tblIR TableDataIR, w storage.ExternalFileWriter) error {
 	fileRowIter := tblIR.Rows()
 	if !fileRowIter.HasNext() {
 		return nil
@@ -148,7 +148,7 @@ func WriteInsert(pCtx context.Context, cfg *Config, meta TableMeta, tblIR TableD
 	wp := newWriterPipe(w, cfg.FileSize, cfg.StatementSize, cfg.Labels)
 
 	// use context.Background here to make sure writerPipe can deplete all the chunks in pipeline
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := tcontext.Background().WithLogger(pCtx.L()).WithCancel()
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
@@ -197,7 +197,7 @@ func WriteInsert(pCtx context.Context, cfg *Config, meta TableMeta, tblIR TableD
 			lastBfSize := bf.Len()
 			if selectedField != "" {
 				if err = fileRowIter.Decode(row); err != nil {
-					log.Error("fail to scan from sql.Row", zap.Error(err))
+					pCtx.L().Error("fail to scan from sql.Row", zap.Error(err))
 					return errors.Trace(err)
 				}
 				row.WriteToBuffer(bf, escapeBackslash)
@@ -228,7 +228,7 @@ func WriteInsert(pCtx context.Context, cfg *Config, meta TableMeta, tblIR TableD
 					if bfCap := bf.Cap(); bfCap < lengthLimit {
 						bf.Grow(lengthLimit - bfCap)
 					}
-					finishedRowsCounter.With(cfg.Labels).Add(float64(counter - lastCounter))
+					AddCounter(finishedRowsCounter, cfg.Labels, float64(counter-lastCounter))
 					lastCounter = counter
 				}
 			}
@@ -241,7 +241,7 @@ func WriteInsert(pCtx context.Context, cfg *Config, meta TableMeta, tblIR TableD
 			break
 		}
 	}
-	log.Debug("finish dumping table(chunk)",
+	pCtx.L().Debug("finish dumping table(chunk)",
 		zap.String("database", meta.DatabaseName()),
 		zap.String("table", meta.TableName()),
 		zap.Uint64("total rows", counter))
@@ -252,7 +252,7 @@ func WriteInsert(pCtx context.Context, cfg *Config, meta TableMeta, tblIR TableD
 	<-wp.closed
 	summary.CollectSuccessUnit(summary.TotalBytes, 1, wp.finishedFileSize)
 	summary.CollectSuccessUnit("total rows", 1, counter)
-	finishedRowsCounter.With(cfg.Labels).Add(float64(counter - lastCounter))
+	AddCounter(finishedRowsCounter, cfg.Labels, float64(counter-lastCounter))
 	if err = fileRowIter.Error(); err != nil {
 		return errors.Trace(err)
 	}
@@ -260,7 +260,7 @@ func WriteInsert(pCtx context.Context, cfg *Config, meta TableMeta, tblIR TableD
 }
 
 // WriteInsertInCsv writes TableDataIR to a storage.ExternalFileWriter in csv type
-func WriteInsertInCsv(pCtx context.Context, cfg *Config, meta TableMeta, tblIR TableDataIR, w storage.ExternalFileWriter) error {
+func WriteInsertInCsv(pCtx *tcontext.Context, cfg *Config, meta TableMeta, tblIR TableDataIR, w storage.ExternalFileWriter) error {
 	fileRowIter := tblIR.Rows()
 	if !fileRowIter.HasNext() {
 		return nil
@@ -279,7 +279,7 @@ func WriteInsertInCsv(pCtx context.Context, cfg *Config, meta TableMeta, tblIR T
 	}
 
 	// use context.Background here to make sure writerPipe can deplete all the chunks in pipeline
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := tcontext.Background().WithLogger(pCtx.L()).WithCancel()
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
@@ -317,7 +317,7 @@ func WriteInsertInCsv(pCtx context.Context, cfg *Config, meta TableMeta, tblIR T
 		lastBfSize := bf.Len()
 		if selectedFields != "" {
 			if err = fileRowIter.Decode(row); err != nil {
-				log.Error("fail to scan from sql.Row", zap.Error(err))
+				pCtx.L().Error("fail to scan from sql.Row", zap.Error(err))
 				return errors.Trace(err)
 			}
 			row.WriteToBufferInCsv(bf, escapeBackslash, opt)
@@ -337,7 +337,7 @@ func WriteInsertInCsv(pCtx context.Context, cfg *Config, meta TableMeta, tblIR T
 				if bfCap := bf.Cap(); bfCap < lengthLimit {
 					bf.Grow(lengthLimit - bfCap)
 				}
-				finishedRowsCounter.With(cfg.Labels).Add(float64(counter - lastCounter))
+				AddCounter(finishedRowsCounter, cfg.Labels, float64(counter-lastCounter))
 				lastCounter = counter
 			}
 		}
@@ -348,7 +348,7 @@ func WriteInsertInCsv(pCtx context.Context, cfg *Config, meta TableMeta, tblIR T
 		}
 	}
 
-	log.Debug("finish dumping table(chunk)",
+	pCtx.L().Debug("finish dumping table(chunk)",
 		zap.String("database", meta.DatabaseName()),
 		zap.String("table", meta.TableName()),
 		zap.Uint64("total rows", counter))
@@ -359,37 +359,37 @@ func WriteInsertInCsv(pCtx context.Context, cfg *Config, meta TableMeta, tblIR T
 	<-wp.closed
 	summary.CollectSuccessUnit(summary.TotalBytes, 1, wp.finishedFileSize)
 	summary.CollectSuccessUnit("total rows", 1, counter)
-	finishedRowsCounter.With(cfg.Labels).Add(float64(counter - lastCounter))
+	AddCounter(finishedRowsCounter, cfg.Labels, float64(counter-lastCounter))
 	if err = fileRowIter.Error(); err != nil {
 		return errors.Trace(err)
 	}
 	return wp.Error()
 }
 
-func write(ctx context.Context, writer storage.ExternalFileWriter, str string) error {
-	_, err := writer.Write(ctx, []byte(str))
+func write(tctx *tcontext.Context, writer storage.ExternalFileWriter, str string) error {
+	_, err := writer.Write(tctx, []byte(str))
 	if err != nil {
 		// str might be very long, only output the first 200 chars
 		outputLength := len(str)
 		if outputLength >= 200 {
 			outputLength = 200
 		}
-		log.Error("writing failed",
+		tctx.L().Error("writing failed",
 			zap.String("string", str[:outputLength]),
 			zap.Error(err))
 	}
 	return errors.Trace(err)
 }
 
-func writeBytes(ctx context.Context, writer storage.ExternalFileWriter, p []byte) error {
-	_, err := writer.Write(ctx, p)
+func writeBytes(tctx *tcontext.Context, writer storage.ExternalFileWriter, p []byte) error {
+	_, err := writer.Write(tctx, p)
 	if err != nil {
 		// str might be very long, only output the first 200 chars
 		outputLength := len(p)
 		if outputLength >= 200 {
 			outputLength = 200
 		}
-		log.Error("writing failed",
+		tctx.L().Error("writing failed",
 			zap.ByteString("string", p[:outputLength]),
 			zap.String("writer", fmt.Sprintf("%#v", writer)),
 			zap.Error(err))
@@ -397,31 +397,31 @@ func writeBytes(ctx context.Context, writer storage.ExternalFileWriter, p []byte
 	return errors.Trace(err)
 }
 
-func buildFileWriter(ctx context.Context, s storage.ExternalStorage, fileName string, compressType storage.CompressType) (storage.ExternalFileWriter, func(ctx context.Context), error) {
+func buildFileWriter(tctx *tcontext.Context, s storage.ExternalStorage, fileName string, compressType storage.CompressType) (storage.ExternalFileWriter, func(ctx context.Context), error) {
 	fileName += compressFileSuffix(compressType)
 	fullPath := path.Join(s.URI(), fileName)
-	writer, err := storage.WithCompression(s, compressType).Create(ctx, fileName)
+	writer, err := storage.WithCompression(s, compressType).Create(tctx, fileName)
 	if err != nil {
-		log.Error("open file failed",
+		tctx.L().Error("open file failed",
 			zap.String("path", fullPath),
 			zap.Error(err))
 		return nil, nil, errors.Trace(err)
 	}
-	log.Debug("opened file", zap.String("path", fullPath))
+	tctx.L().Debug("opened file", zap.String("path", fullPath))
 	tearDownRoutine := func(ctx context.Context) {
 		err := writer.Close(ctx)
 		if err == nil {
 			return
 		}
 		err = errors.Trace(err)
-		log.Error("close file failed",
+		tctx.L().Error("close file failed",
 			zap.String("path", fullPath),
 			zap.Error(err))
 	}
 	return writer, tearDownRoutine, nil
 }
 
-func buildInterceptFileWriter(pCtx context.Context, s storage.ExternalStorage, fileName string, compressType storage.CompressType) (storage.ExternalFileWriter, func(context.Context)) {
+func buildInterceptFileWriter(pCtx *tcontext.Context, s storage.ExternalStorage, fileName string, compressType storage.CompressType) (storage.ExternalFileWriter, func(context.Context)) {
 	fileName += compressFileSuffix(compressType)
 	var writer storage.ExternalFileWriter
 	fullPath := path.Join(s.URI(), fileName)
@@ -431,13 +431,13 @@ func buildInterceptFileWriter(pCtx context.Context, s storage.ExternalStorage, f
 		// which will cause a context canceled error when closing gcs's Writer
 		w, err := storage.WithCompression(s, compressType).Create(pCtx, fileName)
 		if err != nil {
-			log.Error("open file failed",
+			pCtx.L().Error("open file failed",
 				zap.String("path", fullPath),
 				zap.Error(err))
 			return newWriterError(err)
 		}
 		writer = w
-		log.Debug("opened file", zap.String("path", fullPath))
+		pCtx.L().Debug("opened file", zap.String("path", fullPath))
 		fileWriter.ExternalFileWriter = writer
 		return nil
 	}
@@ -447,10 +447,10 @@ func buildInterceptFileWriter(pCtx context.Context, s storage.ExternalStorage, f
 		if writer == nil {
 			return
 		}
-		log.Debug("tear down lazy file writer...", zap.String("path", fullPath))
+		pCtx.L().Debug("tear down lazy file writer...", zap.String("path", fullPath))
 		err := writer.Close(ctx)
 		if err != nil {
-			log.Error("close file failed",
+			pCtx.L().Error("close file failed",
 				zap.String("path", fullPath),
 				zap.Error(err))
 		}
@@ -588,7 +588,7 @@ func (f FileFormat) Extension() string {
 }
 
 // WriteInsert writes TableDataIR to a storage.ExternalFileWriter in sql/csv type
-func (f FileFormat) WriteInsert(pCtx context.Context, cfg *Config, meta TableMeta, tblIR TableDataIR, w storage.ExternalFileWriter) error {
+func (f FileFormat) WriteInsert(pCtx *tcontext.Context, cfg *Config, meta TableMeta, tblIR TableDataIR, w storage.ExternalFileWriter) error {
 	switch f {
 	case FileFormatSQLText:
 		return WriteInsert(pCtx, cfg, meta, tblIR, w)
diff --git a/v4/log/log.go b/v4/log/log.go
index f134bcb..4632634 100644
--- a/v4/log/log.go
+++ b/v4/log/log.go
@@ -6,13 +6,9 @@ import (
 	"github.com/pingcap/errors"
 	pclog "github.com/pingcap/log"
 	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 )
 
-var (
-	appLogger = Logger{zap.NewNop()}
-	appLevel  = zap.NewAtomicLevel()
-)
+var appLogger = Logger{zap.NewNop()}
 
 // Logger wraps the zap logger.
 type Logger struct {
@@ -42,8 +38,8 @@ type Config struct {
 }
 
 // InitAppLogger inits the wrapped logger from config.
-func InitAppLogger(cfg *Config) error {
-	logger, props, err := pclog.InitLogger(&pclog.Config{
+func InitAppLogger(cfg *Config) (Logger, error) {
+	logger, _, err := pclog.InitLogger(&pclog.Config{
 		Level: cfg.Level,
 		File: pclog.FileLogConfig{
 			Filename:   cfg.File,
@@ -54,50 +50,12 @@ func InitAppLogger(cfg *Config) error {
 		Format: cfg.Format,
 	})
 	if err != nil {
-		return errors.Trace(err)
+		return appLogger, errors.Trace(err)
 	}
-	logger = logger.WithOptions(zap.AddCallerSkip(1))
-	appLogger = Logger{logger}
-	appLevel = props.Level
-	return nil
-}
-
-// SetAppLogger sets the wrapped logger from config.
-func SetAppLogger(logger *zap.Logger) {
-	appLogger = Logger{logger}
-}
-
-// ChangeAppLogLevel changes the wrapped logger's log level.
-func ChangeAppLogLevel(level zapcore.Level) {
-	appLevel.SetLevel(level)
-}
-
-// Info wraps *zap.Logger's Info function.
-func Info(msg string, fields ...zap.Field) {
-	appLogger.Info(msg, fields...)
-}
-
-// Warn wraps *zap.Logger's Warn function.
-func Warn(msg string, fields ...zap.Field) {
-	appLogger.Warn(msg, fields...)
-}
-
-// Error wraps *zap.Logger's Error function.
-func Error(msg string, fields ...zap.Field) {
-	appLogger.Error(msg, fields...)
-}
-
-// Debug wraps *zap.Logger's Debug function.
-func Debug(msg string, fields ...zap.Field) {
-	appLogger.Debug(msg, fields...)
-}
-
-// Fatal wraps *zap.Logger's Fatal function.
-func Fatal(msg string, fields ...zap.Field) {
-	appLogger.Fatal(msg, fields...)
+	return Logger{logger.WithOptions(zap.AddStacktrace(zap.DPanicLevel))}, nil
 }
 
-// Panic wraps *zap.Logger's Panic function.
-func Panic(msg string, fields ...zap.Field) {
-	appLogger.Panic(msg, fields...)
+// NewAppLogger returns the wrapped logger from config.
+func NewAppLogger(logger *zap.Logger) Logger {
+	return Logger{logger}
 }


Test patch:
diff --git a/v4/export/block_allow_list_test.go b/v4/export/block_allow_list_test.go
index bc15e16..b82a3c4 100644
--- a/v4/export/block_allow_list_test.go
+++ b/v4/export/block_allow_list_test.go
@@ -5,6 +5,8 @@ package export
 import (
 	"strings"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	. "github.com/pingcap/check"
 	"github.com/pingcap/tidb-tools/pkg/filter"
 	tf "github.com/pingcap/tidb-tools/pkg/table-filter"
@@ -35,7 +37,7 @@ func (s *testBWListSuite) TestFilterTables(c *C) {
 	}
 
 	conf.TableFilter = tf.NewSchemasFilter("xxx")
-	filterTables(conf)
+	filterTables(tcontext.Background(), conf)
 	c.Assert(conf.Tables, HasLen, 1)
 	c.Assert(conf.Tables, DeepEquals, expectedDBTables)
 }
@@ -53,14 +55,14 @@ func (s *testBWListSuite) TestFilterDatabaseWithNoTable(c *C) {
 		TableFilter:       tf.NewSchemasFilter("yyy"),
 		DumpEmptyDatabase: true,
 	}
-	filterTables(conf)
+	filterTables(tcontext.Background(), conf)
 	c.Assert(conf.Tables, HasLen, 0)
 
 	dbTables["xxx"] = []*TableInfo{}
 	expectedDBTables["xxx"] = []*TableInfo{}
 	conf.Tables = dbTables
 	conf.TableFilter = tf.NewSchemasFilter("xxx")
-	filterTables(conf)
+	filterTables(tcontext.Background(), conf)
 	c.Assert(conf.Tables, HasLen, 1)
 	c.Assert(conf.Tables, DeepEquals, expectedDBTables)
 
@@ -68,7 +70,7 @@ func (s *testBWListSuite) TestFilterDatabaseWithNoTable(c *C) {
 	expectedDBTables = DatabaseTables{}
 	conf.Tables = dbTables
 	conf.DumpEmptyDatabase = false
-	filterTables(conf)
+	filterTables(tcontext.Background(), conf)
 	c.Assert(conf.Tables, HasLen, 0)
 	c.Assert(conf.Tables, DeepEquals, expectedDBTables)
 }
diff --git a/v4/export/consistency_test.go b/v4/export/consistency_test.go
index 8c135bf..f434449 100644
--- a/v4/export/consistency_test.go
+++ b/v4/export/consistency_test.go
@@ -7,6 +7,8 @@ import (
 	"errors"
 	"strings"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/go-sql-driver/mysql"
 	. "github.com/pingcap/check"
@@ -22,9 +24,9 @@ func (s *testConsistencySuite) assertNil(err error, c *C) {
 	}
 }
 
-func (s *testConsistencySuite) assertLifetimeErrNil(ctx context.Context, ctrl ConsistencyController, c *C) {
-	s.assertNil(ctrl.Setup(ctx), c)
-	s.assertNil(ctrl.TearDown(ctx), c)
+func (s *testConsistencySuite) assertLifetimeErrNil(tctx *tcontext.Context, ctrl ConsistencyController, c *C) {
+	s.assertNil(ctrl.Setup(tctx), c)
+	s.assertNil(ctrl.TearDown(tctx), c)
 }
 
 func (s *testConsistencySuite) TestConsistencyController(c *C) {
@@ -33,6 +35,7 @@ func (s *testConsistencySuite) TestConsistencyController(c *C) {
 	defer db.Close()
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
+	tctx := tcontext.Background().WithContext(ctx)
 	conf := DefaultConfig()
 	resultOk := sqlmock.NewResult(0, 1)
 
@@ -40,7 +43,7 @@ func (s *testConsistencySuite) TestConsistencyController(c *C) {
 	ctrl, _ := NewConsistencyController(ctx, conf, db)
 	_, ok := ctrl.(*ConsistencyNone)
 	c.Assert(ok, IsTrue)
-	s.assertLifetimeErrNil(ctx, ctrl, c)
+	s.assertLifetimeErrNil(tctx, ctrl, c)
 
 	conf.Consistency = consistencyTypeFlush
 	mock.ExpectExec("FLUSH TABLES WITH READ LOCK").WillReturnResult(resultOk)
@@ -48,7 +51,7 @@ func (s *testConsistencySuite) TestConsistencyController(c *C) {
 	ctrl, _ = NewConsistencyController(ctx, conf, db)
 	_, ok = ctrl.(*ConsistencyFlushTableWithReadLock)
 	c.Assert(ok, IsTrue)
-	s.assertLifetimeErrNil(ctx, ctrl, c)
+	s.assertLifetimeErrNil(tctx, ctrl, c)
 	if err = mock.ExpectationsWereMet(); err != nil {
 		c.Fatal(err.Error())
 	}
@@ -58,7 +61,7 @@ func (s *testConsistencySuite) TestConsistencyController(c *C) {
 	ctrl, _ = NewConsistencyController(ctx, conf, db)
 	_, ok = ctrl.(*ConsistencyNone)
 	c.Assert(ok, IsTrue)
-	s.assertLifetimeErrNil(ctx, ctrl, c)
+	s.assertLifetimeErrNil(tctx, ctrl, c)
 
 	conf.Consistency = consistencyTypeLock
 	conf.Tables = NewDatabaseTables().
@@ -69,7 +72,7 @@ func (s *testConsistencySuite) TestConsistencyController(c *C) {
 	ctrl, _ = NewConsistencyController(ctx, conf, db)
 	_, ok = ctrl.(*ConsistencyLockDumpingTables)
 	c.Assert(ok, IsTrue)
-	s.assertLifetimeErrNil(ctx, ctrl, c)
+	s.assertLifetimeErrNil(tctx, ctrl, c)
 	if err = mock.ExpectationsWereMet(); err != nil {
 		c.Fatal(err.Error())
 	}
@@ -81,6 +84,7 @@ func (s *testConsistencySuite) TestConsistencyLockControllerRetry(c *C) {
 	defer db.Close()
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
+	tctx := tcontext.Background().WithContext(ctx)
 	conf := DefaultConfig()
 	resultOk := sqlmock.NewResult(0, 1)
 
@@ -95,7 +99,7 @@ func (s *testConsistencySuite) TestConsistencyLockControllerRetry(c *C) {
 	ctrl, _ := NewConsistencyController(ctx, conf, db)
 	_, ok := ctrl.(*ConsistencyLockDumpingTables)
 	c.Assert(ok, IsTrue)
-	s.assertLifetimeErrNil(ctx, ctrl, c)
+	s.assertLifetimeErrNil(tctx, ctrl, c)
 	if err = mock.ExpectationsWereMet(); err != nil {
 		c.Fatal(err.Error())
 	}
@@ -129,6 +133,7 @@ func (s *testConsistencySuite) TestConsistencyControllerError(c *C) {
 	defer db.Close()
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
+	tctx := tcontext.Background().WithContext(ctx)
 	conf := DefaultConfig()
 
 	conf.Consistency = "invalid_str"
@@ -146,7 +151,7 @@ func (s *testConsistencySuite) TestConsistencyControllerError(c *C) {
 	conf.Consistency = consistencyTypeFlush
 	conf.ServerInfo.ServerType = ServerTypeTiDB
 	ctrl, _ := NewConsistencyController(ctx, conf, db)
-	err = ctrl.Setup(ctx)
+	err = ctrl.Setup(tctx)
 	c.Assert(err, NotNil)
 
 	// lock table fail
@@ -154,6 +159,6 @@ func (s *testConsistencySuite) TestConsistencyControllerError(c *C) {
 	conf.Tables = NewDatabaseTables().AppendTables("db", "t")
 	mock.ExpectExec("LOCK TABLE").WillReturnError(errors.New(""))
 	ctrl, _ = NewConsistencyController(ctx, conf, db)
-	err = ctrl.Setup(ctx)
+	err = ctrl.Setup(tctx)
 	c.Assert(err, NotNil)
 }
diff --git a/v4/export/metadata_test.go b/v4/export/metadata_test.go
index 8583a3f..0974b9c 100644
--- a/v4/export/metadata_test.go
+++ b/v4/export/metadata_test.go
@@ -7,6 +7,8 @@ import (
 	"errors"
 	"fmt"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/pingcap/br/pkg/storage"
 	. "github.com/pingcap/check"
@@ -36,7 +38,7 @@ func (s *testMetaDataSuite) TestMysqlMetaData(c *C) {
 	mock.ExpectQuery("SHOW SLAVE STATUS").WillReturnRows(
 		sqlmock.NewRows([]string{"exec_master_log_pos", "relay_master_log_file", "master_host", "Executed_Gtid_Set", "Seconds_Behind_Master"}))
 
-	m := newGlobalMetadata(s.createStorage(c), "")
+	m := newGlobalMetadata(tcontext.Background(), s.createStorage(c), "")
 	c.Assert(m.recordGlobalMetaData(conn, ServerTypeMySQL, false), IsNil)
 
 	c.Assert(m.buffer.String(), Equals, "SHOW MASTER STATUS:\n"+
@@ -71,7 +73,7 @@ func (s *testMetaDataSuite) TestMetaDataAfterConn(c *C) {
 		sqlmock.NewRows([]string{"exec_master_log_pos", "relay_master_log_file", "master_host", "Executed_Gtid_Set", "Seconds_Behind_Master"}))
 	mock.ExpectQuery("SHOW MASTER STATUS").WillReturnRows(rows2)
 
-	m := newGlobalMetadata(s.createStorage(c), "")
+	m := newGlobalMetadata(tcontext.Background(), s.createStorage(c), "")
 	c.Assert(m.recordGlobalMetaData(conn, ServerTypeMySQL, false), IsNil)
 	c.Assert(m.recordGlobalMetaData(conn, ServerTypeMySQL, true), IsNil)
 	m.buffer.Write(m.afterConnBuffer.Bytes())
@@ -102,7 +104,7 @@ func (s *testMetaDataSuite) TestMysqlWithFollowersMetaData(c *C) {
 	mock.ExpectQuery("SELECT @@default_master_connection").WillReturnError(fmt.Errorf("mock error"))
 	mock.ExpectQuery("SHOW SLAVE STATUS").WillReturnRows(followerRows)
 
-	m := newGlobalMetadata(s.createStorage(c), "")
+	m := newGlobalMetadata(tcontext.Background(), s.createStorage(c), "")
 	c.Assert(m.recordGlobalMetaData(conn, ServerTypeMySQL, false), IsNil)
 
 	c.Assert(m.buffer.String(), Equals, "SHOW MASTER STATUS:\n"+
@@ -130,7 +132,7 @@ func (s *testMetaDataSuite) TestMysqlWithNullFollowersMetaData(c *C) {
 	mock.ExpectQuery("SELECT @@default_master_connection").WillReturnError(fmt.Errorf("mock error"))
 	mock.ExpectQuery("SHOW SLAVE STATUS").WillReturnRows(sqlmock.NewRows([]string{"SQL_Remaining_Delay"}).AddRow(nil))
 
-	m := newGlobalMetadata(s.createStorage(c), "")
+	m := newGlobalMetadata(tcontext.Background(), s.createStorage(c), "")
 	c.Assert(m.recordGlobalMetaData(conn, ServerTypeMySQL, false), IsNil)
 
 	c.Assert(m.buffer.String(), Equals, "SHOW MASTER STATUS:\n"+
@@ -157,7 +159,7 @@ func (s *testMetaDataSuite) TestMariaDBMetaData(c *C) {
 		AddRow(gtidSet)
 	mock.ExpectQuery("SELECT @@global.gtid_binlog_pos").WillReturnRows(rows)
 	mock.ExpectQuery("SHOW SLAVE STATUS").WillReturnRows(rows)
-	m := newGlobalMetadata(s.createStorage(c), "")
+	m := newGlobalMetadata(tcontext.Background(), s.createStorage(c), "")
 	c.Assert(m.recordGlobalMetaData(conn, ServerTypeMariaDB, false), IsNil)
 
 	c.Assert(mock.ExpectationsWereMet(), IsNil)
@@ -181,7 +183,7 @@ func (s *testMetaDataSuite) TestMariaDBWithFollowersMetaData(c *C) {
 			AddRow("connection_1"))
 	mock.ExpectQuery("SHOW ALL SLAVES STATUS").WillReturnRows(followerRows)
 
-	m := newGlobalMetadata(s.createStorage(c), "")
+	m := newGlobalMetadata(tcontext.Background(), s.createStorage(c), "")
 	c.Assert(m.recordGlobalMetaData(conn, ServerTypeMySQL, false), IsNil)
 
 	c.Assert(m.buffer.String(), Equals, "SHOW MASTER STATUS:\n"+
@@ -219,7 +221,7 @@ func (s *testMetaDataSuite) TestEarlierMysqlMetaData(c *C) {
 	mock.ExpectQuery("SHOW SLAVE STATUS").WillReturnRows(
 		sqlmock.NewRows([]string{"exec_master_log_pos", "relay_master_log_file", "master_host", "Executed_Gtid_Set", "Seconds_Behind_Master"}))
 
-	m := newGlobalMetadata(s.createStorage(c), "")
+	m := newGlobalMetadata(tcontext.Background(), s.createStorage(c), "")
 	c.Assert(m.recordGlobalMetaData(conn, ServerTypeMySQL, false), IsNil)
 
 	c.Assert(m.buffer.String(), Equals, "SHOW MASTER STATUS:\n"+
@@ -242,7 +244,7 @@ func (s *testMetaDataSuite) TestTiDBSnapshotMetaData(c *C) {
 		AddRow(logFile, pos, "", "")
 	mock.ExpectQuery("SHOW MASTER STATUS").WillReturnRows(rows)
 
-	m := newGlobalMetadata(s.createStorage(c), "")
+	m := newGlobalMetadata(tcontext.Background(), s.createStorage(c), "")
 	c.Assert(m.recordGlobalMetaData(conn, ServerTypeTiDB, false), IsNil)
 	c.Assert(m.buffer.String(), Equals, "SHOW MASTER STATUS:\n"+
 		"\tLog: tidb-binlog\n"+
@@ -253,7 +255,7 @@ func (s *testMetaDataSuite) TestTiDBSnapshotMetaData(c *C) {
 	rows = sqlmock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB"}).
 		AddRow(logFile, pos, "", "")
 	mock.ExpectQuery("SHOW MASTER STATUS").WillReturnRows(rows)
-	m = newGlobalMetadata(s.createStorage(c), snapshot)
+	m = newGlobalMetadata(tcontext.Background(), s.createStorage(c), snapshot)
 	c.Assert(m.recordGlobalMetaData(conn, ServerTypeTiDB, false), IsNil)
 	c.Assert(m.buffer.String(), Equals, "SHOW MASTER STATUS:\n"+
 		"\tLog: tidb-binlog\n"+
@@ -271,7 +273,7 @@ func (s *testMetaDataSuite) TestNoPrivilege(c *C) {
 
 	mock.ExpectQuery("SHOW MASTER STATUS").WillReturnError(errors.New("lack SUPER or REPLICATION CLIENT privilege"))
 
-	m := newGlobalMetadata(s.createStorage(c), "")
+	m := newGlobalMetadata(tcontext.Background(), s.createStorage(c), "")
 	// some consistencyType will ignore this error, this test make sure no extra message is written
 	c.Assert(m.recordGlobalMetaData(conn, ServerTypeTiDB, false), NotNil)
 	c.Assert(m.buffer.String(), Equals, "")
diff --git a/v4/export/sql_test.go b/v4/export/sql_test.go
index 7512cc1..28dce05 100644
--- a/v4/export/sql_test.go
+++ b/v4/export/sql_test.go
@@ -6,6 +6,8 @@ import (
 	"context"
 	"errors"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/coreos/go-semver/semver"
 	. "github.com/pingcap/check"
@@ -42,7 +44,7 @@ func (s *testSQLSuite) TestDetectServerInfo(c *C) {
 
 		verStr, err := SelectVersion(db)
 		c.Assert(err, IsNil, cmt)
-		info := ParseServerInfo(verStr)
+		info := ParseServerInfo(tcontext.Background(), verStr)
 		c.Assert(info.ServerType, Equals, serverTp, cmt)
 		c.Assert(info.ServerVersion == nil, Equals, expectVer == nil, cmt)
 		if info.ServerVersion == nil {
diff --git a/v4/export/writer_test.go b/v4/export/writer_test.go
index 14e1e7a..8f55d46 100644
--- a/v4/export/writer_test.go
+++ b/v4/export/writer_test.go
@@ -9,6 +9,8 @@ import (
 	"os"
 	"path"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/pingcap/br/pkg/storage"
 	. "github.com/pingcap/check"
@@ -27,7 +29,7 @@ func (s *testWriterSuite) newWriter(conf *Config, c *C) *Writer {
 	c.Assert(err, IsNil)
 	conn, err := db.Conn(context.Background())
 	c.Assert(err, IsNil)
-	return NewWriter(context.Background(), 0, conf, conn, extStore)
+	return NewWriter(tcontext.Background(), 0, conf, conn, extStore)
 }
 
 func (s *testWriterSuite) TestWriteDatabaseMeta(c *C) {
diff --git a/v4/export/writer_util_test.go b/v4/export/writer_util_test.go
index 4374e76..d3b6920 100644
--- a/v4/export/writer_util_test.go
+++ b/v4/export/writer_util_test.go
@@ -3,12 +3,13 @@
 package export
 
 import (
-	"context"
 	"database/sql/driver"
 	"fmt"
 	"strings"
 	"testing"
 
+	tcontext "github.com/pingcap/dumpling/v4/context"
+
 	"github.com/pingcap/br/pkg/storage"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/errors"
@@ -39,7 +40,7 @@ func (s *testUtilSuite) TestWriteMeta(c *C) {
 	meta := newMockMetaIR("t1", createTableStmt, specCmts)
 	writer := storage.NewBufferWriter()
 
-	err := WriteMeta(context.Background(), meta, writer)
+	err := WriteMeta(tcontext.Background(), meta, writer)
 	c.Assert(err, IsNil)
 	expected := "/*!40103 SET TIME_ZONE='+00:00' */;\n" +
 		"CREATE TABLE `t1` (\n" +
@@ -64,7 +65,7 @@ func (s *testUtilSuite) TestWriteInsert(c *C) {
 	bf := storage.NewBufferWriter()
 
 	conf := configForWriteSQL(UnspecifiedSize, UnspecifiedSize)
-	err := WriteInsert(context.Background(), conf, tableIR, tableIR, bf)
+	err := WriteInsert(tcontext.Background(), conf, tableIR, tableIR, bf)
 	c.Assert(err, IsNil)
 	expected := "/*!40101 SET NAMES binary*/;\n" +
 		"/*!40014 SET FOREIGN_KEY_CHECKS=0*/;\n" +
@@ -95,7 +96,7 @@ func (s *testUtilSuite) TestWriteInsertReturnsError(c *C) {
 	bf := storage.NewBufferWriter()
 
 	conf := configForWriteSQL(UnspecifiedSize, UnspecifiedSize)
-	err := WriteInsert(context.Background(), conf, tableIR, tableIR, bf)
+	err := WriteInsert(tcontext.Background(), conf, tableIR, tableIR, bf)
 	c.Assert(err, Equals, rowErr)
 	expected := "/*!40101 SET NAMES binary*/;\n" +
 		"/*!40014 SET FOREIGN_KEY_CHECKS=0*/;\n" +
@@ -120,7 +121,7 @@ func (s *testUtilSuite) TestWriteInsertInCsv(c *C) {
 	// test nullValue
 	opt := &csvOption{separator: []byte(","), delimiter: doubleQuotationMark, nullValue: "\\N"}
 	conf := configForWriteCSV(true, opt)
-	err := WriteInsertInCsv(context.Background(), conf, tableIR, tableIR, bf)
+	err := WriteInsertInCsv(tcontext.Background(), conf, tableIR, tableIR, bf)
 	c.Assert(err, IsNil)
 	expected := "1,\"male\",\"bob@mail.com\",\"020-1234\",\\N\n" +
 		"2,\"female\",\"sarah@mail.com\",\"020-1253\",\"healthy\"\n" +
@@ -133,7 +134,7 @@ func (s *testUtilSuite) TestWriteInsertInCsv(c *C) {
 	opt.delimiter = quotationMark
 	tableIR = newMockTableIR("test", "employee", data, nil, colTypes)
 	conf = configForWriteCSV(true, opt)
-	err = WriteInsertInCsv(context.Background(), conf, tableIR, tableIR, bf)
+	err = WriteInsertInCsv(tcontext.Background(), conf, tableIR, tableIR, bf)
 	c.Assert(err, IsNil)
 	expected = "1,'male','bob@mail.com','020-1234',\\N\n" +
 		"2,'female','sarah@mail.com','020-1253','healthy'\n" +
@@ -146,7 +147,7 @@ func (s *testUtilSuite) TestWriteInsertInCsv(c *C) {
 	opt.separator = []byte(";")
 	tableIR = newMockTableIR("test", "employee", data, nil, colTypes)
 	conf = configForWriteCSV(true, opt)
-	err = WriteInsertInCsv(context.Background(), conf, tableIR, tableIR, bf)
+	err = WriteInsertInCsv(tcontext.Background(), conf, tableIR, tableIR, bf)
 	c.Assert(err, IsNil)
 	expected = "1;'male';'bob@mail.com';'020-1234';\\N\n" +
 		"2;'female';'sarah@mail.com';'020-1253';'healthy'\n" +
@@ -161,7 +162,7 @@ func (s *testUtilSuite) TestWriteInsertInCsv(c *C) {
 	tableIR = newMockTableIR("test", "employee", data, nil, colTypes)
 	tableIR.colNames = []string{"id", "gender", "email", "phone_number", "status"}
 	conf = configForWriteCSV(false, opt)
-	err = WriteInsertInCsv(context.Background(), conf, tableIR, tableIR, bf)
+	err = WriteInsertInCsv(tcontext.Background(), conf, tableIR, tableIR, bf)
 	c.Assert(err, IsNil)
 	expected = "maidma&;,?magenderma&;,?maemamailma&;,?maphone_numberma&;,?mastatusma\n" +
 		"1&;,?mamamalema&;,?mabob@mamail.comma&;,?ma020-1234ma&;,?\\N\n" +
@@ -187,7 +188,7 @@ func (s *testUtilSuite) TestSQLDataTypes(c *C) {
 		bf := storage.NewBufferWriter()
 
 		conf := configForWriteSQL(UnspecifiedSize, UnspecifiedSize)
-		err := WriteInsert(context.Background(), conf, tableIR, tableIR, bf)
+		err := WriteInsert(tcontext.Background(), conf, tableIR, tableIR, bf)
 		c.Assert(err, IsNil)
 		lines := strings.Split(bf.String(), "\n")
 		c.Assert(len(lines), Equals, 3)
@@ -201,7 +202,7 @@ func (s *testUtilSuite) TestWrite(c *C) {
 	exp := []string{"test", "loooooooooooooooooooong", "poison_error"}
 
 	for i, s := range src {
-		err := write(context.Background(), mocksw, s)
+		err := write(tcontext.Background(), mocksw, s)
 		if err != nil {
 			c.Assert(err.Error(), Equals, exp[i])
 		} else {
@@ -209,7 +210,7 @@ func (s *testUtilSuite) TestWrite(c *C) {
 			c.Assert(mocksw.buf, Equals, exp[i])
 		}
 	}
-	err := write(context.Background(), mocksw, "test")
+	err := write(tcontext.Background(), mocksw, "test")
 	c.Assert(err, IsNil)
 }