allow backup engine to be specified on Backup() API

Signed-off-by: Renan Rangel <[email protected]>

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/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"
 )
 
@@ -70,3 +78,63 @@ func setDefaultCompressionFlag() {
 	mysqlctl.ExternalDecompressorCmd = ""
 	mysqlctl.ManifestExternalDecompressorCmd = ""
 }
+
+func TestBackupEngineSelector(t *testing.T) {
+	defer setDefaultCompressionFlag()
+	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 {
+	output, err := localCluster.VtctldClientProcess.ExecuteCommandWithOutput("GetBackups", shardKsName)
+
+	// split the backups response into a slice of strings
+	backups := strings.Split(strings.TrimSpace(output), "\n")
+
+	// open the Manifest and retrieve the backup engine that was used
+	f, err := os.Open(path.Join(localCluster.CurrentVTDATAROOT,
+		"backups", keyspaceName, shardName,
+		backups[len(backups)-1], "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
+}

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

go/vt/mysqlctl/backup_test.go

@@ -498,7 +498,7 @@ func TestRestoreManifestMySQLVersionValidation(t *testing.T) {
 
 			manifest := BackupManifest{
 				BackupTime:   time.Now().Add(-1 * time.Hour).Format(time.RFC3339),
-				BackupMethod: "fake",
+				BackupMethod: fakeBackupEngineName,
 				Keyspace:     "test",
 				Shard:        "-",
 				MySQLVersion: tc.fromVersion,
@@ -599,7 +599,7 @@ func createFakeBackupRestoreEnv(t *testing.T) *fakeBackupRestoreEnv {
 
 	manifest := BackupManifest{
 		BackupTime:   FormatRFC3339(time.Now().Add(-1 * time.Hour)),
-		BackupMethod: "fake",
+		BackupMethod: fakeBackupEngineName,
 		Keyspace:     "test",
 		Shard:        "-",
 		MySQLVersion: "8.0.32",
@@ -612,8 +612,8 @@ func createFakeBackupRestoreEnv(t *testing.T) *fakeBackupRestoreEnv {
 	testBackupEngine.ExecuteRestoreReturn = FakeBackupEngineExecuteRestoreReturn{&manifest, nil}
 
 	previousBackupEngineImplementation := backupEngineImplementation
-	BackupRestoreEngineMap["fake"] = &testBackupEngine
-	backupEngineImplementation = "fake"
+	BackupRestoreEngineMap[fakeBackupEngineName] = &testBackupEngine
+	backupEngineImplementation = fakeBackupEngineName
 
 	testBackupStorage := FakeBackupStorage{}
 	testBackupStorage.ListBackupsReturn = FakeBackupStorageListBackupsReturn{
@@ -628,9 +628,9 @@ func createFakeBackupRestoreEnv(t *testing.T) *fakeBackupRestoreEnv {
 	testBackupStorage.StartBackupReturn = FakeBackupStorageStartBackupReturn{&FakeBackupHandle{}, nil}
 	testBackupStorage.WithParamsReturn = &testBackupStorage
 
-	backupstorage.BackupStorageMap["fake"] = &testBackupStorage
+	backupstorage.BackupStorageMap[fakeBackupEngineName] = &testBackupStorage
 	previousBackupStorageImplementation := backupstorage.BackupStorageImplementation
-	backupstorage.BackupStorageImplementation = "fake"
+	backupstorage.BackupStorageImplementation = fakeBackupEngineName
 
 	// all restore integration tests must be leak checked
 	t.Cleanup(func() {
@@ -641,10 +641,10 @@ func createFakeBackupRestoreEnv(t *testing.T) *fakeBackupRestoreEnv {
 		backupstats.DeprecatedBackupDurationS.Reset()
 		backupstats.DeprecatedRestoreDurationS.Reset()
 
-		delete(BackupRestoreEngineMap, "fake")
+		delete(BackupRestoreEngineMap, fakeBackupEngineName)
 		backupEngineImplementation = previousBackupEngineImplementation
 
-		delete(backupstorage.BackupStorageMap, "fake")
+		delete(backupstorage.BackupStorageMap, fakeBackupEngineName)
 		backupstorage.BackupStorageImplementation = previousBackupStorageImplementation
 		mysqld.Close()
 		sqldb.Close()

go/vt/mysqlctl/backupengine.go

@@ -23,6 +23,7 @@ import (
 	"os"
 	"path"
 	"path/filepath"
+	"slices"
 	"strings"
 	"time"
 
@@ -57,6 +58,7 @@ const (
 type BackupEngine interface {
 	ExecuteBackup(ctx context.Context, params BackupParams, bh backupstorage.BackupHandle) (BackupResult, error)
 	ShouldDrainForBackup(req *tabletmanagerdatapb.BackupRequest) bool
+	Name() string
 }
 
 // BackupParams is the struct that holds all params passed to ExecuteBackup
@@ -87,6 +89,8 @@ type BackupParams struct {
 	UpgradeSafe bool
 	// MysqlShutdownTimeout defines how long we wait during MySQL shutdown if that is part of the backup process.
 	MysqlShutdownTimeout time.Duration
+	// BackupEngine allows us to override which backup engine should be used for a request
+	BackupEngine *string
 }
 
 func (b *BackupParams) Copy() BackupParams {
@@ -143,6 +147,8 @@ type RestoreParams struct {
 	Stats backupstats.Stats
 	// MysqlShutdownTimeout defines how long we wait during MySQL shutdown if that is part of the backup process.
 	MysqlShutdownTimeout time.Duration
+	// IgnoredBackupEngines will hold any backup engines we should ignore when restoring backups
+	IgnoredBackupEngines []string
 }
 
 func (p *RestoreParams) Copy() RestoreParams {
@@ -213,8 +219,14 @@ func registerBackupEngineFlags(fs *pflag.FlagSet) {
 // a particular backup by calling GetRestoreEngine().
 //
 // This must only be called after flags have been parsed.
-func GetBackupEngine() (BackupEngine, error) {
-	name := backupEngineImplementation
+func GetBackupEngine(backupEngine *string) (BackupEngine, error) {
+	var name string
+	if backupEngine == nil {
+		name = backupEngineImplementation
+	} else {
+		name = *backupEngine
+	}
+
 	be, ok := BackupRestoreEngineMap[name]
 	if !ok {
 		return nil, vterrors.Errorf(vtrpc.Code_NOT_FOUND, "unknown BackupEngine implementation %q", name)
@@ -514,6 +526,12 @@ func FindBackupToRestore(ctx context.Context, params RestoreParams, bhs []backup
 			params.Logger.Warningf("Possibly incomplete backup %v in directory %v on BackupStorage: can't read MANIFEST: %v)", bh.Name(), backupDir, err)
 			continue
 		}
+
+		if slices.Contains(params.IgnoredBackupEngines, bm.BackupMethod) {
+			params.Logger.Warningf("Ignoring backup %v because it is using %q backup engine", bh.Name(), bm.BackupMethod)
+			continue
+		}
+
 		// the manifest is valid
 		manifests[i] = bm // manifests's order is insignificant, it will be sorted later on
 		manifestHandleMap.Map(bm, bh)

go/vt/mysqlctl/builtinbackupengine.go

@@ -1184,6 +1184,8 @@ func (be *BuiltinBackupEngine) ShouldDrainForBackup(req *tabletmanagerdatapb.Bac
 	return true
 }
 
+func (be *BuiltinBackupEngine) Name() string { return builtinBackupEngineName }
+
 func getPrimaryPosition(ctx context.Context, tmc tmclient.TabletManagerClient, ts *topo.Server, keyspace, shard string) (replication.Position, error) {
 	si, err := ts.GetShard(ctx, keyspace, shard)
 	if err != nil {

go/vt/mysqlctl/fakebackupengine.go

@@ -24,6 +24,8 @@ import (
 	tabletmanagerdatapb "vitess.io/vitess/go/vt/proto/tabletmanagerdata"
 )
 
+const fakeBackupEngineName = "fake"
+
 type FakeBackupEngine struct {
 	ExecuteBackupCalls         []FakeBackupEngineExecuteBackupCall
 	ExecuteBackupDuration      time.Duration
@@ -91,3 +93,5 @@ func (be *FakeBackupEngine) ShouldDrainForBackup(req *tabletmanagerdatapb.Backup
 	be.ShouldDrainForBackupCalls = be.ShouldDrainForBackupCalls + 1
 	return be.ShouldDrainForBackupReturn
 }
+
+func (be *FakeBackupEngine) Name() string { return fakeBackupEngineName }

go/vt/mysqlctl/xtrabackupengine.go

@@ -956,6 +956,8 @@ func (be *XtrabackupEngine) ShouldDrainForBackup(req *tabletmanagerdatapb.Backup
 	return false
 }
 
+func (be *XtrabackupEngine) Name() string { return xtrabackupEngineName }
+
 func init() {
 	BackupRestoreEngineMap[xtrabackupEngineName] = &XtrabackupEngine{}
 }