Fix (#198)

* fix(input.mysql): no fatal when columns and data mismatch
moiot · Aug 22, 2019 · 7a18d70 · 7a18d70
1 parent 886b6bd
commit 7a18d70
Show file tree

Hide file tree

Showing 8 changed files with 33 additions and 24 deletions.
diff --git a/docs/2.0/01-quick-start-en.md b/docs/2.0/01-quick-start-en.md
@@ -30,12 +30,9 @@ This document takes synchronizing the data of the local MySQL instance and data
 [Configure the Go environment](https://golang.org/doc/install) and compile:
 
 ```bash
-mkdir -p $GOPATH/src/github.com/moiot/ && cd $GOPATH/src/github.com/moiot/
-
 git clone https://github.com/moiot/gravity.git
 
 cd gravity && make
-
 ```
 
 ## Step 3: Write config file 
@@ -134,4 +131,8 @@ bin/gravity -config mysql2mysql.toml
 or docker
 ```bash
 docker run -v ${PWD}/config.toml:/etc/gravity/config.toml -d --net=host moiot/gravity:latest
-```
+```
+
+## Step 5: Monitoring
+Gravity uses [Prometheus](https://prometheus.io) and [Grafana](https://grafana.com/) for monitoring. 
+The running port(default 8080) provides standard prometheus metrics endpoint(/metrics). Grafana Dashboards are available under deploy/grafana folder in the source tree.
diff --git a/docs/2.0/01-quick-start.md b/docs/2.0/01-quick-start.md
@@ -25,14 +25,13 @@ CREATE TABLE `test`.`test_target_table` (
 
 
 ```bash
-mkdir -p $GOPATH/src/github.com/moiot/ && cd $GOPATH/src/github.com/moiot/
-
 git clone https://github.com/moiot/gravity.git
 
 cd gravity && make
-
 ```
 
+Gravity 使用了 [go mod](https://github.com/golang/go/wiki/Modules)，中国大陆地区用户建议设置 `export GOPROXY=https://goproxy.io` 或其他代理。
+将代码克隆到 GOPATH 路径下的用户需要设置 `export GO111MODULE=on`。
 
 #### MySQL 到 MySQL 同步
 
@@ -119,4 +118,9 @@ bin/gravity -config mysql2mysql.toml
 从 docker
 ```bash
 docker run -v ${PWD}/config.toml:/etc/gravity/config.toml -d --net=host moiot/gravity:latest
-```
+```
+
+## 监控
+Gravity 使用 [Prometheus](https://prometheus.io) 和 [Grafana](https://grafana.com/) 实现监控功能。
+在运行端口（默认8080）上提供了 Prometheus 标准的指标抓取路径`/metrics`。
+在源码路径`deploy/grafana`下提供了 Grafana dashboard 供导入。 
diff --git a/docs/2.0/example-mysql2kafka.toml b/docs/2.0/example-mysql2kafka.toml
@@ -8,6 +8,9 @@
 # 来匹配 filter 和 output 的路由，filter/output 里的每一个 match 函数
 # 都匹配才算满足匹配规则
 #
+name = "mysql2mysqlDemo"
+version = "1.0"
+
 [input]
 type = "mysql"
 mode = "stream"

diff --git a/docs/2.0/example-mysql2mysql-full.toml b/docs/2.0/example-mysql2mysql-full.toml
@@ -8,6 +8,9 @@
 # 来匹配 filter 和 output 的路由，filter/output 里的每一个 match 函数
 # 都匹配才算满足匹配规则
 #
+name = "mysql2mysqlDemo"
+version = "1.0"
+
 [input]
 type = "mysql"
 mode = "stream"

diff --git a/docs/2.0/example-mysql2mysql-inc.toml b/docs/2.0/example-mysql2mysql-inc.toml
@@ -8,6 +8,8 @@
 # 来匹配 filter 和 output 的路由，filter/output 里的每一个 match 函数
 # 都匹配才算满足匹配规则
 #
+name = "mysql2mysqlDemo"
+version = "1.0"
 
 [input]
 type = "mysql"

diff --git a/go.mod b/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/Shopify/sarama v1.19.0
 	github.com/aws/aws-sdk-go v1.16.11 // indirect
 	github.com/coreos/etcd v3.3.10+incompatible
-	github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548 // indirect
+	github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548
 	github.com/denisenkom/go-mssqldb v0.0.0-20190412130859-3b1d194e553a // indirect
 	github.com/erikstmartin/go-testdb v0.0.0-20160219214506-8d10e4a1bae5 // indirect
 	github.com/fortytw2/leaktest v1.3.0 // indirect

diff --git a/pkg/inputs/mysqlstream/msg.go b/pkg/inputs/mysqlstream/msg.go
@@ -6,22 +6,18 @@ import (
 	"time"
 
 	"github.com/OneOfOne/xxhash"
-
-	"github.com/moiot/gravity/pkg/config"
-
+	"github.com/cznic/mathutil"
+	"github.com/juju/errors"
 	"github.com/mitchellh/hashstructure"
-
 	"github.com/pingcap/parser/ast"
-
-	"github.com/moiot/gravity/pkg/mysql"
-	"github.com/moiot/gravity/pkg/utils"
-
-	"github.com/juju/errors"
 	"github.com/siddontang/go-mysql/replication"
 	log "github.com/sirupsen/logrus"
 
+	"github.com/moiot/gravity/pkg/config"
 	"github.com/moiot/gravity/pkg/core"
+	"github.com/moiot/gravity/pkg/mysql"
 	"github.com/moiot/gravity/pkg/schema_store"
+	"github.com/moiot/gravity/pkg/utils"
 )
 
 type binlogOp string
@@ -313,7 +309,7 @@ func NewDeleteMsgs(
 		dmlMsg.Operation = core.Delete
 
 		data := make(map[string]interface{})
-		for i := 0; i < len(columns); i++ {
+		for i := 0; i < mathutil.Min(len(row), len(columns)); i++ {
 			data[columns[i].Name] = deserialize(row[i], columns[i])
 		}
 		dmlMsg.Data = data

diff --git a/pkg/inputs/mysqlstream/utils.go b/pkg/inputs/mysqlstream/utils.go
@@ -22,7 +22,7 @@ func extractSchemaNameFromDDLQueryEvent(p *parser.Parser, ev *replication.QueryE
 	stmt, err := p.ParseOneStmt(string(ev.Query), "", "")
 	if err != nil {
 		log.Errorf("sql parser: %s. error: %v", string(ev.Query), err.Error())
-		return []string{string(ev.Schema)}, []string{""}, nil
+		return []string{string(ev.Schema)}, []string{""}, []ast.StmtNode{nil}
 	}
 
 	switch v := stmt.(type) {
@@ -42,10 +42,10 @@ func extractSchemaNameFromDDLQueryEvent(p *parser.Parser, ev *replication.QueryE
 		for i := range v.Tables {
 			db = append(db, v.Tables[i].Schema.String())
 			table = append(table, v.Tables[i].Name.String())
-			copy := *v
-			copy.Tables = nil
-			copy.Tables = append(copy.Tables, v.Tables[i])
-			node = append(node, &copy)
+			dropTableStmt := *v
+			dropTableStmt.Tables = nil
+			dropTableStmt.Tables = append(dropTableStmt.Tables, v.Tables[i])
+			node = append(node, &dropTableStmt)
 		}
 	case *ast.AlterTableStmt:
 		db = append(db, v.Table.Schema.String())