From 3cb465b3218f72ca1aac698fb89145eee8fa6301 Mon Sep 17 00:00:00 2001 From: GMHDBJD <35025882+GMHDBJD@users.noreply.github.com> Date: Tue, 26 Jan 2021 21:54:30 +0800 Subject: [PATCH] relay: align with MySQL when using GTID (#1390) --- _utils/terror_gen/errors_release.txt | 1 + chaos/cases/conf/source3.yaml | 2 +- errors.toml | 6 + pkg/binlog/event/event.go | 16 ++ pkg/binlog/reader/util.go | 36 ++++ pkg/binlog/reader/util_test.go | 29 +++ pkg/streamer/reader.go | 216 +++++++++----------- pkg/streamer/reader_test.go | 295 +++++++++++++++++++-------- pkg/terror/error_list.go | 5 + relay/relay.go | 35 +++- relay/relay_test.go | 14 +- relay/transformer/transformer.go | 2 + 12 files changed, 434 insertions(+), 223 deletions(-) diff --git a/_utils/terror_gen/errors_release.txt b/_utils/terror_gen/errors_release.txt index 1650d0a98e..22561a4cb3 100644 --- a/_utils/terror_gen/errors_release.txt +++ b/_utils/terror_gen/errors_release.txt @@ -127,6 +127,7 @@ ErrNoUUIDDirMatchGTID,[code=11120:class=functional:scope=internal:level=high], " ErrNoRelayPosMatchGTID,[code=11121:class=functional:scope=internal:level=high], "Message: no relay pos match gtid %s" ErrReaderReachEndOfFile,[code=11122:class=functional:scope=internal:level=low] ErrMetadataNoBinlogLoc,[code=11123:class=functional:scope=upstream:level=low], "Message: didn't found binlog location in dumped metadata file %s, Workaround: Please check log of dump unit, there maybe errors when read upstream binlog status" +ErrPreviousGTIDNotExist,[code=11124:class=functional:scope=internal:level=high], "Message: no previous gtid event from binlog %s" ErrConfigCheckItemNotSupport,[code=20001:class=config:scope=internal:level=medium], "Message: checking item %s is not supported\n%s, Workaround: Please check `ignore-checking-items` config in task configuration file, which can be set including `all`/`dump_privilege`/`replication_privilege`/`version`/`binlog_enable`/`binlog_format`/`binlog_row_image`/`table_schema`/`schema_of_shard_tables`/`auto_increment_ID`." ErrConfigTomlTransform,[code=20002:class=config:scope=internal:level=medium], "Message: %s, Workaround: Please check the configuration file has correct TOML format." ErrConfigYamlTransform,[code=20003:class=config:scope=internal:level=medium], "Message: %s, Workaround: Please check the configuration file has correct YAML format." diff --git a/chaos/cases/conf/source3.yaml b/chaos/cases/conf/source3.yaml index d944aebcc0..2d4bff1ec9 100644 --- a/chaos/cases/conf/source3.yaml +++ b/chaos/cases/conf/source3.yaml @@ -1,5 +1,5 @@ source-id: "replica-03" -enable-gtid: false +enable-gtid: true enable-relay: true from: diff --git a/errors.toml b/errors.toml index 1e635f7f5c..a1ef08ecb3 100644 --- a/errors.toml +++ b/errors.toml @@ -772,6 +772,12 @@ description = "" workaround = "Please check log of dump unit, there maybe errors when read upstream binlog status" tags = ["upstream", "low"] +[error.DM-functional-11124] +message = "no previous gtid event from binlog %s" +description = "" +workaround = "" +tags = ["internal", "high"] + [error.DM-config-20001] message = "checking item %s is not supported\n%s" description = "" diff --git a/pkg/binlog/event/event.go b/pkg/binlog/event/event.go index eb3995358d..2c6c20324e 100644 --- a/pkg/binlog/event/event.go +++ b/pkg/binlog/event/event.go @@ -851,3 +851,19 @@ func GenDummyEvent(header *replication.EventHeader, latestPos uint32, eventSize ev, err := GenQueryEvent(&headerClone, latestPos, 0, 0, 0, nil, nil, queryBytes) return ev, err } + +// GenHeartbeatEvent generates a heartbeat event. +// ref: https://dev.mysql.com/doc/internals/en/heartbeat-event.html +func GenHeartbeatEvent(header *replication.EventHeader) *replication.BinlogEvent { + // modify header + headerClone := *header // do a copy + headerClone.Flags = 0 + headerClone.EventSize = 39 + headerClone.Timestamp = 0 + headerClone.EventType = replication.HEARTBEAT_EVENT + + eventBytes := make([]byte, 39) + ev := &replication.BinlogEvent{Header: &headerClone, Event: &replication.GenericEvent{Data: eventBytes}} + + return ev +} diff --git a/pkg/binlog/reader/util.go b/pkg/binlog/reader/util.go index 2a45978931..01aaee33db 100644 --- a/pkg/binlog/reader/util.go +++ b/pkg/binlog/reader/util.go @@ -140,3 +140,39 @@ func GetGTIDsForPos(ctx context.Context, r Reader, endPos gmysql.Position) (gtid } } } + +// GetPreviousGTIDFromGTIDSet tries to get previous GTID sets from Previous_GTID_EVENT GTID for the specified GITD Set. +// events should be [fake_rotate_event,format_description_event,previous_gtids_event/mariadb_gtid_list_event] +func GetPreviousGTIDFromGTIDSet(ctx context.Context, r Reader, gset gtid.Set) (gtid.Set, error) { + err := r.StartSyncByGTID(gset) + if err != nil { + return nil, err + } + defer r.Close() + + for { + var e *replication.BinlogEvent + e, err = r.GetEvent(ctx) + if err != nil { + return nil, err + } + + switch e.Header.EventType { + case replication.ROTATE_EVENT: + if e.Header.Timestamp == 0 || e.Header.LogPos == 0 { // fake rotate event + continue + } + return nil, terror.ErrPreviousGTIDNotExist.Generate(gset.String()) + case replication.FORMAT_DESCRIPTION_EVENT: + continue + case replication.PREVIOUS_GTIDS_EVENT: + previousGset, err := event.GTIDsFromPreviousGTIDsEvent(e) + return previousGset, err + case replication.MARIADB_GTID_LIST_EVENT: + previousGset, err := event.GTIDsFromMariaDBGTIDListEvent(e) + return previousGset, err + default: + return nil, terror.ErrPreviousGTIDNotExist.Generate(gset.String()) + } + } +} diff --git a/pkg/binlog/reader/util_test.go b/pkg/binlog/reader/util_test.go index dd0558f021..c4c3d5f50d 100644 --- a/pkg/binlog/reader/util_test.go +++ b/pkg/binlog/reader/util_test.go @@ -62,3 +62,32 @@ func (t *testTCPReaderSuite) TestGetGTIDsForPos(c *C) { c.Assert(err, ErrorMatches, ".*invalid position .* or GTID not enabled in upstream.*") c.Assert(gs, IsNil) } + +// added to testTCPReaderSuite to re-use DB connection. +func (t *testTCPReaderSuite) TestGetPreviousGTIDFromGTIDSet(c *C) { + var ( + cfg = replication.BinlogSyncerConfig{ + ServerID: serverIDs[1], + Flavor: flavor, + Host: t.host, + Port: uint16(t.port), + User: t.user, + Password: t.password, + UseDecimal: true, + VerifyChecksum: true, + } + ctx, cancel = context.WithTimeout(context.Background(), utils.DefaultDBTimeout) + ) + defer cancel() + + _, endGS, err := utils.GetMasterStatus(ctx, t.db, flavor) + c.Assert(err, IsNil) + + r1 := NewTCPReader(cfg) + c.Assert(r1, NotNil) + defer r1.Close() + + gs, err := GetPreviousGTIDFromGTIDSet(ctx, r1, endGS) + c.Assert(err, IsNil) + c.Assert(endGS.Contain(gs), IsTrue) +} diff --git a/pkg/streamer/reader.go b/pkg/streamer/reader.go index fe99e1c1c7..bb40760f0f 100644 --- a/pkg/streamer/reader.go +++ b/pkg/streamer/reader.go @@ -23,7 +23,6 @@ import ( "sync" "time" - "github.com/BurntSushi/toml" "github.com/google/uuid" "github.com/pingcap/errors" "github.com/siddontang/go-mysql/mysql" @@ -31,8 +30,10 @@ import ( "go.uber.org/zap" "github.com/pingcap/dm/pkg/binlog" + "github.com/pingcap/dm/pkg/binlog/event" "github.com/pingcap/dm/pkg/binlog/reader" tcontext "github.com/pingcap/dm/pkg/context" + "github.com/pingcap/dm/pkg/gtid" "github.com/pingcap/dm/pkg/log" "github.com/pingcap/dm/pkg/terror" "github.com/pingcap/dm/pkg/utils" @@ -118,59 +119,21 @@ func (r *BinlogReader) checkRelayPos(pos mysql.Position) error { return nil } -// getUUIDByGTID gets uuid subdir which contain the gtid set -func (r *BinlogReader) getUUIDByGTID(gset mysql.GTIDSet) (string, error) { - // get flush logs from oldest to newest - for _, uuid := range r.uuids { - filename := path.Join(r.cfg.RelayDir, uuid, utils.MetaFilename) - var meta Meta - _, err := toml.DecodeFile(filename, &meta) - if err != nil { - return "", terror.ErrRelayLoadMetaData.Delegate(err) - } - - gs, err := mysql.ParseGTIDSet(r.cfg.Flavor, meta.BinlogGTID) - if err != nil { - return "", terror.ErrRelayLoadMetaData.Delegate(err) - } - if gs.Contain(gset) { - r.tctx.L().Info("get uuid subdir by gtid", zap.Stringer("GTID Set", gset), zap.String("uuid", uuid), zap.Stringer("latest GTID Set in subdir", gs)) - return uuid, nil - } - } - - // TODO: use a better mechanism to call relay.meta.Flush - // get the meta save in memory - relayMetaHub := GetRelayMetaHub() - relayMeta := relayMetaHub.GetMeta() - - if len(relayMeta.UUID) > 0 { - gs, err := mysql.ParseGTIDSet(r.cfg.Flavor, relayMeta.BinlogGTID) - if err != nil { - return "", terror.ErrRelayLoadMetaData.Delegate(err) - } - if gs.Contain(gset) { - r.tctx.L().Info("get uuid subdir by gtid", zap.Stringer("GTID Set", gset), zap.String("uuid", relayMeta.UUID)) - return relayMeta.UUID, nil - } - } - return "", terror.ErrNoUUIDDirMatchGTID.Generate(gset.String()) -} - -// GetFilePosByGTID tries to get Pos by GTID for file -func (r *BinlogReader) GetFilePosByGTID(ctx context.Context, filePath string, gset mysql.GTIDSet) (uint32, error) { +// IsGTIDCoverPreviousFiles check whether gset contains file's previous_gset +func (r *BinlogReader) IsGTIDCoverPreviousFiles(ctx context.Context, filePath string, gset mysql.GTIDSet) (bool, error) { fileReader := reader.NewFileReader(&reader.FileReaderConfig{Timezone: r.cfg.Timezone}) defer fileReader.Close() err := fileReader.StartSyncByPos(mysql.Position{Name: filePath, Pos: 4}) if err != nil { - return 0, err + return false, err } - lastPos := uint32(0) + var gs gtid.Set + for { select { case <-ctx.Done(): - return 0, nil + return false, nil default: } @@ -179,95 +142,66 @@ func (r *BinlogReader) GetFilePosByGTID(ctx context.Context, filePath string, gs cancel() if err != nil { // reach end of file + // Maybe we can only Parse the first three fakeRotate, Format_desc and Previous_gtids events. if terror.ErrReaderReachEndOfFile.Equal(err) { - return lastPos, nil + return false, terror.ErrPreviousGTIDNotExist.Generate(filePath) } - return 0, err + return false, err } - switch ev := e.Event.(type) { - case *replication.PreviousGTIDsEvent: - // nil previous gtid event, continue to parse file - if len(ev.GTIDSets) == 0 { - break - } - gs, err := mysql.ParseGTIDSet(r.cfg.Flavor, ev.GTIDSets) - if err != nil { - return 0, err - } - // if PreviousGITDsEvent contain but not equal the gset, go to previous file - if gs.Contain(gset) { - // continue to parse file if gset equals gs - if gset.Contain(gs) { - break - } - return 0, nil - } - case *replication.RotateEvent: - // should not happen - if e.Header.Timestamp != 0 && e.Header.LogPos != 0 { - return lastPos, nil - } + if e.Header.EventType == replication.PREVIOUS_GTIDS_EVENT { + gs, err = event.GTIDsFromPreviousGTIDsEvent(e) + } else if e.Header.EventType == replication.MARIADB_GTID_LIST_EVENT { + gs, err = event.GTIDsFromMariaDBGTIDListEvent(e) + } else { continue - case *replication.GTIDEvent: - u, _ := uuid.FromBytes(ev.SID) - gs, err := mysql.ParseGTIDSet(r.cfg.Flavor, fmt.Sprintf("%s:%d", u.String(), ev.GNO)) - if err != nil { - return 0, err - } - // meet first gtid event greater than gset - if !gset.Contain(gs) { - return lastPos, nil - } - case *replication.MariadbGTIDEvent: - GTID := ev.GTID - gs, err := mysql.ParseGTIDSet(r.cfg.Flavor, fmt.Sprintf("%d-%d-%d", GTID.DomainID, GTID.ServerID, GTID.SequenceNumber)) - if err != nil { - return 0, err - } - // meet first gtid event greater than gset - if !gset.Contain(gs) { - return lastPos, nil - } } - lastPos = e.Header.LogPos + + if err != nil { + return false, err + } + return gset.Contain(gs.Origin()), nil } } -// getPosByGTID gets position by gtid +// getPosByGTID gets file position by gtid, result should be (filename, 4) func (r *BinlogReader) getPosByGTID(gset mysql.GTIDSet) (*mysql.Position, error) { - uuid, err := r.getUUIDByGTID(gset) - if err != nil { - return nil, err - } - _, suffix, err := utils.ParseSuffixForUUID(uuid) - if err != nil { - return nil, err - } - - uuidDir := path.Join(r.cfg.RelayDir, uuid) - allFiles, err := CollectAllBinlogFiles(uuidDir) - if err != nil { - return nil, err - } + // start from newest uuid dir + for i := len(r.uuids) - 1; i >= 0; i-- { + uuid := r.uuids[i] + _, suffix, err := utils.ParseSuffixForUUID(uuid) + if err != nil { + return nil, err + } - // iterate files from the newest one - for i := len(allFiles) - 1; i >= 0; i-- { - file := allFiles[i] - filePath := path.Join(uuidDir, file) - pos, err := r.GetFilePosByGTID(r.tctx.Ctx, filePath, gset) + uuidDir := path.Join(r.cfg.RelayDir, uuid) + allFiles, err := CollectAllBinlogFiles(uuidDir) if err != nil { return nil, err } - if pos != 0 { - fileName, err := binlog.ParseFilename(file) + + // iterate files from the newest one + for i := len(allFiles) - 1; i >= 0; i-- { + file := allFiles[i] + filePath := path.Join(uuidDir, file) + // if input `gset` not contain previous_gtids_event's gset (complementary set of `gset` overlap with + // previous_gtids_event), that means there're some needed events in previous files. + // so we go to previous one + contain, err := r.IsGTIDCoverPreviousFiles(r.tctx.Ctx, filePath, gset) if err != nil { return nil, err } - return &mysql.Position{ - Name: binlog.ConstructFilenameWithUUIDSuffix(fileName, utils.SuffixIntToStr(suffix)), - Pos: uint32(pos), - }, nil + if contain { + fileName, err := binlog.ParseFilename(file) + if err != nil { + return nil, err + } + // Start at the beginning of the file + return &mysql.Position{ + Name: binlog.ConstructFilenameWithUUIDSuffix(fileName, utils.SuffixIntToStr(suffix)), + Pos: 4, + }, nil + } } } return nil, terror.ErrNoRelayPosMatchGTID.Generate(gset.String()) @@ -499,6 +433,7 @@ func (r *BinlogReader) parseFile( uuidSuffix := utils.SuffixIntToStr(suffixInt) // current UUID's suffix, which will be added to binlog name latestPos = offset // set to argument passed in + replaceWithHeartbeat := false onEventFunc := func(e *replication.BinlogEvent) error { r.tctx.L().Debug("read event", zap.Reflect("header", e.Header)) @@ -535,9 +470,9 @@ func (r *BinlogReader) parseFile( break } u, _ := uuid.FromBytes(ev.SID) - err2 := r.advanceCurrentGtidSet(fmt.Sprintf("%s:%d", u.String(), ev.GNO)) - if err2 != nil { - return errors.Trace(err2) + replaceWithHeartbeat, err = r.advanceCurrentGtidSet(fmt.Sprintf("%s:%d", u.String(), ev.GNO)) + if err != nil { + return errors.Trace(err) } latestPos = int64(e.Header.LogPos) case *replication.MariadbGTIDEvent: @@ -546,9 +481,9 @@ func (r *BinlogReader) parseFile( break } GTID := ev.GTID - err2 := r.advanceCurrentGtidSet(fmt.Sprintf("%d-%d-%d", GTID.DomainID, GTID.ServerID, GTID.SequenceNumber)) - if err2 != nil { - return errors.Trace(err2) + replaceWithHeartbeat, err = r.advanceCurrentGtidSet(fmt.Sprintf("%d-%d-%d", GTID.DomainID, GTID.ServerID, GTID.SequenceNumber)) + if err != nil { + return errors.Trace(err) } latestPos = int64(e.Header.LogPos) case *replication.XIDEvent: @@ -562,6 +497,22 @@ func (r *BinlogReader) parseFile( latestPos = int64(e.Header.LogPos) } + // align with MySQL + // if an event's gtid has been contained by given gset + // replace it with HEARTBEAT event + // for Mariadb, it will bee replaced with MARIADB_GTID_LIST_EVENT + // In DM, we replace both of them with HEARTBEAT event + if replaceWithHeartbeat { + switch e.Event.(type) { + // Only replace transaction event + // Other events such as FormatDescriptionEvent, RotateEvent, etc. should be the same as before + case *replication.RowsEvent, *replication.QueryEvent, *replication.GTIDEvent, *replication.XIDEvent, *replication.TableMapEvent: + // replace with heartbeat event + e = event.GenHeartbeatEvent(e.Header) + default: + } + } + select { case s.ch <- e: case <-ctx.Done(): @@ -666,16 +617,31 @@ func (r *BinlogReader) getCurrentGtidSet() mysql.GTIDSet { return r.currGset.Clone() } -func (r *BinlogReader) advanceCurrentGtidSet(gtid string) error { +// advanceCurrentGtidSet advance gtid set and return whether currGset not updated +func (r *BinlogReader) advanceCurrentGtidSet(gtid string) (bool, error) { if r.currGset == nil { r.currGset = r.prevGset.Clone() } + // Special treatment for Maridb + // MaridbGTIDSet.Update(gtid) will replace gset with given gtid + // ref https://github.com/siddontang/go-mysql/blob/0c5789dd0bd378b4b84f99b320a2d35a80d8858f/mysql/mariadb_gtid.go#L96 + if r.cfg.Flavor == mysql.MariaDBFlavor { + gset, err := mysql.ParseMariadbGTIDSet(gtid) + if err != nil { + return false, err + } + if r.currGset.Contain(gset) { + return true, nil + } + } prev := r.currGset.Clone() err := r.currGset.Update(gtid) if err == nil { if !r.currGset.Equal(prev) { r.prevGset = prev + return false, nil } + return true, nil } - return err + return false, err } diff --git a/pkg/streamer/reader_test.go b/pkg/streamer/reader_test.go index 040916cc28..f9d9e7d44f 100644 --- a/pkg/streamer/reader_test.go +++ b/pkg/streamer/reader_test.go @@ -628,52 +628,58 @@ func (t *testReaderSuite) TestStartSyncByGTID(c *C) { lastGTID gtid.Set previousGset, _ = gtid.ParserGTID(mysql.MySQLFlavor, "") ) - type FileEventType struct { - filename string - eventTypes []replication.EventType + + type EventResult struct { + eventType replication.EventType + result string // filename result of getPosByGTID + } + + type FileEventResult struct { + filename string + eventResults []EventResult } testCase := []struct { - serverUUID string - uuid string - gtidStr string - fileEventTypes []FileEventType + serverUUID string + uuid string + gtidStr string + fileEventResult []FileEventResult }{ { "ba8f633f-1f15-11eb-b1c7-0242ac110002", "ba8f633f-1f15-11eb-b1c7-0242ac110002.000001", "ba8f633f-1f15-11eb-b1c7-0242ac110002:0", - []FileEventType{ + []FileEventResult{ { "mysql.000001", - []replication.EventType{ - replication.PREVIOUS_GTIDS_EVENT, - replication.QUERY_EVENT, - replication.XID_EVENT, - replication.QUERY_EVENT, - replication.XID_EVENT, - replication.ROTATE_EVENT, + []EventResult{ + {replication.PREVIOUS_GTIDS_EVENT, ""}, + {replication.QUERY_EVENT, "mysql|000001.000001"}, + {replication.XID_EVENT, "mysql|000001.000001"}, + {replication.QUERY_EVENT, "mysql|000001.000001"}, + {replication.XID_EVENT, "mysql|000001.000002"}, // next binlog filename + {replication.ROTATE_EVENT, ""}, }, }, { "mysql.000002", - []replication.EventType{ - replication.PREVIOUS_GTIDS_EVENT, - replication.QUERY_EVENT, - replication.XID_EVENT, - replication.QUERY_EVENT, - replication.XID_EVENT, - replication.ROTATE_EVENT, + []EventResult{ + {replication.PREVIOUS_GTIDS_EVENT, ""}, + {replication.QUERY_EVENT, "mysql|000001.000002"}, + {replication.XID_EVENT, "mysql|000001.000002"}, + {replication.QUERY_EVENT, "mysql|000001.000002"}, + {replication.XID_EVENT, "mysql|000001.000003"}, // next binlog filename + {replication.ROTATE_EVENT, ""}, }, }, { "mysql.000003", - []replication.EventType{ - replication.PREVIOUS_GTIDS_EVENT, - replication.QUERY_EVENT, - replication.XID_EVENT, - replication.QUERY_EVENT, - replication.XID_EVENT, + []EventResult{ + {replication.PREVIOUS_GTIDS_EVENT, ""}, + {replication.QUERY_EVENT, "mysql|000001.000003"}, + {replication.XID_EVENT, "mysql|000001.000003"}, + {replication.QUERY_EVENT, "mysql|000001.000003"}, + {replication.XID_EVENT, "mysql|000002.000001"}, // next subdir }, }, }, @@ -682,31 +688,31 @@ func (t *testReaderSuite) TestStartSyncByGTID(c *C) { "bf6227a7-1f15-11eb-9afb-0242ac110004", "bf6227a7-1f15-11eb-9afb-0242ac110004.000002", "bf6227a7-1f15-11eb-9afb-0242ac110004:20", - []FileEventType{ + []FileEventResult{ { "mysql.000001", - []replication.EventType{ - replication.PREVIOUS_GTIDS_EVENT, - replication.QUERY_EVENT, - replication.XID_EVENT, - replication.QUERY_EVENT, - replication.XID_EVENT, - replication.ROTATE_EVENT, + []EventResult{ + {replication.PREVIOUS_GTIDS_EVENT, ""}, + {replication.QUERY_EVENT, "mysql|000002.000001"}, + {replication.XID_EVENT, "mysql|000002.000001"}, + {replication.QUERY_EVENT, "mysql|000002.000001"}, + {replication.XID_EVENT, "mysql|000002.000002"}, + {replication.ROTATE_EVENT, ""}, }, }, { "mysql.000002", - []replication.EventType{ - replication.PREVIOUS_GTIDS_EVENT, - replication.QUERY_EVENT, - replication.XID_EVENT, - replication.QUERY_EVENT, - replication.XID_EVENT, - replication.ROTATE_EVENT, + []EventResult{ + {replication.PREVIOUS_GTIDS_EVENT, ""}, + {replication.QUERY_EVENT, "mysql|000002.000002"}, + {replication.XID_EVENT, "mysql|000002.000002"}, + {replication.QUERY_EVENT, "mysql|000002.000002"}, + {replication.XID_EVENT, "mysql|000003.000001"}, + {replication.ROTATE_EVENT, ""}, }, }, { "mysql.000003", - []replication.EventType{ - replication.PREVIOUS_GTIDS_EVENT, + []EventResult{ + {replication.PREVIOUS_GTIDS_EVENT, ""}, }, }, }, @@ -714,15 +720,15 @@ func (t *testReaderSuite) TestStartSyncByGTID(c *C) { "bcbf9d42-1f15-11eb-a41c-0242ac110003", "bcbf9d42-1f15-11eb-a41c-0242ac110003.000003", "bcbf9d42-1f15-11eb-a41c-0242ac110003:30", - []FileEventType{ + []FileEventResult{ { "mysql.000001", - []replication.EventType{ - replication.PREVIOUS_GTIDS_EVENT, - replication.QUERY_EVENT, - replication.XID_EVENT, - replication.QUERY_EVENT, - replication.XID_EVENT, + []EventResult{ + {replication.PREVIOUS_GTIDS_EVENT, ""}, + {replication.QUERY_EVENT, "mysql|000003.000001"}, + {replication.XID_EVENT, "mysql|000003.000001"}, + {replication.QUERY_EVENT, "mysql|000003.000001"}, + {replication.XID_EVENT, "mysql|000003.000001"}, }, }, }, @@ -739,12 +745,10 @@ func (t *testReaderSuite) TestStartSyncByGTID(c *C) { c.Assert(err, IsNil) var allEvents []*replication.BinlogEvent + var allResults []string - // prePosMap record previous uuid's last file's size - // when master switch, we get the previous uuid's last pos now - prePosMap := make(map[string]uint32) + // generate binlog file for _, subDir := range testCase { - prePosMap[subDir.serverUUID] = lastPos lastPos = 4 lastGTID, err = gtid.ParserGTID(mysql.MySQLFlavor, subDir.gtidStr) c.Assert(err, IsNil) @@ -752,12 +756,21 @@ func (t *testReaderSuite) TestStartSyncByGTID(c *C) { err = os.MkdirAll(uuidDir, 0700) c.Assert(err, IsNil) - for _, fileEventType := range subDir.fileEventTypes { - events, lastPos, lastGTID, previousGset = t.genEvents(c, fileEventType.eventTypes, lastPos, lastGTID, previousGset) + for _, fileEventResult := range subDir.fileEventResult { + eventTypes := []replication.EventType{} + for _, eventResult := range fileEventResult.eventResults { + eventTypes = append(eventTypes, eventResult.eventType) + if len(eventResult.result) != 0 { + allResults = append(allResults, eventResult.result) + } + } + + // generate events + events, lastPos, lastGTID, previousGset = t.genEvents(c, eventTypes, lastPos, lastGTID, previousGset) allEvents = append(allEvents, events...) // write binlog file - f, err2 := os.OpenFile(path.Join(uuidDir, fileEventType.filename), os.O_CREATE|os.O_WRONLY, 0600) + f, err2 := os.OpenFile(path.Join(uuidDir, fileEventResult.filename), os.O_CREATE|os.O_WRONLY, 0600) c.Assert(err2, IsNil) _, err = f.Write(replication.BinLogFileHeader) c.Assert(err, IsNil) @@ -768,7 +781,7 @@ func (t *testReaderSuite) TestStartSyncByGTID(c *C) { f.Close() // write meta file - meta := Meta{BinLogName: fileEventType.filename, BinLogPos: lastPos, BinlogGTID: previousGset.String()} + meta := Meta{BinLogName: fileEventResult.filename, BinLogPos: lastPos, BinlogGTID: previousGset.String()} metaFile, err2 := os.Create(path.Join(uuidDir, utils.MetaFilename)) c.Assert(err2, IsNil) err = toml.NewEncoder(metaFile).Encode(meta) @@ -788,45 +801,117 @@ func (t *testReaderSuite) TestStartSyncByGTID(c *C) { for { ev, err2 := s.GetEvent(ctx) c.Assert(err2, IsNil) - if ev.Header.Timestamp == 0 || ev.Header.LogPos == 0 { + if ev.Header.Timestamp == 0 && ev.Header.LogPos == 0 { continue // ignore fake event } obtainBaseEvents = append(obtainBaseEvents, ev) - // start after the first nil previous event - if len(obtainBaseEvents) == len(allEvents)-1 { + // start from the first format description event + if len(obtainBaseEvents) == len(allEvents) { break } } - preGset, err := gtid.ParserGTID(mysql.MySQLFlavor, "") + preGset, err := mysql.ParseGTIDSet(mysql.MySQLFlavor, "") c.Assert(err, IsNil) - // allEvents: [FORMAT_DESCRIPTION_EVENT, PREVIOUS_GTIDS_EVENT, GTID_EVENT, QUERY_EVENT...] - // obtainBaseEvents: [FORMAT_DESCRIPTION_EVENT(generated), GTID_EVENT, QUERY_EVENT...] + + gtidEventCount := 0 for i, event := range obtainBaseEvents { - if i == 0 { - c.Assert(event.Header.EventType, Equals, replication.FORMAT_DESCRIPTION_EVENT) - continue - } - c.Assert(event.Header, DeepEquals, allEvents[i+1].Header) - switch ev := event.Event.(type) { - case *replication.GTIDEvent: - pos, err2 := r.getPosByGTID(preGset.Origin().Clone()) + c.Assert(event.Header, DeepEquals, allEvents[i].Header) + if ev, ok := event.Event.(*replication.GTIDEvent); ok { u, _ := uuid.FromBytes(ev.SID) + c.Assert(preGset.Update(fmt.Sprintf("%s:%d", u.String(), ev.GNO)), IsNil) + + // get pos by preGset + pos, err2 := r.getPosByGTID(preGset.Clone()) c.Assert(err2, IsNil) - // new uuid dir - if len(preGset.String()) != 0 && !strings.Contains(preGset.String(), u.String()) { - c.Assert(pos.Pos, Equals, prePosMap[u.String()], Commentf("a %d", i)) - } else { - c.Assert(pos.Pos, Equals, event.Header.LogPos-event.Header.EventSize, Commentf("b %d", i)) - } - case *replication.QueryEvent: - err2 := preGset.Set(ev.GSet) - c.Assert(err2, IsNil) - case *replication.XIDEvent: - err2 := preGset.Set(ev.GSet) - c.Assert(err2, IsNil) + // check result + c.Assert(pos.Name, Equals, allResults[gtidEventCount]) + c.Assert(pos.Pos, Equals, uint32(4)) + gtidEventCount++ + } + } + + r.Close() + r = NewBinlogReader(log.L(), cfg) + + excludeStrs := []string{} + // exclude first uuid + excludeServerUUID := testCase[0].serverUUID + excludeUUID := testCase[0].uuid + for _, s := range strings.Split(preGset.String(), ",") { + if !strings.Contains(s, excludeServerUUID) { + excludeStrs = append(excludeStrs, s) + } + } + excludeStr := strings.Join(excludeStrs, ",") + excludeGset, err := mysql.ParseGTIDSet(mysql.MySQLFlavor, excludeStr) + c.Assert(err, IsNil) + + // StartSyncByGtid exclude first uuid + s, err = r.StartSyncByGTID(excludeGset) + c.Assert(err, IsNil) + obtainBaseEvents = []*replication.BinlogEvent{} + + for { + ev, err2 := s.GetEvent(ctx) + c.Assert(err2, IsNil) + if ev.Header.Timestamp == 0 && ev.Header.LogPos == 0 { + continue // ignore fake event + } + obtainBaseEvents = append(obtainBaseEvents, ev) + // start from the first format description event + if len(obtainBaseEvents) == len(allEvents) { + break + } + } + + gset := excludeGset.Clone() + // every gtid event not from first uuid should become heartbeat event + for i, event := range obtainBaseEvents { + switch event.Header.EventType { + case replication.HEARTBEAT_EVENT: + c.Assert(event.Header.LogPos, Equals, allEvents[i].Header.LogPos) + case replication.GTID_EVENT: + // check gtid event comes from first uuid subdir + ev, _ := event.Event.(*replication.GTIDEvent) + u, _ := uuid.FromBytes(ev.SID) + c.Assert(u.String(), Equals, excludeServerUUID) + c.Assert(event.Header, DeepEquals, allEvents[i].Header) + c.Assert(gset.Update(fmt.Sprintf("%s:%d", u.String(), ev.GNO)), IsNil) + default: + c.Assert(event.Header, DeepEquals, allEvents[i].Header) } } + // gset same as preGset now + c.Assert(gset.Equal(preGset), IsTrue) + + // purge first uuid subdir's first binlog file + c.Assert(os.Remove(path.Join(baseDir, excludeUUID, "mysql.000001")), IsNil) + + r.Close() + r = NewBinlogReader(log.L(), cfg) + _, err = r.StartSyncByGTID(preGset) + c.Assert(err, IsNil) + + r.Close() + r = NewBinlogReader(log.L(), cfg) + _, err = r.StartSyncByGTID(excludeGset) + // error because file has been purge + c.Assert(terror.ErrNoRelayPosMatchGTID.Equal(err), IsTrue) + + // purge first uuid subdir + c.Assert(os.RemoveAll(path.Join(baseDir, excludeUUID)), IsNil) + + r.Close() + r = NewBinlogReader(log.L(), cfg) + _, err = r.StartSyncByGTID(preGset) + c.Assert(err, IsNil) + + r.Close() + r = NewBinlogReader(log.L(), cfg) + _, err = r.StartSyncByGTID(excludeGset) + // error because subdir has been purge + c.Assert(err, ErrorMatches, ".*no such file or directory.*") } func (t *testReaderSuite) TestStartSyncError(c *C) { @@ -854,7 +939,7 @@ func (t *testReaderSuite) TestStartSyncError(c *C) { c.Assert(s, IsNil) s, err = r.StartSyncByGTID(t.lastGTID.Origin().Clone()) - c.Assert(err, ErrorMatches, ".*no relay subdir match gtid.*") + c.Assert(err, ErrorMatches, ".*no relay pos match gtid.*") c.Assert(s, IsNil) // write UUIDs into index file @@ -869,7 +954,7 @@ func (t *testReaderSuite) TestStartSyncError(c *C) { c.Assert(s, IsNil) s, err = r.StartSyncByGTID(t.lastGTID.Origin().Clone()) - c.Assert(err, ErrorMatches, ".*load meta data.*no such file or directory.*") + c.Assert(err, ErrorMatches, ".*no such file or directory.*") c.Assert(s, IsNil) // can not re-start the reader @@ -899,6 +984,40 @@ func (t *testReaderSuite) TestStartSyncError(c *C) { c.Assert(s, IsNil) } +func (t *testReaderSuite) TestAdvanceCurrentGTIDSet(c *C) { + var ( + baseDir = c.MkDir() + cfg = &BinlogReaderConfig{RelayDir: baseDir, Flavor: mysql.MySQLFlavor} + r = NewBinlogReader(log.L(), cfg) + mysqlGset, _ = mysql.ParseMysqlGTIDSet("b60868af-5a6f-11e9-9ea3-0242ac160006:1-6") + mariadbGset, _ = mysql.ParseMariadbGTIDSet("0-1-5") + ) + r.prevGset = mysqlGset.Clone() + r.currGset = nil + notUpdated, err := r.advanceCurrentGtidSet("b60868af-5a6f-11e9-9ea3-0242ac160006:6") + c.Assert(err, IsNil) + c.Assert(notUpdated, IsTrue) + c.Assert(mysqlGset.Equal(r.currGset), IsTrue) + notUpdated, err = r.advanceCurrentGtidSet("b60868af-5a6f-11e9-9ea3-0242ac160006:7") + c.Assert(err, IsNil) + c.Assert(notUpdated, IsFalse) + c.Assert(mysqlGset.Equal(r.prevGset), IsTrue) + c.Assert(r.currGset.String(), Equals, "b60868af-5a6f-11e9-9ea3-0242ac160006:1-7") + + r.cfg.Flavor = mysql.MariaDBFlavor + r.prevGset = mariadbGset.Clone() + r.currGset = nil + notUpdated, err = r.advanceCurrentGtidSet("0-1-3") + c.Assert(err, IsNil) + c.Assert(notUpdated, IsTrue) + c.Assert(mariadbGset.Equal(r.currGset), IsTrue) + notUpdated, err = r.advanceCurrentGtidSet("0-1-6") + c.Assert(err, IsNil) + c.Assert(notUpdated, IsFalse) + c.Assert(mariadbGset.Equal(r.prevGset), IsTrue) + c.Assert(r.currGset.String(), Equals, "0-1-6") +} + func (t *testReaderSuite) genBinlogEvents(c *C, latestPos uint32, latestGTID gtid.Set) ([]*replication.BinlogEvent, uint32, gtid.Set) { var ( header = &replication.EventHeader{ diff --git a/pkg/terror/error_list.go b/pkg/terror/error_list.go index 7612e28b41..c43b18e96a 100644 --- a/pkg/terror/error_list.go +++ b/pkg/terror/error_list.go @@ -182,6 +182,9 @@ const ( // pkg/dumpling codeMetadataNoBinlogLoc + + // pkg/streamer + codePreviousGTIDNotExist ) // Config related error code list @@ -780,6 +783,8 @@ var ( // pkg/dumplling ErrMetadataNoBinlogLoc = New(codeMetadataNoBinlogLoc, ClassFunctional, ScopeUpstream, LevelLow, "didn't found binlog location in dumped metadata file %s", "Please check log of dump unit, there maybe errors when read upstream binlog status") + ErrPreviousGTIDNotExist = New(codePreviousGTIDNotExist, ClassFunctional, ScopeInternal, LevelHigh, "no previous gtid event from binlog %s", "") + // Config related error ErrConfigCheckItemNotSupport = New(codeConfigCheckItemNotSupport, ClassConfig, ScopeInternal, LevelMedium, "checking item %s is not supported\n%s", "Please check `ignore-checking-items` config in task configuration file, which can be set including `all`/`dump_privilege`/`replication_privilege`/`version`/`binlog_enable`/`binlog_format`/`binlog_row_image`/`table_schema`/`schema_of_shard_tables`/`auto_increment_ID`.") ErrConfigTomlTransform = New(codeConfigTomlTransform, ClassConfig, ScopeInternal, LevelMedium, "%s", "Please check the configuration file has correct TOML format.") diff --git a/relay/relay.go b/relay/relay.go index a45ed0acdb..12b1fd19cf 100755 --- a/relay/relay.go +++ b/relay/relay.go @@ -37,6 +37,7 @@ import ( "github.com/pingcap/dm/dm/unit" "github.com/pingcap/dm/pkg/binlog" "github.com/pingcap/dm/pkg/binlog/common" + binlogReader "github.com/pingcap/dm/pkg/binlog/reader" "github.com/pingcap/dm/pkg/conn" fr "github.com/pingcap/dm/pkg/func-rollback" "github.com/pingcap/dm/pkg/gtid" @@ -565,17 +566,27 @@ func (r *Relay) reSetupMeta(ctx context.Context) error { } // try adjust meta with start pos from config - adjusted, err := r.meta.AdjustWithStartPos(r.cfg.BinLogName, r.cfg.BinlogGTID, r.cfg.EnableGTID, latestPosName, latestGTIDStr) + _, err = r.meta.AdjustWithStartPos(r.cfg.BinLogName, r.cfg.BinlogGTID, r.cfg.EnableGTID, latestPosName, latestGTIDStr) if err != nil { return err } - if adjusted { - _, pos := r.meta.Pos() - _, gtid := r.meta.GTID() - r.logger.Info("adjusted meta to start pos", zap.Reflect("start pos", pos), zap.Stringer("start pos's binlog gtid", gtid)) + _, pos := r.meta.Pos() + _, gtid := r.meta.GTID() + if r.cfg.EnableGTID { + // Adjust given gtid + // This means we always pull the binlog from the beginning of file. + gtid, err = r.adjustGTID(ctx, gtid) + if err != nil { + return terror.Annotate(err, "fail to adjust gtid for relay") + } + err = r.SaveMeta(pos, gtid) + if err != nil { + return err + } } + r.logger.Info("adjusted meta to start pos", zap.Reflect("start pos", pos), zap.Stringer("start pos's binlog gtid", gtid)) r.updateMetricsRelaySubDirIndex() return nil @@ -909,3 +920,17 @@ func (r *Relay) setSyncConfig() error { r.syncerCfg = syncerCfg return nil } + +// AdjustGTID implements Relay.AdjustGTID +func (r *Relay) adjustGTID(ctx context.Context, gset gtid.Set) (gtid.Set, error) { + // setup a TCP binlog reader (because no relay can be used when upgrading). + syncCfg := r.syncerCfg + randomServerID, err := utils.ReuseServerID(ctx, r.cfg.ServerID, r.db) + if err != nil { + return nil, terror.Annotate(err, "fail to get random server id when relay adjust gtid") + } + syncCfg.ServerID = randomServerID + + tcpReader := binlogReader.NewTCPReader(syncCfg) + return binlogReader.GetPreviousGTIDFromGTIDSet(ctx, tcpReader, gset) +} diff --git a/relay/relay_test.go b/relay/relay_test.go index 4aa5517b98..3816e66f0d 100644 --- a/relay/relay_test.go +++ b/relay/relay_test.go @@ -511,25 +511,31 @@ func (t *testRelaySuite) TestReSetupMeta(c *C) { r.cfg.EnableGTID = true r.cfg.BinlogGTID = "24ecd093-8cec-11e9-aa0d-0242ac170002:1-23" r.cfg.BinLogName = "mysql-bin.000005" + + c.Assert(r.setSyncConfig(), IsNil) + // all adjusted gset should be empty since we didn't flush logs + emptyGTID, err := gtid.ParserGTID(r.cfg.Flavor, "") + c.Assert(err, IsNil) + c.Assert(r.reSetupMeta(ctx), IsNil) uuid001 := fmt.Sprintf("%s.000001", uuid) - t.verifyMetadata(c, r, uuid001, gmysql.Position{Name: r.cfg.BinLogName, Pos: 4}, r.cfg.BinlogGTID, []string{uuid001}) + t.verifyMetadata(c, r, uuid001, gmysql.Position{Name: r.cfg.BinLogName, Pos: 4}, emptyGTID.String(), []string{uuid001}) // re-setup meta again, often happen when connecting a server behind a VIP. c.Assert(r.reSetupMeta(ctx), IsNil) uuid002 := fmt.Sprintf("%s.000002", uuid) - t.verifyMetadata(c, r, uuid002, minCheckpoint, r.cfg.BinlogGTID, []string{uuid001, uuid002}) + t.verifyMetadata(c, r, uuid002, minCheckpoint, emptyGTID.String(), []string{uuid001, uuid002}) r.cfg.BinLogName = "mysql-bin.000002" r.cfg.BinlogGTID = "24ecd093-8cec-11e9-aa0d-0242ac170002:1-50,24ecd093-8cec-11e9-aa0d-0242ac170003:1-50" r.cfg.UUIDSuffix = 2 c.Assert(r.reSetupMeta(ctx), IsNil) - t.verifyMetadata(c, r, uuid002, gmysql.Position{Name: r.cfg.BinLogName, Pos: 4}, r.cfg.BinlogGTID, []string{uuid002}) + t.verifyMetadata(c, r, uuid002, gmysql.Position{Name: r.cfg.BinLogName, Pos: 4}, emptyGTID.String(), []string{uuid002}) // re-setup meta again, often happen when connecting a server behind a VIP. c.Assert(r.reSetupMeta(ctx), IsNil) uuid003 := fmt.Sprintf("%s.000003", uuid) - t.verifyMetadata(c, r, uuid003, minCheckpoint, r.cfg.BinlogGTID, []string{uuid002, uuid003}) + t.verifyMetadata(c, r, uuid003, minCheckpoint, emptyGTID.String(), []string{uuid002, uuid003}) } func (t *testRelaySuite) verifyMetadata(c *C, r *Relay, uuidExpected string, diff --git a/relay/transformer/transformer.go b/relay/transformer/transformer.go index 6682883212..f164ce823f 100644 --- a/relay/transformer/transformer.go +++ b/relay/transformer/transformer.go @@ -69,6 +69,8 @@ func (t *transformer) Transform(e *replication.BinlogEvent) Result { switch ev := e.Event.(type) { case *replication.PreviousGTIDsEvent: result.CanSaveGTID = true + case *replication.MariadbGTIDListEvent: + result.CanSaveGTID = true case *replication.RotateEvent: result.LogPos = uint32(ev.Position) // next event's position result.NextLogName = string(ev.NextLogName) // for RotateEvent, update binlog name