restore: Run PITR through mysqlctl

go/cmd/vtbackup/vtbackup.go

@@ -373,7 +373,10 @@ func takeBackup(ctx context.Context, topoServer *topo.Server, backupStorage back
 	}
 	durationByPhase.Set("RestoreLastBackup", int64(time.Since(restoreAt).Seconds()))
 
-	// Disable redo logging (if we can) before we start replication.
+	// As of MySQL 8.0.21, you can disable redo logging using the ALTER INSTANCE
+	// DISABLE INNODB REDO_LOG statement. This functionality is intended for
+	// loading data into a new MySQL instance. Disabling redo logging speeds up
+	// data loading by avoiding redo log writes and doublewrite buffering.
 	disabledRedoLog := false
 	if disableRedoLog {
 		if err := mysqld.DisableRedoLog(ctx); err != nil {

go/vt/mysqlctl/backup.go

@@ -388,7 +388,7 @@ func Restore(ctx context.Context, params RestoreParams) (*BackupManifest, error)
 	}
 
 	params.Logger.Infof("Restore: running mysql_upgrade")
-	if err := params.Mysqld.RunMysqlUpgrade(); err != nil {
+	if err := params.Mysqld.RunMysqlUpgrade(ctx); err != nil {
 		return nil, vterrors.Wrap(err, "mysql_upgrade failed")
 	}
 

go/vt/mysqlctl/builtinbackupengine.go

@@ -847,7 +847,7 @@ func (be *BuiltinBackupEngine) executeRestoreIncrementalBackup(ctx context.Conte
 		if err != nil {
 			return vterrors.Wrap(err, "failed to restore file")
 		}
-		if err := mysqld.applyBinlogFile(binlogFile, params.RestoreToPos.GTIDSet); err != nil {
+		if err := mysqld.ApplyBinlogFile(ctx, binlogFile, params.RestoreToPos); err != nil {
 			return vterrors.Wrap(err, "failed to extract binlog file")
 		}
 		defer os.Remove(binlogFile)

go/vt/mysqlctl/capabilityset.go

@@ -36,6 +36,7 @@ type capabilitySet struct {
 }
 
 func newCapabilitySet(f MySQLFlavor, v ServerVersion) (c capabilitySet) {
+	noSocketFile()
 	c.flavor = f
 	c.version = v
 	return
@@ -51,18 +52,6 @@ func (c *capabilitySet) hasMaria104InstallDb() bool {
 	return c.isMariaDB() && c.version.atLeast(ServerVersion{Major: 10, Minor: 4, Patch: 0})
 }
 
-// hasDisableRedoLog tells you if the version of MySQL in use can disable redo logging.
-//
-// As of MySQL 8.0.21, you can disable redo logging using the ALTER INSTANCE
-// DISABLE INNODB REDO_LOG statement. This functionality is intended for
-// loading data into a new MySQL instance. Disabling redo logging speeds up
-// data loading by avoiding redo log writes and doublewrite buffering.
-//
-//	https://dev.mysql.com/doc/refman/8.0/en/innodb-redo-log.html#innodb-disable-redo-logging
-func (c *capabilitySet) hasDisableRedoLog() bool {
-	return c.isMySQLLike() && c.version.atLeast(ServerVersion{Major: 8, Minor: 0, Patch: 21})
-}
-
 // IsMySQLLike tests if the server is either MySQL
 // or Percona Server. At least currently, Vitess doesn't
 // make use of any specific Percona Server features.

go/vt/mysqlctl/fakemysqldaemon.go

@@ -221,7 +221,12 @@ func (fmd *FakeMysqlDaemon) Shutdown(ctx context.Context, cnf *Mycnf, waitForMys
 }
 
 // RunMysqlUpgrade is part of the MysqlDaemon interface
-func (fmd *FakeMysqlDaemon) RunMysqlUpgrade() error {
+func (fmd *FakeMysqlDaemon) RunMysqlUpgrade(ctx context.Context) error {
+	return nil
+}
+
+// ApplyBinlogFile is part of the MysqlDaemon interface
+func (fmd *FakeMysqlDaemon) ApplyBinlogFile(ctx context.Context, binlogFile string, restorePos mysql.Position) error {
 	return nil
 }
 
@@ -668,12 +673,12 @@ func (fmd *FakeMysqlDaemon) SemiSyncReplicationStatus() (bool, error) {
 	return fmd.SemiSyncReplicaEnabled, nil
 }
 
-// GetVersionString is part of the MysqlDeamon interface.
-func (fmd *FakeMysqlDaemon) GetVersionString() string {
+// GetVersionString is part of the MysqlDaemon interface.
+func (fmd *FakeMysqlDaemon) GetVersionString(ctx context.Context) string {
 	return ""
 }
 
-// GetVersionComment is part of the MysqlDeamon interface.
+// GetVersionComment is part of the MysqlDaemon interface.
 func (fmd *FakeMysqlDaemon) GetVersionComment(ctx context.Context) string {
 	return ""
 }

go/vt/mysqlctl/grpcmysqlctlclient/client.go

@@ -19,6 +19,7 @@ limitations under the License.
 package grpcmysqlctlclient
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"time"
@@ -27,8 +28,6 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
-	"context"
-
 	"vitess.io/vitess/go/vt/grpcclient"
 	"vitess.io/vitess/go/vt/mysqlctl/mysqlctlclient"
 
@@ -84,6 +83,18 @@ func (c *client) RunMysqlUpgrade(ctx context.Context) error {
 	})
 }
 
+// ApplyBinlogFile is part of the MysqlctlClient interface.
+func (c *client) ApplyBinlogFile(ctx context.Context, binlogFileName, binlogRestorePosition string) error {
+	req := &mysqlctlpb.ApplyBinlogFileRequest{
+		BinlogFileName:        binlogFileName,
+		BinlogRestorePosition: binlogRestorePosition,
+	}
+	return c.withRetry(ctx, func() error {
+		_, err := c.c.ApplyBinlogFile(ctx, req)
+		return err
+	})
+}
+
 // ReinitConfig is part of the MysqlctlClient interface.
 func (c *client) ReinitConfig(ctx context.Context) error {
 	return c.withRetry(ctx, func() error {

go/vt/mysqlctl/grpcmysqlctlserver/server.go

@@ -21,12 +21,12 @@ side of the remote execution of mysqlctl commands.
 package grpcmysqlctlserver
 
 import (
-	"google.golang.org/grpc"
-
 	"context"
 
-	"vitess.io/vitess/go/vt/mysqlctl"
+	"google.golang.org/grpc"
 
+	"vitess.io/vitess/go/mysql"
+	"vitess.io/vitess/go/vt/mysqlctl"
 	mysqlctlpb "vitess.io/vitess/go/vt/proto/mysqlctl"
 )
 
@@ -48,8 +48,17 @@ func (s *server) Shutdown(ctx context.Context, request *mysqlctlpb.ShutdownReque
 }
 
 // RunMysqlUpgrade implements the server side of the MysqlctlClient interface.
-func (s *server) RunMysqlUpgrade(ctx context.Context, request *mysqlctlpb.RunMysqlUpgradeRequest) (*mysqlctlpb.RunMysqlUpgradeResponse, error) {
-	return &mysqlctlpb.RunMysqlUpgradeResponse{}, s.mysqld.RunMysqlUpgrade()
+func (s *server) RunMysqlUpgrade(ctx context.Context, _ *mysqlctlpb.RunMysqlUpgradeRequest) (*mysqlctlpb.RunMysqlUpgradeResponse, error) {
+	return &mysqlctlpb.RunMysqlUpgradeResponse{}, s.mysqld.RunMysqlUpgrade(ctx)
+}
+
+// RunMysqlUpgrade implements the server side of the MysqlctlClient interface.
+func (s *server) ApplyBinlogFile(ctx context.Context, request *mysqlctlpb.ApplyBinlogFileRequest) (*mysqlctlpb.ApplyBinlogFileResponse, error) {
+	pos, err := mysql.DecodePosition(request.BinlogRestorePosition)
+	if err != nil {
+		return nil, err
+	}
+	return &mysqlctlpb.ApplyBinlogFileResponse{}, s.mysqld.ApplyBinlogFile(ctx, request.BinlogFileName, pos)
 }
 
 // ReinitConfig implements the server side of the MysqlctlClient interface.

go/vt/mysqlctl/mysql_daemon.go

@@ -33,7 +33,8 @@ type MysqlDaemon interface {
 	// methods related to mysql running or not
 	Start(ctx context.Context, cnf *Mycnf, mysqldArgs ...string) error
 	Shutdown(ctx context.Context, cnf *Mycnf, waitForMysqld bool) error
-	RunMysqlUpgrade() error
+	RunMysqlUpgrade(ctx context.Context) error
+	ApplyBinlogFile(ctx context.Context, binlogFile string, restorePos mysql.Position) error
 	ReinitConfig(ctx context.Context, cnf *Mycnf) error
 	Wait(ctx context.Context, cnf *Mycnf) error
 
@@ -102,7 +103,7 @@ type MysqlDaemon interface {
 	GetAllPrivsConnection(ctx context.Context) (*dbconnpool.DBConnection, error)
 
 	// GetVersionString returns the database version as a string
-	GetVersionString() string
+	GetVersionString(ctx context.Context) string
 
 	// GetVersionComment returns the version comment
 	GetVersionComment(ctx context.Context) string

go/vt/mysqlctl/mysqlctlclient/interface.go

@@ -49,6 +49,9 @@ type MysqlctlClient interface {
 	// RunMysqlUpgrade calls Mysqld.RunMysqlUpgrade remotely.
 	RunMysqlUpgrade(ctx context.Context) error
 
+	// ApplyBinlogFile calls Mysqld.ApplyBinlogFile remotely.
+	ApplyBinlogFile(ctx context.Context, binlogFileName, binlogRestorePosition string) error
+
 	// ReinitConfig calls Mysqld.ReinitConfig remotely.
 	ReinitConfig(ctx context.Context) error
 

go/vt/mysqlctl/mysqld.go

@@ -157,6 +157,12 @@ func NewMysqld(dbcfgs *dbconfigs.DBConfigs) *Mysqld {
 		log.Info("mysqld is unmanaged or remote. Skipping flavor detection")
 		return result
 	}
+
+	if socketFile != "" {
+		log.Info("mysqld is remote. Skipping flavor detection")
+		return result
+	}
+
 	version, getErr := GetVersionString()
 	f, v, err := ParseVersionString(version)
 
@@ -228,6 +234,7 @@ func GetVersionFromEnv() (flavor MySQLFlavor, ver ServerVersion, err error) {
 
 // GetVersionString runs mysqld --version and returns its output as a string
 func GetVersionString() (string, error) {
+	noSocketFile()
 	mysqlRoot, err := vtenv.VtMysqlRoot()
 	if err != nil {
 		return "", err
@@ -277,7 +284,7 @@ func ParseVersionString(version string) (flavor MySQLFlavor, ver ServerVersion,
 // RunMysqlUpgrade will run the mysql_upgrade program on the current
 // install.  Will be called only when mysqld is running with no
 // network and no grant tables.
-func (mysqld *Mysqld) RunMysqlUpgrade() error {
+func (mysqld *Mysqld) RunMysqlUpgrade(ctx context.Context) error {
 	// Execute as remote action on mysqlctld if requested.
 	if socketFile != "" {
 		log.Infof("executing Mysqld.RunMysqlUpgrade() remotely via mysqlctld server: %v", socketFile)
@@ -286,7 +293,7 @@ func (mysqld *Mysqld) RunMysqlUpgrade() error {
 			return fmt.Errorf("can't dial mysqlctld: %v", err)
 		}
 		defer client.Close()
-		return client.RunMysqlUpgrade(context.TODO())
+		return client.RunMysqlUpgrade(ctx)
 	}
 
 	if mysqld.capabilities.hasMySQLUpgradeInServer() {
@@ -635,6 +642,7 @@ func execCmd(name string, args, env []string, dir string, input io.Reader) (cmd
 // binaryPath does a limited path lookup for a command,
 // searching only within sbin and bin in the given root.
 func binaryPath(root, binary string) (string, error) {
+	noSocketFile()
 	subdirs := []string{"sbin", "bin", "libexec", "scripts"}
 	for _, subdir := range subdirs {
 		binPath := path.Join(root, subdir, binary)
@@ -1176,9 +1184,20 @@ func buildLdPaths() ([]string, error) {
 	return ldPaths, nil
 }
 
-// GetVersionString is part of the MysqlDaemon interface.
-func (mysqld *Mysqld) GetVersionString() string {
-	return fmt.Sprintf("%d.%d.%d", mysqld.capabilities.version.Major, mysqld.capabilities.version.Minor, mysqld.capabilities.version.Patch)
+// GetVersionString is part of the MysqlExecutor interface.
+func (mysqld *Mysqld) GetVersionString(ctx context.Context) string {
+	qr, err := mysqld.FetchSuperQuery(ctx, "select @@global.version")
+	if err != nil {
+		log.Errorf("Error fetching MySQL version: %v", err)
+		return ""
+	}
+	if len(qr.Rows) != 1 {
+		log.Errorf("Unexpected number of rows: %v", qr.Rows)
+		return ""
+	}
+	res := qr.Named().Row()
+	version, _ := res.ToString("@@global.version")
+	return version
 }
 
 // GetVersionComment gets the version comment.
@@ -1195,9 +1214,18 @@ func (mysqld *Mysqld) GetVersionComment(ctx context.Context) string {
 	return versionComment
 }
 
-// applyBinlogFile extracts a binary log file and applies it to MySQL. It is the equivalent of:
+// ApplyBinlogFile extracts a binary log file and applies it to MySQL. It is the equivalent of:
 // $ mysqlbinlog --include-gtids binlog.file | mysql
-func (mysqld *Mysqld) applyBinlogFile(binlogFile string, includeGTIDs mysql.GTIDSet) error {
+func (mysqld *Mysqld) ApplyBinlogFile(ctx context.Context, binlogFile string, restorePos mysql.Position) error {
+	if socketFile != "" {
+		log.Infof("executing Mysqld.ApplyBinlogFile() remotely via mysqlctld server: %v", socketFile)
+		client, err := mysqlctlclient.New("unix", socketFile)
+		if err != nil {
+			return fmt.Errorf("can't dial mysqlctld: %v", err)
+		}
+		defer client.Close()
+		return client.ApplyBinlogFile(ctx, binlogFile, mysql.EncodePosition(restorePos))
+	}
 	var pipe io.ReadCloser
 	var mysqlbinlogCmd *exec.Cmd
 	var mysqlCmd *exec.Cmd
@@ -1216,7 +1244,7 @@ func (mysqld *Mysqld) applyBinlogFile(binlogFile string, includeGTIDs mysql.GTID
 			return err
 		}
 		args := []string{}
-		if gtids := includeGTIDs.String(); gtids != "" {
+		if gtids := restorePos.GTIDSet.String(); gtids != "" {
 			args = append(args,
 				"--include-gtids",
 				gtids,
@@ -1228,7 +1256,7 @@ func (mysqld *Mysqld) applyBinlogFile(binlogFile string, includeGTIDs mysql.GTID
 		mysqlbinlogCmd = exec.Command(name, args...)
 		mysqlbinlogCmd.Dir = dir
 		mysqlbinlogCmd.Env = env
-		log.Infof("applyBinlogFile: running %#v", mysqlbinlogCmd)
+		log.Infof("ApplyBinlogFile: running %#v", mysqlbinlogCmd)
 		pipe, err = mysqlbinlogCmd.StdoutPipe() // to be piped into mysql
 		if err != nil {
 			return err
@@ -1255,21 +1283,21 @@ func (mysqld *Mysqld) applyBinlogFile(binlogFile string, includeGTIDs mysql.GTID
 		// We disable super_read_only, in case it is in the default MySQL startup
 		// parameters.  We do it blindly, since this will fail on MariaDB, which doesn't
 		// have super_read_only This is safe, since we're restarting MySQL after the restore anyway
-		log.Infof("applyBinlogFile: disabling super_read_only")
+		log.Infof("ApplyBinlogFile: disabling super_read_only")
 		resetFunc, err := mysqld.SetSuperReadOnly(false)
 		if err != nil {
 			if sqlErr, ok := err.(*mysql.SQLError); ok && sqlErr.Number() == mysql.ERUnknownSystemVariable {
-				log.Warningf("applyBinlogFile: server does not know about super_read_only, continuing anyway...")
+				log.Warningf("ApplyBinlogFile: server does not know about super_read_only, continuing anyway...")
 			} else {
-				log.Errorf("applyBinlogFile: unexpected error while trying to set super_read_only: %v", err)
+				log.Errorf("ApplyBinlogFile: unexpected error while trying to set super_read_only: %v", err)
 				return err
 			}
 		}
 		if resetFunc != nil {
 			defer func() {
 				err := resetFunc()
 				if err != nil {
-					log.Error("Not able to set super_read_only to its original value during applyBinlogFile.")
+					log.Error("Not able to set super_read_only to its original value during ApplyBinlogFile.")
 				}
 			}()
 		}
@@ -1295,3 +1323,12 @@ func (mysqld *Mysqld) applyBinlogFile(binlogFile string, includeGTIDs mysql.GTID
 	}
 	return nil
 }
+
+// noSocketFile panics if socketFile is set. This is to prevent
+// incorrect use of settings not supported when we're running
+// remote through mysqlctl.
+func noSocketFile() {
+	if socketFile != "" {
+		panic("Running remotely through mysqlctl, socketFile must not be set")
+	}
+}

go/vt/mysqlctl/redo_log.go

@@ -21,10 +21,6 @@ import (
 	"fmt"
 )
 
-func (mysqld *Mysqld) BinaryHasDisableRedoLog() bool {
-	return mysqld.capabilities.hasDisableRedoLog()
-}
-
 func (mysqld *Mysqld) DisableRedoLog(ctx context.Context) error {
 	return mysqld.ExecuteSuperQuery(ctx, "ALTER INSTANCE DISABLE INNODB REDO_LOG")
 }