Enhancement to allow syncing binlogs using a specified GTID set

Author: Deepak Gupta

replication/backup.go

@@ -27,6 +27,20 @@ func (b *BinlogSyncer) StartBackup(backupDir string, p mysql.Position, timeout t
 	}
 }
 
+func (b *BinlogSyncer) StartBackupGTID(backupDir string, gset mysql.GTIDSet, timeout time.Duration) error {
+	err := os.MkdirAll(backupDir, 0o755)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if b.cfg.SynchronousEventHandler == nil {
+		return b.StartBackupWithHandlerAndGTID(gset, timeout, func(filename string) (io.WriteCloser, error) {
+			return os.OpenFile(path.Join(backupDir, filename), os.O_CREATE|os.O_WRONLY, 0o644)
+		})
+	} else {
+		return b.StartSynchronousBackupWithGTID(gset, timeout)
+	}
+}
+
 // StartBackupWithHandler starts the backup process for the binary log using the specified position and handler.
 // The process will continue until the timeout is reached or an error occurs.
 // This method should not be used together with SynchronousEventHandler.
@@ -54,52 +68,73 @@ func (b *BinlogSyncer) StartBackupWithHandler(p mysql.Position, timeout time.Dur
 	backupHandler := &BackupEventHandler{
 		handler: handler,
 	}
-
 	s, err := b.StartSync(p)
 	if err != nil {
 		return errors.Trace(err)
 	}
+	return processWithHandler(b, s, backupHandler, timeout)
+}
 
