Repository: pingcap/tidb-lightning (go)

Issue:
lightning warning repeatedly without exit while certs are wrong
## Bug Report

Please answer these questions before submitting your issue. Thanks!

1. What did you do? If possible, provide a recipe for reproducing the error.
tidb-lightning.toml with wrong certs in security section
```
tidb.tls = "false"
```

2. What did you expect to see?
Import successfully and no error

3. What did you see instead?
lightning output warning repeatedly, and it can't be stopped by ctrl-c
```
root@agent:/data1/cy/src/tidb-lightning# ./bin/tidb-lightning -c tidb-lightning.toml
Verbose debug logs will be written to tidb-lightning.log

[2020/11/10 07:04:59.497 +00:00] [INFO] [client.go:166] ["[pd] create pd client with endpoints"] [pd-address="[db-pd.tidb1326054958621003776.svc:2379]"]
[2020/11/10 07:04:59.505 +00:00] [WARN] [base_client.go:180] ["[pd] failed to get cluster id"] [url=http://db-pd.tidb1326054958621003776.svc:2379] [error="[PD:client:ErrClientGetMember]error:rpc error: code = Unavailable desc = connection error: desc = \"transport: authen
tication handshake failed: x509: certificate signed by unknown authority (possibly because of \\\"crypto/rsa: verification error\\\" while trying to verify candidate authority certificate \\\"TiDB CA\\\")\" target:db-pd.tidb1326054958621003776.svc:2379 status:TRANSIENT_FA
ILURE"]
[2020/11/10 07:05:00.505 +00:00] [WARN] [base_client.go:180] ["[pd] failed to get cluster id"] [url=http://db-pd.tidb1326054958621003776.svc:2379] [error="[PD:client:ErrClientGetMember]error:rpc error: code = Unavailable desc = connection error: desc = \"transport: authen
tication handshake failed: x509: certificate signed by unknown authority (possibly because of \\\"crypto/rsa: verification error\\\" while trying to verify candidate authority certificate \\\"TiDB CA\\\")\" target:db-pd.tidb1326054958621003776.svc:2379 status:TRANSIENT_FA
ILURE"]
[2020/11/10 07:05:01.506 +00:00] [WARN] [base_client.go:180] ["[pd] failed to get cluster id"] [url=http://db-pd.tidb1326054958621003776.svc:2379] [error="[PD:client:ErrClientGetMember]error:rpc error: code = Unavailable desc = connection error: desc = \"transport: authen
tication handshake failed: x509: certificate signed by unknown authority (possibly because of \\\"crypto/rsa: verification error\\\" while trying to verify candidate authority certificate \\\"TiDB CA\\\")\" target:db-pd.tidb1326054958621003776.svc:2379 status:TRANSIENT_FA
ILURE"]
```

4. Versions of the cluster

    - TiDB-Lightning version (run `tidb-lightning -V`):

        ```
        ./bin/tidb-lightning -V
       Release Version: v4.0.8-10-g51d0e7b
       Git Commit Hash: 51d0e7bf56f7b5007fbdb310be7b7ef5e68ffcc9
       Git Branch: master
       UTC Build Time: 2020-11-10 04:38:28
       Go Version: go version go1.13.15 linux/amd64
        ```
        
    - TiKV-Importer version (run `tikv-importer -V`)

        ```
        (paste TiKV-Importer version here)
        ```
        
    - TiKV version (run `tikv-server -V`):

        ```
        (paste TiKV version here)
        ```

    - TiDB cluster version (execute `SELECT tidb_version();` in a MySQL client):

        ```
        (paste TiDB cluster version here)
        ```

    - Other interesting information (system version, hardware config, etc):

        >
        >


5. Operation logs
   - Please upload `tidb-lightning.log` for TiDB-Lightning if possible
   - Please upload `tikv-importer.log` from TiKV-Importer if possible
   - Other interesting logs
   
6. Configuration of the cluster and the task
   - `tidb-lightning.toml` for TiDB-Lightning if possible
```
[lightning]

status-addr = ":8289"

server-mode = false


index-concurrency = 2
table-concurrency = 6

level = "info"
file = "tidb-lightning.log"
max-size = 128 # MB
max-days = 28
max-backups = 14

[security]
ca-path = "2768/client-ca.crt"
cert-path = "2768/client-tls.crt"
key-path = "2768/client-tls.key"

[checkpoint]
enable = true
schema = "tidb_lightning_checkpoint"
driver = "file"

[tikv-importer]
backend = "local"
addr = "127.0.0.1:8287"
sorted-kv-dir = "/data2/sortkv"

[mydumper]
read-block-size = 65536 # Byte (default = 64 KB)

batch-import-ratio = 0.75

data-source-dir = "s3://cy-massive-data/uuid_1t_csv?region=us-east-1"
no-schema=true

case-sensitive = false

strict-format = false


filter = ['*.*', '!mysql.*', '!sys.*', '!INFORMATION_SCHEMA.*', '!PERFORMANCE_SCHEMA.*', '!METRICS_SCHEMA.*', '!INSPECTION_SCHEMA.*']

[mydumper.csv]
separator = ','
delimiter = '"'
header = false
not-null = false
null = '\N'
backslash-escape = true
trim-last-separator = false


[tidb]
host = "db-tidb.tidb1326054958621003776.svc"
port = 4000
user = "root"
password = "12345678"
status-port = 10080
pd-addr = "db-pd.tidb1326054958621003776.svc:2379"
log-level = "error"


tls = "false"

build-stats-concurrency = 20
distsql-scan-concurrency = 100
index-serial-scan-concurrency = 20
checksum-table-concurrency = 16

[tidb.security]
ca-path = "2768/client-ca.crt"
cert-path = "2768/client-tls.crt"
key-path = "2768/client-tls.key"

[post-restore]
checksum = "required"
analyze = "optional"
level-1-compact = false
compact = false

[cron]
switch-mode = "5m"
log-progress = "5m"
```
   - `tikv-importer.toml` for TiKV-Importer if possible
   - `inventory.ini` if deployed by Ansible

