replication,cmd: improve flavor handling (#946)

For `go-mysqlbinlog` and `go-canal`: return an error for invalid
flavors.

For `go-mysqlbinlog`: log a non-critical error if the server version has "MariaDB"
in it but the flavor isn't set to "mariadb".

Co-authored-by: lance6716 <[email protected]>

cmd/go-canal/main.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/go-mysql-org/go-mysql/canal"
 	"github.com/go-mysql-org/go-mysql/mysql"
+	"github.com/pingcap/errors"
 )
 
 var (
@@ -41,6 +42,12 @@ var (
 func main() {
 	flag.Parse()
 
+	err := mysql.ValidateFlavor(*flavor)
+	if err != nil {
+		fmt.Printf("Flavor error: %v\n", errors.ErrorStack(err))
+		return
+	}
+
 	cfg := canal.NewDefaultConfig()
 	cfg.Addr = net.JoinHostPort(*host, strconv.Itoa(*port))
 	cfg.User = *user

cmd/go-mysqlbinlog/main.go

@@ -50,6 +50,12 @@ func main() {
 		MaxReconnectAttempts: 10,
 	}
 
+	err := mysql.ValidateFlavor(*flavor)
+	if err != nil {
+		fmt.Printf("Flavor error: %v\n", errors.ErrorStack(err))
+		return
+	}
+
 	b := replication.NewBinlogSyncer(cfg)
 
 	pos := mysql.Position{Name: *file, Pos: uint32(*pos)}

mysql/mysql_test.go

@@ -337,3 +337,26 @@ func mysqlGTIDfromString(t *testing.T, gtidStr string) MysqlGTIDSet {
 
 	return *gtid.(*MysqlGTIDSet)
 }
+
+func TestValidateFlavor(t *testing.T) {
+	tbls := []struct {
+		flavor string
+		valid  bool
+	}{
+		{"mysql", true},
+		{"mariadb", true},
+		{"maria", false},
+		{"MariaDB", true},
+		{"msql", false},
+		{"mArIAdb", true},
+	}
+
+	for _, f := range tbls {
+		err := ValidateFlavor(f.flavor)
+		if f.valid == true {
+			require.NoError(t, err)
+		} else {
+			require.Error(t, err)
+		}
+	}
+}

mysql/validate.go

@@ -0,0 +1,17 @@
+package mysql
+
+import (
+	"fmt"
+	"strings"
+)
+
+func ValidateFlavor(flavor string) error {
+	switch strings.ToLower(flavor) {
+	case MySQLFlavor:
+		return nil
+	case MariaDBFlavor:
+		return nil
+	default:
+		return fmt.Errorf("%s is not a valid flavor", flavor)
+	}
+}

replication/binlogsyncer.go

@@ -8,6 +8,7 @@ import (
 	"net"
 	"os"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -424,6 +425,18 @@ func (b *BinlogSyncer) GetNextPosition() Position {
 	return b.nextPos
 }
 
+func (b *BinlogSyncer) checkFlavor() {
+	serverVersion := b.c.GetServerVersion()
+	if b.cfg.Flavor != MariaDBFlavor &&
+		strings.Contains(b.c.GetServerVersion(), "MariaDB") {
+		// Setting the flavor to `mysql` causes MariaDB to try and behave
+		// in a MySQL compatible way. In this mode MariaDB won't use
+		// MariaDB specific binlog event types, but may used dummy events instead.
+		b.cfg.Logger.Errorf("misconfigured flavor (%s) for server %s",
+			b.cfg.Flavor, serverVersion)
+	}
+}
+
 // StartSync starts syncing from the `pos` position.
 func (b *BinlogSyncer) StartSync(pos Position) (*BinlogStreamer, error) {
 	b.cfg.Logger.Infof("begin to sync binlog from position %s", pos)
@@ -439,6 +452,8 @@ func (b *BinlogSyncer) StartSync(pos Position) (*BinlogStreamer, error) {
 		return nil, errors.Trace(err)
 	}
 
+	b.checkFlavor()
+
 	return b.startDumpStream(), nil
 }
 
@@ -477,6 +492,8 @@ func (b *BinlogSyncer) StartSyncGTID(gset GTIDSet) (*BinlogStreamer, error) {
 		return nil, err
 	}
 
+	b.checkFlavor()
+
 	return b.startDumpStream(), nil
 }