diff --git a/CHANGELOG.md b/CHANGELOG.md index 1943fe2f..a05bf0a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ All notable changes to this project will be documented in this file. -## [Unreleased] +## [1.4.0 - unreleased] ### Added diff --git a/Makefile b/Makefile index 4098bf8a..8ef4be3e 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ # Variables to be built into the binary -VERSION := 1.3.1 +VERSION := 1.4.0 # This variable can be overwritten by the caller DATETIME ?= $(shell date -u +%Y%m%d%H%M%S) diff --git a/config.go b/config.go index ed75858f..4cf51d0e 100644 --- a/config.go +++ b/config.go @@ -692,9 +692,19 @@ type Config struct { // Make sure you have binlog_row_image=FULL when turning on this SkipBinlogRowImageCheck bool - // BinlogCoordinateMode is "file_position" (default) or "gtid" (experimental; - // requires gtid_mode=ON on the source, and on the target when target - // verification is enabled). Resume from state is not yet supported in gtid mode. + // BinlogCoordinateMode selects how binlog coordinates are tracked. + // + // Valid values: + // - "" or "file_position": the classic (file, position) coordinate. This + // is the default. + // - "gtid": GTID-set based coordinates (experimental, MySQL only). + // Requires @@GLOBAL.gtid_mode=ON on the source, on the target when + // target verification is enabled, and on the replication master when + // running from a replica. + // + // In GTID mode, LastSuccessfulBinlogPos still reports the streamed + // file/position, while FinalBinlogPos remains unset; use + // LastSuccessfulBinlogCoordinate and FinalBinlogCoordinate instead. BinlogCoordinateMode BinlogCoordinateType // This config is necessary for inline verification for a special case of diff --git a/copydb/config.go b/copydb/config.go index e3742a3e..4aa89f5f 100644 --- a/copydb/config.go +++ b/copydb/config.go @@ -70,6 +70,13 @@ type Config struct { // SELECT file, position FROM meta.ptheartbeat WHERE server_id = master_server_id ReplicatedMasterPositionQuery string + // This is the SQL query used to read the master's executed GTID set that + // has been replicated to the Source. Only used in GTID mode; must return + // a single column with the GTID set string, e.g. + // + // SELECT gtid_executed FROM meta.ptheartbeat WHERE server_id = master_server_id + ReplicatedMasterGTIDQuery string + // The duration to wait for the replication to catchup before aborting. Only use if RunFerryFromReplica is true. WaitForReplicationTimeout string diff --git a/copydb/copydb.go b/copydb/copydb.go index b858de06..491ea3a2 100644 --- a/copydb/copydb.go +++ b/copydb/copydb.go @@ -114,8 +114,6 @@ func (this *CopydbFerry) initializeWaitUntilReplicaIsCaughtUpToMasterConnection( return err } - positionFetcher := ghostferry.ReplicatedMasterPositionViaCustomQuery{Query: this.config.ReplicatedMasterPositionQuery} - var timeout time.Duration if this.config.WaitForReplicationTimeout == "" { timeout = time.Duration(0) @@ -126,11 +124,29 @@ func (this *CopydbFerry) initializeWaitUntilReplicaIsCaughtUpToMasterConnection( } } - this.Ferry.WaitUntilReplicaIsCaughtUpToMaster = &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ - MasterDB: masterDB, - Timeout: timeout, - ReplicatedMasterPositionFetcher: positionFetcher, + wait := &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ + MasterDB: masterDB, + Timeout: timeout, + BinlogCoordinateMode: this.config.BinlogCoordinateMode, } + + if this.config.BinlogCoordinateMode == ghostferry.BinlogCoordinateGTID { + if this.config.ReplicatedMasterGTIDQuery == "" { + return fmt.Errorf("ReplicatedMasterGTIDQuery must be set when running from a replica in GTID mode") + } + wait.ReplicatedMasterCoordinateFetcher = ghostferry.ReplicatedMasterGTIDViaCustomQuery{ + Query: this.config.ReplicatedMasterGTIDQuery, + } + } else { + if this.config.ReplicatedMasterPositionQuery == "" { + return fmt.Errorf("ReplicatedMasterPositionQuery must be set when running from a replica") + } + wait.ReplicatedMasterPositionFetcher = ghostferry.ReplicatedMasterPositionViaCustomQuery{ + Query: this.config.ReplicatedMasterPositionQuery, + } + } + + this.Ferry.WaitUntilReplicaIsCaughtUpToMaster = wait return nil } diff --git a/ferry.go b/ferry.go index 1f8ca534..1074bcc2 100644 --- a/ferry.go +++ b/ferry.go @@ -422,6 +422,7 @@ func (f *Ferry) Initialize() (err error) { // of the MySQL databases. if f.WaitUntilReplicaIsCaughtUpToMaster != nil { f.WaitUntilReplicaIsCaughtUpToMaster.ReplicaDB = f.SourceDB + f.WaitUntilReplicaIsCaughtUpToMaster.BinlogCoordinateMode = f.Config.BinlogCoordinateMode if f.WaitUntilReplicaIsCaughtUpToMaster.MasterDB == nil { err = errors.New("must specify a MasterDB") @@ -435,8 +436,22 @@ func (f *Ferry) Initialize() (err error) { return err } - // Ensures the query to check for position is executable. - _, err = f.WaitUntilReplicaIsCaughtUpToMaster.IsCaughtUp(zeroPosition, 1) + // In GTID mode the master's executed GTID set is the cutover target, so the master must have GTID mode enabled. + if f.Config.BinlogCoordinateMode == BinlogCoordinateGTID { + if err = CheckServerGTIDModeEnabled(f.WaitUntilReplicaIsCaughtUpToMaster.MasterDB); err != nil { + f.logger.WithError(err).Error("source master is not configured for GTID mode") + return err + } + } + + // Ensures the query to check for the replicated coordinate is executable. + var probeTarget BinlogCoordinate + if f.Config.BinlogCoordinateMode == BinlogCoordinateGTID { + probeTarget = NewGTIDCoordinate("") + } else { + probeTarget = NewFilePositionCoordinate(zeroPosition) + } + _, err = f.WaitUntilReplicaIsCaughtUpToMaster.IsCaughtUpToCoordinate(probeTarget, 1) if err != nil { f.logger.WithError(err).Error("cannot check replicated master position on the source database") return err diff --git a/sharding/config.go b/sharding/config.go index a3e6ab74..b8b7d932 100644 --- a/sharding/config.go +++ b/sharding/config.go @@ -17,7 +17,9 @@ type Config struct { SourceReplicationMaster *ghostferry.DatabaseConfig ReplicatedMasterPositionQuery string - RunFerryFromReplica bool + // Used instead of ReplicatedMasterPositionQuery in GTID mode; must return a single column with the master's executed GTID set. + ReplicatedMasterGTIDQuery string + RunFerryFromReplica bool StatsDAddress string diff --git a/sharding/sharding.go b/sharding/sharding.go index 4a08efbc..a21a5237 100644 --- a/sharding/sharding.go +++ b/sharding/sharding.go @@ -264,12 +264,25 @@ func (r *ShardingFerry) initializeWaitUntilReplicaIsCaughtUpToMasterConnection() return err } - positionFetcher := ghostferry.ReplicatedMasterPositionViaCustomQuery{Query: r.config.ReplicatedMasterPositionQuery} + wait := &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ + MasterDB: masterDB, + BinlogCoordinateMode: r.config.BinlogCoordinateMode, + } - r.Ferry.WaitUntilReplicaIsCaughtUpToMaster = &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ - MasterDB: masterDB, - ReplicatedMasterPositionFetcher: positionFetcher, + if r.config.BinlogCoordinateMode == ghostferry.BinlogCoordinateGTID { + if r.config.ReplicatedMasterGTIDQuery == "" { + return fmt.Errorf("ReplicatedMasterGTIDQuery must be set when running from a replica in GTID mode") + } + wait.ReplicatedMasterCoordinateFetcher = ghostferry.ReplicatedMasterGTIDViaCustomQuery{ + Query: r.config.ReplicatedMasterGTIDQuery, + } + } else { + wait.ReplicatedMasterPositionFetcher = ghostferry.ReplicatedMasterPositionViaCustomQuery{ + Query: r.config.ReplicatedMasterPositionQuery, + } } + + r.Ferry.WaitUntilReplicaIsCaughtUpToMaster = wait return nil } diff --git a/test/go/race_conditions_integration_test.go b/test/go/race_conditions_integration_test.go index 15ec599c..3736eaff 100644 --- a/test/go/race_conditions_integration_test.go +++ b/test/go/race_conditions_integration_test.go @@ -2,6 +2,7 @@ package test import ( "fmt" + "os" "testing" "time" @@ -84,6 +85,10 @@ func TestUpdateBinlogSelectCopy(t *testing.T) { } func TestMasterChangingBeforeStoppingBinlogStreaming(t *testing.T) { + if os.Getenv("GHOSTFERRY_BINLOG_COORDINATE_MODE") == string(ghostferry.BinlogCoordinateGTID) { + t.Skip("master-demotion detection is mode-independent; covered by the file/position run") + } + ferry := testhelpers.NewTestFerry() sourceDB, err := ferry.Source.SqlDB(nil) diff --git a/test/go/replication_config_test.go b/test/go/replication_config_test.go index 088d0e2e..cbceb14f 100644 --- a/test/go/replication_config_test.go +++ b/test/go/replication_config_test.go @@ -2,10 +2,12 @@ package test import ( "fmt" - sql "github.com/Shopify/ghostferry/sqlwrapper" "math" + "os" "testing" + sql "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/Shopify/ghostferry" "github.com/Shopify/ghostferry/testhelpers" "github.com/stretchr/testify/suite" @@ -16,6 +18,8 @@ type ReplicationConfigTestSuite struct { SourceDB *sql.DB ReplicatedMasterPositionFetcher *ghostferry.ReplicatedMasterPositionViaCustomQuery + ReplicatedMasterGTIDFetcher *ghostferry.ReplicatedMasterGTIDViaCustomQuery + gtidMode bool } func (t *ReplicationConfigTestSuite) SetupTest() { @@ -25,14 +29,27 @@ func (t *ReplicationConfigTestSuite) SetupTest() { t.SourceDB, err = t.Ferry.Source.SqlDB(nil) t.Require().Nil(err) - t.ReplicatedMasterPositionFetcher = &ghostferry.ReplicatedMasterPositionViaCustomQuery{ - Query: "SELECT 'mysql-bin.000003', 483685", + t.gtidMode = os.Getenv("GHOSTFERRY_BINLOG_COORDINATE_MODE") == string(ghostferry.BinlogCoordinateGTID) + + wait := &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ + MasterDB: t.SourceDB, + BinlogCoordinateMode: t.Ferry.Config.BinlogCoordinateMode, } - t.Ferry.WaitUntilReplicaIsCaughtUpToMaster = &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ - MasterDB: t.SourceDB, - ReplicatedMasterPositionFetcher: t.ReplicatedMasterPositionFetcher, + if t.gtidMode { + // A high, unreachable GTID set so the replica is never "caught up". + t.ReplicatedMasterGTIDFetcher = &ghostferry.ReplicatedMasterGTIDViaCustomQuery{ + Query: "SELECT '3e11fa47-71ca-11e1-9e33-c80aa9429562:1-1'", + } + wait.ReplicatedMasterCoordinateFetcher = t.ReplicatedMasterGTIDFetcher + } else { + t.ReplicatedMasterPositionFetcher = &ghostferry.ReplicatedMasterPositionViaCustomQuery{ + Query: "SELECT 'mysql-bin.000003', 483685", + } + wait.ReplicatedMasterPositionFetcher = t.ReplicatedMasterPositionFetcher } + + t.Ferry.WaitUntilReplicaIsCaughtUpToMaster = wait } func (t *ReplicationConfigTestSuite) TearDownTest() { @@ -57,6 +74,16 @@ func (t *ReplicationConfigTestSuite) TestErrorsIfItsRunFromAReplicaWithoutSettin } func (t *ReplicationConfigTestSuite) TestErrorsIfPositionFetcherQueryIsNotProvided() { + if t.gtidMode { + // The GTID fetcher scans a single column, so a two-column query is the error condition. + t.ReplicatedMasterGTIDFetcher.Query = "SELECT 1, 2" + + err := t.Ferry.Initialize() + t.Require().NotNil(err) + t.Require().Equal("sql: expected 2 destination arguments in Scan, not 1", err.Error()) + return + } + t.ReplicatedMasterPositionFetcher.Query = "SELECT 1" err := t.Ferry.Initialize() @@ -73,6 +100,15 @@ func (t *ReplicationConfigTestSuite) TestErrorsIfProvidedMasterIsReadOnly() { } func (t *ReplicationConfigTestSuite) TestCanInitializeFerryWithValidConfig() { + if t.gtidMode { + // Initialize only probes that the query is executable, so a valid GTID set must succeed. + t.ReplicatedMasterGTIDFetcher.Query = "SELECT '3e11fa47-71ca-11e1-9e33-c80aa9429562:1-999999999'" + + err := t.Ferry.Initialize() + t.Require().Nil(err) + return + } + t.ReplicatedMasterPositionFetcher.Query = fmt.Sprintf("SELECT 'mysql-bin.999999',%d", math.MaxUint32) err := t.Ferry.Initialize() diff --git a/test/go/wait_until_replica_is_caught_up_to_master_test.go b/test/go/wait_until_replica_is_caught_up_to_master_test.go index 9770e85c..c1e8c314 100644 --- a/test/go/wait_until_replica_is_caught_up_to_master_test.go +++ b/test/go/wait_until_replica_is_caught_up_to_master_test.go @@ -79,3 +79,65 @@ func (s *WaitUntilReplicaIsCaughtUpToMasterSuite) TestIsCaughtUpIsCorrect() { func TestWaitUntilReplicaIsCaughtUpToMaster(t *testing.T) { suite.Run(t, new(WaitUntilReplicaIsCaughtUpToMasterSuite)) } + +type WaitUntilReplicaIsCaughtUpToMasterGTIDSuite struct { + suite.Suite + + w *ghostferry.WaitUntilReplicaIsCaughtUpToMaster +} + +func (s *WaitUntilReplicaIsCaughtUpToMasterGTIDSuite) SetupTest() { + config := testhelpers.NewTestConfig() + + masterDB, err := config.Source.SqlDB(nil) + s.Require().Nil(err) + replicaDB, err := config.Target.SqlDB(nil) + s.Require().Nil(err) + + s.w = &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ + MasterDB: masterDB, + ReplicaDB: replicaDB, + BinlogCoordinateMode: ghostferry.BinlogCoordinateGTID, + ReplicatedMasterCoordinateFetcher: ghostferry.ReplicatedMasterGTIDViaCustomQuery{ + Query: "SELECT gtid_executed FROM meta.heartbeat_gtid WHERE server_id = 1", + }, + } + + s.recreateTable(s.w.MasterDB) + s.recreateTable(s.w.ReplicaDB) +} + +func (s *WaitUntilReplicaIsCaughtUpToMasterGTIDSuite) recreateTable(db *sql.DB) { + _, err := db.Exec("DROP DATABASE IF EXISTS meta") + s.Require().Nil(err) + + _, err = db.Exec("CREATE DATABASE meta") + s.Require().Nil(err) + + _, err = db.Exec("CREATE TABLE meta.heartbeat_gtid (server_id int unsigned NOT NULL, gtid_executed longtext, PRIMARY KEY (server_id))") + s.Require().Nil(err) +} + +func (s *WaitUntilReplicaIsCaughtUpToMasterGTIDSuite) updateHeartbeatGTID(db *sql.DB, gtidSet string) { + _, err := db.Exec("REPLACE INTO meta.heartbeat_gtid (server_id, gtid_executed) VALUES (1, ?)", gtidSet) + s.Require().Nil(err) +} + +func (s *WaitUntilReplicaIsCaughtUpToMasterGTIDSuite) TestIsCaughtUpToCoordinateGTID() { + uuid := "3e11fa47-71ca-11e1-9e33-c80aa9429562" + target := ghostferry.NewGTIDCoordinate(uuid + ":1-100") + + s.updateHeartbeatGTID(s.w.ReplicaDB, uuid+":1-50") + isCaughtUp, err := s.w.IsCaughtUpToCoordinate(target, 1) + s.Require().Nil(err) + s.Require().False(isCaughtUp) + + s.updateHeartbeatGTID(s.w.ReplicaDB, uuid+":1-150") + isCaughtUp, err = s.w.IsCaughtUpToCoordinate(target, 1) + s.Require().Nil(err) + s.Require().True(isCaughtUp) +} + +func TestWaitUntilReplicaIsCaughtUpToMasterGTID(t *testing.T) { + suite.Run(t, new(WaitUntilReplicaIsCaughtUpToMasterGTIDSuite)) +} diff --git a/wait_until_replica_is_caught_up_to_master.go b/wait_until_replica_is_caught_up_to_master.go index a8329e76..674dd9dc 100644 --- a/wait_until_replica_is_caught_up_to_master.go +++ b/wait_until_replica_is_caught_up_to_master.go @@ -14,6 +14,10 @@ type ReplicatedMasterPositionFetcher interface { Current(*sql.DB) (mysql.Position, error) } +type ReplicatedMasterCoordinateFetcher interface { + CurrentCoordinate(*sql.DB) (BinlogCoordinate, error) +} + // Selects the master position that we have replicated until from a heartbeat // table of sort. type ReplicatedMasterPositionViaCustomQuery struct { @@ -41,27 +45,107 @@ func (r ReplicatedMasterPositionViaCustomQuery) Current(replicaDB *sql.DB) (mysq return NewMysqlPosition(file, pos, err, replicaDB) } +func (r ReplicatedMasterPositionViaCustomQuery) CurrentCoordinate(replicaDB *sql.DB) (BinlogCoordinate, error) { + pos, err := r.Current(replicaDB) + if err != nil { + return BinlogCoordinate{}, err + } + return NewFilePositionCoordinate(pos), nil +} + +// ReplicatedMasterGTIDViaCustomQuery selects the GTID set the replica has +// replicated up to from a heartbeat table of sort. +type ReplicatedMasterGTIDViaCustomQuery struct { + // The custom query must return a single row with a single column: the + // master's executed GTID set as a string, e.g. + // "SELECT gtid_executed FROM meta.ptheartbeat WHERE server_id = %d". + // Do not return the replica's own applied GTID set if it differs from the + // master's executed set. + Query string +} + +func (r ReplicatedMasterGTIDViaCustomQuery) CurrentCoordinate(replicaDB *sql.DB) (BinlogCoordinate, error) { + var gtidSet string + row := replicaDB.QueryRow(r.Query) + if err := row.Scan(>idSet); err != nil { + return BinlogCoordinate{}, err + } + if _, err := mysql.ParseMysqlGTIDSet(gtidSet); err != nil { + return BinlogCoordinate{}, err + } + return NewGTIDCoordinate(gtidSet), nil +} + // Only set the MasterDB and ReplicatedMasterPosition options in your code as // the others will be overwritten by the ferry. type WaitUntilReplicaIsCaughtUpToMaster struct { MasterDB *sql.DB ReplicatedMasterPositionFetcher ReplicatedMasterPositionFetcher - Timeout time.Duration + + // Takes precedence over ReplicatedMasterPositionFetcher; required in GTID mode. + ReplicatedMasterCoordinateFetcher ReplicatedMasterCoordinateFetcher + + BinlogCoordinateMode BinlogCoordinateType + + Timeout time.Duration ReplicaDB *sql.DB logger Logger } +func (w *WaitUntilReplicaIsCaughtUpToMaster) coordinateMode() BinlogCoordinateType { + if w.BinlogCoordinateMode == "" { + return BinlogCoordinateFilePosition + } + return w.BinlogCoordinateMode +} + +func (w *WaitUntilReplicaIsCaughtUpToMaster) coordinateFetcher() ReplicatedMasterCoordinateFetcher { + if w.ReplicatedMasterCoordinateFetcher != nil { + return w.ReplicatedMasterCoordinateFetcher + } + if w.ReplicatedMasterPositionFetcher != nil { + if cf, ok := w.ReplicatedMasterPositionFetcher.(ReplicatedMasterCoordinateFetcher); ok { + return cf + } + return positionFetcherAdapter{w.ReplicatedMasterPositionFetcher} + } + return nil +} + +type positionFetcherAdapter struct { + fetcher ReplicatedMasterPositionFetcher +} + +func (a positionFetcherAdapter) CurrentCoordinate(replicaDB *sql.DB) (BinlogCoordinate, error) { + pos, err := a.fetcher.Current(replicaDB) + if err != nil { + return BinlogCoordinate{}, err + } + return NewFilePositionCoordinate(pos), nil +} + +// IsCaughtUp is the backwards-compatible file/position variant of IsCaughtUpToCoordinate. func (w *WaitUntilReplicaIsCaughtUpToMaster) IsCaughtUp(targetMasterPos mysql.Position, retries int) (bool, error) { + return w.IsCaughtUpToCoordinate(NewFilePositionCoordinate(targetMasterPos), retries) +} + +// IsCaughtUpToCoordinate reports whether the replica has replicated up to the given target coordinate. +func (w *WaitUntilReplicaIsCaughtUpToMaster) IsCaughtUpToCoordinate(targetMaster BinlogCoordinate, retries int) (bool, error) { if w.logger == nil { w.logger = LogWithField("tag", "wait_replica") } - var currentReplicatedMasterPos mysql.Position - err := WithRetries(retries, 600*time.Millisecond, w.logger, "read replicated master binlog position", func() error { + fetcher := w.coordinateFetcher() + if fetcher == nil { + return false, errors.New("no replicated master coordinate fetcher configured") + } + + var current BinlogCoordinate + err := WithRetries(retries, 600*time.Millisecond, w.logger, "read replicated master coordinate", func() error { var err error - currentReplicatedMasterPos, err = w.ReplicatedMasterPositionFetcher.Current(w.ReplicaDB) + current, err = fetcher.CurrentCoordinate(w.ReplicaDB) return err }) @@ -69,12 +153,17 @@ func (w *WaitUntilReplicaIsCaughtUpToMaster) IsCaughtUp(targetMasterPos mysql.Po return false, err } - if currentReplicatedMasterPos.Compare(targetMasterPos) >= 0 { - w.logger.Infof("target master position reached by replica: %v >= %v\n", currentReplicatedMasterPos, targetMasterPos) + reached, err := current.HasReached(targetMaster) + if err != nil { + return false, err + } + + if reached { + w.logger.Infof("target master coordinate reached by replica: %v has reached %v\n", current, targetMaster) return true, nil } - w.logger.Debugf("replicated master position is: %v < %v\n", currentReplicatedMasterPos, targetMasterPos) + w.logger.Debugf("replicated master coordinate %v has not yet reached %v\n", current, targetMaster) return false, nil } @@ -87,10 +176,16 @@ func (w *WaitUntilReplicaIsCaughtUpToMaster) Wait() error { start := time.Now() - var targetMasterPos mysql.Position - err := WithRetries(100, 600*time.Millisecond, w.logger, "read master binlog position", func() error { + var targetMaster BinlogCoordinate + err := WithRetries(100, 600*time.Millisecond, w.logger, "read master coordinate", func() error { var err error - targetMasterPos, err = ShowMasterStatusBinlogPosition(w.MasterDB) + if w.coordinateMode() == BinlogCoordinateGTID { + targetMaster, err = ReadCurrentGTIDCoordinate(w.MasterDB) + return err + } + var pos mysql.Position + pos, err = ShowMasterStatusBinlogPosition(w.MasterDB) + targetMaster = NewFilePositionCoordinate(pos) return err }) @@ -99,10 +194,10 @@ func (w *WaitUntilReplicaIsCaughtUpToMaster) Wait() error { return err } - w.logger.Infof("target master position is: %v\n", targetMasterPos) + w.logger.Infof("target master coordinate is: %v\n", targetMaster) for { - isCaughtUp, err := w.IsCaughtUp(targetMasterPos, 100) + isCaughtUp, err := w.IsCaughtUpToCoordinate(targetMaster, 100) if err != nil { w.logger.WithError(err).Error("failed to get replica binlog coordinates") return err