Improve MySQL restore performance

cmd/fdb/backup_push.go

@@ -25,7 +25,7 @@ var backupPushCmd = &cobra.Command{
 
 		uploader, err := internal.ConfigureUploader()
 		tracelog.ErrorLogger.FatalOnError(err)
-		uploader.UploadingFolder = uploader.UploadingFolder.GetSubFolder(utility.BaseBackupPath)
+		uploader.ChangeDirectory(utility.BaseBackupPath)
 
 		backupCmd, err := internal.GetCommandSetting(internal.NameStreamCreateCmd)
 		tracelog.ErrorLogger.FatalOnError(err)

cmd/mongo/backup_fetch.go

@@ -8,7 +8,6 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/wal-g/tracelog"
 	"github.com/wal-g/wal-g/internal"
-	"github.com/wal-g/wal-g/internal/databases/mongo"
 	"github.com/wal-g/wal-g/utility"
 )
 
@@ -32,8 +31,10 @@ var backupFetchCmd = &cobra.Command{
 		restoreCmd.Stdout = os.Stdout
 		restoreCmd.Stderr = os.Stderr
 
-		err = mongo.HandleBackupFetch(ctx, folder, args[0], restoreCmd)
+		backupSelector, err := internal.NewBackupNameSelector(args[0], true)
 		tracelog.ErrorLogger.FatalOnError(err)
+
+		internal.HandleBackupFetch(folder, backupSelector, internal.GetBackupToCommandFetcher(restoreCmd))
 	},
 }
 

cmd/mongo/backup_push.go

