Migrate to GitHub actions

.github/workflows/ci.yml

@@ -0,0 +1,45 @@
+name: runTestsAndLinters
+on: [push, pull_request]
+
+jobs:
+  test:
+    strategy:
+      matrix:
+        go: [ 1.16, 1.15 ]
+    name: Tests Go ${{ matrix.go }}
+    runs-on: ubuntu-18.04
+
+    steps:
+      - name: Setup MySQL
+        run: |
+          echo -n "mysql -V: " ; mysql -V
+          echo -n "mysqldump -V: " ; mysqldump -V
+
+          echo -e '[mysqld]\nserver-id=1\nlog-bin=mysql\nbinlog-format=row\ngtid-mode=ON\nenforce_gtid_consistency=ON\n' | sudo tee /etc/mysql/conf.d/replication.cnf
+          sudo service mysql start
+          sudo mysql -h 127.0.0.1 -uroot -proot -e "use mysql; update user set authentication_string=PASSWORD('') where User='root'; update user set plugin='mysql_native_password'; FLUSH PRIVILEGES;"
+          # create ssl/rsa files for mysql ssl support
+          sudo mysql_ssl_rsa_setup --uid=mysql
+          mysql -e "CREATE DATABASE IF NOT EXISTS test;" -uroot
+          mysql -e "SHOW VARIABLES LIKE 'log_bin'" -uroot
+      - name: Prepare for Go
+        run: |
+          sudo apt-get install -y make gcc
+      - name: Install Go
+        uses: actions/setup-go@v2
+        with:
+          go-version: ${{ matrix.go }}
+      - name: Checkout code
+        uses: actions/checkout@v1
+      - name: Run tests
+        run: go test ./...
+
+  golangci:
+    name: golangci
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: golangci-lint
+        uses: golangci/golangci-lint-action@v2
+        with:
+          version: latest

.golangci.yml

@@ -0,0 +1,28 @@
+linters:
+  disable-all: true
+  enable:
+    # All code is ready for:
+    - deadcode
+    - errcheck
+    - staticcheck
+    - structcheck
+    - typecheck
+    - unused
+    - varcheck
+    - misspell
+    - nolintlint
+    - goimports
+    - nakedret
+    - unconvert
+    - whitespace
+    # ToDo:
+    #- gosimple
+    #- govet
+    #- ineffassign
+    #- gocritic
+    #- golint
+linters-settings:
+  nolintlint:
+    allow-unused: false
+    allow-leading-space: false
+    require-specific: true

canal/canal.go

@@ -249,7 +249,7 @@ func (c *Canal) Close() {
 	c.conn = nil
 	c.connLock.Unlock()
 
-	c.eventHandler.OnPosSynced(c.master.Position(), c.master.GTIDSet(), true)
+	_ = c.eventHandler.OnPosSynced(c.master.Position(), c.master.GTIDSet(), true)
 }
 
 func (c *Canal) WaitDumpDone() <-chan struct{} {

canal/canal_test.go

@@ -283,8 +283,8 @@ func TestDropTableExp(t *testing.T) {
 		}
 	}
 }
