Add GTID support for binlog streaming

doc/command-line-flags.md

@@ -160,6 +160,10 @@ Table name prefix to be used on the temporary tables.
 
 Add this flag when executing on a 1st generation Google Cloud Platform (GCP).
 
+### gtid
+
+Add this flag to enable support for [MySQL replication GTIDs](https://dev.mysql.com/doc/refman/5.7/en/replication-gtids-concepts.html) for replication positioning. This requires `gtid_mode` and `enforce_gtid_consistency` to be set to `ON`.
+
 ### heartbeat-interval-millis
 
 Default 100. See [`subsecond-lag`](subsecond-lag.md) for details.

go/base/context.go

@@ -108,6 +108,7 @@ type MigrationContext struct {
 	// This is useful when connecting to a MySQL instance where the external port
 	// may not match the internal port.
 	SkipPortValidation bool
+	UseGTIDs           bool
 
 	config            ContextConfig
 	configMutex       *sync.Mutex

go/base/context_test.go

@@ -1,5 +1,5 @@
 /*
-   Copyright 2022 GitHub Inc.
+   Copyright 2021 GitHub Inc.
 	 See https://github.com/github/gh-ost/blob/master/LICENSE
 */
 

go/binlog/binlog_entry.go

@@ -1,29 +1,20 @@
 /*
-   Copyright 2016 GitHub Inc.
+   Copyright 2022 GitHub Inc.
 	 See https://github.com/github/gh-ost/blob/master/LICENSE
 */
 
 package binlog
 
 import (
 	"fmt"
+
 	"github.com/github/gh-ost/go/mysql"
 )
 
 // BinlogEntry describes an entry in the binary log
 type BinlogEntry struct {
 	Coordinates mysql.BinlogCoordinates
-	EndLogPos   uint64
-
-	DmlEvent *BinlogDMLEvent
-}
-
-// NewBinlogEntry creates an empty, ready to go BinlogEntry object
-func NewBinlogEntry(logFile string, logPos uint64) *BinlogEntry {
-	binlogEntry := &BinlogEntry{
-		Coordinates: mysql.BinlogCoordinates{LogFile: logFile, LogPos: int64(logPos)},
-	}
-	return binlogEntry
+	DmlEvent    *BinlogDMLEvent
 }
 
 // NewBinlogEntryAt creates an empty, ready to go BinlogEntry object
@@ -34,13 +25,6 @@ func NewBinlogEntryAt(coordinates mysql.BinlogCoordinates) *BinlogEntry {
 	return binlogEntry
 }
 
-// Duplicate creates and returns a new binlog entry, with some of the attributes pre-assigned
-func (this *BinlogEntry) Duplicate() *BinlogEntry {
-	binlogEntry := NewBinlogEntry(this.Coordinates.LogFile, uint64(this.Coordinates.LogPos))
-	binlogEntry.EndLogPos = this.EndLogPos
-	return binlogEntry
-}
-
 // String() returns a string representation of this binlog entry
 func (this *BinlogEntry) String() string {
 	return fmt.Sprintf("[BinlogEntry at %+v; dml:%+v]", this.Coordinates, this.DmlEvent)

go/binlog/gomysql_reader.go

@@ -17,25 +17,27 @@ import (
 
 	gomysql "github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/go-mysql-org/go-mysql/replication"
+	uuid "github.com/google/uuid"
 	"golang.org/x/net/context"
 )
 
 type GoMySQLReader struct {
-	migrationContext         *base.MigrationContext
-	connectionConfig         *mysql.ConnectionConfig
-	binlogSyncer             *replication.BinlogSyncer
-	binlogStreamer           *replication.BinlogStreamer
-	currentCoordinates       mysql.BinlogCoordinates
-	currentCoordinatesMutex  *sync.Mutex
-	LastAppliedRowsEventHint mysql.BinlogCoordinates
+	migrationContext        *base.MigrationContext
+	connectionConfig        *mysql.ConnectionConfig
+	binlogSyncer            *replication.BinlogSyncer
+	binlogStreamer          *replication.BinlogStreamer
+	currentCoordinates      mysql.BinlogCoordinates
+	currentCoordinatesMutex *sync.Mutex
+	// LastTrxCoords are the coordinates of the last transaction completely read.
+	// If using the file coordinates it is binlog position of the transaction's XID event.
+	LastTrxCoords mysql.BinlogCoordinates
 }
 
 func NewGoMySQLReader(migrationContext *base.MigrationContext) *GoMySQLReader {
 	connectionConfig := migrationContext.InspectorConnectionConfig
 	return &GoMySQLReader{
 		migrationContext:        migrationContext,
 		connectionConfig:        connectionConfig,
-		currentCoordinates:      mysql.BinlogCoordinates{},
 		currentCoordinatesMutex: &sync.Mutex{},
 		binlogSyncer: replication.NewBinlogSyncer(replication.BinlogSyncerConfig{
 			ServerID:                uint32(migrationContext.ReplicaServerId),
@@ -46,8 +48,8 @@ func NewGoMySQLReader(migrationContext *base.MigrationContext) *GoMySQLReader {
 			Password:                connectionConfig.Password,
 			TLSConfig:               connectionConfig.TLSConfig(),
 			UseDecimal:              true,
-			MaxReconnectAttempts:    migrationContext.BinlogSyncerMaxReconnectAttempts,
 			TimestampStringLocation: time.UTC,
+			MaxReconnectAttempts:    migrationContext.BinlogSyncerMaxReconnectAttempts,
 		}),
 	}
 }
@@ -58,35 +60,33 @@ func (this *GoMySQLReader) ConnectBinlogStreamer(coordinates mysql.BinlogCoordin
 		return this.migrationContext.Log.Errorf("Empty coordinates at ConnectBinlogStreamer()")
 	}
 
+	this.currentCoordinatesMutex.Lock()
+	defer this.currentCoordinatesMutex.Unlock()
 	this.currentCoordinates = coordinates
-	this.migrationContext.Log.Infof("Connecting binlog streamer at %+v", this.currentCoordinates)
-	// Start sync with specified binlog file and position
-	this.binlogStreamer, err = this.binlogSyncer.StartSync(gomysql.Position{
-		Name: this.currentCoordinates.LogFile,
-		Pos:  uint32(this.currentCoordinates.LogPos),
-	})
-
+	this.migrationContext.Log.Infof("Connecting binlog streamer at %+v", coordinates)
+
+	// Start sync with specified GTID set or binlog file and position
+	if this.migrationContext.UseGTIDs {
+		coords := coordinates.(*mysql.GTIDBinlogCoordinates)
+		this.binlogStreamer, err = this.binlogSyncer.StartSyncGTID(coords.GTIDSet)
+	} else {
+		coords := this.currentCoordinates.(*mysql.FileBinlogCoordinates)
+		this.binlogStreamer, err = this.binlogSyncer.StartSync(gomysql.Position{
+			Name: coords.LogFile,
+			Pos:  uint32(coords.LogPos)},
+		)
+	}
 	return err
 }
 
-func (this *GoMySQLReader) GetCurrentBinlogCoordinates() *mysql.BinlogCoordinates {
+func (this *GoMySQLReader) GetCurrentBinlogCoordinates() mysql.BinlogCoordinates {
 	this.currentCoordinatesMutex.Lock()
 	defer this.currentCoordinatesMutex.Unlock()
-	returnCoordinates := this.currentCoordinates
-	return &returnCoordinates
+	return this.currentCoordinates.Clone()
 }
 
-// StreamEvents
 func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEvent *replication.RowsEvent, entriesChannel chan<- *BinlogEntry) error {
-	if this.currentCoordinates.IsLogPosOverflowBeyond4Bytes(&this.LastAppliedRowsEventHint) {
-		return fmt.Errorf("Unexpected rows event at %+v, the binlog end_log_pos is overflow 4 bytes", this.currentCoordinates)
-	}
-
-	if this.currentCoordinates.SmallerThanOrEquals(&this.LastAppliedRowsEventHint) {
-		this.migrationContext.Log.Debugf("Skipping handled query at %+v", this.currentCoordinates)
-		return nil
-	}
-
+	currentCoords := this.GetCurrentBinlogCoordinates()
 	dml := ToEventDML(ev.Header.EventType.String())
 	if dml == NotDML {
 		return fmt.Errorf("Unknown DML type: %s", ev.Header.EventType.String())
@@ -97,7 +97,7 @@ func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEven
 			// We do both at the same time
 			continue
 		}
-		binlogEntry := NewBinlogEntryAt(this.currentCoordinates)
+		binlogEntry := NewBinlogEntryAt(currentCoords)
 		binlogEntry.DmlEvent = NewBinlogDMLEvent(
 			string(rowsEvent.Table.Schema),
 			string(rowsEvent.Table.Table),
@@ -118,13 +118,13 @@ func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEven
 				binlogEntry.DmlEvent.WhereColumnValues = sql.ToColumnValues(row)
 			}
 		}
+
 		// The channel will do the throttling. Whoever is reading from the channel
 		// decides whether action is taken synchronously (meaning we wait before
 		// next iteration) or asynchronously (we keep pushing more events)
 		// In reality, reads will be synchronous
 		entriesChannel <- binlogEntry
 	}
-	this.LastAppliedRowsEventHint = this.currentCoordinates
 	return nil
 }
 
@@ -141,23 +141,56 @@ func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesCha
 		if err != nil {
 			return err
 		}
-		func() {
+
+		// Update binlog coords if using file-based coords.
+		// GTID coordinates are updated on receiving GTID events.
+		if !this.migrationContext.UseGTIDs {
 			this.currentCoordinatesMutex.Lock()
-			defer this.currentCoordinatesMutex.Unlock()
-			this.currentCoordinates.LogPos = int64(ev.Header.LogPos)
-			this.currentCoordinates.EventSize = int64(ev.Header.EventSize)
-		}()
+			coords := this.currentCoordinates.(*mysql.FileBinlogCoordinates)
+			prevCoords := coords.Clone().(*mysql.FileBinlogCoordinates)
+			coords.LogPos = int64(ev.Header.LogPos)
+			coords.EventSize = int64(ev.Header.EventSize)
+			if coords.IsLogPosOverflowBeyond4Bytes(prevCoords) {
+				this.currentCoordinatesMutex.Unlock()
+				return fmt.Errorf("Unexpected rows event at %+v, the binlog end_log_pos is overflow 4 bytes", coords)
+			}
+			this.currentCoordinatesMutex.Unlock()
+		}
 
-		switch binlogEvent := ev.Event.(type) {
+		switch event := ev.Event.(type) {
+		case *replication.GTIDEvent:
+			if !this.migrationContext.UseGTIDs {
+				continue
+			}
+			sid, err := uuid.FromBytes(event.SID)
+			if err != nil {
+				return err
+			}
+			this.currentCoordinatesMutex.Lock()
+			if this.LastTrxCoords != nil {
+				this.currentCoordinates = this.LastTrxCoords.Clone()
+			}
+			coords := this.currentCoordinates.(*mysql.GTIDBinlogCoordinates)
+			trxGset := gomysql.NewUUIDSet(sid, gomysql.Interval{Start: event.GNO, Stop: event.GNO + 1})
+			coords.GTIDSet.AddSet(trxGset)
+			this.currentCoordinatesMutex.Unlock()
 		case *replication.RotateEvent:
-			func() {
-				this.currentCoordinatesMutex.Lock()
-				defer this.currentCoordinatesMutex.Unlock()
-				this.currentCoordinates.LogFile = string(binlogEvent.NextLogName)
-			}()
-			this.migrationContext.Log.Infof("rotate to next log from %s:%d to %s", this.currentCoordinates.LogFile, int64(ev.Header.LogPos), binlogEvent.NextLogName)
+			if this.migrationContext.UseGTIDs {
+				continue
+			}
+			this.currentCoordinatesMutex.Lock()
+			coords := this.currentCoordinates.(*mysql.FileBinlogCoordinates)
+			coords.LogFile = string(event.NextLogName)
+			this.migrationContext.Log.Infof("rotate to next log from %s:%d to %s", coords.LogFile, int64(ev.Header.LogPos), event.NextLogName)
+			this.currentCoordinatesMutex.Unlock()
+		case *replication.XIDEvent:
+			if this.migrationContext.UseGTIDs {
+				this.LastTrxCoords = &mysql.GTIDBinlogCoordinates{GTIDSet: event.GSet.(*gomysql.MysqlGTIDSet)}
+			} else {
+				this.LastTrxCoords = this.currentCoordinates.Clone()
+			}
 		case *replication.RowsEvent:
-			if err := this.handleRowsEvent(ev, binlogEvent, entriesChannel); err != nil {
+			if err := this.handleRowsEvent(ev, event, entriesChannel); err != nil {
 				return err
 			}
 		}

go/cmd/gh-ost/main.go

@@ -87,6 +87,7 @@ func main() {
 	flag.BoolVar(&migrationContext.AliyunRDS, "aliyun-rds", false, "set to 'true' when you execute on Aliyun RDS.")
 	flag.BoolVar(&migrationContext.GoogleCloudPlatform, "gcp", false, "set to 'true' when you execute on a 1st generation Google Cloud Platform (GCP).")
 	flag.BoolVar(&migrationContext.AzureMySQL, "azure", false, "set to 'true' when you execute on Azure Database on MySQL.")
+	flag.BoolVar(&migrationContext.UseGTIDs, "gtid", false, "(experimental) set to 'true' to use MySQL GTIDs for binlog positioning.")
 
 	executeFlag := flag.Bool("execute", false, "actually execute the alter & migrate the table. Default is noop: do some tests and exit")
 	flag.BoolVar(&migrationContext.TestOnReplica, "test-on-replica", false, "Have the migration run on a replica, not on the master. At the end of migration replication is stopped, and tables are swapped and immediately swap-revert. Replication remains stopped and you can compare the two tables for building trust")

go/logic/applier.go

@@ -1,5 +1,5 @@
 /*
-   Copyright 2022 GitHub Inc.
+   Copyright 2021 GitHub Inc.
 	 See https://github.com/github/gh-ost/blob/master/LICENSE
 */
 
@@ -990,11 +990,11 @@ func (this *Applier) StopReplication() error {
 		return err
 	}
 
-	readBinlogCoordinates, executeBinlogCoordinates, err := mysql.GetReplicationBinlogCoordinates(this.migrationContext.ApplierMySQLVersion, this.db)
+	readBinlogCoordinates, executeBinlogCoordinates, err := mysql.GetReplicationBinlogCoordinates(this.migrationContext.ApplierMySQLVersion, this.db, this.migrationContext.UseGTIDs)
 	if err != nil {
 		return err
 	}
-	this.migrationContext.Log.Infof("Replication IO thread at %+v. SQL thread is at %+v", *readBinlogCoordinates, *executeBinlogCoordinates)
+	this.migrationContext.Log.Infof("Replication IO thread at %+v. SQL thread is at %+v", readBinlogCoordinates, executeBinlogCoordinates)
 	return nil
 }
 

go/logic/inspect.go

@@ -73,6 +73,11 @@ func (this *Inspector) InitDBConnections() (err error) {
 	if err := this.validateBinlogs(); err != nil {
 		return err
 	}
+	if this.migrationContext.UseGTIDs {
+		if err := this.validateGTIDConfig(); err != nil {
+			return err
+		}
+	}
 	if err := this.applyBinlogFormat(); err != nil {
 		return err
 	}
@@ -379,7 +384,7 @@ func (this *Inspector) applyBinlogFormat() error {
 
 // validateBinlogs checks that binary log configuration is good to go
 func (this *Inspector) validateBinlogs() error {
-	query := `select /* gh-ost */ @@global.log_bin, @@global.binlog_format`
+	query := `select /* gh-ost */@@global.log_bin, @@global.binlog_format`
 	var hasBinaryLogs bool
 	if err := this.db.QueryRow(query).Scan(&hasBinaryLogs, &this.migrationContext.OriginalBinlogFormat); err != nil {
 		return err
@@ -418,6 +423,22 @@ func (this *Inspector) validateBinlogs() error {
 	return nil
 }
 
+// validateGTIDConfig checks that the GTID configuration is good to go
+func (this *Inspector) validateGTIDConfig() error {
+	var gtidMode, enforceGtidConsistency string
+	query := `select @@global.gtid_mode, @@global.enforce_gtid_consistency`
+	if err := this.db.QueryRow(query).Scan(&gtidMode, &enforceGtidConsistency); err != nil {
+		return err
+	}
+	enforceGtidConsistency = strings.ToUpper(enforceGtidConsistency)
+	if strings.ToUpper(gtidMode) != "ON" || (enforceGtidConsistency != "ON" && enforceGtidConsistency != "1") {
+		return fmt.Errorf("%s must have gtid_mode=ON and enforce_gtid_consistency=ON to use GTID support", this.connectionConfig.Key.String())
+	}
+
+	this.migrationContext.Log.Infof("gtid config validated on %s", this.connectionConfig.Key.String())
+	return nil
+}
+
 // validateLogSlaveUpdates checks that binary log log_slave_updates is set. This test is not required when migrating on replica or when migrating directly on master
 func (this *Inspector) validateLogSlaveUpdates() error {
 	query := `select /* gh-ost */ @@global.log_slave_updates`

go/logic/migrator.go

@@ -1075,7 +1075,7 @@ func (this *Migrator) printStatus(rule PrintStatusRule, writers ...io.Writer) {
 		return
 	}
 
-	currentBinlogCoordinates := *this.eventsStreamer.GetCurrentBinlogCoordinates()
+	currentBinlogCoordinates := this.eventsStreamer.GetCurrentBinlogCoordinates()
 
 	status := fmt.Sprintf("Copy: %d/%d %.1f%%; Applied: %d; Backlog: %d/%d; Time: %+v(total), %+v(copy); streamer: %+v; Lag: %.2fs, HeartbeatLag: %.2fs, State: %s; ETA: %s",
 		totalRowsCopied, rowsEstimate, progressPct,
@@ -1140,7 +1140,7 @@ func (this *Migrator) initiateStreaming() error {
 			if atomic.LoadInt64(&this.finishedMigrating) > 0 {
 				return
 			}
-			this.migrationContext.SetRecentBinlogCoordinates(*this.eventsStreamer.GetCurrentBinlogCoordinates())
+			this.migrationContext.SetRecentBinlogCoordinates(this.eventsStreamer.GetCurrentBinlogCoordinates())
 		}
 	}()
 	return nil

go/logic/server.go

@@ -1,5 +1,5 @@
 /*
-   Copyright 2022 GitHub Inc.
+   Copyright 2021 GitHub Inc.
 	 See https://github.com/github/gh-ost/blob/master/LICENSE
 */