support TiDB bidirectional input

Author: Ryan-Git

Makefile

@@ -49,8 +49,6 @@ run-dev:
 
 build:
 	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/gravity cmd/gravity/main.go
-	#$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/padder cmd/padder/main.go
-
 
 build-linux:
 	GOARCH=amd64 GOOS=linux $(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/gravity-linux-amd64 cmd/gravity/main.go
@@ -74,14 +72,21 @@ proto:
 	@ which protoc >/dev/null || brew install protobuf
 	@ which protoc-gen-gofast >/dev/null || go get github.com/gogo/protobuf/protoc-gen-gofast
 
-	protoc -I=protocol/msgpb -I=${GOPATH}/src -I=${GOPATH}/src/github.com/gogo/protobuf/protobuf --gofast_out=\
+	protoc -I=protocol/msgpb -I=${GOPATH}/src -I=${GOPATH}/src/github.com/gogo/protobuf/protobuf --gofast_out=.\
 	plugins=grpc,\
 	Mgoogle/protobuf/any.proto=github.com/gogo/protobuf/types,\
 	Mgoogle/protobuf/struct.proto=github.com/gogo/protobuf/types,\
 	Mgoogle/protobuf/timestamp.proto=github.com/gogo/protobuf/types,\
 	Mgoogle/protobuf/wrappers.proto=github.com/gogo/protobuf/types:./pkg/protocol/msgpb \
 	protocol/msgpb/message.proto
 
+	protoc -I=protocol/tidb -I=${GOPATH}/src -I=${GOPATH}/src/github.com/gogo/protobuf/protobuf --gofast_out=.\
+	Mgoogle/protobuf/any.proto=github.com/gogo/protobuf/types,\
+	Mgoogle/protobuf/struct.proto=github.com/gogo/protobuf/types,\
+	Mgoogle/protobuf/timestamp.proto=github.com/gogo/protobuf/types,\
+	Mgoogle/protobuf/wrappers.proto=github.com/gogo/protobuf/types:./pkg/protocol/tidb \
+	protocol/tidb/tidb.proto
+
 
 mock:
 	mockgen -destination ./mock/binlog_checker/mock.go github.com/moiot/gravity/pkg/inputs/helper/binlog_checker BinlogChecker

docs/2.0/example-tidb2tidb.toml

@@ -0,0 +1,58 @@
+name = "tidb2tidbDemo"
+version = "1.0"
+
+[input]
+type = "tidbkafka"
+mode = "stream"
+
+[input.config]
+ignore-bidirectional-data = true
+
+[input.config.position-repo]
+type = "mysql-repo"
+[input.config.position-repo.config.source]
+host = "127.0.0.1"
+username = "root"
+password = ""
+port = 4000
+
+[input.config.source-db]
+host = "127.0.0.1"
+username = "root"
+password = ""
+port = 4000
+
+[input.config.source-kafka]
+topics = ["obinlog"]
+consume-from = "oldest"
+group-id = "tidb2tidbDemo"
+[input.config.source-kafka.brokers]
+broker-addrs = ["localhost:9092", "localhost:9093", "localhost:9094"]
+
+[output]
+type = "mysql"
+
+[output.config]
+enable-ddl = true
+
+[output.config.target]
+host = "127.0.0.1"
+username = "root"
+password = ""
+port = 4000
+max-open = 30 # optional, max connections
+max-idle = 30 # optional, suggest to be the same as max-open
+
+# The definition of the routing rule
+[[output.config.routes]]
+match-schema = "test"
+match-table = "t"
+target-table = "t2"
+
+[scheduler]
+type = "batch-table-scheduler"
+[scheduler.config]
+nr-worker = 30
+batch-size = 1000
+queue-size = 1024
+sliding-window-size = 16384

pkg/config/config.go

@@ -121,6 +121,7 @@ type SourceTiDBConfig struct {
 	// OffsetStoreConfig       *SourceProbeCfg    `mapstructure:"offset-store" toml:"offset-store" json:"offset-store"`
 	PositionRepo            *GenericPluginConfig `mapstructure:"position-repo" toml:"position-repo" json:"position-repo"`
 	IgnoreBiDirectionalData bool                 `mapstructure:"ignore-bidirectional-data" toml:"ignore-bidirectional-data" json:"ignore-bidirectional-data"`
+	FailOnTxnTags           []string             `mapstructure:"fail-on-txn-tags" toml:"fail-on-txn-tags" json:"fail-on-txn-tags"`
 }
 
 type GtmConfig struct {

pkg/core/msg.go

@@ -225,7 +225,7 @@ func IsInternalTrafficV2(db string, tbl string) bool {
 }
 
 func MatchTxnTagPipelineName(patterns []string, msg *Msg) (string, bool) {
-	if IsInternalTrafficV2(msg.Database, msg.Table) {
+	if len(patterns) > 0 && IsInternalTrafficV2(msg.Database, msg.Table) {
 		pipelineName := msg.DmlMsg.Data["pipeline_name"].(string)
 		for _, pattern := range patterns {
 			if utils.Glob(pattern, pipelineName) {

pkg/inputs/plugins.go

@@ -7,10 +7,12 @@ import (
 	"github.com/moiot/gravity/pkg/inputs/mysql"
 	_ "github.com/moiot/gravity/pkg/inputs/mysqlbatch"
 	_ "github.com/moiot/gravity/pkg/inputs/mysqlstream"
+	"github.com/moiot/gravity/pkg/inputs/tidb_kafka"
 	_ "github.com/moiot/gravity/pkg/inputs/tidb_kafka"
 )
 
 const (
 	Mongo = mongo.Name
 	Mysql = mysql.Name
+	TiDB  = tidb_kafka.TiDBKafka
 )