parse gtid for parseHandler if MySQL works in GTID_MODE, and begin to startWithGTID after mysqldump is done

canal/canal.go

@@ -120,6 +120,36 @@ func NewCanal(cfg *Config) (*Canal, error) {
 	return c, nil
 }
 
+// validateSetGtidPurged: check config parameter set-gtid-purged is supported by MySQL at the moment
+// an error is supposed to return,if set-gtid-purged:auto is set, but MySQL's gtid_mode is off, or GTID is not supported by MySQL server,
+// otherwise, nil is returned.
+func (c *Canal) validateSetGtidPurged() error {
+	gtidPuged := strings.ToLower(c.cfg.Dump.GtidPurged)
+	if gtidPuged == "none" {
+		return nil
+	} else if gtidPuged == "auto" {
+		isOn, err := c.IsGtidModeEnabled()
+		if !isOn {
+			return err
+		}
+		return nil
+	}
+
+	return errors.Errorf("set-gtid-purged: none or auto can be set, current is %s", gtidPuged)
+}
+
+// IsGtidModeEnabled: return true when gtid_mode of MySQL set to on, otherwise return false
+func (c *Canal) IsGtidModeEnabled() (bool, error) {
+	res, err := c.Execute(`SHOW GLOBAL VARIABLES LIKE "gtid_mode";`)
+	if err != nil {
+		return false, errors.Trace(err)
+	} else if f, _ := res.GetString(0, 1); strings.ToLower(f) != "on" {
+		// if MySQL's gtid_mode is set to off or gtid_mode is not supported
+		return false, errors.Errorf("set-gtid-purged: %s,gtid_mode should be on, but now is %s", c.cfg.Dump.GtidPurged, f)
+	}
+	return true, nil
+}
+
 func (c *Canal) prepareDumper() error {
 	var err error
 	dumpPath := c.cfg.Dump.ExecutionPath
@@ -128,6 +158,12 @@ func (c *Canal) prepareDumper() error {
 		return nil
 	}
 
+	// validate c.cfg.Dump.GtidPurged)
+	err = c.validateSetGtidPurged()
+	if err != nil {
+		return err
+	}
+
 	if c.dumper, err = dump.NewDumper(dumpPath,
 		c.cfg.Addr, c.cfg.User, c.cfg.Password); err != nil {
 		return errors.Trace(err)
@@ -153,6 +189,7 @@ func (c *Canal) prepareDumper() error {
 
 	c.dumper.SetWhere(c.cfg.Dump.Where)
 	c.dumper.SkipMasterData(c.cfg.Dump.SkipMasterData)
+	c.dumper.SetGtidPurged(strings.ToLower(c.cfg.Dump.GtidPurged))
 	c.dumper.SetMaxAllowedPacket(c.cfg.Dump.MaxAllowedPacketMB)
 	c.dumper.SetProtocol(c.cfg.Dump.Protocol)
 	c.dumper.SetExtraOptions(c.cfg.Dump.ExtraOptions)

canal/canal_test.go

@@ -14,7 +14,10 @@ import (
 	"github.com/siddontang/go-mysql/replication"
 )
 
-var testHost = flag.String("host", "127.0.0.1", "MySQL host")
+var (
+	ErrSkip  = errors.New("Handler error, but skipped")
+	testHost = flag.String("host", "127.0.0.1", "MySQL host")
+)
 
 func Test(t *testing.T) {
 	TestingT(t)

canal/config.go

@@ -34,6 +34,12 @@ type DumpConfig struct {
 	// 'FLUSH TABLES WITH READ LOCK'
 	SkipMasterData bool `toml:"skip_master_data"`
 
+	// Only MySQL's GTID is supported, cause mysqldump's parameter is a little different between MySQL and MariaDB
+	// mysqldump of MySQL uses --set-gtid-purged, MariaDB uses --gtid instead.
+	// GtidPurged:auto when MySQL supports GTID,and GtidPurged:none for gtid is disabled,or gtid is not supported,
+	// or you strongly wants file-position replication.
+	GtidPurged string `toml:"set_gtid_purged"`
+
 	// Set to change the default max_allowed_packet size
 	MaxAllowedPacketMB int `toml:"max_allowed_packet_mb"`
 
@@ -116,6 +122,8 @@ func NewDefaultConfig() *Config {
 	c.Dump.ExecutionPath = "mysqldump"
 	c.Dump.DiscardErr = true
 	c.Dump.SkipMasterData = false
+	// add default value to disable mysqldump --set-gtid-purged
+	c.Dump.GtidPurged = "none"
 
 	return c
 }

canal/dump.go

@@ -27,6 +27,15 @@ func (h *dumpParseHandler) BinLog(name string, pos uint64) error {
 	return nil
 }
 
+func (h *dumpParseHandler) GtidSet(gtidsets string) (err error) {
+	if h.gset != nil {
+		err = h.gset.Update(gtidsets)
+	} else {
+		h.gset, err = mysql.ParseGTIDSet("mysql", gtidsets)
+	}
+	return err
+}
+
 func (h *dumpParseHandler) Data(db string, table string, values []string) error {
 	if err := h.c.ctx.Err(); err != nil {
 		return err
@@ -167,6 +176,7 @@ func (c *Canal) dump() error {
 
 	pos := mysql.Position{Name: h.name, Pos: uint32(h.pos)}
 	c.master.Update(pos)
+	c.master.UpdateGTIDSet(h.gset)
 	if err := c.eventHandler.OnPosSynced(pos, c.master.GTIDSet(), true); err != nil {
 		return errors.Trace(err)
 	}

canal/sync.go

@@ -16,7 +16,7 @@ import (
 
 func (c *Canal) startSyncer() (*replication.BinlogStreamer, error) {
 	gset := c.master.GTIDSet()
-	if gset == nil {
+	if gset == nil || (gset != nil && gset.String() == "") {
 		pos := c.master.Position()
 		s, err := c.syncer.StartSync(pos)
 		if err != nil {

dump/dump.go

@@ -39,6 +39,7 @@ type Dumper struct {
 
 	masterDataSkipped bool
 	maxAllowedPacket  int
+	gtidPurged        string
 	hexBlob           bool
 }
 
@@ -94,6 +95,11 @@ func (d *Dumper) SkipMasterData(v bool) {
 	d.masterDataSkipped = v
 }
 
+// SetGtidPurged: none, auto; none for gtid is disabled or "version too low", auto for gtid_mode=on;
+func (d *Dumper) SetGtidPurged(gtid string) {
+	d.gtidPurged = gtid
+}
+
 func (d *Dumper) SetMaxAllowedPacket(i int) {
 	d.maxAllowedPacket = i
 }
@@ -150,6 +156,10 @@ func (d *Dumper) Dump(w io.Writer) error {
 		args = append(args, "--master-data")
 	}
 
+	if d.gtidPurged == "auto" {
+		args = append(args, fmt.Sprintf("--set-gtid-purged=%s", d.gtidPurged))
+	}
+
 	if d.maxAllowedPacket > 0 {
 		// mysqldump param should be --max-allowed-packet=%dM not be --max_allowed_packet=%dM
 		args = append(args, fmt.Sprintf("--max-allowed-packet=%dM", d.maxAllowedPacket))
@@ -226,6 +236,7 @@ func (d *Dumper) DumpAndParse(h ParseHandler) error {
 
 	done := make(chan error, 1)
 	go func() {
+		// TODO: set_gtid_purged indicate if parse SET @@GLOBAL.GTID_PURGED='c0977f88-3104-11e9-81e1-00505690245b:1-274559' OR NOT;
 		err := Parse(r, h, !d.masterDataSkipped)
 		r.CloseWithError(err)
 		done <- err

dump/dump_test.go

@@ -1,14 +1,21 @@
 package dump
 
 import (
+	"bufio"
 	"bytes"
 	"flag"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
+	"regexp"
 	"strings"
 	"testing"
 
+	"github.com/pingcap/errors"
+	"github.com/siddontang/go-log/log"
+	"github.com/siddontang/go-mysql/mysql"
+
 	. "github.com/pingcap/check"
 	"github.com/siddontang/go-mysql/client"
 )
@@ -36,6 +43,7 @@ func (s *schemaTestSuite) SetUpSuite(c *C) {
 	c.Assert(err, IsNil)
 
 	s.d, err = NewDumper(*execution, fmt.Sprintf("%s:%d", *host, *port), "root", "")
+	s.d.gtidPurged = "none"
 	c.Assert(err, IsNil)
 	c.Assert(s.d, NotNil)
 
@@ -119,10 +127,105 @@ func (h *testParseHandler) BinLog(name string, pos uint64) error {
 	return nil
 }
 
+func (h *testParseHandler) GtidSet(gtidsets string) (err error) {
+	return nil
+}
+
 func (h *testParseHandler) Data(schema string, table string, values []string) error {
 	return nil
 }
 
+type GtidParseTest struct {
+	gset mysql.GTIDSet
+}
+
+func (h *GtidParseTest) UpdateGtidSet(gtidStr string) (err error) {
+	if h.gset != nil {
+		err = h.gset.Update(gtidStr)
+	} else {
+		h.gset, err = mysql.ParseGTIDSet("mysql", gtidStr)
+	}
+	return err
+}
+
+func TestParseGtidStrFromMysqlDump(t *testing.T) {
+	binlogExp := regexp.MustCompile("^CHANGE MASTER TO MASTER_LOG_FILE='(.+)', MASTER_LOG_POS=(\\d+);")
+	tbls := []struct {
+		input    string
+		expected string
+	}{
+		{`SET @@GLOBAL.GTID_PURGED='071a84e8-b253-11e8-8472-005056a27e86:1-76,
+2337be48-0456-11e9-bd1c-00505690543b:1-7,
+41d816cd-0455-11e9-be42-005056901a22:1-2,
+5f1eea9e-b1e5-11e8-bc77-005056a221ed:1-144609156,
+75848cdb-8131-11e7-b6fc-1c1b0de85e7b:1-151378598,
+780ad602-0456-11e9-8bcd-005056901a22:1-516653148,
+92809ddd-1e3c-11e9-9d04-00505690f6ab:1-11858565,
+c59598c7-0467-11e9-bbbe-005056901a22:1-226464969,
+cbd7809d-0433-11e9-b1cf-00505690543b:1-18233950,
+cca778e9-8cdf-11e8-94d0-005056a247b1:1-303899574,
+cf80679b-7695-11e8-8873-1c1b0d9a4ab9:1-12836047,
+d0951f24-1e21-11e9-bb2e-00505690b730:1-4758092,
+e7574090-b123-11e8-8bb4-005056a29643:1-12'`, "071a84e8-b253-11e8-8472-005056a27e86:1-76,2337be48-0456-11e9-bd1c-00505690543b:1-7,41d816cd-0455-11e9-be42-005056901a22:1-2,5f1eea9e-b1e5-11e8-bc77-005056a221ed:1-144609156,75848cdb-8131-11e7-b6fc-1c1b0de85e7b:1-151378598,780ad602-0456-11e9-8bcd-005056901a22:1-516653148,92809ddd-1e3c-11e9-9d04-00505690f6ab:1-11858565,c59598c7-0467-11e9-bbbe-005056901a22:1-226464969,cbd7809d-0433-11e9-b1cf-00505690543b:1-18233950,cca778e9-8cdf-11e8-94d0-005056a247b1:1-303899574,cf80679b-7695-11e8-8873-1c1b0d9a4ab9:1-12836047,d0951f24-1e21-11e9-bb2e-00505690b730:1-4758092,e7574090-b123-11e8-8bb4-005056a29643:1-12"},
+		{`SET @@GLOBAL.GTID_PURGED='071a84e8-b253-11e8-8472-005056a27e86:1-76,
+2337be48-0456-11e9-bd1c-00505690543b:1-7';`, "071a84e8-b253-11e8-8472-005056a27e86:1-76,2337be48-0456-11e9-bd1c-00505690543b:1-7"},
+		{`SET @@GLOBAL.GTID_PURGED='c0977f88-3104-11e9-81e1-00505690245b:1-274559';`, "c0977f88-3104-11e9-81e1-00505690245b:1-274559"},
+		{`CHANGE MASTER TO MASTER_LOG_FILE='mysql-bin.008995', MASTER_LOG_POS=102052485;`, ""},
+	}
+
+	for _, tt := range tbls {
+		reader := strings.NewReader(tt.input)
+		newReader := bufio.NewReader(reader)
+		var gtidParse = new(GtidParseTest)
+		var binlogParsed bool
+		var gtidDoneParsed bool
+		var mutilGtidParsed bool
+		parseBinlogPos := true
+		for {
+			bytes, _, err := newReader.ReadLine()
+			line := string(bytes)
+			if err == io.EOF {
+				break
+			}
+
+			// begin parsed gtid
+			if parseBinlogPos && !gtidDoneParsed && !binlogParsed {
+				gtidStr, IsMultiSetReturned, IsDoneOfGtidParsed := ParseGtidStrFromMysqlDump(line, mutilGtidParsed)
+				if err := gtidParse.UpdateGtidSet(gtidStr); err != nil {
+					mutilGtidParsed = IsMultiSetReturned
+					gtidDoneParsed = IsDoneOfGtidParsed
+					if err != nil {
+						errors.Errorf("ParseGtidSetFromMysqlDump err: %v", err)
+					}
+				}
+
+				if parseBinlogPos && !binlogParsed {
+					if m := binlogExp.FindAllStringSubmatch(line, -1); len(m) == 1 {
+						binlogParsed = true
+						gtidDoneParsed = true
+					}
+				}
+
+			}
+		}
+
+		if tt.expected == "" {
+			if gtidParse.gset != nil && gtidParse.gset.String() != "" {
+				log.Fatalf("expected nil, but get %v", gtidParse.gset)
+			}
+			continue
+		}
+		expectedGtidset, err := mysql.ParseGTIDSet("mysql", tt.expected)
+		if err != nil {
+			log.Fatalf("Gtid:%s failed parsed, err: %v", tt.expected, err)
+		}
+		if !expectedGtidset.Equal(gtidParse.gset) {
+			log.Fatalf("expected:%v , but get: %v", expectedGtidset, gtidParse.gset)
+		}
+	}
+
+}
+
 func (s *parserTestSuite) TestParseFindTable(c *C) {
 	tbl := []struct {
 		sql   string