sync-diff-inspector: add session configuration in toml #847
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -20,13 +20,17 @@ check-struct-only = false | |
| port = 3306 | ||
| user = "root" | ||
| password = "" | ||
| session.max_execution_time = 3600 | ||
| session.net_read_timeout = 60 | ||
| # MySQL doesn't has snapshot config | ||
|
|
||
| [data-sources.tidb0] | ||
| host = "127.0.0.1" | ||
| port = 4000 | ||
| user = "root" | ||
| password = "" | ||
| session.tidb_opt_prefer_range_scan = "ON" | ||
| session.max_execution_time = 86400 | ||
|
|
||
| # Support tls connection | ||
| # security.ca-path = "..." | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -49,6 +49,8 @@ check-struct-only = false | |
| port = 4000 | ||
| user = "root" | ||
| password = "" | ||
| session.tidb_opt_prefer_range_scan = "ON" | ||
| session.max_execution_time = 86400 | ||
| # remove comment if use tidb's snapshot data | ||
| # snapshot = "2016-10-08 16:45:26" | ||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -50,10 +50,10 @@ func TestParseConfig(t *testing.T) { | |
|
|
||
| // we might not use the same config to run this test. e.g. MYSQL_PORT can be 4000 | ||
| require.JSONEq(t, cfg.String(), | ||
| "{\"check-thread-count\":4,\"split-thread-count\":5,\"export-fix-sql\":true,\"check-struct-only\":false,\"dm-addr\":\"\",\"dm-task\":\"\",\"data-sources\":{\"mysql1\":{\"host\":\"127.0.0.1\",\"port\":3306,\"user\":\"root\",\"password\":\"******\",\"sql-mode\":\"\",\"snapshot\":\"\",\"security\":null,\"route-rules\":[\"rule1\",\"rule2\"],\"Router\":{\"Selector\":{}},\"Conn\":null,\"session\":null},\"mysql2\":{\"host\":\"127.0.0.1\",\"port\":3306,\"user\":\"root\",\"password\":\"******\",\"sql-mode\":\"\",\"snapshot\":\"\",\"security\":null,\"route-rules\":[\"rule1\",\"rule2\"],\"Router\":{\"Selector\":{}},\"Conn\":null,\"session\":null},\"mysql3\":{\"host\":\"127.0.0.1\",\"port\":3306,\"user\":\"root\",\"password\":\"******\",\"sql-mode\":\"\",\"snapshot\":\"\",\"security\":null,\"route-rules\":[\"rule1\",\"rule3\"],\"Router\":{\"Selector\":{}},\"Conn\":null,\"session\":null},\"tidb0\":{\"host\":\"127.0.0.1\",\"port\":4000,\"user\":\"root\",\"password\":\"******\",\"sql-mode\":\"\",\"snapshot\":\"\",\"security\":null,\"route-rules\":null,\"Router\":{\"Selector\":{}},\"Conn\":null,\"session\":{\"max_execution_time\":86400,\"tidb_opt_prefer_range_scan\":\"ON\"}}},\"routes\":{\"rule1\":{\"schema-pattern\":\"test_*\",\"table-pattern\":\"t_*\",\"target-schema\":\"test\",\"target-table\":\"t\"},\"rule2\":{\"schema-pattern\":\"test2_*\",\"table-pattern\":\"t2_*\",\"target-schema\":\"test2\",\"target-table\":\"t2\"},\"rule3\":{\"schema-pattern\":\"test2_*\",\"table-pattern\":\"t2_*\",\"target-schema\":\"test\",\"target-table\":\"t\"}},\"table-configs\":{\"config1\":{\"target-tables\":[\"schema*.table*\",\"test2.t2\"],\"Schema\":\"\",\"Table\":\"\",\"ConfigIndex\":0,\"HasMatched\":false,\"IgnoreColumns\":[\"\",\"\"],\"Fields\":[\"\"],\"Range\":\"age \\u003e 10 AND age \\u003c 20\",\"TargetTableInfo\":null,\"Collation\":\"\",\"chunk-size\":0}},\"task\":{\"source-instances\":[\"mysql1\",\"mysql2\",\"mysql3\"],\"source-routes\":null,\"target-instance\":\"tidb0\",\"target-check-tables\":[\"schema*.table*\",\"!c.*\",\"test2.t2\"],\"target-configs\":[\"config1\"],\"output-dir\":\"/tmp/output/config\",\"SourceInstances\":[{\"host\":\"127.0.0.1\",\"port\":3306,\"user\":\"root\",\"password\":\"******\",\"sql-mode\":\"\",\"snapshot\":\"\",\"security\":null,\"route-rules\":[\"rule1\",\"rule2\"],\"Router\":{\"Selector\":{}},\"Conn\":null,\"session\":null},{\"host\":\"127.0.0.1\",\"port\":3306,\"user\":\"root\",\"password\":\"******\",\"sql-mode\":\"\",\"snapshot\":\"\",\"security\":null,\"route-rules\":[\"rule1\",\"rule2\"],\"Router\":{\"Selector\":{}},\"Conn\":null,\"session\":null},{\"host\":\"127.0.0.1\",\"port\":3306,\"user\":\"root\",\"password\":\"******\",\"sql-mode\":\"\",\"snapshot\":\"\",\"security\":null,\"route-rules\":[\"rule1\",\"rule3\"],\"Router\":{\"Selector\":{}},\"Conn\":null,\"session\":null}],\"TargetInstance\":{\"host\":\"127.0.0.1\",\"port\":4000,\"user\":\"root\",\"password\":\"******\",\"sql-mode\":\"\",\"snapshot\":\"\",\"security\":null,\"route-rules\":null,\"Router\":{\"Selector\":{}},\"Conn\":null,\"session\":{\"max_execution_time\":86400,\"tidb_opt_prefer_range_scan\":\"ON\"}},\"TargetTableConfigs\":[{\"target-tables\":[\"schema*.table*\",\"test2.t2\"],\"Schema\":\"\",\"Table\":\"\",\"ConfigIndex\":0,\"HasMatched\":false,\"IgnoreColumns\":[\"\",\"\"],\"Fields\":[\"\"],\"Range\":\"age \\u003e 10 AND age \\u003c 20\",\"TargetTableInfo\":null,\"Collation\":\"\",\"chunk-size\":0}],\"TargetCheckTables\":[{},{},{}],\"FixDir\":\"/tmp/output/config/fix-on-tidb0\",\"CheckpointDir\":\"/tmp/output/config/checkpoint\",\"HashFile\":\"\"},\"ConfigFile\":\"config_sharding.toml\",\"PrintVersion\":false}") | ||
| hash, err := cfg.Task.ComputeConfigHash() | ||
| require.NoError(t, err) | ||
| require.Equal(t, hash, "553515158a9da2724195f6f71eb8daeebb605bc7de8f3dd9b7447468d6f385b9") | ||
|
|
||
| require.True(t, cfg.TableConfigs["config1"].Valid()) | ||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -16,11 +16,15 @@ package common | |||||
| import ( | ||||||
| "database/sql" | ||||||
| "encoding/base64" | ||||||
| "fmt" | ||||||
|
|
||||||
| "github.com/go-sql-driver/mysql" | ||||||
| "github.com/pingcap/errors" | ||||||
| "github.com/pingcap/failpoint" | ||||||
| "github.com/pingcap/log" | ||||||
| "github.com/pingcap/tidb-tools/sync_diff_inspector/config" | ||||||
| tmysql "github.com/pingcap/tidb/pkg/parser/mysql" | ||||||
| "go.uber.org/zap" | ||||||
| ) | ||||||
|
|
||||||
| func tryConnectMySQL(cfg *mysql.Config) (*sql.DB, error) { | ||||||
|
|
@@ -43,8 +47,39 @@ func tryConnectMySQL(cfg *mysql.Config) (*sql.DB, error) { | |||||
| return db, nil | ||||||
| } | ||||||
|
|
||||||
| func verifyParams(db *sql.DB, sessionCfg *config.SessionConfig) error { | ||||||
| if sessionCfg == nil { | ||||||
| return nil | ||||||
| } | ||||||
| for param, value := range *sessionCfg { | ||||||
| res, err := db.Query(fmt.Sprintf("show session variables like '%s'", param)) | ||||||
|
There was a problem hiding this comment.
Suggested change
|
||||||
| if err != nil { | ||||||
| return err | ||||||
| } | ||||||
| //nolint: errcheck | ||||||
| defer res.Close() | ||||||
| if res.Next() { | ||||||
| var paramName, actual string | ||||||
| if err := res.Scan(¶mName, &actual); err != nil { | ||||||
| return err | ||||||
| } | ||||||
| expected := fmt.Sprintf("%v", value) | ||||||
|
There was a problem hiding this comment.
Suggested change
|
||||||
| if actual != expected { | ||||||
| log.Warn("The session variable was set, but the database returned a different value", | ||||||
| zap.String("variable", param), | ||||||
| zap.String("configured_value", expected), | ||||||
| zap.String("effective_value", actual), | ||||||
| ) | ||||||
| } | ||||||
| } else { | ||||||
| return fmt.Errorf("parameter %s not found", param) | ||||||
| } | ||||||
| } | ||||||
| return nil | ||||||
| } | ||||||
|
|
||||||
| // ConnectMySQL creates sql.DB used for select data | ||||||
| func ConnectMySQL(sessionCfg *config.SessionConfig, cfg *mysql.Config, num int) (db *sql.DB, err error) { | ||||||
| defer func() { | ||||||
| if err == nil && db != nil { | ||||||
| // SetMaxOpenConns and SetMaxIdleConns for connection to avoid error like | ||||||
|
|
@@ -55,7 +90,7 @@ func ConnectMySQL(cfg *mysql.Config, num int) (db *sql.DB, err error) { | |||||
| }() | ||||||
| // Try plain password first. | ||||||
| db, firstErr := tryConnectMySQL(cfg) | ||||||
| if firstErr == nil && verifyParams(db, sessionCfg) == nil { | ||||||
| return db, nil | ||||||
| } | ||||||
| // If access is denied and password is encoded by base64, try the decoded string as well. | ||||||
|
|
@@ -64,7 +99,7 @@ func ConnectMySQL(cfg *mysql.Config, num int) (db *sql.DB, err error) { | |||||
| if password, decodeErr := base64.StdEncoding.DecodeString(cfg.Passwd); decodeErr == nil && string(password) != cfg.Passwd { | ||||||
| cfg.Passwd = string(password) | ||||||
| db2, err := tryConnectMySQL(cfg) | ||||||
| if err == nil && verifyParams(db2, sessionCfg) == nil { | ||||||
| return db2, nil | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -40,9 +40,9 @@ func TestConnect(t *testing.T) { | |
| User: "root", | ||
| Password: utils.SecretString(plainPsw), | ||
| } | ||
| _, err := ConnectMySQL(nil, dataSource.ToDriverConfig(), 2) | ||
| require.NoError(t, err) | ||
| dataSource.Password = utils.SecretString(base64.StdEncoding.EncodeToString([]byte(plainPsw))) | ||
| _, err = ConnectMySQL(nil, dataSource.ToDriverConfig(), 2) | ||
| require.NoError(t, err) | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||||||
|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -16,6 +16,7 @@ package source | |||||||||
| import ( | ||||||||||
| "context" | ||||||||||
| "database/sql" | ||||||||||
| "fmt" | ||||||||||
| "sort" | ||||||||||
| "strings" | ||||||||||
| "time" | ||||||||||
|
|
@@ -237,7 +238,7 @@ func buildSourceFromCfg(ctx context.Context, tableDiffs []*common.TableDiff, con | |||||||||
| } | ||||||||||
|
|
||||||||||
| func getAutoSnapshotPosition(cfg *mysql.Config) (string, string, error) { | ||||||||||
| tmpConn, err := common.ConnectMySQL(nil, cfg, 2) | ||||||||||
| if err != nil { | ||||||||||
| return "", "", errors.Annotatef(err, "connecting to auto-position tidb_snapshot failed") | ||||||||||
| } | ||||||||||
|
|
@@ -269,22 +270,36 @@ func initDBConn(ctx context.Context, cfg *config.Config) error { | |||||||||
| } | ||||||||||
| // we had `cfg.SplitThreadCount` producers and `cfg.CheckThreadCount` consumer to use db connections maybe and `cfg.CheckThreadCount` splitter to split buckets. | ||||||||||
| // so the connection count need to be cfg.SplitThreadCount + cfg.CheckThreadCount + cfg.CheckThreadCount. | ||||||||||
| targetConn, err := common.ConnectMySQL( | ||||||||||
| &cfg.Task.TargetInstance.SessionConfig, | ||||||||||
| cfg.Task.TargetInstance.ToDriverConfig(), | ||||||||||
| cfg.SplitThreadCount+2*cfg.CheckThreadCount, | ||||||||||
| ) | ||||||||||
| if err != nil { | ||||||||||
| log.Error(fmt.Sprintf("failed to configure session for data source '%s'", cfg.Task.Target), | ||||||||||
| zap.String("error", err.Error()), | ||||||||||
| ) | ||||||||||
|
There was a problem hiding this comment.
Suggested change
|
||||||||||
| return errors.Trace(err) | ||||||||||
| } | ||||||||||
|
|
||||||||||
| cfg.Task.TargetInstance.Conn = targetConn | ||||||||||
|
|
||||||||||
| for sourceIdx, source := range cfg.Task.SourceInstances { | ||||||||||
| // If it is still set to AUTO it means it was not set on the target. | ||||||||||
| // We require it to be set to AUTO on both. | ||||||||||
| if source.IsAutoSnapshot() { | ||||||||||
| return errors.Errorf("'auto' snapshot should be set on both target and source") | ||||||||||
| } | ||||||||||
| // connect source db with target db time_zone | ||||||||||
| conn, err := common.ConnectMySQL( | ||||||||||
| &source.SessionConfig, | ||||||||||
| source.ToDriverConfig(), | ||||||||||
| cfg.SplitThreadCount+2*cfg.CheckThreadCount, | ||||||||||
| ) | ||||||||||
| if err != nil { | ||||||||||
| log.Error(fmt.Sprintf("failed to configure session for data source '%s'", cfg.Task.Source[sourceIdx]), | ||||||||||
| zap.String("error", err.Error()), | ||||||||||
| ) | ||||||||||
|
There was a problem hiding this comment.
Suggested change
|
||||||||||
| return errors.Trace(err) | ||||||||||
| } | ||||||||||
| source.Conn = conn | ||||||||||
|
|
||||||||||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -28,6 +28,8 @@ check-struct-only = false | |
| port = 4000 | ||
| user = "root" | ||
| password = "" | ||
| session.tidb_opt_prefer_range_scan = "ON" | ||
| session.max_execution_time = 86400 | ||
| # remove comment if use tidb's snapshot data | ||
| # snapshot = "2016-10-08 16:45:26" | ||
|
|
||
|
|
||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.