-	defer func() {
-		if backupHandler.w != nil {
-			closeErr := backupHandler.w.Close()
-			if retErr == nil {
-				retErr = closeErr
-			}
-		}
-	}()
+// StartBackupWithHandlerAndGTID starts the backup process for the binary log using the specified GTID set and handler.
+//   - gset: The GTID set from which to begin the backup.
+//   - timeout: The maximum duration to wait for new binlog events before stopping the backup process.
+//     If set to 0, a default very long timeout (30 days) is used instead.
+//   - handler: A function that takes a binlog filename and returns an WriteCloser for writing raw events to.
+func (b *BinlogSyncer) StartBackupWithHandlerAndGTID(gset mysql.GTIDSet, timeout time.Duration,
+	handler func(binlogFilename string) (io.WriteCloser, error),
+) (retErr error) {
+	if timeout == 0 {
+		// a very long timeout here
+		timeout = 30 * 3600 * 24 * time.Second
+	}
+	if b.cfg.SynchronousEventHandler != nil {
+		return errors.New("StartBackupWithHandlerAndGTID cannot be used when SynchronousEventHandler is set. Use StartSynchronousBackupWithGTID instead.")
+	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
+	// Force use raw mode
+	b.parser.SetRawMode(true)
 
-	for {
-		select {
-		case <-ctx.Done():
-			return nil
-		case <-b.ctx.Done():
-			return nil
-		case err := <-s.ech:
-			return errors.Trace(err)
-		case e := <-s.ch:
-			err = backupHandler.HandleEvent(e)
-			if err != nil {
-				return errors.Trace(err)
-			}
-		}
+	// Set up the backup event handler
+	backupHandler := &BackupEventHandler{
+		handler: handler,
+	}
+
+	s, err := b.StartSyncGTID(gset)
+	if err != nil {
+		return errors.Trace(err)
 	}
+	return processWithHandler(b, s, backupHandler, timeout)
 }
 
 // StartSynchronousBackup starts the backup process using the SynchronousEventHandler in the BinlogSyncerConfig.
 func (b *BinlogSyncer) StartSynchronousBackup(p mysql.Position, timeout time.Duration) error {
 	if b.cfg.SynchronousEventHandler == nil {
 		return errors.New("SynchronousEventHandler must be set in BinlogSyncerConfig to use StartSynchronousBackup")
 	}
-
 	s, err := b.StartSync(p)
 	if err != nil {
 		return errors.Trace(err)
 	}
 
+	return process(b, s, timeout)
+}
+
+// StartSynchronousBackupWithGTID starts the backup process using the SynchronousEventHandler in the BinlogSyncerConfig with a specified GTID set.
+func (b *BinlogSyncer) StartSynchronousBackupWithGTID(gset mysql.GTIDSet, timeout time.Duration) error {
+
+	if b.cfg.SynchronousEventHandler == nil {
+		return errors.New("SynchronousEventHandler must be set in BinlogSyncerConfig to use StartSynchronousBackupWithGTID")
+	}
+
+	s, err := b.StartSyncGTID(gset)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	return process(b, s, timeout)
+}
+
+func process(b *BinlogSyncer, s *BinlogStreamer, timeout time.Duration) error {
 	var ctx context.Context
 	var cancel context.CancelFunc
 
@@ -123,6 +158,35 @@ func (b *BinlogSyncer) StartSynchronousBackup(p mysql.Position, timeout time.Dur
 	}
 }
 
+func processWithHandler(b *BinlogSyncer, s *BinlogStreamer, backupHandler *BackupEventHandler, timeout time.Duration) (retErr error) {
+	defer func() {
+		if backupHandler.w != nil {
+			closeErr := backupHandler.w.Close()
+			if retErr == nil {
+				retErr = closeErr
+			}
+		}
+	}()
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-b.ctx.Done():
+			return nil
+		case err := <-s.ech:
+			return errors.Trace(err)
+		case e := <-s.ch:
+			err := backupHandler.HandleEvent(e)
+			if err != nil {
+				return errors.Trace(err)
+			}
+		}
+	}
+}
+
 // BackupEventHandler handles writing events for backup
 type BackupEventHandler struct {
 	handler func(binlogFilename string) (io.WriteCloser, error)

replication/backup_test.go

@@ -2,6 +2,8 @@ package replication
 
 import (
 	"context"
+	"fmt"
+	"github.com/google/uuid"
 	"io"
 	"os"
 	"path"
@@ -61,6 +63,16 @@ func (t *testSyncerSuite) TestSyncBackup() {
 	testBackup(t, true) // true indicates synchronous mode
 }
 
+// TestAsyncBackupWithGTID runs the backup process in asynchronous mode with GTID and verifies binlog file creation.
+func (t *testSyncerSuite) TestAsyncBackupWithGTID() {
+	testBackUpWithGTID(t, false) // false indicates asynchronous mode
+}
+
+// TestSyncBackupWithGTID runs the backup process in synchronous mode with GTID and verifies binlog file creation.
+func (t *testSyncerSuite) TestSyncBackupWithGTID() {
+	testBackUpWithGTID(t, true) // true indicates synchronous mode
+}
+
 // testBackup is a helper function that runs the backup process in the specified mode and checks if binlog files are written correctly.
 func testBackup(t *testSyncerSuite, isSynchronous bool) {
 	t.setupTest(mysql.MySQLFlavor)
@@ -111,6 +123,71 @@ func testBackup(t *testSyncerSuite, isSynchronous bool) {
 	}
 }
 
+func testBackUpWithGTID(t *testSyncerSuite, isSynchronous bool) {
+	t.setupTest(mysql.MySQLFlavor)
+	t.b.cfg.SemiSyncEnabled = false // Ensure semi-sync is disabled
+
+	binlogDir := "./var"
+	os.RemoveAll(binlogDir)
+	timeout := 3 * time.Second
+
+	if isSynchronous {
+		// Set up a BackupEventHandler for synchronous mode
+		backupHandler := NewBackupEventHandler(
+			func(filename string) (io.WriteCloser, error) {
+				return os.OpenFile(path.Join(binlogDir, filename), os.O_CREATE|os.O_WRONLY, 0o644)
+			},
+		)
+		t.b.cfg.SynchronousEventHandler = backupHandler
+	} else {
+		// Ensure SynchronousEventHandler is nil for asynchronous mode
+		t.b.cfg.SynchronousEventHandler = nil
+	}
+
+	r, err := t.c.Execute("SELECT @@gtid_mode")
+	require.NoError(t.T(), err)
+	modeOn, _ := r.GetString(0, 0)
+	if modeOn != "ON" {
+		t.T().Skip("GTID mode is not ON")
+	}
+
+	r, err = t.c.Execute("SHOW GLOBAL VARIABLES LIKE 'SERVER_UUID'")
+	require.NoError(t.T(), err)
+
+	var masterUuid uuid.UUID
+	if s, _ := r.GetString(0, 1); len(s) > 0 && s != "NONE" {
+		masterUuid, err = uuid.Parse(s)
+		require.NoError(t.T(), err)
+	}
+
+	set, _ := mysql.ParseMysqlGTIDSet(fmt.Sprintf("%s:%d-%d", masterUuid.String(), 1, 2))
+	done := make(chan bool)
+
+	// Start the backup process in a goroutine
+	go func() {
+		err := t.b.StartBackupGTID(binlogDir, set, timeout)
+		require.NoError(t.T(), err)
+		done <- true
+	}()
+
+	failTimeout := 2 * timeout
+	ctx, cancel := context.WithTimeout(context.Background(), failTimeout)
+	defer cancel()
+
+	// Wait for the backup to complete or timeout
+	select {
+	case <-done:
+		files, err := os.ReadDir(binlogDir)
+		require.NoError(t.T(), err, "Failed to read binlog directory")
+		require.Greater(t.T(), len(files), 0, "Binlog files were not written to the directory")
+		mode := modeLabel(isSynchronous)
+		t.T().Logf("Backup completed successfully in %s mode using GTID with %d binlog file(s).", mode, len(files))
+	case <-ctx.Done():
+		mode := modeLabel(isSynchronous)
+		t.T().Fatalf("Timeout error during backup in %s mode.", mode)
+	}
+}
+
 func modeLabel(isSynchronous bool) string {
 	if isSynchronous {
 		return "synchronous"