@@ -41,15 +41,15 @@ var backupPushCmd = &cobra.Command{
 		mongoClient, err := client.NewMongoClient(ctx, mongodbURL)
 		tracelog.ErrorLogger.FatalOnError(err)
 
-		uplProvider, err := internal.ConfigureUploader()
+		uplProvider, err := internal.ConfigureSplitUploader()
 		tracelog.ErrorLogger.FatalOnError(err)
-		uplProvider.UploadingFolder = uplProvider.UploadingFolder.GetSubFolder(utility.BaseBackupPath)
+		uplProvider.ChangeDirectory(utility.BaseBackupPath)
 
 		backupCmd, err := internal.GetCommandSettingContext(ctx, internal.NameStreamCreateCmd)
 		tracelog.ErrorLogger.FatalOnError(err)
 		backupCmd.Stderr = os.Stderr
 		uploader := archive.NewStorageUploader(uplProvider)
-		metaConstructor := archive.NewBackupMongoMetaConstructor(ctx, mongoClient, uplProvider.UploadingFolder, permanent)
+		metaConstructor := archive.NewBackupMongoMetaConstructor(ctx, mongoClient, uplProvider.Folder(), permanent)
 
 		err = mongo.HandleBackupPush(uploader, metaConstructor, backupCmd)
 		tracelog.ErrorLogger.FatalfOnError("Backup creation failed: %v", err)

cmd/mongo/oplog_push.go

@@ -58,7 +58,7 @@ func runOplogPush(ctx context.Context, pushArgs oplogPushRunArgs, statsArgs oplo
 	if err != nil {
 		return err
 	}
-	uplProvider.UploadingFolder = uplProvider.UploadingFolder.GetSubFolder(models.OplogArchBasePath)
+	uplProvider.ChangeDirectory(models.OplogArchBasePath)
 	uploader := archive.NewStorageUploader(uplProvider)
 
 	// set up mongodb client and oplog fetcher

cmd/mysql/backup_push.go

@@ -6,6 +6,7 @@ import (
 	"github.com/wal-g/tracelog"
 	"github.com/wal-g/wal-g/internal"
 	"github.com/wal-g/wal-g/internal/databases/mysql"
+	"github.com/wal-g/wal-g/utility"
 )
 
 const (
@@ -27,7 +28,8 @@ var (
 			tracelog.ErrorLogger.FatalOnError(err)
 		},
 		Run: func(cmd *cobra.Command, args []string) {
-			uploader, err := internal.ConfigureUploader()
+			uploader, err := internal.ConfigureSplitUploader()
+			uploader.ChangeDirectory(utility.BaseBackupPath)
 			tracelog.ErrorLogger.FatalOnError(err)
 			backupCmd, err := internal.GetCommandSetting(internal.NameStreamCreateCmd)
 			tracelog.ErrorLogger.FatalOnError(err)

internal/backup_fetch_handler.go

@@ -3,6 +3,7 @@ package internal
 import (
 	"bytes"
 	"fmt"
+	"io"
 	"os/exec"
 
 	"github.com/wal-g/tracelog"
@@ -16,6 +17,8 @@ type BackupNonExistenceError struct {
 	error
 }
 
+type StreamFetcher = func(backup Backup, writeCloser io.WriteCloser) error
+
 func NewBackupNonExistenceError(backupName string) BackupNonExistenceError {
 	return BackupNonExistenceError{errors.Errorf("Backup '%s' does not exist.", backupName)}
 }
@@ -24,15 +27,21 @@ func (err BackupNonExistenceError) Error() string {
 	return fmt.Sprintf(tracelog.GetErrorFormatter(), err.error)
 }
 
-func GetCommandStreamFetcher(cmd *exec.Cmd) func(folder storage.Folder, backup Backup) {
+// GetBackupToCommandFetcher returns function that copies all bytes from backup to cmd's stdin
+func GetBackupToCommandFetcher(cmd *exec.Cmd) func(folder storage.Folder, backup Backup) {
 	return func(folder storage.Folder, backup Backup) {
 		stdin, err := cmd.StdinPipe()
 		tracelog.ErrorLogger.FatalfOnError("Failed to fetch backup: %v\n", err)
 		stderr := &bytes.Buffer{}
 		cmd.Stderr = stderr
 		err = cmd.Start()
 		tracelog.ErrorLogger.FatalfOnError("Failed to start restore command: %v\n", err)
-		err = downloadAndDecompressStream(backup, stdin)
+
+		fetcher, err := GetBackupStreamFetcher(backup)
+		tracelog.ErrorLogger.FatalfOnError("Failed to detect backup format: %v\n", err)
+
+		err = fetcher(backup, stdin)
+
 		cmdErr := cmd.Wait()
 		if err != nil || cmdErr != nil {
 			tracelog.ErrorLogger.Printf("Restore command output:\n%s", stderr.String())
@@ -58,7 +67,7 @@ func StreamBackupToCommandStdin(cmd *exec.Cmd, backup Backup) error {
 	if err != nil {
 		return fmt.Errorf("failed to start command: %v", err)
 	}
-	err = downloadAndDecompressStream(backup, stdin)
+	err = DownloadAndDecompressStream(backup, stdin)
 	if err != nil {
 		return errors.Wrap(err, "failed to download and decompress stream")
 	}
@@ -79,7 +88,7 @@ func HandleBackupFetch(folder storage.Folder,
 	fetcher func(folder storage.Folder, backup Backup)) {
 	backupName, err := targetBackupSelector.Select(folder)
 	tracelog.ErrorLogger.FatalOnError(err)
-	tracelog.DebugLogger.Printf("HandleBackupFetch(%s, folder,)\n", backupName)
+	tracelog.DebugLogger.Printf("HandleBackupFetch(%s)\n", backupName)
 	backup, err := GetBackupByName(backupName, utility.BaseBackupPath, folder)
 	tracelog.ErrorLogger.FatalfOnError("Failed to fetch backup: %v\n", err)
 

internal/backup_util.go

@@ -134,6 +134,14 @@ func SentinelNameFromBackup(backupName string) string {
 	return backupName + utility.SentinelSuffix
 }
 
+func MetadataNameFromBackup(backupName string) string {
+	return backupName + "/" + utility.MetadataFileName
+}
+
+func StreamMetadataNameFromBackup(backupName string) string {
+	return backupName + "/" + utility.StreamMetadataFileName
+}
+
 // UnwrapLatestModifier checks if LATEST is provided instead of backupName
 // if so, replaces it with the name of the latest backup
 func UnwrapLatestModifier(backupName string, folder storage.Folder) (string, error) {

internal/config.go

@@ -77,6 +77,8 @@ const (
 	PrefetchDir                  = "WALG_PREFETCH_DIR"
 	PgReadyRename                = "PG_READY_RENAME"
 	SerializerTypeSetting        = "WALG_SERIALIZER_TYPE"
+	StreamSplitterPartitions     = "WALG_STREAM_SPLITTER_PARTITIONS"
+	StreamSplitterBlockSize      = "WALG_STREAM_SPLITTER_BLOCK_SIZE"
 
 	MongoDBUriSetting               = "MONGODB_URI"
 	MongoDBLastWriteUpdateInterval  = "MONGODB_LAST_WRITE_UPDATE_INTERVAL"
@@ -164,6 +166,11 @@ var (
 		OplogArchiveTimeoutInterval:    "60s",
 		OplogArchiveAfterSize:          "16777216", // 32 << (10 * 2)
 		MongoDBLastWriteUpdateInterval: "3s",
+		StreamSplitterBlockSize:        "1048576",
+	}
+
+	MysqlDefaultSettings = map[string]string{
+		StreamSplitterBlockSize: "1048576",
 	}
 
 	SQLServerDefaultSettings = map[string]string{
@@ -325,6 +332,8 @@ var (
 		OplogPushWaitForBecomePrimary:  true,
 		OplogPushPrimaryCheckInterval:  true,
 		OplogPITRDiscoveryInterval:     true,
+		StreamSplitterBlockSize:        true,
+		StreamSplitterPartitions:       true,
 	}
 
 	SQLServerAllowedSettings = map[string]bool{
@@ -347,6 +356,8 @@ var (
 		MysqlBackupPrepareCmd:      true,
 		MysqlTakeBinlogsFromMaster: true,
 		MysqlCheckGTIDs:            true,
+		StreamSplitterPartitions:   true,
+		StreamSplitterBlockSize:    true,
 	}
 
 	RedisAllowedSettings = map[string]bool{
@@ -377,6 +388,8 @@ func ConfigureSettings(currentType string) {
 			dbSpecificDefaultSettings = PGDefaultSettings
 		case MONGO:
 			dbSpecificDefaultSettings = MongoDefaultSettings
+		case MYSQL:
+			dbSpecificDefaultSettings = MysqlDefaultSettings
 		case SQLSERVER:
 			dbSpecificDefaultSettings = SQLServerDefaultSettings
 		case GP:

internal/configure.go

@@ -261,6 +261,24 @@ func ConfigureUploaderWithoutCompressMethod() (uploader *Uploader, err error) {
 	return uploader, err
 }
 
+func ConfigureSplitUploader() (uploader UploaderProvider, err error) {
+	folder, err := ConfigureFolder()
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to configure folder")
+	}
+
+	compressor, err := ConfigureCompressor()
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to configure compression")
+	}
+
+	var partitions = viper.GetInt(StreamSplitterPartitions)
+	var blockSize = viper.GetSizeInBytes(StreamSplitterBlockSize)
+
+	uploader = NewSplitStreamUploader(compressor, folder, partitions, int(blockSize))
+	return uploader, err
+}
+
 // ConfigureCrypter uses environment variables to create and configure a crypter.
 // In case no configuration in environment variables found, return `<nil>` value.
 func ConfigureCrypter() crypto.Crypter {

internal/databases/fdb/backup_fetch_handler.go

@@ -12,5 +12,5 @@ func HandleBackupFetch(ctx context.Context,
 	folder storage.Folder,
 	targetBackupSelector internal.BackupSelector,
 	restoreCmd *exec.Cmd) {
-	internal.HandleBackupFetch(folder, targetBackupSelector, internal.GetCommandStreamFetcher(restoreCmd))
+	internal.HandleBackupFetch(folder, targetBackupSelector, internal.GetBackupToCommandFetcher(restoreCmd))
 }

internal/databases/mongo/backup_fetch_handler.go

@@ -1,18 +1 @@
 package mongo
-
-import (
-	"context"
-	"os/exec"
-
-	"github.com/wal-g/wal-g/internal"
-	"github.com/wal-g/wal-g/pkg/storages/storage"
-	"github.com/wal-g/wal-g/utility"
-)
-
-func HandleBackupFetch(ctx context.Context, folder storage.Folder, backupName string, restoreCmd *exec.Cmd) error {
-	backup, err := internal.GetBackupByName(backupName, utility.BaseBackupPath, folder)
-	if err != nil {
-		return err
-	}
-	return internal.StreamBackupToCommandStdin(restoreCmd, backup)
-}

internal/databases/mysql/backup_fetch_handler.go

@@ -12,7 +12,9 @@ func HandleBackupFetch(folder storage.Folder,
 	targetBackupSelector internal.BackupSelector,
 	restoreCmd *exec.Cmd,
 	prepareCmd *exec.Cmd) {
-	internal.HandleBackupFetch(folder, targetBackupSelector, internal.GetCommandStreamFetcher(restoreCmd))
+	internal.HandleBackupFetch(folder, targetBackupSelector, internal.GetBackupToCommandFetcher(restoreCmd))
+
+	// Prepare Backup
 	if prepareCmd != nil {
 		err := prepareCmd.Run()
 		tracelog.ErrorLogger.FatalfOnError("failed to prepare fetched backup: %v", err)

internal/databases/mysql/backup_list_handler.go

@@ -33,6 +33,7 @@ type BackupDetail struct {
 	UserData    interface{} `json:"user_data,omitempty"`
 }
 
+//nolint:gocritic,hugeParam
 func NewBackupDetail(backupTime internal.BackupTime, sentinel StreamSentinelDto) BackupDetail {
 	return BackupDetail{
 		BackupName:       backupTime.BackupName,

internal/databases/mysql/backup_push_handler.go

@@ -10,9 +10,7 @@ import (
 	"github.com/wal-g/wal-g/utility"
 )
 
-func HandleBackupPush(uploader *internal.Uploader, backupCmd *exec.Cmd, isPermanent bool, userDataRaw string) {
-	uploader.UploadingFolder = uploader.UploadingFolder.GetSubFolder(utility.BaseBackupPath)
-
+func HandleBackupPush(uploader internal.UploaderProvider, backupCmd *exec.Cmd, isPermanent bool, userDataRaw string) {
 	db, err := getMySQLConnection()
 	tracelog.ErrorLogger.FatalOnError(err)
 	defer utility.LoggedClose(db, "")

internal/databases/mysql/binlog_push_handler.go

@@ -29,9 +29,9 @@ type LogsCache struct {
 }
 
 //gocyclo:ignore
-func HandleBinlogPush(uploader *internal.Uploader, untilBinlog string, checkGTIDs bool) {
-	rootFolder := uploader.UploadingFolder
-	uploader.UploadingFolder = uploader.UploadingFolder.GetSubFolder(BinlogPath)
+func HandleBinlogPush(uploader internal.UploaderProvider, untilBinlog string, checkGTIDs bool) {
+	rootFolder := uploader.Folder()
+	uploader.ChangeDirectory(BinlogPath)
 
 	db, err := getMySQLConnection()
 	tracelog.ErrorLogger.FatalOnError(err)
@@ -134,7 +134,7 @@ func getMySQLBinlogsFolder(db *sql.DB) (string, error) {
 	return path.Dir(logBinBasename), nil
 }
 
-func archiveBinLog(uploader *internal.Uploader, dataDir string, binLog string) error {
+func archiveBinLog(uploader internal.UploaderProvider, dataDir string, binLog string) error {
 	tracelog.InfoLogger.Printf("Archiving %v\n", binLog)
 
 	filename := path.Join(dataDir, binLog)

internal/databases/mysql/mysql.go

@@ -169,6 +169,7 @@ type StreamSentinelDto struct {
 
 	IsPermanent bool        `json:"IsPermanent,omitempty"`
 	UserData    interface{} `json:"UserData,omitempty"`
+
 	//todo: add other fields from internal.GenericMetadata
 }
 

internal/extract.go

@@ -74,6 +74,27 @@ func (e EmptyWriteIgnorer) Write(p []byte) (int, error) {
 	return e.WriteCloser.Write(p)
 }
 
+type DevNullWriter struct {
+	io.WriteCloser
+	statPrinter sync.Once
+	totalBytes  int64
+}
+
+func (e *DevNullWriter) Write(p []byte) (int, error) {
+	e.statPrinter.Do(func() {
+		go func() {
+			for {
+				time.Sleep(1 * time.Second)
+				tracelog.ErrorLogger.Printf("/dev/null size %d", e.totalBytes)
+			}
+		}()
+	})
+	e.totalBytes += int64(len(p))
+	return len(p), nil
+}
+
+var _ io.Writer = &DevNullWriter{}
+
 // TODO : unit tests
 // Extract exactly one tar bundle.
 func extractOne(tarInterpreter TarInterpreter, source io.Reader) error {

internal/fetch_helper.go

@@ -9,11 +9,10 @@ import (
 	"os/user"
 	"path/filepath"
 
-	"github.com/wal-g/wal-g/internal/ioextensions"
-
 	"github.com/pkg/errors"
 	"github.com/wal-g/tracelog"
 	"github.com/wal-g/wal-g/internal/compression"
+	"github.com/wal-g/wal-g/internal/ioextensions"
 	"github.com/wal-g/wal-g/pkg/storages/storage"
 	"github.com/wal-g/wal-g/utility"
 )
@@ -53,8 +52,8 @@ func DownloadFile(folder storage.Folder, filename, ext string, writeCloser io.Wr
 	return nil
 }
 
-func TryDownloadFile(folder storage.Folder, path string) (walFileReader io.ReadCloser, exists bool, err error) {
-	walFileReader, err = folder.ReadObject(path)
+func TryDownloadFile(folder storage.Folder, path string) (fileReader io.ReadCloser, exists bool, err error) {
+	fileReader, err = folder.ReadObject(path)
 	if err == nil {
 		exists = true
 		return