many bug fixes (moiot#150)

* os.Exit(1) when configuration file changed

* introduce official mongo driver. mgo is not maintained anymore, and mgo have bugs when using find for range query. (sometimes it does not return any result when the query is right)

* Change metric name from EnterInput to Start, and calculate Start before initiate query in mongobatch

* Fix mysql batch scan when primary key contains non-acii char.

* setup mongo replica set in docker test

* Some minor refactoring.

Author: ming535

.gitignore

@@ -7,14 +7,14 @@ cmd/scratch
 profile
 *.idea
 *.meta
-config.toml
-master_config.toml
+
+
 *_dev.toml
 *_dev.json
 
 config_dev
-nuclear/config_dev
-scanner/config_dev
+
+
 
 *.log
 .schema-store

Dockerfile.mongo.setup

@@ -0,0 +1,3 @@
+FROM mongo:4.1
+
+RUN echo "rs.initiate();" > /docker-entrypoint-initdb.d/replica-init.js

cmd/verifier/main.go

@@ -8,6 +8,8 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/moiot/gravity/pkg/config"
+
 	_ "github.com/go-sql-driver/mysql"
 
 	"sort"
@@ -31,17 +33,17 @@ import (
 )
 
 type Config struct {
-	Source      *utils.DBConfig `toml:"source"`
-	Target      *utils.DBConfig `toml:"target"`
-	SourceTable string          `toml:"source-table"`
-	TargetTable string          `toml:"target-table"`
-	Shading     bool            `toml:"shading"`
-	ColName     string          `toml:"col-name"`
-	ColType     string          `toml:"col-type"`
-	MinString   string          `toml:"min-string"`
-	MaxString   string          `toml:"max-string"`
-	ValPrefix   string          `toml:"prefix"`
-	Parallel    int             `toml:"parallel"`
+	Source      *config.DBConfig `toml:"source"`
+	Target      *config.DBConfig `toml:"target"`
+	SourceTable string           `toml:"source-table"`
+	TargetTable string           `toml:"target-table"`
+	Shading     bool             `toml:"shading"`
+	ColName     string           `toml:"col-name"`
+	ColType     string           `toml:"col-type"`
+	MinString   string           `toml:"min-string"`
+	MaxString   string           `toml:"max-string"`
+	ValPrefix   string           `toml:"prefix"`
+	Parallel    int              `toml:"parallel"`
 }
 
 func main() {

dcp/barrier/barrier.go

@@ -4,6 +4,8 @@ import (
 	"database/sql"
 	"time"
 
+	"github.com/moiot/gravity/pkg/config"
+
 	"github.com/moiot/gravity/pkg/utils"
 	log "github.com/sirupsen/logrus"
 )
@@ -17,7 +19,7 @@ const (
 )
 
 type Config struct {
-	Db            utils.DBConfig
+	Db            config.DBConfig
 	TickerSeconds int
 	TestDB        *sql.DB
 }

docker-compose-gravity-test.yml

@@ -24,8 +24,10 @@ services:
       - ./mycnf:/etc/mysql/conf.d
 
   mongo:
-    image: mongo:4.1
     container_name: mongo
+    build:
+      context: ./
+      dockerfile: Dockerfile.mongo.setup
     ports:
       - 27017
     logging:
@@ -71,4 +73,9 @@ services:
       - MONGO_HOST=mongo
       - KAFKA_BROKER=kafka:9092
       - ELASTICSEARCH_URLS=http://elasticsearch:9200
-    command: ["bash", "./wait-for-it.sh", "source-db:3306","-t", "0",  "--", "./wait-for-it.sh", "target-db:3306","-t", "0",  "--", "./wait-for-it.sh", "mongo:27017", "-t", "0", "--", "./wait-for-it.sh", "kafka:9092", "-t", "0", "--", "./wait-for-it.sh", "elasticsearch:9200", "-t", "0",  "--", "make", "go-test"]
+    command: ["bash", "./wait-for-it.sh", "source-db:3306","-t", "0",
+              "--", "./wait-for-it.sh", "target-db:3306","-t", "0",
+              "--", "./wait-for-it.sh", "mongo:27017", "-t", "0",
+              "--", "./wait-for-it.sh", "kafka:9092", "-t", "0",
+              "--", "./wait-for-it.sh", "elasticsearch:9200", "-t", "0",
+              "--", "make", "go-test"]

go.mod

@@ -51,6 +51,9 @@ require (
 	github.com/siddontang/go-mysql v0.0.0-20190312052122-c6ab05a85eb8
 	github.com/sirupsen/logrus v1.2.0
 	github.com/stretchr/testify v1.3.0
+	github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c // indirect
+	github.com/xdg/stringprep v1.0.0 // indirect
+	go.mongodb.org/mongo-driver v1.0.1
 	golang.org/x/net v0.0.0-20190311183353-d8887717615a
 	google.golang.org/grpc v1.19.0
 	gopkg.in/DATA-DOG/go-sqlmock.v1 v1.3.0

go.sum

@@ -76,6 +76,7 @@ github.com/go-ole/go-ole v1.2.2/go.mod h1:pnvuG7BrDMZ8ifMurTQmxwhQM/odqm9sSqNe5B
 github.com/go-sql-driver/mysql v0.0.0-20170715192408-3955978caca4/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-sql-driver/mysql v1.4.0 h1:7LxgVwFb2hIQtMm87NdgAVfXjnt4OePseqT1tKx+opk=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
+github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/gogo/protobuf v0.0.0-20180717141946-636bf0302bc9/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
@@ -300,8 +301,14 @@ github.com/ulikunitz/xz v0.5.5 h1:pFrO0lVpTBXLpYw+pnLj6TbvHuyjXMfjGeCwSqCVwok=
 github.com/ulikunitz/xz v0.5.5/go.mod h1:2bypXElzHzzJZwzH67Y6wb67pO62Rzfn7BSiF4ABRW8=
 github.com/unrolled/render v0.0.0-20180914162206-b9786414de4d h1:ggUgChAeyge4NZ4QUw6lhHsVymzwSDJOZcE0s2X8S20=
 github.com/unrolled/render v0.0.0-20180914162206-b9786414de4d/go.mod h1:tu82oB5W2ykJRVioYsB+IQKcft7ryBr7w12qMBUPyXg=
+github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c h1:u40Z8hqBAAQyv+vATcGgV0YCnDjqSL7/q/JyPhhJSPk=
+github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
+github.com/xdg/stringprep v1.0.0 h1:d9X0esnoa3dFsV0FG35rAT0RIhYFlPq7MiP+DW89La0=
+github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
 github.com/xiang90/probing v0.0.0-20160813154853-07dd2e8dfe18 h1:MPPkRncZLN9Kh4MEFmbnK4h3BD7AUmskWv2+EeZJCCs=
 github.com/xiang90/probing v0.0.0-20160813154853-07dd2e8dfe18/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
+go.mongodb.org/mongo-driver v1.0.1 h1:r2xNB8juGGrZVcIjX2TpY7HUfz+pNYq+GIuC9h6URZg=
+go.mongodb.org/mongo-driver v1.0.1/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qLUO4lqsUM=
 go.opencensus.io v0.20.1/go.mod h1:6WKK9ahsWS3RSO+PY9ZHZUfv2irvY6gN279GOPZjmmk=
 go.uber.org/atomic v1.3.2 h1:2Oa65PReHzfn29GpvgsYwloV9AVFHPDk8tYxt2c2tr4=
 go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=

integration_test/mongokafka/main_test.go

@@ -4,6 +4,8 @@ import (
 	"os"
 	"testing"
 
+	"github.com/moiot/gravity/pkg/utils"
+
 	"github.com/Shopify/sarama"
 
 	"github.com/moiot/gravity/pkg/consts"
@@ -13,7 +15,6 @@ import (
 
 	mgo "gopkg.in/mgo.v2"
 
-	"github.com/moiot/gravity/pkg/mongo"
 	"github.com/moiot/gravity/pkg/mongo_test"
 )
 
@@ -24,7 +25,7 @@ var kafkaAdmin sarama.ClusterAdmin
 
 func TestMain(m *testing.M) {
 	mongoCfg := mongo_test.TestConfig()
-	s, err := mongo.CreateMongoSession(&mongoCfg)
+	s, err := utils.CreateMongoSession(&mongoCfg)
 	if err != nil {
 		panic(err)
 	}

integration_test/mongomysql/main_test.go

@@ -4,11 +4,12 @@ import (
 	"os"
 	"testing"
 
+	"github.com/moiot/gravity/pkg/utils"
+
 	mgo "gopkg.in/mgo.v2"
 
 	"github.com/moiot/gravity/pkg/consts"
 
-	"github.com/moiot/gravity/pkg/mongo"
 	"github.com/moiot/gravity/pkg/mongo_test"
 )
 
@@ -17,7 +18,7 @@ var db *mgo.Database
 
 func TestMain(m *testing.M) {
 	mongoCfg := mongo_test.TestConfig()
-	s, err := mongo.CreateMongoSession(&mongoCfg)
+	s, err := utils.CreateMongoSession(&mongoCfg)
 	if err != nil {
 		panic(err)
 	}

padder/config/config.go

@@ -3,11 +3,12 @@ package config
 import (
 	"flag"
 
+	"github.com/moiot/gravity/pkg/config"
+
 	"github.com/BurntSushi/toml"
 	"github.com/juju/errors"
 
 	"github.com/moiot/gravity/pkg/logutil"
-	"github.com/moiot/gravity/pkg/utils"
 )
 
 type Config struct {
@@ -43,8 +44,8 @@ func (cfg *Config) ParseCmd(arguments []string) error {
 }
 
 type MySQLConfig struct {
-	Target        *utils.DBConfig            `toml:"target" json:"target"`
-	StartPosition *utils.MySQLBinlogPosition `toml:"start-position" json:"start-position"`
+	Target        *config.DBConfig            `toml:"target" json:"target"`
+	StartPosition *config.MySQLBinlogPosition `toml:"start-position" json:"start-position"`
 }
 
 func (c *Config) CreateConfigFromFile(path string) error {

pkg/config/config.go

@@ -12,7 +12,6 @@ import (
 	"gopkg.in/mgo.v2/bson"
 
 	"github.com/moiot/gravity/pkg/logutil"
-	"github.com/moiot/gravity/pkg/utils"
 )
 
 var DefaultBinlogSyncerTimeout = "10s"
@@ -69,7 +68,7 @@ type PipelineConfig struct {
 
 	TableConfig []*TableConfig `toml:"table-config" json:"table-config"`
 
-	TargetMySQL *utils.DBConfig `toml:"target-mysql" json:"target-mysql"`
+	TargetMySQL *DBConfig `toml:"target-mysql" json:"target-mysql"`
 
 	TargetMySQLWorkerCfg *TargetMySQLWorkerConfig `toml:"target-mysql-worker" json:"target-mysql-worker"`
 
@@ -93,8 +92,8 @@ type SourceKafkaConfig struct {
 }
 
 type SourceProbeCfg struct {
-	SourceMySQL *utils.DBConfig `mapstructure:"mysql" toml:"mysql" json:"mysql"`
-	Annotation  string          `mapstructure:"annotation" toml:"annotation" json:"annotation"`
+	SourceMySQL *DBConfig `mapstructure:"mysql" toml:"mysql" json:"mysql"`
+	Annotation  string    `mapstructure:"annotation" toml:"annotation" json:"annotation"`
 }
 
 type MongoPosition bson.MongoTimestamp
@@ -110,14 +109,14 @@ type MongoConfigs struct {
 }
 
 type MySQLConfig struct {
-	IgnoreBiDirectionalData bool                       `mapstructure:"ignore-bidirectional-data" toml:"ignore-bidirectional-data" json:"ignore-bidirectional-data"`
-	Source                  *utils.DBConfig            `mapstructure:"source" toml:"source" json:"source"`
-	SourceSlave             *utils.DBConfig            `mapstructure:"source-slave" toml:"source-slave" json:"source-slave"`
-	StartPosition           *utils.MySQLBinlogPosition `mapstructure:"start-position" toml:"start-position" json:"start-position"`
+	IgnoreBiDirectionalData bool                 `mapstructure:"ignore-bidirectional-data" toml:"ignore-bidirectional-data" json:"ignore-bidirectional-data"`
+	Source                  *DBConfig            `mapstructure:"source" toml:"source" json:"source"`
+	SourceSlave             *DBConfig            `mapstructure:"source-slave" toml:"source-slave" json:"source-slave"`
+	StartPosition           *MySQLBinlogPosition `mapstructure:"start-position" toml:"start-position" json:"start-position"`
 }
 
 type SourceTiDBConfig struct {
-	SourceDB    *utils.DBConfig    `mapstructure:"source-db" toml:"source-db" json:"source-db"`
+	SourceDB    *DBConfig          `mapstructure:"source-db" toml:"source-db" json:"source-db"`
 	SourceKafka *SourceKafkaConfig `mapstructure:"source-kafka" toml:"source-kafka" json:"source-kafka"`
 	// OffsetStoreConfig       *SourceProbeCfg    `mapstructure:"offset-store" toml:"offset-store" json:"offset-store"`
 	PositionRepo            *GenericPluginConfig `mapstructure:"position-repo" toml:"position-repo" json:"position-repo"`
@@ -140,6 +139,44 @@ type MongoConnConfig struct {
 	Direct   bool   `mapstructure:"direct" toml:"direct" json:"direct"`
 }
 
+func (cfg MongoConnConfig) URI() string {
+	username := ""
+	if cfg.Username != "" {
+		username = cfg.Username
+	}
+
+	password := ""
+	if cfg.Password != "" {
+		password = cfg.Password
+	}
+
+	host := "localhost"
+	if cfg.Host != "" {
+		host = cfg.Host
+	}
+
+	port := 27017
+	if cfg.Port != 0 {
+		port = cfg.Port
+	}
+
+	db := cfg.Database
+
+	var url string
+	if username == "" || password == "" {
+		url = fmt.Sprintf("mongodb://%s:%d/%s", host, port, db)
+	} else {
+		url = fmt.Sprintf("mongodb://%s:%s@%s:%d/%s", username, password, host, port, db)
+	}
+
+	//If not specified, the connection will timeout, probably because the replica set has not been initialized yet.
+	if cfg.Direct {
+		url += "?connect=direct"
+	}
+
+	return url
+}
+
 type TargetMySQLWorkerConfig struct {
 	EnableDDL          bool     `toml:"enable-ddl" json:"enable-ddl"`
 	UseBidirection     bool     `toml:"use-bidirection" json:"use-bidirection"`

pkg/config/mysql.go

@@ -0,0 +1,76 @@
+/*
+ *
+ * // Copyright 2019 , Beijing Mobike Technology Co., Ltd.
+ * //
+ * // Licensed under the Apache License, Version 2.0 (the "License");
+ * // you may not use this file except in compliance with the License.
+ * // You may obtain a copy of the License at
+ * //
+ * //     http://www.apache.org/licenses/LICENSE-2.0
+ * //
+ * // Unless required by applicable law or agreed to in writing, software
+ * // distributed under the License is distributed on an "AS IS" BASIS,
+ * // See the License for the specific language governing permissions and
+ * // limitations under the License.
+ */
+
+package config
+
+import (
+	"time"
+
+	"github.com/juju/errors"
+)
+
+// DBConfig is the DB configuration.
+type DBConfig struct {
+	Host                   string        `toml:"host" json:"host" mapstructure:"host"`
+	Location               string        `toml:"location" json:"location" mapstructure:"location"`
+	Username               string        `toml:"username" json:"username" mapstructure:"username"`
+	Password               string        `toml:"password" json:"password" mapstructure:"password"`
+	Port                   int           `toml:"port" json:"port" mapstructure:"port"`
+	Schema                 string        `toml:"schema" json:"schema" mapstructure:"schema"`
+	MaxIdle                int           `toml:"max-idle" json:"max-idle" mapstructure:"max-idle"`
+	MaxOpen                int           `toml:"max-open" json:"max-open" mapstructure:"max-open"`
+	MaxLifeTimeDurationStr string        `toml:"max-life-time-duration" json:"max-life-time-duration" mapstructure:"max-life-time-duration"`
+	MaxLifeTimeDuration    time.Duration `toml:"-" json:"-" mapstructure:"-"`
+}
+
+func (dbc *DBConfig) ValidateAndSetDefault() error {
+	// Sets the location for time.Time values (when using parseTime=true). "Local" sets the system's location. See time.LoadLocation for details.
+	// Note that this sets the location for time.Time values but does not change MySQL's time_zone setting.
+	// For that see the time_zone system variable, which can also be set as a DSN parameter.
+	if dbc.Location == "" {
+		dbc.Location = time.Local.String()
+	}
+
+	// set default values of connection related settings
+	// assume the response time of db is 2ms, then
+	// then a single connection can have tps of 500 TPS
+	if dbc.MaxOpen == 0 {
+		dbc.MaxOpen = 200
+	}
+
+	if dbc.MaxIdle == 0 {
+		dbc.MaxIdle = dbc.MaxOpen
+	}
+
+	var err error
+	if dbc.MaxLifeTimeDurationStr == "" {
+		dbc.MaxLifeTimeDurationStr = "15m"
+		dbc.MaxLifeTimeDuration = 15 * time.Minute
+	} else {
+		dbc.MaxLifeTimeDuration, err = time.ParseDuration(dbc.MaxLifeTimeDurationStr)
+		if err != nil {
+			return errors.Trace(err)
+		}
+	}
+
+	return nil
+}
+
+type MySQLBinlogPosition struct {
+	BinLogFileName string `toml:"binlog-name" json:"binlog-name" mapstructure:"binlog-name"`
+	BinLogFilePos  uint32 `toml:"binlog-pos" json:"binlog-pos" mapstructure:"binlog-pos"`
+	BinlogGTID     string `toml:"binlog-gtid" json:"binlog-gtid" mapstructure:"binlog-gtid"`
+}

pkg/config/table.go

@@ -1,7 +1,5 @@
 package config
 
-import "github.com/moiot/gravity/pkg/utils"
-
 type TableConfig struct {
 	Schema string `toml:"schema" json:"schema"`
 	Table  string `toml:"table" json:"table"`
@@ -14,12 +12,3 @@ type TableConfig struct {
 	ScanColumn string `toml:"scan-column" json:"scan-column"`
 	ScanType   string `toml:"scan-type" json:"scan-type"`
 }
-
-func GetTableConfig(tableConfig []TableConfig, schema string, table string) *TableConfig {
-	for i := range tableConfig {
-		if utils.Glob(tableConfig[i].Schema, schema) && utils.Glob(tableConfig[i].Table, table) {
-			return &tableConfig[i]
-		}
-	}
-	return nil
-}