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

canal/canal.go

@@ -120,6 +120,23 @@ func NewCanal(cfg *Config) (*Canal, error) {
 	return c, nil
 }
 
+func (c *Canal) validateSetGtidPurged() error {
+	gtidPuged := strings.ToLower(c.cfg.Dump.GtidPurged)
+	if gtidPuged == "none" {
+		return nil
+	} else if gtidPuged == "auto" {
+		res, err := c.Execute(`SHOW GLOBAL VARIABLES LIKE "gtid_mode";`)
+		if err != nil {
+			return errors.Trace(err)
+		} else if f, _ := res.GetString(0, 1); f != "ON" {
+			return errors.Errorf("set-gtid-purged: %s, gtid_mode should be ON, but now is %s", c.cfg.Dump.GtidPurged, f)
+		}
+		return nil
+	}
+
+	return errors.Errorf("set-gtid-purged: none or auto  can be set, current is %s", gtidPuged)
+}
+
 func (c *Canal) prepareDumper() error {
 	var err error
 	dumpPath := c.cfg.Dump.ExecutionPath
@@ -128,6 +145,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 +176,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)
 	// Use hex blob for mysqldump

canal/canal_test.go

@@ -1,8 +1,13 @@
 package canal
 
 import (
+	"bufio"
 	"flag"
 	"fmt"
+	"io"
+	"regexp"
+	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -14,7 +19,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)
@@ -74,6 +82,126 @@ func (s *canalTestSuite) SetUpSuite(c *C) {
 	}()
 }
 
+func TestCanalHandler(t *testing.T) {
+	oneGtidExp := regexp.MustCompile("SET @@GLOBAL.GTID_PURGED='(.+)'")
+	mutilGtidStartExp := regexp.MustCompile("SET @@GLOBAL.GTID_PURGED='(.+),")
+	midUuidSet := regexp.MustCompile("(^\\w{8}(-\\w{4}){3}-\\w{12}:\\d+-\\d+),")
+	endUuidSet := regexp.MustCompile("(^\\w{8}(-\\w{4}){3}-\\w{12}:\\d+-\\d+)'")
+	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 {
+		h := dumpParseHandler{}
+		reader := strings.NewReader(tt.input)
+		newReader := bufio.NewReader(reader)
+		var binlogParsed bool
+		var gtidDoneParsed bool
+		var mutilGtidParsed bool
+		parseBinlogPos := true
+		for {
+			bytes, _, err := newReader.ReadLine()
+			line := string(bytes)
+			if err != io.EOF {
+				fmt.Println(string(line))
+			} else {
+				break
+			}
+			if parseBinlogPos && !gtidDoneParsed && !binlogParsed {
+				if m := oneGtidExp.FindAllStringSubmatch(line, -1); len(m) == 1 {
+					gset := m[0][1]
+					if err := h.UpdateGtidFromPurged(gset); err != nil {
+						errors.Trace(err)
+					}
+					gtidDoneParsed = true
+				}
+				if m := mutilGtidStartExp.FindAllStringSubmatch(line, -1); len(m) == 1 {
+					gset := m[0][1]
+					if err := h.UpdateGtidFromPurged(gset); err != nil {
+						errors.Trace(err)
+					}
+					mutilGtidParsed = true
+				}
+
+				if mutilGtidParsed && !gtidDoneParsed {
+					if m := midUuidSet.FindAllStringSubmatch(line, -1); len(m) == 1 {
+						gset := m[0][1]
+						if err := h.UpdateGtidFromPurged(gset); err != nil {
+							errors.Trace(err)
+						}
+
+					}
+
+					if m := endUuidSet.FindAllStringSubmatch(line, -1); len(m) == 1 {
+						gset := m[0][1]
+						if err := h.UpdateGtidFromPurged(gset); err != nil {
+							errors.Trace(err)
+						}
+						gtidDoneParsed = true
+					}
+
+				}
+			}
+
+			if parseBinlogPos && !binlogParsed {
+				if m := binlogExp.FindAllStringSubmatch(line, -1); len(m) == 1 {
+					name := m[0][1]
+					pos, err := strconv.ParseUint(m[0][2], 10, 64)
+					if err != nil {
+						errors.Errorf("parse binlog %v err, invalid number", line)
+					}
+
+					if err = h.BinLog(name, pos); err != nil && err != ErrSkip {
+						errors.Trace(err)
+					}
+
+					binlogParsed = true
+					gtidDoneParsed = true
+				}
+			}
+
+		}
+
+		if tt.expected == "" {
+			if h.gset != nil {
+				log.Fatalf("expected nil, but get %v", h.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(h.gset) {
+			log.Fatalf("expected:%v , but get: %v", expectedGtidset, h.gset)
+		}
+
+		//	c.Assert(expectedGtidset.Equal(h.gset), IsTrue)
+		//	c.Logf("parsed gtidset: %v", h.gset)
+	}
+
+}
+
 func (s *canalTestSuite) TearDownSuite(c *C) {
 	// To test the heartbeat and read timeout,so need to sleep 1 seconds without data transmission
 	c.Logf("Start testing the heartbeat and read timeout")

canal/config.go

@@ -34,6 +34,9 @@ type DumpConfig struct {
 	// 'FLUSH TABLES WITH READ LOCK'
 	SkipMasterData bool `toml:"skip_master_data"`
 
+	// set --set-gtid-purged none, auto; none for gtid is disabled or "version too low", auto for gtid_mode=on;
+	GtidPurged string `toml:"set_gtid_purged"`
+
 	// Set to change the default max_allowed_packet size
 	MaxAllowedPacketMB int `toml:"max_allowed_packet_mb"`
 
@@ -113,6 +116,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) UpdateGtidFromPurged(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
@@ -158,6 +167,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)
 	}

dump/dump.go

@@ -37,6 +37,7 @@ type Dumper struct {
 
 	masterDataSkipped bool
 	maxAllowedPacket  int
+	gtidPurged        string
 	hexBlob           bool
 }
 
@@ -87,6 +88,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
 }
@@ -143,6 +149,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))
@@ -215,6 +225,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

