backport logical backups

go/flags/endtoend/vtbackup.txt

@@ -120,6 +120,12 @@ Usage of vtbackup:
       --mycnf_slow_log_path string                      mysql slow query log path
       --mycnf_socket_file string                        mysql socket file
       --mycnf_tmp_dir string                            mysql tmp directory
+      --mysql-shell-backup-location string              location where the backup will be stored
+      --mysql-shell-dump-flags string                   flags to pass to mysql shell dump utility. This should be a JSON string and will be saved in the MANIFEST (default "{\"threads\": 4}")
+      --mysql-shell-flags string                        execution flags to pass to mysqlsh binary to be used during dump/load (default "--defaults-file=/dev/null --js -h localhost")
+      --mysql-shell-load-flags string                   flags to pass to mysql shell load utility. This should be a JSON string (default "{\"threads\": 4, \"loadUsers\": true, \"updateGtidSet\": \"replace\", \"skipBinlog\": true, \"progressFile\": \"\"}")
+      --mysql-shell-should-drain                        decide if we should drain while taking a backup or continue to serving traffic
+      --mysql-shell-speedup-restore                     speed up restore by disabling redo logging and double write buffer during the restore process
       --mysql_port int                                  mysql port (default 3306)
       --mysql_server_version string                     MySQL server version to advertise.
       --mysql_socket string                             path to the mysql socket

go/flags/endtoend/vttablet.txt

@@ -194,6 +194,12 @@ Usage of vttablet:
       --mycnf_slow_log_path string                                                mysql slow query log path
       --mycnf_socket_file string                                                  mysql socket file
       --mycnf_tmp_dir string                                                      mysql tmp directory
+      --mysql-shell-backup-location string                                        location where the backup will be stored
+      --mysql-shell-dump-flags string                                             flags to pass to mysql shell dump utility. This should be a JSON string and will be saved in the MANIFEST (default "{\"threads\": 4}")
+      --mysql-shell-flags string                                                  execution flags to pass to mysqlsh binary to be used during dump/load (default "--defaults-file=/dev/null --js -h localhost")
+      --mysql-shell-load-flags string                                             flags to pass to mysql shell load utility. This should be a JSON string (default "{\"threads\": 4, \"loadUsers\": true, \"updateGtidSet\": \"replace\", \"skipBinlog\": true, \"progressFile\": \"\"}")
+      --mysql-shell-should-drain                                                  decide if we should drain while taking a backup or continue to serving traffic
+      --mysql-shell-speedup-restore                                               speed up restore by disabling redo logging and double write buffer during the restore process
       --mysql_server_version string                                               MySQL server version to advertise.
       --mysqlctl_mycnf_template string                                            template file to use for generating the my.cnf file during server init
       --mysqlctl_socket string                                                    socket file to use for remote mysqlctl actions (empty for local actions)

go/flags/endtoend/vttestserver.txt

@@ -72,6 +72,12 @@ Usage of vttestserver:
       --logtostderr                                                      log to standard error instead of files
       --max_table_shard_size int                                         The maximum number of initial rows in a table shard. Ignored if--initialize_with_random_data is false. The actual number is chosen randomly (default 10000)
       --min_table_shard_size int                                         The minimum number of initial rows in a table shard. Ignored if--initialize_with_random_data is false. The actual number is chosen randomly. (default 1000)
+      --mysql-shell-backup-location string                               location where the backup will be stored
+      --mysql-shell-dump-flags string                                    flags to pass to mysql shell dump utility. This should be a JSON string and will be saved in the MANIFEST (default "{\"threads\": 4}")
+      --mysql-shell-flags string                                         execution flags to pass to mysqlsh binary to be used during dump/load (default "--defaults-file=/dev/null --js -h localhost")
+      --mysql-shell-load-flags string                                    flags to pass to mysql shell load utility. This should be a JSON string (default "{\"threads\": 4, \"loadUsers\": true, \"updateGtidSet\": \"replace\", \"skipBinlog\": true, \"progressFile\": \"\"}")
+      --mysql-shell-should-drain                                         decide if we should drain while taking a backup or continue to serving traffic
+      --mysql-shell-speedup-restore                                      speed up restore by disabling redo logging and double write buffer during the restore process
       --mysql_bind_host string                                           which host to bind vtgate mysql listener to (default "localhost")
       --mysql_only                                                       If this flag is set only mysql is initialized. The rest of the vitess components are not started. Also, the output specifies the mysql unix socket instead of the vtgate port.
       --mysql_server_version string                                      MySQL server version to advertise.

