Incremental backup and point in time recovery for XtraBackup

.github/workflows/cluster_endtoend_backup_pitr.yml

@@ -84,15 +84,16 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       run: |
 
-        # Get key to latest MySQL repo
-        sudo apt-key adv --keyserver keyserver.ubuntu.com --recv-keys 467B942D3A79BD29
-        # Setup MySQL 8.0
-        wget -c https://dev.mysql.com/get/mysql-apt-config_0.8.24-1_all.deb
-        echo mysql-apt-config mysql-apt-config/select-server select mysql-8.0 | sudo debconf-set-selections
-        sudo DEBIAN_FRONTEND="noninteractive" dpkg -i mysql-apt-config*
+        # Setup Percona Server for MySQL 8.0
         sudo apt-get update
+        sudo apt-get install -y lsb-release gnupg2 curl
+        wget https://repo.percona.com/apt/percona-release_latest.$(lsb_release -sc)_all.deb
+        sudo DEBIAN_FRONTEND="noninteractive" dpkg -i percona-release_latest.$(lsb_release -sc)_all.deb
+        sudo percona-release setup ps80
+        sudo apt-get update
+
         # Install everything else we need, and configure
-        sudo apt-get install -y mysql-server mysql-client make unzip g++ etcd curl git wget eatmydata xz-utils libncurses5
+        sudo apt-get install -y percona-server-server percona-server-client make unzip g++ etcd git wget eatmydata xz-utils libncurses5
 
         sudo service mysql stop
         sudo service etcd stop
@@ -103,6 +104,8 @@ jobs:
         # install JUnit report formatter
         go install github.com/vitessio/go-junit-report@HEAD
 
+        sudo apt-get install percona-xtrabackup-80 lz4
+
     - name: Setup launchable dependencies
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true' && github.base_ref == 'main'
       run: |

.github/workflows/cluster_endtoend_backup_pitr_mysql57.yml

@@ -13,6 +13,10 @@ env:
   LAUNCHABLE_WORKSPACE: "vitess-app"
   GITHUB_PR_HEAD_SHA: "${{ github.event.pull_request.head.sha }}"
 
+  # This is used if we need to pin the xtrabackup version used in tests.
+  # If this is NOT set then the latest version available will be used.
+  #XTRABACKUP_VERSION: "2.4.24-1"
+
 jobs:
   build:
     name: Run endtoend tests on Cluster (backup_pitr) mysql57
@@ -114,6 +118,18 @@ jobs:
         # install JUnit report formatter
         go install github.com/vitessio/go-junit-report@HEAD
 
+        wget "https://repo.percona.com/apt/percona-release_latest.$(lsb_release -sc)_all.deb"
+        sudo apt-get install -y gnupg2
+        sudo dpkg -i "percona-release_latest.$(lsb_release -sc)_all.deb"
+        sudo apt-get update
+        if [[ -n $XTRABACKUP_VERSION ]]; then
+          debfile="percona-xtrabackup-24_$XTRABACKUP_VERSION.$(lsb_release -sc)_amd64.deb"
+          wget "https://repo.percona.com/pxb-24/apt/pool/main/p/percona-xtrabackup-24/$debfile"
+          sudo apt install -y "./$debfile"
+        else
+          sudo apt-get install -y percona-xtrabackup-24
+        fi
+
     - name: Setup launchable dependencies
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true' && github.base_ref == 'main'
       run: |

go/test/endtoend/backup/pitr/backup_mysqlctld_pitr_test.go

@@ -52,10 +52,30 @@ func waitForReplica(t *testing.T) {
 }
 
 // TestIncrementalBackupMysqlctld - tests incremental backups using myslctld