@@ -36,6 +36,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,6 +120,10 @@ func (h *testParseHandler) BinLog(name string, pos uint64) error {
 	return nil
 }
 
+func (h *testParseHandler) UpdateGtidFromPurged(gtidsets string) (err error) {
+	return nil
+}
+
 func (h *testParseHandler) Data(schema string, table string, values []string) error {
 	return nil
 }

dump/parser.go

@@ -19,15 +19,24 @@ var (
 type ParseHandler interface {
 	// Parse CHANGE MASTER TO MASTER_LOG_FILE=name, MASTER_LOG_POS=pos;
 	BinLog(name string, pos uint64) error
-
+	UpdateGtidFromPurged(gtidsets string) error
 	Data(schema string, table string, values []string) error
 }
 
+var oneGtidExp *regexp.Regexp
+var mutilGtidStartExp *regexp.Regexp
+var midUuidSet *regexp.Regexp
+var endUuidSet *regexp.Regexp
 var binlogExp *regexp.Regexp
 var useExp *regexp.Regexp
 var valuesExp *regexp.Regexp
 
 func init() {
+	oneGtidExp = regexp.MustCompile("SET @@GLOBAL.GTID_PURGED='(.+)'")
+	mutilGtidStartExp = regexp.MustCompile("SET @@GLOBAL.GTID_PURGED='(.+),")
+	midUuidSet = regexp.MustCompile("(^\\w{8}(-\\w{4}){3}-\\w{12}:\\d+-\\d+),")
+	endUuidSet = regexp.MustCompile("(^\\w{8}(-\\w{4}){3}-\\w{12}:\\d+-\\d+)'")
+
 	binlogExp = regexp.MustCompile("^CHANGE MASTER TO MASTER_LOG_FILE='(.+)', MASTER_LOG_POS=(\\d+);")
 	useExp = regexp.MustCompile("^USE `(.+)`;")
 	valuesExp = regexp.MustCompile("^INSERT INTO `(.+?)` VALUES \\((.+)\\);$")
@@ -40,6 +49,8 @@ func Parse(r io.Reader, h ParseHandler, parseBinlogPos bool) error {
 
 	var db string
 	var binlogParsed bool
+	var gtidDoneParsed bool
+	var mutilGtidParsed bool
 
 	for {
 		line, err := rb.ReadString('\n')
@@ -54,6 +65,43 @@ func Parse(r io.Reader, h ParseHandler, parseBinlogPos bool) error {
 			return c == '\r' || c == '\n'
 		})
 
+		// parsed gtid set from mysqldump,refer to canal_test.go TestDumperHandler function
+		if parseBinlogPos && !gtidDoneParsed && !binlogParsed {
+			if m := oneGtidExp.FindAllStringSubmatch(line, -1); len(m) == 1 {
+				gset := m[0][1]
+				if err := h.UpdateGtidFromPurged(gset); err != nil {
+					return errors.Trace(err)
+				}
+				gtidDoneParsed = true
+			}
+			if m := mutilGtidStartExp.FindAllStringSubmatch(line, -1); len(m) == 1 {
+				gset := m[0][1]
+				if err := h.UpdateGtidFromPurged(gset); err != nil {
+					return errors.Trace(err)
+				}
+				mutilGtidParsed = true
+			}
+
+			if mutilGtidParsed && !gtidDoneParsed {
+				if m := midUuidSet.FindAllStringSubmatch(line, -1); len(m) == 1 {
+					gset := m[0][1]
+					if err := h.UpdateGtidFromPurged(gset); err != nil {
+						return errors.Trace(err)
+					}
+
+				}
+
+				if m := endUuidSet.FindAllStringSubmatch(line, -1); len(m) == 1 {
+					gset := m[0][1]
+					if err := h.UpdateGtidFromPurged(gset); err != nil {
+						return errors.Trace(err)
+					}
+					gtidDoneParsed = true
+				}
+
+			}
+		}
+
 		if parseBinlogPos && !binlogParsed {
 			if m := binlogExp.FindAllStringSubmatch(line, -1); len(m) == 1 {
 				name := m[0][1]
@@ -67,6 +115,7 @@ func Parse(r io.Reader, h ParseHandler, parseBinlogPos bool) error {
 				}
 
 				binlogParsed = true
+				gtidDoneParsed = true
 			}
 		}