fixing issue with xtrabackup and long gtids (vitessio#16304)

Signed-off-by: Renan Rangel <rrangel@slack-corp.com>

Author: rvrangel

go/vt/mysqlctl/xtrabackupengine.go

@@ -71,6 +71,7 @@ const (
 	writerBufferSize     = 2 * 1024 * 1024 /*2 MiB*/
 	xtrabackupBinaryName = "xtrabackup"
 	xtrabackupEngineName = "xtrabackup"
+	xtrabackupInfoFile   = "xtrabackup_info"
 	xbstream             = "xbstream"
 )
 
@@ -292,14 +293,14 @@ func (be *XtrabackupEngine) backupFiles(
 	numStripes int,
 	flavor string,
 ) (replicationPosition replication.Position, finalErr error) {
-
 	backupProgram := path.Join(xtrabackupEnginePath, xtrabackupBinaryName)
 	flagsToExec := []string{"--defaults-file=" + params.Cnf.Path,
 		"--backup",
 		"--socket=" + params.Cnf.SocketFile,
 		"--slave-info",
 		"--user=" + xtrabackupUser,
 		"--target-dir=" + params.Cnf.TmpDir,
+		"--extra-lsndir=" + params.Cnf.TmpDir,
 	}
 	if xtrabackupStreamMode != "" {
 		flagsToExec = append(flagsToExec, "--stream="+xtrabackupStreamMode)
@@ -398,27 +399,14 @@ func (be *XtrabackupEngine) backupFiles(
 	// the replication position. Note that if we don't read stderr as we go, the
 	// xtrabackup process gets blocked when the write buffer fills up.
 	stderrBuilder := &strings.Builder{}
-	posBuilder := &strings.Builder{}
 	stderrDone := make(chan struct{})
 	go func() {
 		defer close(stderrDone)
 
 		scanner := bufio.NewScanner(backupErr)
-		capture := false
 		for scanner.Scan() {
 			line := scanner.Text()
 			params.Logger.Infof("xtrabackup stderr: %s", line)
-
-			// Wait until we see the first line of the binlog position.
-			// Then capture all subsequent lines. We need multiple lines since
-			// the value we're looking for has newlines in it.
-			if !capture {
-				if !strings.Contains(line, "MySQL binlog position") {
-					continue
-				}
-				capture = true
-			}
-			fmt.Fprintln(posBuilder, line)
 		}
 		if err := scanner.Err(); err != nil {
 			params.Logger.Errorf("error reading from xtrabackup stderr: %v", err)
@@ -462,8 +450,7 @@ func (be *XtrabackupEngine) backupFiles(
 		return replicationPosition, vterrors.Wrap(err, fmt.Sprintf("xtrabackup failed with error. Output=%s", sterrOutput))
 	}
 
-	posOutput := posBuilder.String()
-	replicationPosition, rerr := findReplicationPosition(posOutput, flavor, params.Logger)
+	replicationPosition, rerr := findReplicationPositionFromXtrabackupInfo(params.Cnf.TmpDir, flavor, params.Logger)
 	if rerr != nil {
 		return replicationPosition, vterrors.Wrap(rerr, "backup failed trying to find replication position")
 	}
@@ -751,6 +738,22 @@ func (be *XtrabackupEngine) extractFiles(ctx context.Context, logger logutil.Log
 	return nil
 }
 
+func findReplicationPositionFromXtrabackupInfo(directory, flavor string, logger logutil.Logger) (replication.Position, error) {
+	f, err := os.Open(path.Join(directory, xtrabackupInfoFile))
+	if err != nil {
+		return replication.Position{}, vterrors.Errorf(vtrpc.Code_INVALID_ARGUMENT,
+			"couldn't open %q to read GTID position", path.Join(directory, xtrabackupInfoFile))
+	}
+	defer f.Close()
+
+	contents, err := io.ReadAll(f)
+	if err != nil {
+		return replication.Position{}, vterrors.Errorf(vtrpc.Code_INVALID_ARGUMENT, "couldn't read GTID position from %q", f.Name())
+	}
+
+	return findReplicationPosition(string(contents), flavor, logger)
+}
+
 var xtrabackupReplicationPositionRegexp = regexp.MustCompile(`GTID of the last change '([^']*)'`)
 
 func findReplicationPosition(input, flavor string, logger logutil.Logger) (replication.Position, error) {

go/vt/mysqlctl/xtrabackupengine_test.go

@@ -20,6 +20,8 @@ import (
 	"bytes"
 	"io"
 	"math/rand"
+	"os"
+	"path"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -54,26 +56,48 @@ func TestFindReplicationPosition(t *testing.T) {
 	}
 }
 
-func TestFindReplicationPositionNoMatch(t *testing.T) {
+func TestFindReplicationPositionFromXtrabackupInfo(t *testing.T) {
+	input := `tool_version = 8.0.35-30
+	binlog_pos = filename 'vt-0476396352-bin.000005', position '310088991', GTID of the last change '145e508e-ae54-11e9-8ce6-46824dd1815e:1-3,
+	1e51f8be-ae54-11e9-a7c6-4280a041109b:1-3,
+	47b59de1-b368-11e9-b48b-624401d35560:1-152981,
+	557def0a-b368-11e9-84ed-f6fffd91cc57:1-3,
+	599ef589-ae55-11e9-9688-ca1f44501925:1-14857169,
+	b9ce485d-b36b-11e9-9b17-2a6e0a6011f4:1-371262'
+	format = xbstream
+	`
+	want := "145e508e-ae54-11e9-8ce6-46824dd1815e:1-3,1e51f8be-ae54-11e9-a7c6-4280a041109b:1-3,47b59de1-b368-11e9-b48b-624401d35560:1-152981,557def0a-b368-11e9-84ed-f6fffd91cc57:1-3,599ef589-ae55-11e9-9688-ca1f44501925:1-14857169,b9ce485d-b36b-11e9-9b17-2a6e0a6011f4:1-371262"
+
+	tmp, err := os.MkdirTemp(t.TempDir(), "test")
+	assert.NoError(t, err)
+
+	f, err := os.Create(path.Join(tmp, xtrabackupInfoFile))
+	assert.NoError(t, err)
+	_, err = f.WriteString(input)
+	assert.NoError(t, err)
+	assert.NoError(t, f.Close())
+
+	pos, err := findReplicationPositionFromXtrabackupInfo(tmp, "MySQL56", logutil.NewConsoleLogger())
+	assert.NoError(t, err)
+	assert.Equal(t, want, pos.String())
+}
+
+func TestFindReplicationPositionNoMatchFromXtrabackupInfo(t *testing.T) {
 	// Make sure failure to find a match triggers an error.
 	input := `nothing`
 
-	_, err := findReplicationPosition(input, "MySQL56", logutil.NewConsoleLogger())
-	if err == nil {
-		t.Fatalf("expected error from findReplicationPosition but got nil")
-	}
+	_, err := findReplicationPositionFromXtrabackupInfo(input, "MySQL56", logutil.NewConsoleLogger())
+	assert.Error(t, err)
 }
 
-func TestFindReplicationPositionEmptyMatch(t *testing.T) {
+func TestFindReplicationPositionEmptyMatchFromXtrabackupInfo(t *testing.T) {
 	// Make sure failure to find a match triggers an error.
 	input := `GTID of the last change '
 	
 	'`
 
-	_, err := findReplicationPosition(input, "MySQL56", logutil.NewConsoleLogger())
-	if err == nil {
-		t.Fatalf("expected error from findReplicationPosition but got nil")
-	}
+	_, err := findReplicationPositionFromXtrabackupInfo(input, "MySQL56", logutil.NewConsoleLogger())
+	assert.Error(t, err)
 }
 
 func TestStripeRoundTrip(t *testing.T) {