7. Screenshot/exported-PDF of Grafana dashboard or metrics' graph in Prometheus for TiDB-Lightning if possible


Pull request description:
*: replace context.Backend with app context
<!--
Thank you for contributing to TiDB-Lightning! Please read the [CONTRIBUTING](https://github.com/pingcap/tidb-lightning/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve? <!--add issue link with summary if exists-->
Replace some usage of `context.Backend` or `context.TODO` with lightning app context. So we can stop some sub task more quickly.

close #462 

### What is changed and how it works?


### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test
 - Manual test (add detailed scripts or steps below)
 - No code

Side effects

Related changes


Patch:
diff --git a/cmd/tidb-lightning-ctl/main.go b/cmd/tidb-lightning-ctl/main.go
index c3a91bc..ceda277 100644
--- a/cmd/tidb-lightning-ctl/main.go
+++ b/cmd/tidb-lightning-ctl/main.go
@@ -72,11 +72,13 @@ func run() error {
 		fsUsage = fs.Usage
 	}))
 
+	ctx := context.Background()
+
 	cfg := config.NewConfig()
 	if err := cfg.LoadFromGlobal(globalCfg); err != nil {
 		return err
 	}
-	if err := cfg.Adjust(); err != nil {
+	if err := cfg.Adjust(ctx); err != nil {
 		return err
 	}
 
@@ -88,8 +90,6 @@ func run() error {
 		return err
 	}
 
-	ctx := context.Background()
-
 	if *compact {
 		return errors.Trace(compactCluster(ctx, cfg, tls))
 	}
diff --git a/lightning/backend/backend.go b/lightning/backend/backend.go
index 643a1f6..00dcb9f 100644
--- a/lightning/backend/backend.go
+++ b/lightning/backend/backend.go
@@ -125,7 +125,7 @@ type AbstractBackend interface {
 
 	// CheckRequirements performs the check whether the backend satisfies the
 	// version requirements
-	CheckRequirements() error
+	CheckRequirements(ctx context.Context) error
 
 	// FetchRemoteTableModels obtains the models of all tables given the schema
 	// name. The returned table info does not need to be precise if the encoder,
@@ -142,9 +142,9 @@ type AbstractBackend interface {
 	FetchRemoteTableModels(ctx context.Context, schemaName string) ([]*model.TableInfo, error)
 }
 
-func fetchRemoteTableModelsFromTLS(tls *common.TLS, schema string) ([]*model.TableInfo, error) {
+func fetchRemoteTableModelsFromTLS(ctx context.Context, tls *common.TLS, schema string) ([]*model.TableInfo, error) {
 	var tables []*model.TableInfo
-	err := tls.GetJSON("/schema/"+schema, &tables)
+	err := tls.GetJSON(ctx, "/schema/"+schema, &tables)
 	if err != nil {
 		return nil, errors.Annotatef(err, "cannot read schema '%s' from remote", schema)
 	}
@@ -204,8 +204,8 @@ func (be Backend) ShouldPostProcess() bool {
 	return be.abstract.ShouldPostProcess()
 }
 
-func (be Backend) CheckRequirements() error {
-	return be.abstract.CheckRequirements()
+func (be Backend) CheckRequirements(ctx context.Context) error {
+	return be.abstract.CheckRequirements(ctx)
 }
 
 func (be Backend) FetchRemoteTableModels(ctx context.Context, schemaName string) ([]*model.TableInfo, error) {
diff --git a/lightning/backend/importer.go b/lightning/backend/importer.go
index 0b9ec80..d18d9b6 100644
--- a/lightning/backend/importer.go
+++ b/lightning/backend/importer.go
@@ -241,51 +241,52 @@ func (*importer) NewEncoder(tbl table.Table, options *SessionOptions) Encoder {
 	return NewTableKVEncoder(tbl, options)
 }
 
-func (importer *importer) CheckRequirements() error {
-	if err := checkTiDBVersion(importer.tls, requiredTiDBVersion); err != nil {
+func (importer *importer) CheckRequirements(ctx context.Context) error {
+	if err := checkTiDBVersionByTLS(ctx, importer.tls, requiredTiDBVersion); err != nil {
 		return err
 	}
-	if err := checkPDVersion(importer.tls, importer.pdAddr, requiredPDVersion); err != nil {
+	if err := checkPDVersion(ctx, importer.tls, importer.pdAddr, requiredPDVersion); err != nil {
 		return err
 	}
-	if err := checkTiKVVersion(importer.tls, importer.pdAddr, requiredTiKVVersion); err != nil {
+	if err := checkTiKVVersion(ctx, importer.tls, importer.pdAddr, requiredTiKVVersion); err != nil {
 		return err
 	}
 	return nil
 }
 
-func checkTiDBVersion(tls *common.TLS, requiredVersion semver.Version) error {
+func checkTiDBVersionByTLS(ctx context.Context, tls *common.TLS, requiredVersion semver.Version) error {
 	var status struct{ Version string }
-	err := tls.GetJSON("/status", &status)
+	err := tls.GetJSON(ctx, "/status", &status)
 	if err != nil {
 		return err
 	}
 
-	version, err := common.ExtractTiDBVersion(status.Version)
+	return checkTiDBVersion(status.Version, requiredVersion)
+}
+
+func checkTiDBVersion(versionStr string, requiredVersion semver.Version) error {
+	version, err := common.ExtractTiDBVersion(versionStr)
 	if err != nil {
 		return errors.Trace(err)
 	}
 	return checkVersion("TiDB", requiredVersion, *version)
 }
 
-func checkTiDBVersionBySQL(g glue.Glue, requiredVersion semver.Version) error {
+func checkTiDBVersionBySQL(ctx context.Context, g glue.Glue, requiredVersion semver.Version) error {
 	versionStr, err := g.GetSQLExecutor().ObtainStringWithLog(
-		context.Background(),
+		ctx,
 		"SELECT version();",
 		"check TiDB version",
 		log.L())
 	if err != nil {
 		return errors.Trace(err)
 	}
-	version, err := common.ExtractTiDBVersion(versionStr)
-	if err != nil {
-		return errors.Trace(err)
-	}
-	return checkVersion("TiDB", requiredVersion, *version)
+
+	return checkTiDBVersion(versionStr, requiredVersion)
 }
 
-func checkPDVersion(tls *common.TLS, pdAddr string, requiredVersion semver.Version) error {
-	version, err := common.FetchPDVersion(tls, pdAddr)
+func checkPDVersion(ctx context.Context, tls *common.TLS, pdAddr string, requiredVersion semver.Version) error {
+	version, err := common.FetchPDVersion(ctx, tls, pdAddr)
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -293,9 +294,9 @@ func checkPDVersion(tls *common.TLS, pdAddr string, requiredVersion semver.Versi
 	return checkVersion("PD", requiredVersion, *version)
 }
 
-func checkTiKVVersion(tls *common.TLS, pdAddr string, requiredVersion semver.Version) error {
+func checkTiKVVersion(ctx context.Context, tls *common.TLS, pdAddr string, requiredVersion semver.Version) error {
 	return ForAllStores(
-		context.Background(),
+		ctx,
 		tls.WithHost(pdAddr),
 		StoreStateDown,
 		func(c context.Context, store *Store) error {
@@ -322,5 +323,5 @@ func checkVersion(component string, expected, actual semver.Version) error {
 }
 
 func (importer *importer) FetchRemoteTableModels(ctx context.Context, schema string) ([]*model.TableInfo, error) {
-	return fetchRemoteTableModelsFromTLS(importer.tls, schema)
+	return fetchRemoteTableModelsFromTLS(ctx, importer.tls, schema)
 }
diff --git a/lightning/backend/local.go b/lightning/backend/local.go
index c162c37..bf6cb56 100644
--- a/lightning/backend/local.go
+++ b/lightning/backend/local.go
@@ -252,7 +252,7 @@ func NewLocalBackend(
 	enableCheckpoint bool,
 	g glue.Glue,
 ) (Backend, error) {
-	pdCli, err := pd.NewClient([]string{pdAddr}, tls.ToPDSecurityOption())
+	pdCli, err := pd.NewClientWithContext(ctx, []string{pdAddr}, tls.ToPDSecurityOption())
 	if err != nil {
 		return MakeBackend(nil), errors.Annotate(err, "construct pd client failed")
 	}
@@ -1173,21 +1173,21 @@ func (local *local) CleanupEngine(ctx context.Context, engineUUID uuid.UUID) err
 	return nil
 }
 
-func (local *local) CheckRequirements() error {
-	if err := checkTiDBVersionBySQL(local.g, localMinTiDBVersion); err != nil {
+func (local *local) CheckRequirements(ctx context.Context) error {
+	if err := checkTiDBVersionBySQL(ctx, local.g, localMinTiDBVersion); err != nil {
 		return err
 	}
-	if err := checkPDVersion(local.tls, local.pdAddr, localMinPDVersion); err != nil {
+	if err := checkPDVersion(ctx, local.tls, local.pdAddr, localMinPDVersion); err != nil {
 		return err
 	}
-	if err := checkTiKVVersion(local.tls, local.pdAddr, localMinTiKVVersion); err != nil {
+	if err := checkTiKVVersion(ctx, local.tls, local.pdAddr, localMinTiKVVersion); err != nil {
 		return err
 	}
 	return nil
 }
 
 func (local *local) FetchRemoteTableModels(ctx context.Context, schemaName string) ([]*model.TableInfo, error) {
-	return fetchRemoteTableModelsFromTLS(local.tls, schemaName)
+	return fetchRemoteTableModelsFromTLS(ctx, local.tls, schemaName)
 }
 
 func (local *local) WriteRows(
diff --git a/lightning/backend/tidb.go b/lightning/backend/tidb.go
index d3a75d8..748d363 100644
--- a/lightning/backend/tidb.go
+++ b/lightning/backend/tidb.go
@@ -285,7 +285,7 @@ func (be *tidbBackend) ShouldPostProcess() bool {
 	return false
 }
 
-func (be *tidbBackend) CheckRequirements() error {
+func (be *tidbBackend) CheckRequirements(ctx context.Context) error {
 	log.L().Info("skipping check requirements for tidb backend")
 	return nil
 }
diff --git a/lightning/backend/tikv.go b/lightning/backend/tikv.go
index 8e90680..cb12883 100644
--- a/lightning/backend/tikv.go
+++ b/lightning/backend/tikv.go
@@ -113,7 +113,7 @@ func ForAllStores(
 			Store Store
 		}
 	}
-	err := tls.GetJSON("/pd/api/v1/stores", &stores)
+	err := tls.GetJSON(ctx, "/pd/api/v1/stores", &stores)
 	if err != nil {
 		return err
 	}
diff --git a/lightning/common/security.go b/lightning/common/security.go
index a7e7449..4035b5d 100644
--- a/lightning/common/security.go
+++ b/lightning/common/security.go
@@ -14,6 +14,7 @@
 package common
 
 import (
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"io/ioutil"
@@ -144,9 +145,8 @@ func (tc *TLS) WrapListener(l net.Listener) net.Listener {
 	return tls.NewListener(l, tc.inner)
 }
 
-// GetJSON obtains JSON result with the HTTP GET method.
-func (tc *TLS) GetJSON(path string, v interface{}) error {
-	return GetJSON(tc.client, tc.url+path, v)
+func (tc *TLS) GetJSON(ctx context.Context, path string, v interface{}) error {
+	return GetJSON(ctx, tc.client, tc.url+path, v)
 }
 
 func (tc *TLS) ToPDSecurityOption() pd.SecurityOption {
diff --git a/lightning/common/util.go b/lightning/common/util.go
index 909c264..3265990 100644
--- a/lightning/common/util.go
+++ b/lightning/common/util.go
@@ -295,11 +295,17 @@ func EscapeMySQLSingleQuote(builder *strings.Builder, s string) {
 //		return errors.Trace(err)
 //	}
 //	fmt.Println(resp.IP)
-func GetJSON(client *http.Client, url string, v interface{}) error {
-	resp, err := client.Get(url)
+func GetJSON(ctx context.Context, client *http.Client, url string, v interface{}) error {
+	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
 	if err != nil {
 		return errors.Trace(err)
 	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
diff --git a/lightning/common/version.go b/lightning/common/version.go
index f80ed78..5931f63 100644
--- a/lightning/common/version.go
+++ b/lightning/common/version.go
@@ -14,6 +14,7 @@
 package common
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
@@ -63,9 +64,9 @@ func PrintInfo(app string, callback func()) {
 }
 
 // FetchPDVersion get pd version
-func FetchPDVersion(tls *TLS, pdAddr string) (*semver.Version, error) {
+func FetchPDVersion(ctx context.Context, tls *TLS, pdAddr string) (*semver.Version, error) {
 	var rawVersion string
-	err := tls.WithHost(pdAddr).GetJSON("/pd/api/v1/config/cluster-version", &rawVersion)
+	err := tls.WithHost(pdAddr).GetJSON(ctx, "/pd/api/v1/config/cluster-version", &rawVersion)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
diff --git a/lightning/config/config.go b/lightning/config/config.go
index 2ab3c5b..c126dca 100644
--- a/lightning/config/config.go
+++ b/lightning/config/config.go
@@ -14,6 +14,7 @@
 package config
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net"
@@ -459,7 +460,7 @@ func (cfg *Config) LoadFromTOML(data []byte) error {
 }
 
 // Adjust fixes the invalid or unspecified settings to reasonable valid values.
-func (cfg *Config) Adjust() error {
+func (cfg *Config) Adjust(ctx context.Context) error {
 	// Reject problematic CSV configurations.
 	csv := &cfg.Mydumper.CSV
 	if len(csv.Separator) == 0 {
@@ -594,7 +595,7 @@ func (cfg *Config) Adjust() error {
 		}
 
 		var settings tidbcfg.Config
-		err = tls.GetJSON("/settings", &settings)
+		err = tls.GetJSON(ctx, "/settings", &settings)
 		if err != nil {
 			return errors.Annotate(err, "cannot fetch settings from TiDB, please manually fill in `tidb.port` and `tidb.pd-addr`")
 		}
diff --git a/lightning/lightning.go b/lightning/lightning.go
index ed5a74d..df70544 100755
--- a/lightning/lightning.go
+++ b/lightning/lightning.go
@@ -176,7 +176,7 @@ func (l *Lightning) goServe(statusAddr string, realAddrWriter io.Writer) error {
 // - for lightning as a library, taskCtx could be a meaningful context that get canceled outside, and glue could be a
 //   caller implemented glue.
 func (l *Lightning) RunOnce(taskCtx context.Context, taskCfg *config.Config, glue glue.Glue, replaceLogger *zap.Logger) error {
-	if err := taskCfg.Adjust(); err != nil {
+	if err := taskCfg.Adjust(taskCtx); err != nil {
 		return err
 	}
 
@@ -467,7 +467,7 @@ func (l *Lightning) handlePostTask(w http.ResponseWriter, req *http.Request) {
 		writeJSONError(w, http.StatusBadRequest, "cannot parse task (must be TOML)", err)
 		return
 	}
-	if err = cfg.Adjust(); err != nil {
+	if err = cfg.Adjust(l.ctx); err != nil {
 		writeJSONError(w, http.StatusBadRequest, "invalid task configuration", err)
 		return
 	}
diff --git a/lightning/restore/checksum.go b/lightning/restore/checksum.go
index 6ba0a78..8dc4b4d 100644
--- a/lightning/restore/checksum.go
+++ b/lightning/restore/checksum.go
@@ -49,14 +49,14 @@ type ChecksumManager interface {
 	Checksum(ctx context.Context, tableInfo *TidbTableInfo) (*RemoteChecksum, error)
 }
 
-func newChecksumManager(rc *RestoreController) (ChecksumManager, error) {
+func newChecksumManager(ctx context.Context, rc *RestoreController) (ChecksumManager, error) {
 	// if we don't need checksum, just return nil
 	if rc.cfg.TikvImporter.Backend == config.BackendTiDB || rc.cfg.PostRestore.Checksum == config.OpLevelOff {
 		return nil, nil
 	}
 
 	pdAddr := rc.cfg.TiDB.PdAddr
-	pdVersion, err := common.FetchPDVersion(rc.tls, pdAddr)
+	pdVersion, err := common.FetchPDVersion(ctx, rc.tls, pdAddr)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -65,7 +65,7 @@ func newChecksumManager(rc *RestoreController) (ChecksumManager, error) {
 	var manager ChecksumManager
 	if pdVersion.Major >= 4 {
 		tlsOpt := rc.tls.ToPDSecurityOption()
-		pdCli, err := pd.NewClient([]string{pdAddr}, tlsOpt)
+		pdCli, err := pd.NewClientWithContext(ctx, []string{pdAddr}, tlsOpt)
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
diff --git a/lightning/restore/restore.go b/lightning/restore/restore.go
index 011d7c4..be5fd8c 100644
--- a/lightning/restore/restore.go
+++ b/lightning/restore/restore.go
@@ -667,7 +667,7 @@ func (rc *RestoreController) restoreTables(ctx context.Context) error {
 	taskCh := make(chan task, rc.cfg.App.IndexConcurrency)
 	defer close(taskCh)
 
-	manager, err := newChecksumManager(rc)
+	manager, err := newChecksumManager(ctx, rc)
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -1290,12 +1290,12 @@ func (rc *RestoreController) switchTiKVMode(ctx context.Context, mode sstpb.Swit
 	)
 }
 
-func (rc *RestoreController) checkRequirements(_ context.Context) error {
+func (rc *RestoreController) checkRequirements(ctx context.Context) error {
 	// skip requirement check if explicitly turned off
 	if !rc.cfg.App.CheckRequirements {
 		return nil
 	}
-	return rc.backend.CheckRequirements()
+	return rc.backend.CheckRequirements(ctx)
 }
 
 func (rc *RestoreController) setGlobalVariables(ctx context.Context) error {
diff --git a/mock/backend.go b/mock/backend.go
index cebc7d4..6183b75 100644
--- a/mock/backend.go
+++ b/mock/backend.go
@@ -46,17 +46,17 @@ func (m *MockBackend) EXPECT() *MockBackendMockRecorder {
 }
 
 // CheckRequirements mocks base method
-func (m *MockBackend) CheckRequirements() error {
+func (m *MockBackend) CheckRequirements(arg0 context.Context) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "CheckRequirements")
+	ret := m.ctrl.Call(m, "CheckRequirements", arg0)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // CheckRequirements indicates an expected call of CheckRequirements
-func (mr *MockBackendMockRecorder) CheckRequirements() *gomock.Call {
+func (mr *MockBackendMockRecorder) CheckRequirements(arg0 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CheckRequirements", reflect.TypeOf((*MockBackend)(nil).CheckRequirements))
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CheckRequirements", reflect.TypeOf((*MockBackend)(nil).CheckRequirements), arg0)
 }
 
 // CleanupEngine mocks base method


Test patch:
diff --git a/lightning/backend/checkreq_test.go b/lightning/backend/checkreq_test.go
index 53a667b..1753310 100644
--- a/lightning/backend/checkreq_test.go
+++ b/lightning/backend/checkreq_test.go
@@ -14,6 +14,7 @@
 package backend
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -40,6 +41,7 @@ func (s *checkReqSuite) TestCheckVersion(c *C) {
 
 func (s *checkReqSuite) TestCheckTiDBVersion(c *C) {
 	var version string
+	ctx := context.Background()
 
 	mockServer := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		c.Assert(req.URL.Path, Equals, "/status")
@@ -53,14 +55,15 @@ func (s *checkReqSuite) TestCheckTiDBVersion(c *C) {
 	tls := common.NewTLSFromMockServer(mockServer)
 
 	version = "5.7.25-TiDB-v9999.0.0"
-	c.Assert(checkTiDBVersion(tls, requiredTiDBVersion), IsNil)
+	c.Assert(checkTiDBVersionByTLS(ctx, tls, requiredTiDBVersion), IsNil)
 
 	version = "5.7.25-TiDB-v1.0.0"
-	c.Assert(checkTiDBVersion(tls, requiredTiDBVersion), ErrorMatches, "TiDB version too old.*")
+	c.Assert(checkTiDBVersionByTLS(ctx, tls, requiredTiDBVersion), ErrorMatches, "TiDB version too old.*")
 }
 
 func (s *checkReqSuite) TestCheckPDVersion(c *C) {
 	var version string
+	ctx := context.Background()
 
 	mockServer := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		c.Assert(req.URL.Path, Equals, "/pd/api/v1/config/cluster-version")
@@ -74,14 +77,15 @@ func (s *checkReqSuite) TestCheckPDVersion(c *C) {
 	tls := common.NewTLSFromMockServer(mockServer)
 
 	version = "9999.0.0"
-	c.Assert(checkPDVersion(tls, mockURL.Host, requiredPDVersion), IsNil)
+	c.Assert(checkPDVersion(ctx, tls, mockURL.Host, requiredPDVersion), IsNil)
 
 	version = "1.0.0"
-	c.Assert(checkPDVersion(tls, mockURL.Host, requiredPDVersion), ErrorMatches, "PD version too old.*")
+	c.Assert(checkPDVersion(ctx, tls, mockURL.Host, requiredPDVersion), ErrorMatches, "PD version too old.*")
 }
 
 func (s *checkReqSuite) TestCheckTiKVVersion(c *C) {
 	var versions []string
+	ctx := context.Background()
 
 	mockServer := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		c.Assert(req.URL.Path, Equals, "/pd/api/v1/stores")
@@ -108,11 +112,11 @@ func (s *checkReqSuite) TestCheckTiKVVersion(c *C) {
 	tls := common.NewTLSFromMockServer(mockServer)
 
 	versions = []string{"9999.0.0", "9999.0.0"}
-	c.Assert(checkTiKVVersion(tls, mockURL.Host, requiredTiKVVersion), IsNil)
+	c.Assert(checkTiKVVersion(ctx, tls, mockURL.Host, requiredTiKVVersion), IsNil)
 
 	versions = []string{"4.1.0", "v4.1.0-alpha-9-ga27a7dd"}
-	c.Assert(checkTiKVVersion(tls, mockURL.Host, requiredTiKVVersion), IsNil)
+	c.Assert(checkTiKVVersion(ctx, tls, mockURL.Host, requiredTiKVVersion), IsNil)
 
 	versions = []string{"9999.0.0", "1.0.0"}
-	c.Assert(checkTiKVVersion(tls, mockURL.Host, requiredTiKVVersion), ErrorMatches, `TiKV \(at tikv1\.test:20160\) version too old.*`)
+	c.Assert(checkTiKVVersion(ctx, tls, mockURL.Host, requiredTiKVVersion), ErrorMatches, `TiKV \(at tikv1\.test:20160\) version too old.*`)
 }
diff --git a/lightning/common/security_test.go b/lightning/common/security_test.go
index b958d33..490e0e5 100644
--- a/lightning/common/security_test.go
+++ b/lightning/common/security_test.go
@@ -14,6 +14,7 @@
 package common_test
 
 import (
+	"context"
 	"io"
 	"io/ioutil"
 	"net/http"
@@ -40,6 +41,7 @@ func (s *securitySuite) TestGetJSONInsecure(c *C) {
 	mockServer := httptest.NewServer(http.HandlerFunc(respondPathHandler))
 	defer mockServer.Close()
 
+	ctx := context.Background()
 	u, err := url.Parse(mockServer.URL)
 	c.Assert(err, IsNil)
 
@@ -47,10 +49,10 @@ func (s *securitySuite) TestGetJSONInsecure(c *C) {
 	c.Assert(err, IsNil)
 
 	var result struct{ Path string }
-	err = tls.GetJSON("/aaa", &result)
+	err = tls.GetJSON(ctx, "/aaa", &result)
 	c.Assert(err, IsNil)
 	c.Assert(result.Path, Equals, "/aaa")
-	err = tls.GetJSON("/bbbb", &result)
+	err = tls.GetJSON(ctx, "/bbbb", &result)
 	c.Assert(err, IsNil)
 	c.Assert(result.Path, Equals, "/bbbb")
 }
@@ -59,13 +61,14 @@ func (s *securitySuite) TestGetJSONSecure(c *C) {
 	mockServer := httptest.NewTLSServer(http.HandlerFunc(respondPathHandler))
 	defer mockServer.Close()
 
+	ctx := context.Background()
 	tls := common.NewTLSFromMockServer(mockServer)
 
 	var result struct{ Path string }
-	err := tls.GetJSON("/ccc", &result)
+	err := tls.GetJSON(ctx, "/ccc", &result)
 	c.Assert(err, IsNil)
 	c.Assert(result.Path, Equals, "/ccc")
-	err = tls.GetJSON("/dddd", &result)
+	err = tls.GetJSON(ctx, "/dddd", &result)
 	c.Assert(err, IsNil)
 	c.Assert(result.Path, Equals, "/dddd")
 }
diff --git a/lightning/common/util_test.go b/lightning/common/util_test.go
index 9071540..ca022cf 100644
--- a/lightning/common/util_test.go
+++ b/lightning/common/util_test.go
@@ -54,6 +54,7 @@ func (s *utilSuite) TestGetJSON(c *C) {
 		Password: "lightning-ctl",
 	}
 
+	ctx := context.Background()
 	// Mock success response
 	handle := func(res http.ResponseWriter, req *http.Request) {
 		res.WriteHeader(http.StatusOK)
@@ -68,9 +69,9 @@ func (s *utilSuite) TestGetJSON(c *C) {
 	client := &http.Client{Timeout: time.Second}
 
 	response := TestPayload{}
-	err := common.GetJSON(client, "http://not-exists", &response)
+	err := common.GetJSON(ctx, client, "http://not-exists", &response)
 	c.Assert(err, NotNil)
-	err = common.GetJSON(client, testServer.URL, &response)
+	err = common.GetJSON(ctx, client, testServer.URL, &response)
 	c.Assert(err, IsNil)
 	c.Assert(request, DeepEquals, response)
 
@@ -78,7 +79,7 @@ func (s *utilSuite) TestGetJSON(c *C) {
 	handle = func(res http.ResponseWriter, req *http.Request) {
 		res.WriteHeader(http.StatusNoContent)
 	}
-	err = common.GetJSON(client, testServer.URL, &response)
+	err = common.GetJSON(ctx, client, testServer.URL, &response)
 	c.Assert(err, NotNil)
 	c.Assert(err, ErrorMatches, ".*http status code != 200.*")
 }
diff --git a/lightning/config/config_test.go b/lightning/config/config_test.go
index a848daf..0477f07 100644
--- a/lightning/config/config_test.go
+++ b/lightning/config/config_test.go
@@ -15,6 +15,7 @@ package config_test
 
 import (
 	"bytes"
+	"context"
 	"flag"
 	"fmt"
 	"net"
@@ -77,7 +78,7 @@ func (s *configTestSuite) TestAdjustPdAddrAndPort(c *C) {
 	cfg.TiDB.StatusPort = port
 	cfg.Mydumper.SourceDir = "."
 
-	err := cfg.Adjust()
+	err := cfg.Adjust(context.Background())
 	c.Assert(err, IsNil)
 	c.Assert(cfg.TiDB.Port, Equals, 4444)
 	c.Assert(cfg.TiDB.PdAddr, Equals, "123.45.67.89:1234")
@@ -94,7 +95,7 @@ func (s *configTestSuite) TestAdjustPdAddrAndPortViaAdvertiseAddr(c *C) {
 	cfg.TiDB.StatusPort = port
 	cfg.Mydumper.SourceDir = "."
 
-	err := cfg.Adjust()
+	err := cfg.Adjust(context.Background())
 	c.Assert(err, IsNil)
 	c.Assert(cfg.TiDB.Port, Equals, 6666)
 	c.Assert(cfg.TiDB.PdAddr, Equals, "34.34.34.34:3434")
@@ -108,7 +109,7 @@ func (s *configTestSuite) TestAdjustPageNotFound(c *C) {
 	cfg.TiDB.Host = host
 	cfg.TiDB.StatusPort = port
 
-	err := cfg.Adjust()
+	err := cfg.Adjust(context.Background())
 	c.Assert(err, ErrorMatches, "cannot fetch settings from TiDB.*")
 }
 
@@ -121,14 +122,14 @@ func (s *configTestSuite) TestAdjustConnectRefused(c *C) {
 
 	ts.Close() // immediately close to ensure connection refused.
 
-	err := cfg.Adjust()
+	err := cfg.Adjust(context.Background())
 	c.Assert(err, ErrorMatches, "cannot fetch settings from TiDB.*")
 }
 
 func (s *configTestSuite) TestAdjustInvalidBackend(c *C) {
 	cfg := config.NewConfig()
 	cfg.TikvImporter.Backend = "no_such_backend"
-	err := cfg.Adjust()
+	err := cfg.Adjust(context.Background())
 	c.Assert(err, ErrorMatches, "invalid config: unsupported `tikv-importer\\.backend` \\(no_such_backend\\)")
 }
 
@@ -136,17 +137,18 @@ func (s *configTestSuite) TestAdjustFileRoutePath(c *C) {
 	cfg := config.NewConfig()
 	assignMinimalLegalValue(cfg)
 
+	ctx := context.Background()
 	tmpDir := c.MkDir()
 	cfg.Mydumper.SourceDir = tmpDir
 	invalidPath := filepath.Join(tmpDir, "../test123/1.sql")
 	rule := &config.FileRouteRule{Path: invalidPath, Type: "sql", Schema: "test", Table: "tbl"}
 	cfg.Mydumper.FileRouters = []*config.FileRouteRule{rule}
-	err := cfg.Adjust()
+	err := cfg.Adjust(ctx)
 	c.Assert(err, ErrorMatches, fmt.Sprintf("file route path '%s' is not in source dir '%s'", invalidPath, tmpDir))
 
 	relPath := "test_dir/1.sql"
 	rule.Path = filepath.Join(tmpDir, relPath)
-	err = cfg.Adjust()
+	err = cfg.Adjust(ctx)
 	c.Assert(err, IsNil)
 	c.Assert(cfg.Mydumper.FileRouters[0].Path, Equals, relPath)
 }
@@ -159,7 +161,7 @@ func (s *configTestSuite) TestDecodeError(c *C) {
 	cfg.TiDB.Host = host
 	cfg.TiDB.StatusPort = port
 
-	err := cfg.Adjust()
+	err := cfg.Adjust(context.Background())
 	c.Assert(err, ErrorMatches, "cannot fetch settings from TiDB.*")
 }
 
@@ -171,7 +173,7 @@ func (s *configTestSuite) TestInvalidSetting(c *C) {
 	cfg.TiDB.Host = host
 	cfg.TiDB.StatusPort = port
 
-	err := cfg.Adjust()
+	err := cfg.Adjust(context.Background())
 	c.Assert(err, ErrorMatches, "invalid `tidb.port` setting")
 }
 
@@ -183,7 +185,7 @@ func (s *configTestSuite) TestInvalidPDAddr(c *C) {
 	cfg.TiDB.Host = host
 	cfg.TiDB.StatusPort = port
 
-	err := cfg.Adjust()
+	err := cfg.Adjust(context.Background())
 	c.Assert(err, ErrorMatches, "invalid `tidb.pd-addr` setting")
 }
 
@@ -191,7 +193,7 @@ func (s *configTestSuite) TestAdjustWillNotContactServerIfEverythingIsDefined(c
 	cfg := config.NewConfig()
 	assignMinimalLegalValue(cfg)
 
-	err := cfg.Adjust()
+	err := cfg.Adjust(context.Background())
 	c.Assert(err, IsNil)
 	c.Assert(cfg.TiDB.Port, Equals, 4567)
 	c.Assert(cfg.TiDB.PdAddr, Equals, "234.56.78.90:12345")
@@ -201,7 +203,7 @@ func (s *configTestSuite) TestAdjustWillBatchImportRatioInvalid(c *C) {
 	cfg := config.NewConfig()
 	assignMinimalLegalValue(cfg)
 	cfg.Mydumper.BatchImportRatio = -1
-	err := cfg.Adjust()
+	err := cfg.Adjust(context.Background())
 	c.Assert(err, IsNil)
 	c.Assert(cfg.Mydumper.BatchImportRatio, Equals, 0.75)
 }
@@ -280,7 +282,7 @@ func (s *configTestSuite) TestAdjustSecuritySection(c *C) {
 		err := cfg.LoadFromTOML([]byte(tc.input))
 		c.Assert(err, IsNil, comment)
 
-		err = cfg.Adjust()
+		err = cfg.Adjust(context.Background())
 		c.Assert(err, IsNil, comment)
 		c.Assert(cfg.TiDB.Security.CAPath, Equals, tc.expectedCA, comment)
 		c.Assert(cfg.TiDB.TLS, Equals, tc.expectedTLS, comment)
@@ -419,7 +421,7 @@ func (s *configTestSuite) TestInvalidCSV(c *C) {
 		err := cfg.LoadFromTOML([]byte(tc.input))
 		c.Assert(err, IsNil)
 
-		err = cfg.Adjust()
+		err = cfg.Adjust(context.Background())
 		if tc.err != "" {
 			c.Assert(err, ErrorMatches, regexp.QuoteMeta(tc.err), comment)
 		} else {
@@ -517,7 +519,7 @@ func (s *configTestSuite) TestLoadConfig(c *C) {
 
 	taskCfg.Checkpoint.DSN = ""
 	taskCfg.Checkpoint.Driver = config.CheckpointDriverMySQL
-	err = taskCfg.Adjust()
+	err = taskCfg.Adjust(context.Background())
 	c.Assert(err, IsNil)
 	c.Assert(taskCfg.Checkpoint.DSN, Equals, "guest:12345@tcp(172.16.30.11:4001)/?charset=utf8mb4&sql_mode='"+mysql.DefaultSQLMode+"'&maxAllowedPacket=67108864&tls=false")
 
@@ -529,7 +531,7 @@ func (s *configTestSuite) TestDefaultImporterBackendValue(c *C) {
 	cfg := config.NewConfig()
 	assignMinimalLegalValue(cfg)
 	cfg.TikvImporter.Backend = "importer"
-	err := cfg.Adjust()
+	err := cfg.Adjust(context.Background())
 	c.Assert(err, IsNil)
 	c.Assert(cfg.App.IndexConcurrency, Equals, 2)
 	c.Assert(cfg.App.TableConcurrency, Equals, 6)
@@ -540,7 +542,7 @@ func (s *configTestSuite) TestDefaultTidbBackendValue(c *C) {
 	assignMinimalLegalValue(cfg)
 	cfg.TikvImporter.Backend = "tidb"
 	cfg.App.RegionConcurrency = 123
-	err := cfg.Adjust()
+	err := cfg.Adjust(context.Background())
 	c.Assert(err, IsNil)
 	c.Assert(cfg.App.IndexConcurrency, Equals, 123)
 	c.Assert(cfg.App.TableConcurrency, Equals, 123)
@@ -552,7 +554,7 @@ func (s *configTestSuite) TestDefaultCouldBeOverwritten(c *C) {
 	cfg.TikvImporter.Backend = "importer"
 	cfg.App.IndexConcurrency = 20
 	cfg.App.TableConcurrency = 60
-	err := cfg.Adjust()
+	err := cfg.Adjust(context.Background())
 	c.Assert(err, IsNil)
 	c.Assert(cfg.App.IndexConcurrency, Equals, 20)
 	c.Assert(cfg.App.TableConcurrency, Equals, 60)
@@ -638,14 +640,15 @@ func (s *configTestSuite) TestAdjustWithLegacyBlackWhiteList(c *C) {
 	c.Assert(cfg.Mydumper.Filter, DeepEquals, config.DefaultFilter)
 	c.Assert(cfg.HasLegacyBlackWhiteList(), IsFalse)
 
+	ctx := context.Background()
 	cfg.Mydumper.Filter = []string{"test.*"}
-	c.Assert(cfg.Adjust(), IsNil)
+	c.Assert(cfg.Adjust(ctx), IsNil)
 	c.Assert(cfg.HasLegacyBlackWhiteList(), IsFalse)
 
 	cfg.BWList.DoDBs = []string{"test"}
-	c.Assert(cfg.Adjust(), ErrorMatches, "invalid config: `mydumper\\.filter` and `black-white-list` cannot be simultaneously defined")
+	c.Assert(cfg.Adjust(ctx), ErrorMatches, "invalid config: `mydumper\\.filter` and `black-white-list` cannot be simultaneously defined")
 
 	cfg.Mydumper.Filter = config.DefaultFilter
-	c.Assert(cfg.Adjust(), IsNil)
+	c.Assert(cfg.Adjust(ctx), IsNil)
 	c.Assert(cfg.HasLegacyBlackWhiteList(), IsTrue)
 }