go/ioutil/writer.go

@@ -0,0 +1,39 @@
+/*
+Copyright 2022 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+/*
+MeteredWriteCloser and MeteredWriter are respectively, time-and-byte-tracking
+wrappers around WriteCloser and Writer.
+*/
+
+package ioutil
+
+import (
+	"bytes"
+)
+
+// BytesBufferWriter implements io.WriteCloser using an in-memory buffer.
+type BytesBufferWriter struct {
+	*bytes.Buffer
+}
+
+func (m BytesBufferWriter) Close() error {
+	return nil
+}
+
+func NewBytesBufferWriter() BytesBufferWriter {
+	return BytesBufferWriter{bytes.NewBuffer(nil)}
+}

go/test/fuzzing/tablet_manager_fuzzer.go

@@ -22,7 +22,7 @@ import (
 	"vitess.io/vitess/go/mysql/fakesqldb"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/dbconfigs"
-	"vitess.io/vitess/go/vt/mysqlctl/fakemysqldaemon"
+	"vitess.io/vitess/go/vt/mysqlctl"
 	"vitess.io/vitess/go/vt/vttablet/tabletmanager"
 	"vitess.io/vitess/go/vt/vttablet/tabletservermock"
 
@@ -42,7 +42,7 @@ func FuzzTabletManagerExecuteFetchAsDba(data []byte) int {
 	cp := mysql.ConnParams{}
 	db := fakesqldb.New(t)
 	db.AddQueryPattern(".*", &sqltypes.Result{})
-	daemon := fakemysqldaemon.NewFakeMysqlDaemon(db)
+	daemon := mysqlctl.NewFakeMysqlDaemon(db)
 
 	dbName := "dbname"
 	tm := &tabletmanager.TabletManager{

go/vt/mysqlctl/backup.go

@@ -17,16 +17,18 @@ limitations under the License.
 package mysqlctl
 
 import (
+	"bufio"
 	"errors"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
-	"strconv"
 	"strings"
 	"time"
 
 	"github.com/spf13/pflag"
 
+	"vitess.io/vitess/go/vt/logutil"
 	"vitess.io/vitess/go/vt/servenv"
 
 	"context"
@@ -352,21 +354,6 @@ func Restore(ctx context.Context, params RestoreParams) (*BackupManifest, error)
 		return nil, err
 	}
 
-	// We disable super_read_only, in case it is in the default MySQL startup
-	// parameters and will be blocking the writes we need to do in
-	// PopulateMetadataTables().  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
-	params.Logger.Infof("Restore: disabling super_read_only")
-	if err := params.Mysqld.SetSuperReadOnly(false); err != nil {
-		if strings.Contains(err.Error(), strconv.Itoa(mysql.ERUnknownSystemVariable)) {
-			params.Logger.Warningf("Restore: server does not know about super_read_only, continuing anyway...")
-		} else {
-			params.Logger.Errorf("Restore: unexpected error while trying to set super_read_only: %v", err)
-			return nil, err
-		}
-	}
-
 	params.Logger.Infof("Restore: running mysql_upgrade")
 	if err := params.Mysqld.RunMysqlUpgrade(); err != nil {
 		return nil, vterrors.Wrap(err, "mysql_upgrade failed")
@@ -403,3 +390,24 @@ func Restore(ctx context.Context, params RestoreParams) (*BackupManifest, error)
 	restoreDuration.Set(int64(time.Since(startTs).Seconds()))
 	return manifest, nil
 }
+
+// scanLinesToLogger scans full lines from the given Reader and sends them to
+// the given Logger until EOF.
+func scanLinesToLogger(prefix string, reader io.Reader, logger logutil.Logger, doneFunc func()) {
+	defer doneFunc()
+
+	scanner := bufio.NewScanner(reader)
+	for scanner.Scan() {
+		line := scanner.Text()
+		logger.Infof("%s: %s", prefix, line)
+	}
+	if err := scanner.Err(); err != nil {
+		// This is usually run in a background goroutine, so there's no point
+		// returning an error. Just log it.
+		logger.Warningf("error scanning lines from %s: %v", prefix, err)
+	}
+}
+
+func FormatRFC3339(t time.Time) string {
+	return t.Format(time.RFC3339)
+}

go/vt/mysqlctl/backup_test.go

@@ -17,13 +17,19 @@ limitations under the License.
 package mysqlctl
 
 import (
+	"fmt"
+	"io"
 	"os"
 	"path"
 	"reflect"
 	"sort"
+	"sync"
 	"testing"
 
+	"github.com/stretchr/testify/require"
+
 	"vitess.io/vitess/go/mysql"
+	"vitess.io/vitess/go/vt/logutil"
 )
 
 func TestFindFilesToBackupWithoutRedoLog(t *testing.T) {
@@ -214,3 +220,26 @@ type forTest []FileEntry
 func (f forTest) Len() int           { return len(f) }
 func (f forTest) Swap(i, j int)      { f[i], f[j] = f[j], f[i] }
 func (f forTest) Less(i, j int) bool { return f[i].Base+f[i].Name < f[j].Base+f[j].Name }
+
+func TestScanLinesToLogger(t *testing.T) {
+	reader, writer := io.Pipe()
+	logger := logutil.NewMemoryLogger()
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go scanLinesToLogger("test", reader, logger, wg.Done)
+
+	for i := range [100]int{} {
+		_, err := writer.Write([]byte(fmt.Sprintf("foobar %d\n", i)))
+		require.NoError(t, err)
+	}
+
+	writer.Close()
+	wg.Wait()
+
+	require.Equal(t, 100, len(logger.Events))
+
+	for i, event := range logger.Events {
+		require.Equal(t, fmt.Sprintf("test: foobar %d", i), event.Value)
+	}
+}

go/vt/mysqlctl/backupengine.go

@@ -99,6 +99,7 @@ type RestoreParams struct {
 // Returns the manifest of a backup if successful, otherwise returns an error
 type RestoreEngine interface {
 	ExecuteRestore(ctx context.Context, params RestoreParams, bh backupstorage.BackupHandle) (*BackupManifest, error)
+	ShouldStartMySQLAfterRestore() bool
 }
 
 // BackupRestoreEngine is a combination of BackupEngine and RestoreEngine.

go/vt/mysqlctl/builtinbackupengine.go

@@ -770,6 +770,11 @@ func (be *BuiltinBackupEngine) ShouldDrainForBackup() bool {
 	return true
 }
 
+// ShouldStartMySQLAfterRestore signifies if this backup engine needs to restart MySQL once the restore is completed.
+func (be *BuiltinBackupEngine) ShouldStartMySQLAfterRestore() bool {
+	return true
+}
+
 func getPrimaryPosition(ctx context.Context, tmc tmclient.TabletManagerClient, ts *topo.Server, keyspace, shard string) (mysql.Position, error) {
 	si, err := ts.GetShard(ctx, keyspace, shard)
 	if err != nil {

go/vt/mysqlctl/builtinbackupengine_test.go

@@ -19,7 +19,6 @@ import (
 	"vitess.io/vitess/go/mysql/fakesqldb"
 	"vitess.io/vitess/go/vt/logutil"
 	"vitess.io/vitess/go/vt/mysqlctl"
-	"vitess.io/vitess/go/vt/mysqlctl/fakemysqldaemon"
 	"vitess.io/vitess/go/vt/mysqlctl/filebackupstorage"
 	"vitess.io/vitess/go/vt/proto/topodata"
 	"vitess.io/vitess/go/vt/proto/vttime"
@@ -104,7 +103,7 @@ func TestExecuteBackup(t *testing.T) {
 
 	// Spin up a fake daemon to be used in backups. It needs to be allowed to receive:
 	//  "STOP SLAVE", "START SLAVE", in that order.
-	mysqld := fakemysqldaemon.NewFakeMysqlDaemon(fakesqldb.New(t))
+	mysqld := mysqlctl.NewFakeMysqlDaemon(fakesqldb.New(t))
 	mysqld.ExpectedExecuteSuperQueryList = []string{"STOP SLAVE", "START SLAVE"}
 	// mysqld.ShutdownTime = time.Minute
 

go/vt/mysqlctl/fakebackupengine.go

@@ -0,0 +1,92 @@
+/*
+Copyright 2022 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package mysqlctl
+
+import (
+	"context"
+	"time"
+
+	"vitess.io/vitess/go/vt/mysqlctl/backupstorage"
+)
+
+type FakeBackupEngine struct {
+	ExecuteBackupCalls         []FakeBackupEngineExecuteBackupCall
+	ExecuteBackupDuration      time.Duration
+	ExecuteBackupReturn        FakeBackupEngineExecuteBackupReturn
+	ExecuteRestoreCalls        []FakeBackupEngineExecuteRestoreCall
+	ExecuteRestoreDuration     time.Duration
+	ExecuteRestoreReturn       FakeBackupEngineExecuteRestoreReturn
+	ShouldDrainForBackupCalls  int
+	ShouldDrainForBackupReturn bool
+}
+
+type FakeBackupEngineExecuteBackupCall struct {
+	BackupParams BackupParams
+	BackupHandle backupstorage.BackupHandle
+}
+
+type FakeBackupEngineExecuteBackupReturn struct {
+	Ok  bool
+	Err error
+}
+
+type FakeBackupEngineExecuteRestoreCall struct {
+	BackupHandle  backupstorage.BackupHandle
+	RestoreParams RestoreParams
+}
+
+type FakeBackupEngineExecuteRestoreReturn struct {
+	Manifest *BackupManifest
+	Err      error
+}
+
+func (be *FakeBackupEngine) ExecuteBackup(
+	ctx context.Context,
+	params BackupParams,
+	bh backupstorage.BackupHandle,
+) (bool, error) {
+	be.ExecuteBackupCalls = append(be.ExecuteBackupCalls, FakeBackupEngineExecuteBackupCall{params, bh})
+
+	if be.ExecuteBackupDuration > 0 {
+		time.Sleep(be.ExecuteBackupDuration)
+	}
+
+	return be.ExecuteBackupReturn.Ok, be.ExecuteBackupReturn.Err
+}
+
+func (be *FakeBackupEngine) ExecuteRestore(
+	ctx context.Context, params RestoreParams,
+	bh backupstorage.BackupHandle,
+) (*BackupManifest, error) {
+	be.ExecuteRestoreCalls = append(be.ExecuteRestoreCalls, FakeBackupEngineExecuteRestoreCall{bh, params})
+
+	// mark restore as in progress
+	if err := createStateFile(params.Cnf); err != nil {
+		return nil, err
+	}
+
+	if be.ExecuteRestoreDuration > 0 {
+		time.Sleep(be.ExecuteRestoreDuration)
+	}
+
+	return be.ExecuteRestoreReturn.Manifest, be.ExecuteRestoreReturn.Err
+}
+
+func (be *FakeBackupEngine) ShouldDrainForBackup() bool {
+	be.ShouldDrainForBackupCalls = be.ShouldDrainForBackupCalls + 1
+	return be.ShouldDrainForBackupReturn
+}