-func TestWithoutSchemeExp(t *testing.T) {
 
+func TestWithoutSchemeExp(t *testing.T) {
 	cases := []replication.QueryEvent{
 		replication.QueryEvent{
 			Schema: []byte("test"),

canal/sync.go

@@ -227,7 +227,7 @@ func parseStmt(stmt ast.StmtNode) (ns []*node) {
 		}
 		ns = []*node{n}
 	}
-	return
+	return ns
 }
 
 func (c *Canal) updateTable(db, table string) (err error) {

client/auth.go

@@ -46,7 +46,7 @@ func (c *Conn) readInitialHandshake() error {
 	pos := 1 + bytes.IndexByte(data[1:], 0x00) + 1
 
 	// connection id length is 4
-	c.connectionID = uint32(binary.LittleEndian.Uint32(data[pos : pos+4]))
+	c.connectionID = binary.LittleEndian.Uint32(data[pos : pos+4])
 	pos += 4
 
 	c.salt = []byte{}
@@ -199,7 +199,7 @@ func (c *Conn) writeAuthHandshake() error {
 
 	// Charset [1 byte]
 	// use default collation id 33 here, is utf-8
-	data[12] = byte(DEFAULT_COLLATION_ID)
+	data[12] = DEFAULT_COLLATION_ID
 
 	// SSL Connection Request Packet
 	// http://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol::SSLRequest

client/client_test.go

@@ -289,7 +289,6 @@ func (s *clientTestSuite) TestStmt_Select(c *C) {
 
 	e, _ = result.GetStringByName(0, "e")
 	c.Assert(e, Equals, "test1")
-
 }
 
 func (s *clientTestSuite) TestStmt_NULL(c *C) {
@@ -309,11 +308,13 @@ func (s *clientTestSuite) TestStmt_NULL(c *C) {
 
 	str = `select * from mixer_test_stmt where id = ?`
 	stmt, err = s.c.Prepare(str)
+	c.Assert(err, IsNil)
+
 	defer stmt.Close()
 
+	result, err = stmt.Execute(2)
 	c.Assert(err, IsNil)
 
-	result, err = stmt.Execute(2)
 	b, err := result.IsNullByName(0, "id")
 	c.Assert(err, IsNil)
 	c.Assert(b, Equals, false)

client/resp.go

@@ -101,6 +101,10 @@ func (c *Conn) handleAuthResult() error {
 		}
 		c.authPluginName = switchToPlugin
 		auth, addNull, err := c.genAuthResponse(data)
+		if err != nil {
+			return err
+		}
+
 		if err = c.WriteAuthSwitchPacket(auth, addNull); err != nil {
 			return err
 		}
@@ -138,7 +142,7 @@ func (c *Conn) handleAuthResult() error {
 				}
 			}
 		} else {
-			errors.Errorf("invalid packet")
+			return errors.Errorf("invalid packet %x", data[0])
 		}
 	} else if c.authPluginName == AUTH_SHA256_PASSWORD {
 		if len(data) == 0 {
@@ -169,7 +173,6 @@ func (c *Conn) readAuthResult() ([]byte, string, error) {
 	// see: https://insidemysql.com/preparing-your-community-connector-for-mysql-8-part-2-sha256/
 	// packet indicator
 	switch data[0] {
-
 	case OK_HEADER:
 		_, err := c.handleOKPacket(data)
 		return nil, "", err

client/stmt.go

@@ -10,9 +10,8 @@ import (
 )
 
 type Stmt struct {
-	conn  *Conn
-	id    uint32
-	query string
+	conn *Conn
+	id   uint32
 
 	params  int
 	columns int
@@ -55,7 +54,7 @@ func (s *Stmt) write(args ...interface{}) error {
 	//NULL-bitmap, length: (num-params+7)
 	nullBitmap := make([]byte, (paramsNum+7)>>3)
 
-	var length int = int(1 + 4 + 1 + 4 + ((paramsNum + 7) >> 3) + 1 + (paramsNum << 1))
+	length := 1 + 4 + 1 + 4 + ((paramsNum + 7) >> 3) + 1 + (paramsNum << 1)
 
 	var newParamBoundFlag byte = 0
 
@@ -91,26 +90,25 @@ func (s *Stmt) write(args ...interface{}) error {
 		case uint16:
 			paramTypes[i<<1] = MYSQL_TYPE_SHORT
 			paramTypes[(i<<1)+1] = 0x80
-			paramValues[i] = Uint16ToBytes(uint16(v))
+			paramValues[i] = Uint16ToBytes(v)
 		case uint32:
 			paramTypes[i<<1] = MYSQL_TYPE_LONG
 			paramTypes[(i<<1)+1] = 0x80
-			paramValues[i] = Uint32ToBytes(uint32(v))
+			paramValues[i] = Uint32ToBytes(v)
 		case uint:
 			paramTypes[i<<1] = MYSQL_TYPE_LONGLONG
 			paramTypes[(i<<1)+1] = 0x80
 			paramValues[i] = Uint64ToBytes(uint64(v))
 		case uint64:
 			paramTypes[i<<1] = MYSQL_TYPE_LONGLONG
 			paramTypes[(i<<1)+1] = 0x80
-			paramValues[i] = Uint64ToBytes(uint64(v))
+			paramValues[i] = Uint64ToBytes(v)
 		case bool:
 			paramTypes[i<<1] = MYSQL_TYPE_TINY
 			if v {
 				paramValues[i] = []byte{1}
 			} else {
 				paramValues[i] = []byte{0}
-
 			}
 		case float32:
 			paramTypes[i<<1] = MYSQL_TYPE_FLOAT

cmd/go-canal/main.go

@@ -23,8 +23,8 @@ var flavor = flag.String("flavor", "mysql", "Flavor: mysql or mariadb")
 var serverID = flag.Int("server-id", 101, "Unique Server ID")
 var mysqldump = flag.String("mysqldump", "mysqldump", "mysqldump execution path")
 
-var dbs = flag.String("dbs", "test", "dump databases, seperated by comma")
-var tables = flag.String("tables", "", "dump tables, seperated by comma, will overwrite dbs")
+var dbs = flag.String("dbs", "test", "dump databases, separated by comma")
+var tables = flag.String("tables", "", "dump tables, separated by comma, will overwrite dbs")
 var tableDB = flag.String("table_db", "test", "database for dump tables")
 var ignoreTables = flag.String("ignore_tables", "", "ignore tables, must be database.table format, separated by comma")
 
@@ -89,8 +89,7 @@ func main() {
 
 	sc := make(chan os.Signal, 1)
 	signal.Notify(sc,
-		os.Kill,
-		os.Interrupt,
+		syscall.SIGINT,
 		syscall.SIGHUP,
 		syscall.SIGINT,
 		syscall.SIGTERM,

cmd/go-mysqlbinlog/main.go

@@ -77,5 +77,4 @@ func main() {
 			e.Dump(os.Stdout)
 		}
 	}
-
 }

cmd/go-mysqldump/main.go

@@ -16,8 +16,8 @@ var password = flag.String("password", "", "MySQL password")
 var execution = flag.String("exec", "mysqldump", "mysqldump execution path")
 var output = flag.String("o", "", "dump output, empty for stdout")
 
-var dbs = flag.String("dbs", "", "dump databases, seperated by comma")
-var tables = flag.String("tables", "", "dump tables, seperated by comma, will overwrite dbs")
+var dbs = flag.String("dbs", "", "dump databases, separated by comma")
+var tables = flag.String("tables", "", "dump tables, separated by comma, will overwrite dbs")
 var tableDB = flag.String("table_db", "", "database for dump tables")
 var ignoreTables = flag.String("ignore_tables", "", "ignore tables, must be database.table format, separated by comma")
 

dump/dump.go

@@ -1,6 +1,7 @@
 package dump
 
 import (
+	"bytes"
 	"fmt"
 	"io"
 	"os"
@@ -40,6 +41,9 @@ type Dumper struct {
 	masterDataSkipped bool
 	maxAllowedPacket  int
 	hexBlob           bool
+
+	// see detectColumnStatisticsParamSupported
+	isColumnStatisticsParamSupported bool
 }
 
 func NewDumper(executionPath string, addr string, user string, password string) (*Dumper, error) {
@@ -63,12 +67,28 @@ func NewDumper(executionPath string, addr string, user string, password string)
 	d.IgnoreTables = make(map[string][]string)
 	d.ExtraOptions = make([]string, 0, 5)
 	d.masterDataSkipped = false
+	d.isColumnStatisticsParamSupported = d.detectColumnStatisticsParamSupported()
 
 	d.ErrOut = os.Stderr
 
 	return d, nil
 }
 
+// New mysqldump versions try to send queries to information_schema.COLUMN_STATISTICS table which does not exist in old MySQL (<5.x).
+// And we got error: "Unknown table 'COLUMN_STATISTICS' in information_schema (1109)".
+//
+// mysqldump may not send this query if it is started with parameter --column-statistics.
+// But this parameter exists only for versions >=8.0.2 (https://dev.mysql.com/doc/relnotes/mysql/8.0/en/news-8-0-2.html).
+//
+// For environments where the version of mysql-server and mysqldump differs, we try to check this parameter and use it if available.
+func (d *Dumper) detectColumnStatisticsParamSupported() bool {
+	out, err := exec.Command(d.ExecutionPath, `--help`).CombinedOutput()
+	if err != nil {
+		return false
+	}
+	return bytes.Contains(out, []byte(`--column-statistics`))
+}
+
 func (d *Dumper) SetCharset(charset string) {
 	d.Charset = charset
 }
@@ -196,6 +216,10 @@ func (d *Dumper) Dump(w io.Writer) error {
 		args = append(args, d.ExtraOptions...)
 	}
 
+	if d.isColumnStatisticsParamSupported {
+		args = append(args, `--column-statistics=0`)
+	}
+
 	if len(d.Tables) == 0 && len(d.Databases) == 0 {
 		args = append(args, "--all-databases")
 	} else if len(d.Tables) == 0 {
@@ -208,7 +232,10 @@ func (d *Dumper) Dump(w io.Writer) error {
 		// If we only dump some tables, the dump data will not have database name
 		// which makes us hard to parse, so here we add it manually.
 
-		w.Write([]byte(fmt.Sprintf("USE `%s`;\n", d.TableDB)))
+		_, err := w.Write([]byte(fmt.Sprintf("USE `%s`;\n", d.TableDB)))
+		if err != nil {
+			return fmt.Errorf(`could not write USE command: %w`, err)
+		}
 	}
 
 	log.Infof("exec mysqldump with %v", args)
@@ -227,12 +254,12 @@ func (d *Dumper) DumpAndParse(h ParseHandler) error {
 	done := make(chan error, 1)
 	go func() {
 		err := Parse(r, h, !d.masterDataSkipped)
-		r.CloseWithError(err)
+		_ = r.CloseWithError(err)
 		done <- err
 	}()
 
 	err := d.Dump(w)
-	w.CloseWithError(err)
+	_ = w.CloseWithError(err)
 
 	err = <-done