-func TestIncrementalBackupMysqlctld(t *testing.T) {
+func TestIncrementalBackup(t *testing.T) {
 	defer cluster.PanicHandler(t)
+
+	tcases := []struct {
+		name      string
+		setupType int
+		comprss   *backup.CompressionDetails
+	}{
+		{
+			"BuiltinBackup", backup.BuiltinBackup, nil,
+		},
+		{
+			"XtraBackup", backup.XtraBackup, &backup.CompressionDetails{
+				CompressorEngineName: "pgzip",
+			},
+		},
+		{
+			"Mysqlctld", backup.Mysqlctld, nil,
+		},
+	}
+	for _, tcase := range tcases {
+		t.Run(tcase.name, func(t *testing.T) {
 			// setup cluster for the testing
-	code, err := backup.LaunchCluster(backup.Mysqlctld, "xbstream", 0, nil)
+			code, err := backup.LaunchCluster(tcase.setupType, "xbstream", 0, tcase.comprss)
 			require.NoError(t, err, "setup failed with status code %d", code)
 			defer backup.TearDownCluster()
 
@@ -77,6 +97,7 @@ func TestIncrementalBackupMysqlctld(t *testing.T) {
 			t.Run("full backup", func(t *testing.T) {
 				backup.InsertRowOnPrimary(t, "before-full-backup")
 				waitForReplica(t)
+
 				manifest, _ := backup.TestReplicaFullBackup(t)
 				fullBackupPos = manifest.Position
 				require.False(t, fullBackupPos.IsZero())
@@ -168,7 +189,7 @@ func TestIncrementalBackupMysqlctld(t *testing.T) {
 					}
 					require.False(t, manifest.FromPosition.IsZero())
 					require.NotEqual(t, manifest.Position, manifest.FromPosition)
-			require.True(t, manifest.Position.GTIDSet.Contains(manifest.FromPosition.GTIDSet))
+					require.True(t, manifest.Position.GTIDSet.Union(manifest.PurgedPosition.GTIDSet).Contains(manifest.FromPosition.GTIDSet))
 
 					gtidPurgedPos, err := mysql.ParsePosition(mysql.Mysql56FlavorID, backup.GetReplicaGtidPurged(t))
 					require.NoError(t, err)
@@ -178,7 +199,7 @@ func TestIncrementalBackupMysqlctld(t *testing.T) {
 					if !incrementalFromPos.IsZero() {
 						expectFromPosition = incrementalFromPos.GTIDSet.Union(gtidPurgedPos.GTIDSet)
 					}
-			require.Equalf(t, expectFromPosition, fromPositionIncludingPurged, "expected: %v, found: %v", expectFromPosition, fromPositionIncludingPurged)
+					require.Equalf(t, expectFromPosition, fromPositionIncludingPurged, "expected: %v, found: %v, gtid_purged: %v,  manifest.Position: %v", expectFromPosition, fromPositionIncludingPurged, gtidPurgedPos, manifest.Position)
 				})
 			}
 
@@ -209,4 +230,6 @@ func TestIncrementalBackupMysqlctld(t *testing.T) {
 			t.Run("PITR-2", func(t *testing.T) {
 				testRestores(t)
 			})
+		})
+	}
 }

go/vt/mysqlctl/backup.go

@@ -135,16 +135,24 @@ func Backup(ctx context.Context, params BackupParams) error {
 		return vterrors.Wrap(err, "StartBackup failed")
 	}
 
-	be, err := GetBackupEngine()
-	if err != nil {
-		return vterrors.Wrap(err, "failed to find backup engine")
-	}
 	// Scope stats to selected backup engine.
 	beParams := params.Copy()
 	beParams.Stats = params.Stats.Scope(
 		stats.Component(stats.BackupEngine),
 		stats.Implementation(titleCase(backupEngineImplementation)),
 	)
+	var be BackupEngine
+	if isIncrementalBackup(beParams) {
+		// Incremental backups are always done via 'builtin' engine, which copies
+		// appropriate binlog files.
+		be = BackupRestoreEngineMap[builtinBackupEngineName]
+	} else {
+		be, err = GetBackupEngine()
+		if err != nil {
+			return vterrors.Wrap(err, "failed to find backup engine")
+		}
+	}
+
 	// Take the backup, and either AbortBackup or EndBackup.
 	usable, err := be.ExecuteBackup(ctx, beParams, bh)
 	logger := params.Logger
@@ -291,6 +299,43 @@ func ShouldRestore(ctx context.Context, params RestoreParams) (bool, error) {
 	return checkNoDB(ctx, params.Mysqld, params.DbName)
 }
 
+// ensureRestoredGTIDPurgedMatchesManifest sees the following: when you restore a full backup, you want the MySQL server to have
+// @@gtid_purged == <gtid-of-backup>. This then also implies that @@gtid_executed equals same value. This is because we restore without
+// any binary logs.
+func ensureRestoredGTIDPurgedMatchesManifest(ctx context.Context, manifest *BackupManifest, params *RestoreParams) error {
+	if manifest == nil {
+		return nil
+	}
+	if manifest.Position.GTIDSet == nil {
+		return nil
+	}
+	gtid := manifest.Position.GTIDSet.String()
+	if gtid == "" {
+		return nil
+	}
+	// Xtrabackup 2.4's restore seems to set @@gtid_purged to be the @@gtid_purged at the time of backup. But this is not
+	// the desired value. We want to set @@gtid_purged to be the @@gtid_executed of the backup.
+	// As reminder, when restoring from a full backup, setting @@gtid_purged also sets @@gtid_executed.
+	restoredGTIDPurgedPos, err := params.Mysqld.GetGTIDPurged(ctx)
+	if err != nil {
+		return vterrors.Wrapf(err, "failed to read gtid_purged after restore")
+	}
+	if restoredGTIDPurgedPos.Equal(manifest.Position) {
+		return nil
+	}
+	params.Logger.Infof("Restore: @@gtid_purged does not equal manifest's GTID position. Setting @@gtid_purged to %v", gtid)
+	// This is not good. We want to apply a new @@gtid_purged value.
+	query := "RESET MASTER" // required dialect in 5.7
+	if _, err := params.Mysqld.FetchSuperQuery(ctx, query); err != nil {
+		return vterrors.Wrapf(err, "error issuing %v", query)
+	}
+	query = fmt.Sprintf("SET GLOBAL gtid_purged='%s'", gtid)
+	if _, err := params.Mysqld.FetchSuperQuery(ctx, query); err != nil {
+		return vterrors.Wrapf(err, "failed to apply `%s` after restore", query)
+	}
+	return nil
+}
+
 // Restore is the main entry point for backup restore.  If there is no
 // appropriate backup on the BackupStorage, Restore logs an error
 // and returns ErrNoBackup. Any other error is returned.
@@ -382,8 +427,7 @@ func Restore(ctx context.Context, params RestoreParams) (*BackupManifest, error)
 	// of those who can connect.
 	params.Logger.Infof("Restore: starting mysqld for mysql_upgrade")
 	// Note Start will use dba user for waiting, this is fine, it will be allowed.
-	err = params.Mysqld.Start(context.Background(), params.Cnf, "--skip-grant-tables", "--skip-networking")
-	if err != nil {
+	if err := params.Mysqld.Start(context.Background(), params.Cnf, "--skip-grant-tables", "--skip-networking"); err != nil {
 		return nil, err
 	}
 
@@ -395,19 +439,23 @@ func Restore(ctx context.Context, params RestoreParams) (*BackupManifest, error)
 	// The MySQL manual recommends restarting mysqld after running mysql_upgrade,
 	// so that any changes made to system tables take effect.
 	params.Logger.Infof("Restore: restarting mysqld after mysql_upgrade")
-	err = params.Mysqld.Shutdown(context.Background(), params.Cnf, true)
-	if err != nil {
+	if err := params.Mysqld.Shutdown(context.Background(), params.Cnf, true); err != nil {
 		return nil, err
 	}
-	err = params.Mysqld.Start(context.Background(), params.Cnf)
-	if err != nil {
+	if err := params.Mysqld.Start(context.Background(), params.Cnf); err != nil {
+		return nil, err
+	}
+	if err = ensureRestoredGTIDPurgedMatchesManifest(ctx, manifest, &params); err != nil {
 		return nil, err
 	}
 
 	if handles := restorePath.IncrementalBackupHandles(); len(handles) > 0 {
 		params.Logger.Infof("Restore: applying %v incremental backups", len(handles))
+		// Incremental restores are always done via 'builtin' engine, which copies
+		// appropriate binlog files.
+		builtInRE := BackupRestoreEngineMap[builtinBackupEngineName]
 		for _, bh := range handles {
-			manifest, err := re.ExecuteRestore(ctx, params, bh)
+			manifest, err := builtInRE.ExecuteRestore(ctx, params, bh)
 			if err != nil {
 				return nil, err
 			}

go/vt/mysqlctl/backupengine.go

@@ -167,6 +167,11 @@ func init() {
 	}
 }
 
+// isIncrementalBackup is a convenience function to check whether the params indicate an incremental backup request
+func isIncrementalBackup(params BackupParams) bool {
+	return params.IncrementalFromPos != ""
+}
+
 func registerBackupEngineFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&backupEngineImplementation, "backup_engine_implementation", backupEngineImplementation, "Specifies which implementation to use for creating new backups (builtin or xtrabackup). Restores will always be done with whichever engine created a given backup.")
 }

go/vt/mysqlctl/binlogs_gtid.go

@@ -64,7 +64,6 @@ func ChooseBinlogsForIncrementalBackup(
 	incrementalBackupToGTID string,
 	err error,
 ) {
-
 	var prevGTIDsUnion mysql.GTIDSet
 	for i, binlog := range binaryLogs {
 		previousGtids, err := pgtids(ctx, binlog)
@@ -108,6 +107,15 @@ func ChooseBinlogsForIncrementalBackup(
 		if err != nil {
 			return nil, "", "", vterrors.Wrapf(err, "cannot evaluate incremental backup from pos")
 		}
+		if incrementalBackupFromGTID == "" {
+			// This can happen on the very first binary log file. It happens in two scenarios:
+			// 1. This is the first binlog ever in the history of the mysql server; the GTID is truly empty
+			// 2. A full backup was taken and restored, with all binlog scrapped.
+			// We take for granted that the first binary log file covers the
+			// requested "from GTID"
+			incrementalBackupFromGTID = backupFromGTIDSet.String()
+		}
+
 		// The Previous-GTIDs of the binary logs that _follows_ our binary-logs-to-backup indicates
 		// the backup's position.
 		incrementalBackupToGTID, err := pgtids(ctx, binaryLogs[len(binaryLogs)-1])

go/vt/mysqlctl/binlogs_gtid_test.go

@@ -107,6 +107,45 @@ func TestChooseBinlogsForIncrementalBackup(t *testing.T) {
 			backupPos:   "16b1039f-0000-0000-0000-000000000000:1-63",
 			expectError: "Mismatching GTID entries",
 		},
+		{
+			name: "empty previous GTIDs in first binlog covering backup pos",
+			previousGTIDs: map[string]string{
+				"vt-bin.000001": "",
+				"vt-bin.000002": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-60",
+				"vt-bin.000003": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-60",
+				"vt-bin.000004": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-78",
+				"vt-bin.000005": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-243",
+				"vt-bin.000006": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-331",
+			},
+			backupPos:     "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-30",
+			expectBinlogs: []string{"vt-bin.000001", "vt-bin.000002", "vt-bin.000003", "vt-bin.000004", "vt-bin.000005"},
+		},
+		{
+			name: "empty previous GTIDs in first binlog not covering backup pos",
+			previousGTIDs: map[string]string{
+				"vt-bin.000001": "",
+				"vt-bin.000002": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-60",
+				"vt-bin.000003": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-60",
+				"vt-bin.000004": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-78",
+				"vt-bin.000005": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-243",
+				"vt-bin.000006": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-331",
+			},
+			backupPos:     "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-65",
+			expectBinlogs: []string{"vt-bin.000003", "vt-bin.000004", "vt-bin.000005"},
+		},
+		{
+			name: "empty previous GTIDs in first binlog not covering backup pos, 2",
+			previousGTIDs: map[string]string{
+				"vt-bin.000001": "",
+				"vt-bin.000002": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-60",
+				"vt-bin.000003": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-60",
+				"vt-bin.000004": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-78",
+				"vt-bin.000005": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-243",
+				"vt-bin.000006": "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-331",
+			},
+			backupPos:     "16b1039f-22b6-11ed-b765-0a43f95f28a3:1-100",
+			expectBinlogs: []string{"vt-bin.000004", "vt-bin.000005"},
+		},
 		{
 			name: "match with non strictly monotonic sequence",
 			previousGTIDs: map[string]string{
@@ -212,7 +251,9 @@ func TestChooseBinlogsForIncrementalBackup(t *testing.T) {
 			require.NoError(t, err)
 			require.NotEmpty(t, binlogsToBackup)
 			assert.Equal(t, tc.expectBinlogs, binlogsToBackup)
+			if tc.previousGTIDs[binlogsToBackup[0]] != "" {
 				assert.Equal(t, tc.previousGTIDs[binlogsToBackup[0]], fromGTID)
+			}
 			assert.Equal(t, tc.previousGTIDs[binlogs[len(binlogs)-1]], toGTID)
 			assert.NotEqual(t, fromGTID, toGTID)
 		})