diff --git a/binlog_streamer.go b/binlog_streamer.go index d762370d..0dea42ab 100644 --- a/binlog_streamer.go +++ b/binlog_streamer.go @@ -6,6 +6,7 @@ import ( sqlorig "database/sql" "errors" "fmt" + "strings" "time" sql "github.com/Shopify/ghostferry/sqlwrapper" @@ -45,10 +46,24 @@ type BinlogStreamer struct { DatabaseRewrites map[string]string TableRewrites map[string]string + // BinlogCoordinateMode selects whether this streamer tracks file/position + // or GTID coordinates. An empty value is treated as file/position for + // backwards compatibility. + BinlogCoordinateMode BinlogCoordinateType + lastStreamedBinlogPosition mysql.Position lastResumableBinlogPosition mysql.Position stopAtBinlogPosition mysql.Position + // GTID tracking, only maintained when BinlogCoordinateMode is + // BinlogCoordinateGTID. lastStreamedGTIDSet is the committed GTID set seen + // so far. lastResumableGTIDSet is the committed GTID set at the last + // transaction boundary (a safe resume point). stopAtGTIDSet is the target + // executed set to stop at during cutover. + lastStreamedGTIDSet mysql.GTIDSet + lastResumableGTIDSet mysql.GTIDSet + stopAtGTIDSet mysql.GTIDSet + lastProcessedEventTime time.Time lastLagMetricEmittedTime time.Time @@ -148,6 +163,94 @@ func (s *BinlogStreamer) ConnectBinlogStreamerToMysqlFrom(startFromBinlogPositio return s.lastStreamedBinlogPosition, err } +// coordinateMode returns the effective coordinate mode, treating the empty +// value as file/position for backwards compatibility. +func (s *BinlogStreamer) coordinateMode() BinlogCoordinateType { + if s.BinlogCoordinateMode == "" { + return BinlogCoordinateFilePosition + } + return s.BinlogCoordinateMode +} + +// ConnectBinlogStreamerToMysqlWithCoordinate starts streaming from the current +// server coordinate for the configured BinlogCoordinateMode. It is the +// coordinate-typed counterpart of ConnectBinlogStreamerToMysql. +func (s *BinlogStreamer) ConnectBinlogStreamerToMysqlWithCoordinate() (BinlogCoordinate, error) { + s.ensureLogger() + + switch s.coordinateMode() { + case BinlogCoordinateGTID: + coord, err := ReadCurrentGTIDCoordinate(s.DB) + if err != nil { + s.logger.WithError(err).Error("failed to read current executed GTID set") + return BinlogCoordinate{}, err + } + return s.ConnectBinlogStreamerToMysqlSinceCoordinate(coord) + default: + pos, err := s.ConnectBinlogStreamerToMysql() + if err != nil { + return BinlogCoordinate{}, err + } + return NewFilePositionCoordinate(pos), nil + } +} + +// ConnectBinlogStreamerToMysqlSinceCoordinate starts streaming since the given +// coordinate. The coordinate type must match the streamer's configured +// BinlogCoordinateMode. +func (s *BinlogStreamer) ConnectBinlogStreamerToMysqlSinceCoordinate(startFrom BinlogCoordinate) (BinlogCoordinate, error) { + s.ensureLogger() + + switch s.coordinateMode() { + case BinlogCoordinateGTID: + if !startFrom.IsGTID() { + return BinlogCoordinate{}, fmt.Errorf("binlog streamer in GTID mode requires a GTID coordinate, got %q", startFrom.Type) + } + return s.connectBinlogStreamerFromGTID(startFrom) + default: + if !startFrom.IsFilePosition() { + return BinlogCoordinate{}, fmt.Errorf("binlog streamer in file/position mode requires a file/position coordinate, got %q", startFrom.Type) + } + pos, err := s.ConnectBinlogStreamerToMysqlFrom(startFrom.Position()) + if err != nil { + return BinlogCoordinate{}, err + } + return NewFilePositionCoordinate(pos), nil + } +} + +func (s *BinlogStreamer) connectBinlogStreamerFromGTID(startFrom BinlogCoordinate) (BinlogCoordinate, error) { + err := s.createBinlogSyncer() + if err != nil { + return BinlogCoordinate{}, err + } + + gtidSet, err := startFrom.ParsedGTIDSet() + if err != nil { + s.logger.WithError(err).Error("failed to parse starting GTID set") + return BinlogCoordinate{}, err + } + + // Seed both streamed and resumable GTID sets to the starting set. Clone so + // later mutations from event tracking never alias the starting value. + s.lastStreamedGTIDSet = gtidSet.Clone() + s.lastResumableGTIDSet = gtidSet.Clone() + + s.logger.WithFields(Fields{ + "gtid_set": gtidSet.String(), + "host": s.DBConfig.Host, + "port": s.DBConfig.Port, + }).Info("starting binlog streaming from GTID set") + + s.binlogStreamer, err = s.binlogSyncer.StartSyncGTID(gtidSet) + if err != nil { + s.logger.WithError(err).Error("unable to start binlog streamer from GTID set") + return BinlogCoordinate{}, err + } + + return NewGTIDCoordinate(s.lastStreamedGTIDSet.String()), nil +} + // the default event handler is called for replication binLogEvents that do not have a // separate event Handler registered. @@ -200,6 +303,29 @@ func (s *BinlogStreamer) defaultEventHandler(ev *replication.BinlogEvent, query s.logger.WithError(err).Error("failed to handle rows event") s.ErrorHandler.Fatal("binlog_streamer", err) } + case *replication.QueryEvent: + // DDL and administrative statements (CREATE TABLE, GRANT, etc.) commit + // via a QueryEvent rather than an XIDEvent, so their GTID is only + // visible here. Without this, the streamed GTID set would never + // advance past such a statement and a cutover whose stop target + // includes it would hang forever. go-mysql attaches the current + // executed GTID set (including this statement's GTID) to the event. + // + // Transaction-control statements ("BEGIN") also arrive as QueryEvents + // but do NOT commit anything; their GTID is captured at the closing + // XIDEvent instead, so we must skip them here to avoid advancing the + // streamed set before the transaction's rows have been applied. + if s.coordinateMode() == BinlogCoordinateGTID { + qe := ev.Event.(*replication.QueryEvent) + if qe.GSet != nil && !isTransactionControlQuery(qe.Query) { + // A DDL/admin statement is its own transaction; the pre-statement + // committed set is a safe resume point. + if s.lastStreamedGTIDSet != nil { + s.lastResumableGTIDSet = s.lastStreamedGTIDSet.Clone() + } + s.lastStreamedGTIDSet = qe.GSet.Clone() + } + } case *replication.XIDEvent, *replication.GTIDEvent: // With regards to DMLs, we see (at least) the following sequence // of events in the binlog stream: @@ -227,6 +353,28 @@ func (s *BinlogStreamer) defaultEventHandler(ev *replication.BinlogEvent, query // last transaction es.isEventPositionResumable = true + // GTID tracking. go-mysql maintains the current GTID set internally and + // attaches it to XIDEvent.GSet at commit boundaries, so we do not need + // to reconstruct it from raw GTIDEvent SID/GNO. + if s.coordinateMode() == BinlogCoordinateGTID { + switch tev := ev.Event.(type) { + case *replication.GTIDEvent: + // Start of a transaction. A safe resume point is the committed + // set that existed BEFORE this transaction, so that an + // interruption replays the whole in-flight transaction. + if s.lastStreamedGTIDSet != nil { + s.lastResumableGTIDSet = s.lastStreamedGTIDSet.Clone() + } + case *replication.XIDEvent: + // End of a transaction. GSet is the committed GTID set through + // this transaction. Clone to avoid aliasing go-mysql's mutable + // internal set. + if tev.GSet != nil { + s.lastStreamedGTIDSet = tev.GSet.Clone() + } + } + } + // Here we also reset the query event as we are either at the beginning // or the end of the current/next transaction. As such, the query will be // reset following the next RowsQueryEvent before the corresponding RowsEvent(s) @@ -235,6 +383,36 @@ func (s *BinlogStreamer) defaultEventHandler(ev *replication.BinlogEvent, query return query, err } +// shouldContinueStreaming reports whether the Run loop should keep streaming. +// +// It keeps streaming until a stop has been requested AND the stop coordinate +// has been reached. The "have we reached the stop coordinate?" question is +// answered by BinlogCoordinate.HasReached, so this method is identical for +// file/position and GTID; the representation-specific mechanics live on the +// coordinate type. +func (s *BinlogStreamer) shouldContinueStreaming() bool { + if !s.stopRequested { + return true + } + + // Once stopRequested is set, FlushAndStop has already recorded the stop + // coordinate. We must NOT treat a zero/empty stop coordinate as "not yet + // recorded": on a fresh source the executed GTID set (or binlog position) + // can legitimately be empty, and an empty GTID set is a valid stop target + // that any streamed set already contains. Deriving presence from IsZero() + // here would hang cutover forever in that case. + stop := s.GetStopBinlogCoordinate() + + reached, err := s.GetLastStreamedBinlogCoordinate().HasReached(stop) + if err != nil { + // A mismatch or parse error should not silently stop the stream; log + // and keep going so a spurious error can't truncate replication. + s.logger.WithError(err).Warn("could not evaluate stop coordinate; continuing to stream") + return true + } + return !reached +} + func (s *BinlogStreamer) Run() { s.ensureLogger() @@ -242,6 +420,7 @@ func (s *BinlogStreamer) Run() { s.logger.WithFields(Fields{ "stopAtBinlogPosition": s.stopAtBinlogPosition, "lastStreamedBinlogPosition": s.lastStreamedBinlogPosition, + "coordinateMode": s.coordinateMode(), }).Info("exiting binlog streamer") s.binlogSyncer.Close() }() @@ -252,7 +431,7 @@ func (s *BinlogStreamer) Run() { currentFilename := s.lastStreamedBinlogPosition.Name es.nextFilename = s.lastStreamedBinlogPosition.Name s.logger.Info("starting binlog streamer") - for !s.stopRequested || (s.stopRequested && s.lastStreamedBinlogPosition.Compare(s.stopAtBinlogPosition) < 0) { + for s.shouldContinueStreaming() { currentFilename = es.nextFilename var ev *replication.BinlogEvent var timedOut bool @@ -341,22 +520,75 @@ func (s *BinlogStreamer) GetLastStreamedBinlogPosition() mysql.Position { } // GetLastStreamedBinlogCoordinate is the coordinate-typed counterpart of -// GetLastStreamedBinlogPosition. +// GetLastStreamedBinlogPosition. It returns a coordinate matching the +// streamer's configured BinlogCoordinateMode. func (s *BinlogStreamer) GetLastStreamedBinlogCoordinate() BinlogCoordinate { + if s.coordinateMode() == BinlogCoordinateGTID { + if s.lastStreamedGTIDSet == nil { + return NewGTIDCoordinate("") + } + return NewGTIDCoordinate(s.lastStreamedGTIDSet.String()) + } return NewFilePositionCoordinate(s.lastStreamedBinlogPosition) } +// GetStopBinlogCoordinate returns the recorded stop coordinate matching the +// streamer's configured BinlogCoordinateMode. It is zero until FlushAndStop has +// recorded a stop target. +func (s *BinlogStreamer) GetStopBinlogCoordinate() BinlogCoordinate { + if s.coordinateMode() == BinlogCoordinateGTID { + if s.stopAtGTIDSet == nil { + return NewGTIDCoordinate("") + } + return NewGTIDCoordinate(s.stopAtGTIDSet.String()) + } + return NewFilePositionCoordinate(s.stopAtBinlogPosition) +} + +// isTransactionControlQuery reports whether a QueryEvent query is a +// transaction-control statement that does not itself commit data (BEGIN). +// Such statements must not advance the committed GTID set; the enclosing +// transaction commits at its XIDEvent. Note COMMIT/ROLLBACK are normally +// represented as XIDEvents for InnoDB, but are treated defensively here too. +func isTransactionControlQuery(query []byte) bool { + q := strings.ToUpper(strings.TrimSpace(string(query))) + return q == "BEGIN" || q == "COMMIT" || q == "ROLLBACK" +} + func (s *BinlogStreamer) IsAlmostCaughtUp() bool { return time.Now().Sub(s.lastProcessedEventTime) < caughtUpThreshold } func (s *BinlogStreamer) FlushAndStop() { s.logger.Info("requesting binlog streamer to stop") - // Must first read the binlog position before requesting stop - // Otherwise there is a race condition where the stopRequested is - // set to True but the TargetPosition is nil, which would cause - // the BinlogStreamer to immediately exit, as it thinks that it has - // passed the initial target position. + // Must first read the stop coordinate before requesting stop. + // Otherwise there is a race condition where stopRequested is set to true + // but the stop coordinate is still nil/zero, which would cause the + // BinlogStreamer to immediately exit, as it thinks that it has already + // passed the stop coordinate. + if s.coordinateMode() == BinlogCoordinateGTID { + err := WithRetries(100, 600*time.Millisecond, s.logger, "read current executed GTID set", func() error { + gtidSet, err := ReadExecutedGTIDSet(s.DB) + if err != nil { + return err + } + parsed, err := mysql.ParseMysqlGTIDSet(gtidSet) + if err != nil { + return err + } + s.stopAtGTIDSet = parsed + return nil + }) + + if err != nil { + s.ErrorHandler.Fatal("binlog_streamer", err) + } + s.logger.WithField("stop_at_gtid_set", s.stopAtGTIDSet.String()).Info("current stop GTID set was recorded") + + s.stopRequested = true + return + } + err := WithRetries(100, 600*time.Millisecond, s.logger, "read current binlog position", func() error { var err error s.stopAtBinlogPosition, err = ShowMasterStatusBinlogPosition(s.DB) diff --git a/binlog_streamer_gtid_test.go b/binlog_streamer_gtid_test.go new file mode 100644 index 00000000..178c59e5 --- /dev/null +++ b/binlog_streamer_gtid_test.go @@ -0,0 +1,173 @@ +package ghostferry + +import ( + "testing" + + "github.com/go-mysql-org/go-mysql/mysql" + "github.com/go-mysql-org/go-mysql/replication" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + gtidSetLower = "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-57" + gtidSetTarget = "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-100" + gtidSetPast = "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-150" +) + +func mustParseGTID(t *testing.T, s string) mysql.GTIDSet { + t.Helper() + set, err := mysql.ParseMysqlGTIDSet(s) + require.NoError(t, err) + return set +} + +func TestIsTransactionControlQuery(t *testing.T) { + assert.True(t, isTransactionControlQuery([]byte("BEGIN"))) + assert.True(t, isTransactionControlQuery([]byte(" begin "))) + assert.True(t, isTransactionControlQuery([]byte("COMMIT"))) + assert.True(t, isTransactionControlQuery([]byte("ROLLBACK"))) + assert.False(t, isTransactionControlQuery([]byte("CREATE TABLE t (id int)"))) + assert.False(t, isTransactionControlQuery([]byte("GRANT ALL ON *.* TO 'x'@'%'"))) +} + +// TestQueryEventAdvancesStreamedGTID verifies that a DDL/admin QueryEvent +// (which commits without an XIDEvent) advances the streamed GTID set, while a +// BEGIN QueryEvent does not. Without this, a cutover whose stop target includes +// a trailing DDL would hang forever. +func TestQueryEventAdvancesStreamedGTID(t *testing.T) { + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + s.logger = LogWithField("tag", "test") + + ddlSet := mustParseGTID(t, gtidSetTarget) + + // A DDL QueryEvent carrying the executed set advances lastStreamedGTIDSet. + ddlEvent := &replication.BinlogEvent{ + Header: &replication.EventHeader{LogPos: 100}, + Event: &replication.QueryEvent{Query: []byte("CREATE TABLE t (id int)"), GSet: ddlSet}, + } + es := &BinlogEventState{} + _, err := s.defaultEventHandler(ddlEvent, nil, es) + require.NoError(t, err) + require.NotNil(t, s.lastStreamedGTIDSet) + assert.Equal(t, gtidSetTarget, s.lastStreamedGTIDSet.String()) + + // A BEGIN QueryEvent must NOT advance the streamed set. + before := s.lastStreamedGTIDSet.String() + beginEvent := &replication.BinlogEvent{ + Header: &replication.EventHeader{LogPos: 200}, + Event: &replication.QueryEvent{Query: []byte("BEGIN"), GSet: mustParseGTID(t, gtidSetPast)}, + } + _, err = s.defaultEventHandler(beginEvent, nil, es) + require.NoError(t, err) + assert.Equal(t, before, s.lastStreamedGTIDSet.String(), "BEGIN must not advance the streamed GTID set") +} + +func TestCoordinateModeDefaultsToFilePosition(t *testing.T) { + s := &BinlogStreamer{} + assert.Equal(t, BinlogCoordinateFilePosition, s.coordinateMode()) + + s.BinlogCoordinateMode = BinlogCoordinateGTID + assert.Equal(t, BinlogCoordinateGTID, s.coordinateMode()) +} + +func TestShouldContinueStreaming_FilePosition(t *testing.T) { + s := &BinlogStreamer{} + + // No stop requested: always continue. + assert.True(t, s.shouldContinueStreaming()) + + s.stopRequested = true + s.stopAtBinlogPosition = mysql.Position{Name: "mysql-bin.000010", Pos: 100} + + // Streamed position behind stop: continue. + s.lastStreamedBinlogPosition = mysql.Position{Name: "mysql-bin.000010", Pos: 50} + assert.True(t, s.shouldContinueStreaming()) + + // Streamed position reached stop: stop. + s.lastStreamedBinlogPosition = mysql.Position{Name: "mysql-bin.000010", Pos: 100} + assert.False(t, s.shouldContinueStreaming()) +} + +func TestShouldContinueStreaming_GTID(t *testing.T) { + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + + // No stop requested: always continue. + assert.True(t, s.shouldContinueStreaming()) + + s.stopRequested = true + s.stopAtGTIDSet = mustParseGTID(t, gtidSetTarget) + + // No streamed set yet: keep going. + assert.True(t, s.shouldContinueStreaming()) + + // Streamed set does not yet contain target: continue. + s.lastStreamedGTIDSet = mustParseGTID(t, gtidSetLower) + assert.True(t, s.shouldContinueStreaming()) + + // Streamed set exactly reaches target: stop. + s.lastStreamedGTIDSet = mustParseGTID(t, gtidSetTarget) + assert.False(t, s.shouldContinueStreaming()) + + // Streamed set past target: stop. + s.lastStreamedGTIDSet = mustParseGTID(t, gtidSetPast) + assert.False(t, s.shouldContinueStreaming()) +} + +// TestShouldContinueStreaming_GTIDEmptyStopTarget guards the fresh-source +// case: an empty executed GTID set is a valid stop target (not "unset"), and +// any streamed set — including an empty one — has already reached it, so the +// stream must stop rather than hang. +func TestShouldContinueStreaming_GTIDEmptyStopTarget(t *testing.T) { + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + s.stopRequested = true + s.stopAtGTIDSet = mustParseGTID(t, "") // empty executed set on a fresh source + + // Empty streamed set has reached the empty stop target: stop. + assert.False(t, s.shouldContinueStreaming()) + + // A non-empty streamed set also trivially contains the empty target: stop. + s.lastStreamedGTIDSet = mustParseGTID(t, gtidSetTarget) + assert.False(t, s.shouldContinueStreaming()) +} + +func TestGetLastStreamedBinlogCoordinate_GTIDMode(t *testing.T) { + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + + // Nil streamed set yields an empty (zero) GTID coordinate. + coord := s.GetLastStreamedBinlogCoordinate() + assert.True(t, coord.IsGTID()) + assert.True(t, coord.IsZero()) + + s.lastStreamedGTIDSet = mustParseGTID(t, gtidSetTarget) + coord = s.GetLastStreamedBinlogCoordinate() + assert.True(t, coord.IsGTID()) + assert.Equal(t, gtidSetTarget, coord.GTIDSet) +} + +func TestGetLastStreamedBinlogCoordinate_FilePositionMode(t *testing.T) { + s := &BinlogStreamer{} + s.lastStreamedBinlogPosition = mysql.Position{Name: "mysql-bin.000010", Pos: 100} + + coord := s.GetLastStreamedBinlogCoordinate() + assert.True(t, coord.IsFilePosition()) + assert.Equal(t, "mysql-bin.000010", coord.Position().Name) + assert.Equal(t, uint32(100), coord.Position().Pos) +} + +func TestConnectBinlogStreamerSinceCoordinate_TypeMismatch(t *testing.T) { + // GTID mode with a file/position coordinate must be rejected before any DB + // interaction. + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + _, err := s.ConnectBinlogStreamerToMysqlSinceCoordinate( + NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 4}), + ) + require.Error(t, err) + assert.Contains(t, err.Error(), "GTID mode requires a GTID coordinate") + + // File/position mode with a GTID coordinate must also be rejected. + s2 := &BinlogStreamer{} + _, err = s2.ConnectBinlogStreamerToMysqlSinceCoordinate(NewGTIDCoordinate(gtidSetTarget)) + require.Error(t, err) + assert.Contains(t, err.Error(), "file/position mode requires a file/position coordinate") +} diff --git a/ferry.go b/ferry.go index 9666576c..8087a91a 100644 --- a/ferry.go +++ b/ferry.go @@ -147,15 +147,16 @@ func (f *Ferry) newBinlogStreamer(db *sql.DB, dbConf *DatabaseConfig, schemaRewr f.ensureInitialized() return &BinlogStreamer{ - DB: db, - DBConfig: dbConf, - MyServerId: f.Config.MyServerId, - ErrorHandler: f.ErrorHandler, - Filter: f.CopyFilter, - TableSchema: f.Tables, - LogTag: logTag, - DatabaseRewrites: schemaRewrites, - TableRewrites: tableRewrites, + DB: db, + DBConfig: dbConf, + MyServerId: f.Config.MyServerId, + ErrorHandler: f.ErrorHandler, + Filter: f.CopyFilter, + TableSchema: f.Tables, + LogTag: logTag, + DatabaseRewrites: schemaRewrites, + TableRewrites: tableRewrites, + BinlogCoordinateMode: f.Config.BinlogCoordinateMode, } } @@ -378,6 +379,13 @@ func (f *Ferry) Initialize() (err error) { return err } + if f.Config.BinlogCoordinateMode == BinlogCoordinateGTID { + if err = CheckServerGTIDModeEnabled(f.SourceDB); err != nil { + f.logger.WithError(err).Error("source db is not configured for GTID mode") + return err + } + } + f.TargetDB, err = f.Target.SqlDB(f.logger.WithField("dbname", "target")) if err != nil { f.logger.WithError(err).Error("failed to connect to target database") @@ -390,6 +398,15 @@ func (f *Ferry) Initialize() (err error) { return err } + // The target binlog streamer (used by the target verifier) streams in the + // same coordinate mode, so the target must also have GTID mode enabled. + if f.Config.BinlogCoordinateMode == BinlogCoordinateGTID && !f.Config.SkipTargetVerification { + if err = CheckServerGTIDModeEnabled(f.TargetDB); err != nil { + f.logger.WithError(err).Error("target db is not configured for GTID mode") + return err + } + } + if !f.Config.AllowSuperUserOnReadOnly { isReplica, err := CheckDbIsAReplica(f.TargetDB) if err != nil { @@ -588,14 +605,21 @@ func (f *Ferry) Start() error { // miss some records that are inserted between the time the // DataIterator determines the range of IDs to copy and the time that // the starting binlog coordinates are determined. - var sourcePos siddontangmysql.Position - var targetPos siddontangmysql.Position + // Resume-from-state currently only supports file/position coordinates. + // GTID-based resume is introduced in a later stage once GTID coordinates are + // persisted in the serialized state. + if f.StateToResumeFrom != nil && f.Config.BinlogCoordinateMode == BinlogCoordinateGTID { + return fmt.Errorf("resuming from state is not yet supported in GTID binlog coordinate mode") + } + + var sourceCoord BinlogCoordinate + var targetCoord BinlogCoordinate var err error if f.StateToResumeFrom != nil { - sourcePos, err = f.BinlogStreamer.ConnectBinlogStreamerToMysqlFrom(f.StateToResumeFrom.MinSourceBinlogPosition()) + sourceCoord, err = f.BinlogStreamer.ConnectBinlogStreamerToMysqlSinceCoordinate(f.StateToResumeFrom.MinSourceBinlogCoordinate()) } else { - sourcePos, err = f.BinlogStreamer.ConnectBinlogStreamerToMysql() + sourceCoord, err = f.BinlogStreamer.ConnectBinlogStreamerToMysqlWithCoordinate() } if err != nil { return err @@ -603,9 +627,11 @@ func (f *Ferry) Start() error { if !f.Config.SkipTargetVerification { if f.StateToResumeFrom != nil && f.StateToResumeFrom.LastStoredBinlogPositionForTargetVerifier != zeroPosition { - targetPos, err = f.TargetVerifier.BinlogStreamer.ConnectBinlogStreamerToMysqlFrom(f.StateToResumeFrom.LastStoredBinlogPositionForTargetVerifier) + targetCoord, err = f.TargetVerifier.BinlogStreamer.ConnectBinlogStreamerToMysqlSinceCoordinate( + NewFilePositionCoordinate(f.StateToResumeFrom.LastStoredBinlogPositionForTargetVerifier), + ) } else { - targetPos, err = f.TargetVerifier.BinlogStreamer.ConnectBinlogStreamerToMysql() + targetCoord, err = f.TargetVerifier.BinlogStreamer.ConnectBinlogStreamerToMysqlWithCoordinate() } } if err != nil { @@ -615,13 +641,13 @@ func (f *Ferry) Start() error { // If we don't set this now, there is a race condition where Ghostferry // is terminated with some rows copied but no binlog events are written. // This guarentees that we are able to restart from a valid location. - f.StateTracker.UpdateLastResumableSourceBinlogPosition(sourcePos) + f.StateTracker.UpdateLastResumableSourceBinlogCoordinate(sourceCoord) if f.inlineVerifier != nil { - f.StateTracker.UpdateLastResumableSourceBinlogPositionForInlineVerifier(sourcePos) + f.StateTracker.UpdateLastResumableSourceBinlogCoordinateForInlineVerifier(sourceCoord) } if !f.Config.SkipTargetVerification { - f.StateTracker.UpdateLastResumableBinlogPositionForTargetVerifier(targetPos) + f.StateTracker.UpdateLastResumableBinlogCoordinateForTargetVerifier(targetCoord) } return nil