select backup engine in Backup() and ignore engines in RestoreFromBackup()

.github/workflows/cluster_endtoend_vtctlbackup_sharded_clustertest_heavy.yml

@@ -93,15 +93,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 A8D3785C
-        # Setup MySQL 8.0
-        wget -c https://dev.mysql.com/get/mysql-apt-config_0.8.32-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 -qq update
+        sudo apt-get -qq 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 -qq update
+
         # Install everything else we need, and configure
-        sudo apt-get -qq install -y mysql-server mysql-client make unzip g++ etcd curl git wget eatmydata xz-utils libncurses5
+        sudo apt-get -qq 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
@@ -112,6 +113,8 @@ jobs:
         # install JUnit report formatter
         go install github.com/vitessio/go-junit-report@HEAD
 
+        sudo apt-get -qq install -y percona-xtrabackup-80 lz4
+
     - name: Setup launchable dependencies
       if: steps.skip-workflow.outputs.is_draft == 'false' && steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true' && github.base_ref == 'main'
       run: |

go/cmd/vtctldclient/command/backups.go

@@ -35,7 +35,7 @@ import (
 var (
 	// Backup makes a Backup gRPC call to a vtctld.
 	Backup = &cobra.Command{
-		Use:                   "Backup [--concurrency <concurrency>] [--allow-primary] [--incremental-from-pos=<pos>|<backup-name>|auto] [--upgrade-safe] <tablet_alias>",
+		Use:                   "Backup [--concurrency <concurrency>] [--allow-primary] [--incremental-from-pos=<pos>|<backup-name>|auto] [--upgrade-safe] [--backup-engine=enginename] <tablet_alias>",
 		Short:                 "Uses the BackupStorage service on the given tablet to create and store a new backup.",
 		DisableFlagsInUseLine: true,
 		Args:                  cobra.ExactArgs(1),
@@ -70,7 +70,7 @@ If no replica-type tablet can be found, the backup can be taken on the primary i
 	}
 	// RestoreFromBackup makes a RestoreFromBackup gRPC call to a vtctld.
 	RestoreFromBackup = &cobra.Command{
-		Use:                   "RestoreFromBackup [--backup-timestamp|-t <YYYY-mm-DD.HHMMSS>] [--restore-to-pos <pos>] [--dry-run] <tablet_alias>",
+		Use:                   "RestoreFromBackup [--backup-timestamp|-t <YYYY-mm-DD.HHMMSS>] [--restore-to-pos <pos>] [--ignored-backup-engines=enginename,] [--dry-run] <tablet_alias>",
 		Short:                 "Stops mysqld on the specified tablet and restores the data from either the latest backup or closest before `backup-timestamp`.",
 		DisableFlagsInUseLine: true,
 		Args:                  cobra.ExactArgs(1),
@@ -80,6 +80,7 @@ If no replica-type tablet can be found, the backup can be taken on the primary i
 
 var backupOptions = struct {
 	AllowPrimary       bool
+	BackupEngine       string
 	Concurrency        int32
 	IncrementalFromPos string
 	UpgradeSafe        bool
@@ -93,13 +94,19 @@ func commandBackup(cmd *cobra.Command, args []string) error {
 
 	cli.FinishedParsing(cmd)
 
-	stream, err := client.Backup(commandCtx, &vtctldatapb.BackupRequest{
+	req := &vtctldatapb.BackupRequest{
 		TabletAlias:        tabletAlias,
 		AllowPrimary:       backupOptions.AllowPrimary,
 		Concurrency:        backupOptions.Concurrency,
 		IncrementalFromPos: backupOptions.IncrementalFromPos,
 		UpgradeSafe:        backupOptions.UpgradeSafe,
-	})
+	}
+
+	if backupOptions.BackupEngine != "" {
+		req.BackupEngine = &backupOptions.BackupEngine
+	}
+
+	stream, err := client.Backup(commandCtx, req)
 	if err != nil {
 		return err
 	}
@@ -218,10 +225,11 @@ func commandRemoveBackup(cmd *cobra.Command, args []string) error {
 }
 
 var restoreFromBackupOptions = struct {
-	BackupTimestamp    string
-	RestoreToPos       string
-	RestoreToTimestamp string
-	DryRun             bool
+	BackupTimestamp      string
+	IgnoredBackupEngines string
+	RestoreToPos         string
+	RestoreToTimestamp   string
+	DryRun               bool
 }{}
 
 func commandRestoreFromBackup(cmd *cobra.Command, args []string) error {
@@ -258,6 +266,10 @@ func commandRestoreFromBackup(cmd *cobra.Command, args []string) error {
 		req.BackupTime = protoutil.TimeToProto(t)
 	}
 
+	if restoreFromBackupOptions.IgnoredBackupEngines != "" {
+		req.IgnoredBackupEngines = strings.Split(restoreFromBackupOptions.IgnoredBackupEngines, ",")
+	}
+
 	cli.FinishedParsing(cmd)
 
 	stream, err := client.RestoreFromBackup(commandCtx, req)
@@ -282,6 +294,7 @@ func init() {
 	Backup.Flags().BoolVar(&backupOptions.AllowPrimary, "allow-primary", false, "Allow the primary of a shard to be used for the backup. WARNING: If using the builtin backup engine, this will shutdown mysqld on the primary and stop writes for the duration of the backup.")
 	Backup.Flags().Int32Var(&backupOptions.Concurrency, "concurrency", 4, "Specifies the number of compression/checksum jobs to run simultaneously.")
 	Backup.Flags().StringVar(&backupOptions.IncrementalFromPos, "incremental-from-pos", "", "Position, or name of backup from which to create an incremental backup. Default: empty. If given, then this backup becomes an incremental backup from given position or given backup. If value is 'auto', this backup will be taken from the last successful backup position.")
+	Backup.Flags().StringVar(&backupOptions.BackupEngine, "backup-engine", "", "Request a specific backup engine for this backup request. Defaults to the preferred backup engine of the target vttablet")
 
 	Backup.Flags().BoolVar(&backupOptions.UpgradeSafe, "upgrade-safe", false, "Whether to use innodb_fast_shutdown=0 for the backup so it is safe to use for MySQL upgrades.")
 	Root.AddCommand(Backup)
@@ -299,6 +312,7 @@ func init() {
 	Root.AddCommand(RemoveBackup)
 
 	RestoreFromBackup.Flags().StringVarP(&restoreFromBackupOptions.BackupTimestamp, "backup-timestamp", "t", "", "Use the backup taken at, or closest before, this timestamp. Omit to use the latest backup. Timestamp format is \"YYYY-mm-DD.HHMMSS\".")
+	RestoreFromBackup.Flags().StringVar(&restoreFromBackupOptions.IgnoredBackupEngines, "ignored-backup-engines", "", "Ignore backups created with this list of backup engines, sepparated by a comma")
 	RestoreFromBackup.Flags().StringVar(&restoreFromBackupOptions.RestoreToPos, "restore-to-pos", "", "Run a point in time recovery that ends with the given position. This will attempt to use one full backup followed by zero or more incremental backups")
 	RestoreFromBackup.Flags().StringVar(&restoreFromBackupOptions.RestoreToTimestamp, "restore-to-timestamp", "", "Run a point in time recovery that restores up to, and excluding, given timestamp in RFC3339 format (`2006-01-02T15:04:05Z07:00`). This will attempt to use one full backup followed by zero or more incremental backups")
 	RestoreFromBackup.Flags().BoolVar(&restoreFromBackupOptions.DryRun, "dry-run", false, "Only validate restore steps, do not actually restore data")

go/flags/endtoend/vtcombo.txt

@@ -310,6 +310,7 @@ Flags:
       --restore-to-timestamp string                                      (init incremental restore parameter) if set, run a point in time recovery that restores up to the given timestamp, if possible. Given timestamp in RFC3339 format. Example: '2006-01-02T15:04:05Z07:00'
       --restore_concurrency int                                          (init restore parameter) how many concurrent files to restore at once (default 4)
       --restore_from_backup                                              (init restore parameter) will check BackupStorage for a recent backup at startup and start there
+      --restore_from_backup_ignore_engines string                        ignore backups taken with the backup engines provided in this comma-separated list
       --restore_from_backup_ts string                                    (init restore parameter) if set, restore the latest backup taken at or before this timestamp. Example: '2021-04-29.133050'
       --retain_online_ddl_tables duration                                How long should vttablet keep an old migrated table before purging it (default 24h0m0s)
       --sanitize_log_messages                                            Remove potentially sensitive information in tablet INFO, WARNING, and ERROR log messages such as query parameters.

go/flags/endtoend/vttablet.txt

@@ -300,6 +300,7 @@ Flags:
       --restore-to-timestamp string                                      (init incremental restore parameter) if set, run a point in time recovery that restores up to the given timestamp, if possible. Given timestamp in RFC3339 format. Example: '2006-01-02T15:04:05Z07:00'
       --restore_concurrency int                                          (init restore parameter) how many concurrent files to restore at once (default 4)
       --restore_from_backup                                              (init restore parameter) will check BackupStorage for a recent backup at startup and start there
+      --restore_from_backup_ignore_engines string                        ignore backups taken with the backup engines provided in this comma-separated list
       --restore_from_backup_ts string                                    (init restore parameter) if set, restore the latest backup taken at or before this timestamp. Example: '2021-04-29.133050'
       --retain_online_ddl_tables duration                                How long should vttablet keep an old migrated table before purging it (default 24h0m0s)
       --s3_backup_aws_endpoint string                                    endpoint of the S3 backend (region must be provided).

go/test/endtoend/backup/vtctlbackup/backup_test.go

@@ -17,8 +17,16 @@ limitations under the License.
 package vtctlbackup
 
 import (
+	"encoding/json"
+	"io"
+	"os"
+	"path"
+	"strings"
 	"testing"
 
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/test/endtoend/cluster"
 	"vitess.io/vitess/go/vt/mysqlctl"
 )
 
@@ -29,6 +37,7 @@ func TestBuiltinBackup(t *testing.T) {
 
 func TestBuiltinBackupWithZstdCompression(t *testing.T) {
 	defer setDefaultCompressionFlag()
+	defer setDefaultCommonArgs()
 	cDetails := &CompressionDetails{
 		CompressorEngineName:    "zstd",
 		ExternalCompressorCmd:   "zstd",
@@ -41,6 +50,7 @@ func TestBuiltinBackupWithZstdCompression(t *testing.T) {
 
 func TestBuiltinBackupWithExternalZstdCompression(t *testing.T) {
 	defer setDefaultCompressionFlag()
+	defer setDefaultCommonArgs()
 	cDetails := &CompressionDetails{
 		CompressorEngineName:    "external",
 		ExternalCompressorCmd:   "zstd",
@@ -53,6 +63,7 @@ func TestBuiltinBackupWithExternalZstdCompression(t *testing.T) {
 
 func TestBuiltinBackupWithExternalZstdCompressionAndManifestedDecompressor(t *testing.T) {
 	defer setDefaultCompressionFlag()
+	defer setDefaultCommonArgs()
 	cDetails := &CompressionDetails{
 		CompressorEngineName:            "external",
 		ExternalCompressorCmd:           "zstd",
@@ -70,3 +81,132 @@ func setDefaultCompressionFlag() {
 	mysqlctl.ExternalDecompressorCmd = ""
 	mysqlctl.ManifestExternalDecompressorCmd = ""
 }
+
+func setDefaultCommonArgs() { commonTabletArg = getDefaultCommonArgs() }
+
+func TestBackupEngineSelector(t *testing.T) {
+	defer setDefaultCompressionFlag()
+	defer setDefaultCommonArgs()
+	defer cluster.PanicHandler(t)
+
+	// launch the custer with xtrabackup as the default engine
+	code, err := LaunchCluster(XtraBackup, "xbstream", 0, &CompressionDetails{CompressorEngineName: "pgzip"})
+	require.Nilf(t, err, "setup failed with status code %d", code)
+
+	defer TearDownCluster()
+
+	localCluster.DisableVTOrcRecoveries(t)
+	defer func() {
+		localCluster.EnableVTOrcRecoveries(t)
+	}()
+	verifyInitialReplication(t)
+
+	// first try to backup with an alternative engine (builtin)
+	err = localCluster.VtctldClientProcess.ExecuteCommand("Backup", "--allow-primary", "--backup-engine=builtin", primary.Alias)
+	require.NoError(t, err)
+	engineUsed := getBackupEngineOfLastBackup(t)
+	require.Equal(t, "builtin", engineUsed)
+
+	// then try to backup specifying the xtrabackup engine
+	err = localCluster.VtctldClientProcess.ExecuteCommand("Backup", "--allow-primary", "--backup-engine=xtrabackup", primary.Alias)
+	require.NoError(t, err)
+	engineUsed = getBackupEngineOfLastBackup(t)
+	require.Equal(t, "xtrabackup", engineUsed)
+
+	// check that by default we still use the xtrabackup engine if not specified
+	err = localCluster.VtctldClientProcess.ExecuteCommand("Backup", "--allow-primary", primary.Alias)
+	require.NoError(t, err)
+	engineUsed = getBackupEngineOfLastBackup(t)
+	require.Equal(t, "xtrabackup", engineUsed)
+}
+
+// fetch the backup engine used on the last backup triggered by the end-to-end tests.
+func getBackupEngineOfLastBackup(t *testing.T) string {
+	lastBackup := getLastBackup(t)
+
+	// open the Manifest and retrieve the backup engine that was used
+	f, err := os.Open(path.Join(localCluster.CurrentVTDATAROOT,
+		"backups", keyspaceName, shardName,
+		lastBackup, "MANIFEST",
+	))
+	require.NoError(t, err)
+	defer f.Close()
+
+	data, err := io.ReadAll(f)
+	require.NoError(t, err)
+
+	var manifest mysqlctl.BackupManifest
+	err = json.Unmarshal(data, &manifest)
+	require.NoError(t, err)
+
+	return manifest.BackupMethod
+}
+
+func getLastBackup(t *testing.T) string {
+	output, err := localCluster.VtctldClientProcess.ExecuteCommandWithOutput("GetBackups", shardKsName)
+	require.NoError(t, err)
+
+	// split the backups response into a slice of strings
+	backups := strings.Split(strings.TrimSpace(output), "\n")
+
+	return backups[len(backups)-1]
+}
+
+func TestRestoreIgnoreBackups(t *testing.T) {
+	defer setDefaultCompressionFlag()
+	defer setDefaultCommonArgs()
+	defer cluster.PanicHandler(t)
+
+	cDetails := &CompressionDetails{CompressorEngineName: "pgzip"}
+
+	// launch the custer with xtrabackup as the default engine
+	code, err := LaunchCluster(XtraBackup, "xbstream", 0, cDetails)
+	require.Nilf(t, err, "setup failed with status code %d", code)
+
+	defer TearDownCluster()
+
+	localCluster.DisableVTOrcRecoveries(t)
+	defer func() {
+		localCluster.EnableVTOrcRecoveries(t)
+	}()
+	verifyInitialReplication(t)
+
+	// lets take two backups, each using a different backup engine
+	err = localCluster.VtctldClientProcess.ExecuteCommand("Backup", "--allow-primary", "--backup-engine=builtin", primary.Alias)
+	require.NoError(t, err)
+	// firstBackup := getLastBackup(t)
+
+	err = localCluster.VtctldClientProcess.ExecuteCommand("Backup", "--allow-primary", "--backup-engine=xtrabackup", primary.Alias)
+	require.NoError(t, err)
+
+	//  insert more data on the primary
+	_, err = primary.VttabletProcess.QueryTablet("insert into vt_insert_test (msg) values ('test2')", keyspaceName, true)
+	require.NoError(t, err)
+
+	// now bring up the other replica, letting it restore from backup.
+	restoreWaitForBackup(t, "replica", cDetails, true)
+	err = replica2.VttabletProcess.WaitForTabletStatusesForTimeout([]string{"SERVING"}, timeout)
+	require.NoError(t, err)
+
+	// check the new replica has the data
+	cluster.VerifyRowsInTablet(t, replica2, keyspaceName, 2)
+
+	// now lets break the last backup in the shard
+	err = os.Remove(path.Join(localCluster.CurrentVTDATAROOT,
+		"backups", keyspaceName, shardName,
+		getLastBackup(t), "backup.xbstream.gz"))
+	require.NoError(t, err)
+
+	// and try to restore from it
+	err = localCluster.VtctldClientProcess.ExecuteCommand("RestoreFromBackup", replica2.Alias)
+	require.Error(t, err) // this should fail
+
+	// now we retry but trying the earlier backup
+	err = localCluster.VtctldClientProcess.ExecuteCommand("RestoreFromBackup", "--ignored-backup-engines=xtrabackup", replica2.Alias)
+	require.NoError(t, err) // this should succeed
+
+	// make sure we are replicating after the restore is done
+	err = replica2.VttabletProcess.WaitForTabletStatusesForTimeout([]string{"SERVING"}, timeout)
+	require.NoError(t, err)
+	cluster.VerifyRowsInTablet(t, replica2, keyspaceName, 2)
+}

go/test/endtoend/backup/vtctlbackup/backup_utils.go

@@ -74,14 +74,7 @@ var (
 	shardKsName      = fmt.Sprintf("%s/%s", keyspaceName, shardName)
 	dbCredentialFile string
 	shardName        = "0"
-	commonTabletArg  = []string{
-		"--vreplication_retry_delay", "1s",
-		"--degraded_threshold", "5s",
-		"--lock_tables_timeout", "5s",
-		"--watch_replication_stream",
-		"--enable_replication_reporter",
-		"--serving_state_grace_period", "1s",
-	}
+	commonTabletArg  = getDefaultCommonArgs()
 
 	vtInsertTest = `
 		create table vt_insert_test (
@@ -1445,3 +1438,14 @@ func verifyTabletRestoreStats(t *testing.T, vars map[string]any) {
 	}
 	require.Contains(t, bd, "BackupStorage.File.File:Read")
 }
+
+func getDefaultCommonArgs() []string {
+	return []string{
+		"--vreplication_retry_delay", "1s",
+		"--degraded_threshold", "5s",
+		"--lock_tables_timeout", "5s",
+		"--watch_replication_stream",
+		"--enable_replication_reporter",
+		"--serving_state_grace_period", "1s",
+	}
+}

go/vt/mysqlctl/backup.go

@@ -163,12 +163,14 @@ func Backup(ctx context.Context, params BackupParams) error {
 		// appropriate binlog files.
 		be = BackupRestoreEngineMap[builtinBackupEngineName]
 	} else {
-		be, err = GetBackupEngine()
+		be, err = GetBackupEngine(params.BackupEngine)
 		if err != nil {
 			return vterrors.Wrap(err, "failed to find backup engine")
 		}
 	}
 
+	params.Logger.Infof("Using backup engine %q", be.Name())
+
 	// Take the backup, and either AbortBackup or EndBackup.
 	backupResult, err := be.ExecuteBackup(ctx, beParams, bh)
 	logger := params.Logger