GTID 5/5: GTID-aware replica catchup

CHANGELOG.md

@@ -2,7 +2,7 @@
 
 All notable changes to this project will be documented in this file.
 
-## [Unreleased]
+## [1.4.0 - unreleased]
 
 ### Added
 

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)

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

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
 

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
 }
 

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

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
 

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
 }
 

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)

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()

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))
+}