Merge branch 'wal-g:master' into backup-list-failover-storages/0

cmd/common/st/put_object.go

@@ -1,6 +1,9 @@
 package st
 
 import (
+	"io"
+	"os"
+
 	"github.com/spf13/cobra"
 	"github.com/wal-g/tracelog"
 	"github.com/wal-g/wal-g/internal"
@@ -16,23 +19,47 @@ const (
 	noCompressFlag     = "no-compress"
 	overwriteFlag      = "force"
 	overwriteShorthand = "f"
+	readStdinFlag 	   = "read-stdin"
+	readStdinShorthand = "s"
 )
 
+
 // putObjectCmd represents the putObject command
 var putObjectCmd = &cobra.Command{
 	Use:   "put local_path destination_path",
 	Short: putObjectShortDescription,
-	Args:  cobra.ExactArgs(2),
+	Args:  cobra.RangeArgs(1, 2),
 	Run: func(cmd *cobra.Command, args []string) {
-		localPath := args[0]
-		dstPath := args[1]
+
+		if len(args) == 1 && !readStdin {
+			tracelog.ErrorLogger.Fatal("should specify localPath on read-from stdin flag")
+		}
+
+		var dstPath string
+		var reader io.Reader
+
+		if !readStdin {
+			localPath := args[0]
+			dstPath = args[1]
+			fileReadCloser, err := storagetools.OpenLocalFile(localPath)
+			if err != nil {
+				tracelog.ErrorLogger.FatalOnError(err)
+			}
+
+			reader = fileReadCloser
+
+			defer fileReadCloser.Close()
+		} else {
+			dstPath = args[0]
+			reader = os.Stdin
+		}
 
 		err := exec.OnStorage(targetStorage, func(folder storage.Folder) error {
 			uploader, err := internal.ConfigureUploaderToFolder(folder)
 			if err != nil {
 				return err
 			}
-			return storagetools.HandlePutObject(localPath, dstPath, uploader, overwrite, !noEncrypt, !noCompress)
+			return storagetools.HandlePutObject(reader, dstPath, uploader, overwrite, !noEncrypt, !noCompress)
 		})
 		tracelog.ErrorLogger.FatalOnError(err)
 	},
@@ -41,11 +68,14 @@ var putObjectCmd = &cobra.Command{
 var noEncrypt bool
 var noCompress bool
 var overwrite bool
+var readStdin bool
 
 func init() {
 	StorageToolsCmd.AddCommand(putObjectCmd)
 	putObjectCmd.Flags().BoolVar(&noEncrypt, noEncryptFlag, false, "Do not encrypt the object")
 	putObjectCmd.Flags().BoolVar(&noCompress, noCompressFlag, false, "Do not compress the object")
 	putObjectCmd.Flags().BoolVarP(&overwrite, overwriteFlag, overwriteShorthand,
 		false, "Overwrite the existing object")
+	putObjectCmd.Flags().BoolVarP(&readStdin, readStdinFlag, readStdinShorthand,
+			false, "Read file content from STDIN")
 }

cmd/mongo/binary_backup_fetch.go

@@ -20,12 +20,15 @@ const (
 	RsNameDescription            = "Name of replicaset (like rs01)"
 	RsMembersFlag                = "mongo-rs-members"
 	RsMembersDescription         = "Comma separated host:port records from wished rs members (like rs.initiate())"
+	RsMemberIdsFlag              = "mongo-rs-member-ids"
+	RsMemberIdsDescription       = "Comma separated integers for replica IDs of corresponding --mongo-rs-members"
 )
 
 var (
 	minimalConfigPath = ""
 	rsName            = ""
-	rsMembers         = ""
+	rsMembers         []string
+	rsMemberIds       []int
 )
 
 var binaryBackupFetchCmd = &cobra.Command{
@@ -44,14 +47,15 @@ var binaryBackupFetchCmd = &cobra.Command{
 		mongodVersion := args[2]
 
 		err := mongo.HandleBinaryFetchPush(ctx, mongodConfigPath, minimalConfigPath, backupName, mongodVersion, rsName,
-			rsMembers)
+			rsMembers, rsMemberIds)
 		tracelog.ErrorLogger.FatalOnError(err)
 	},
 }
 
 func init() {
 	binaryBackupFetchCmd.Flags().StringVar(&minimalConfigPath, MinimalConfigPathFlag, "", MinimalConfigPathDescription)
 	binaryBackupFetchCmd.Flags().StringVar(&rsName, RsNameFlag, "", RsNameDescription)
-	binaryBackupFetchCmd.Flags().StringVar(&rsMembers, RsMembersFlag, "", RsMembersDescription)
+	binaryBackupFetchCmd.Flags().StringSliceVar(&rsMembers, RsMembersFlag, []string{}, RsMembersDescription)
+	binaryBackupFetchCmd.Flags().IntSliceVar(&rsMemberIds, RsMemberIdsFlag, []int{}, RsMemberIdsDescription)
 	cmd.AddCommand(binaryBackupFetchCmd)
 }

cmd/pg/backup_fetch.go

@@ -61,7 +61,7 @@ var backupFetchCmd = &cobra.Command{
 		tracelog.ErrorLogger.FatalOnError(err)
 
 		folder := multistorage.NewFolder(cache)
-		multistorage.SetPolicies(folder, policies.UniteAllStorages)
+		folder = multistorage.SetPolicies(folder, policies.UniteAllStorages)
 		folder, err = multistorage.UseAllAliveStorages(folder)
 		tracelog.ErrorLogger.FatalOnError(err)
 

docs/PostgreSQL.md

@@ -193,7 +193,7 @@ wal-g backup-push $PGDATA
 ```
 WAL-G will check that command argument, environment variable PGDATA and config setting PGDATA are the same, if set.
 
-If a backup is started from a standby sever, WAL-G will monitor the timeline of the server. If a promotion or timeline change occurs during the backup, the data will be uploaded but not finalized, and WAL-G will exit with an error. The logs will contain the necessary information to finalize the backup, which can then be used if you clearly understand the risks.
+If a backup is started from a standby server, WAL-G will monitor the timeline of the server. If a promotion or timeline change occurs during the backup, the data will be uploaded but not finalized, and WAL-G will exit with an error. The logs will contain the necessary information to finalize the backup, which can then be used if you clearly understand the risks.
 
 ``backup-push`` can also be run with the ``--permanent`` flag, which will mark the backup as permanent and prevent it from being removed when running ``delete``.
 

internal/databases/mongo/binary/mongod.go

@@ -188,7 +188,7 @@ func updateRsConfig(ctx context.Context, localDatabase *mongo.Database, rsConfig
 		return errors.Wrap(err, "unable to read rs config from system.replset")
 	}
 
-	systemRsConfig["members"] = makeBsonRsMembers(rsConfig.RsMembers)
+	systemRsConfig["members"] = makeBsonRsMembers(rsConfig)
 
 	if systemRsConfig["_id"] != rsConfig.RsName {
 		systemRsConfig["_id"] = rsConfig.RsName
@@ -257,13 +257,11 @@ func replaceData(ctx context.Context, collection *mongo.Collection, drop bool, i
 	return nil
 }
 
-func makeBsonRsMembers(rsMembers string) bson.A {
+func makeBsonRsMembers(rsConfig RsConfig) bson.A {
 	bsonMembers := bson.A{}
 
-	if rsMembers != "" {
-		for id, member := range strings.Split(rsMembers, ",") {
-			bsonMembers = append(bsonMembers, bson.M{"_id": id, "host": member})
-		}
+	for i := 0; i != len(rsConfig.RsMembers); i++ {
+		bsonMembers = append(bsonMembers, bson.M{"_id": rsConfig.RsMemberIds[i], "host": rsConfig.RsMembers[i]})
 	}
 
 	return bsonMembers
@@ -301,17 +299,38 @@ type BackupCursorFile struct {
 }
 
 type RsConfig struct {
-	RsName    string
-	RsMembers string
+	RsName      string
+	RsMembers   []string
+	RsMemberIds []int
+}
+
+func NewRsConfig(rsName string, rsMembers []string, rsMemberIds []int) RsConfig {
+	if len(rsMemberIds) == 0 {
+		rsMemberIds = make([]int, len(rsMembers))
+		for i := 0; i < len(rsMembers); i++ {
+			rsMemberIds[i] = i
+		}
+	}
+	return RsConfig{
+		RsName:      rsName,
+		RsMembers:   rsMembers,
+		RsMemberIds: rsMemberIds,
+	}
 }
 
 func (rsConfig RsConfig) Empty() bool {
-	return rsConfig.RsName == "" && rsConfig.RsMembers == ""
+	return rsConfig.RsName == "" && len(rsConfig.RsMembers) == 0
 }
 
 func (rsConfig RsConfig) Validate() error {
-	if rsConfig.RsName == "" && rsConfig.RsMembers != "" || rsConfig.RsName != "" && rsConfig.RsMembers == "" {
+	if rsConfig.RsName == "" && len(rsConfig.RsMembers) > 0 || rsConfig.RsName != "" && len(rsConfig.RsMembers) == 0 {
 		return errors.Errorf("rsConfig should be all empty or full populated, but rsConfig = %+v", rsConfig)
 	}
+	if len(rsConfig.RsMembers) > len(rsConfig.RsMemberIds) {
+		return errors.Errorf("not all replica set members have corresponding ID")
+	}
+	if len(rsConfig.RsMembers) < len(rsConfig.RsMemberIds) {
+		return errors.Errorf("excessive number of replica set IDs")
+	}
 	return nil
 }

internal/databases/mongo/binary/mongod_test.go

@@ -8,13 +8,31 @@ import (
 )
 
 func TestMakeBsonRsMembers(t *testing.T) {
-	assert.Equal(t, bson.A{}, makeBsonRsMembers(""))
-	assert.Equal(t, bson.A{bson.M{"_id": 0, "host": "localhost:1234"}}, makeBsonRsMembers("localhost:1234"))
+	assert.Equal(t, bson.A{}, makeBsonRsMembers(RsConfig{}))
+	assert.Equal(t, bson.A{bson.M{"_id": 0, "host": "localhost:1234"}}, makeBsonRsMembers(RsConfig{
+		RsMembers:   []string{"localhost:1234"},
+		RsMemberIds: []int{0},
+	}))
 	assert.Equal(t,
 		bson.A{
 			bson.M{"_id": 0, "host": "localhost:1234"},
 			bson.M{"_id": 1, "host": "localhost:5678"},
 			bson.M{"_id": 2, "host": "remotehost:9876"},
 		},
-		makeBsonRsMembers("localhost:1234,localhost:5678,remotehost:9876"))
+		makeBsonRsMembers(RsConfig{
+			RsName:      "",
+			RsMembers:   []string{"localhost:1234", "localhost:5678", "remotehost:9876"},
+			RsMemberIds: []int{0, 1, 2},
+		}))
+	assert.Equal(t,
+		bson.A{
+			bson.M{"_id": 4, "host": "localhost:1234"},
+			bson.M{"_id": 5, "host": "localhost:5678"},
+			bson.M{"_id": 0, "host": "remotehost:9876"},
+		},
+		makeBsonRsMembers(RsConfig{
+			RsName:      "",
+			RsMembers:   []string{"localhost:1234", "localhost:5678", "remotehost:9876"},
+			RsMemberIds: []int{4, 5, 0},
+		}))
 }

internal/databases/mongo/binary_backup_fetch_handler.go

@@ -9,7 +9,7 @@ import (
 )
 
 func HandleBinaryFetchPush(ctx context.Context, mongodConfigPath, minimalConfigPath, backupName, restoreMongodVersion,
-	rsName, rsMembers string,
+	rsName string, rsMembers []string, rsMemberIds []int,
 ) error {
 	config, err := binary.CreateMongodConfig(mongodConfigPath)
 	if err != nil {
@@ -36,7 +36,7 @@ func HandleBinaryFetchPush(ctx context.Context, mongodConfigPath, minimalConfigP
 		return err
 	}
 
-	rsConfig := binary.RsConfig{RsName: rsName, RsMembers: rsMembers}
+	rsConfig := binary.NewRsConfig(rsName, rsMembers, rsMemberIds)
 	if err = rsConfig.Validate(); err != nil {
 		return err
 	}

internal/storagetools/put_object_handler.go

@@ -14,26 +14,19 @@ import (
 	"github.com/wal-g/wal-g/internal"
 )
 
-func HandlePutObject(localPath, dstPath string, uploader internal.Uploader, overwrite, encrypt, compress bool) error {
+func HandlePutObject(source io.Reader, dstPath string, uploader internal.Uploader, overwrite, encrypt, compress bool) error {
 	err := checkOverwrite(dstPath, uploader, overwrite)
 	if err != nil {
 		return fmt.Errorf("check file overwrite: %v", err)
 	}
 
-	fileReadCloser, err := openLocalFile(localPath)
-	if err != nil {
-		return fmt.Errorf("open local file: %v", err)
-	}
-
-	defer fileReadCloser.Close()
-
 	storageFolderPath := utility.SanitizePath(filepath.Dir(dstPath))
 	if storageFolderPath != "" {
 		uploader.ChangeDirectory(storageFolderPath)
 	}
 
 	fileName := utility.SanitizePath(filepath.Base(dstPath))
-	err = uploadFile(fileName, fileReadCloser, uploader, encrypt, compress)
+	err = uploadFile(fileName, source, uploader, encrypt, compress)
 	if err != nil {
 		return fmt.Errorf("upload: %v", err)
 	}
@@ -52,7 +45,7 @@ func checkOverwrite(dstPath string, uploader internal.Uploader, overwrite bool)
 	return nil
 }
 
-func openLocalFile(localPath string) (io.ReadCloser, error) {
+func OpenLocalFile(localPath string) (io.ReadCloser, error) {
 	localFile, err := os.Open(localPath)
 	if err != nil {
 		return nil, fmt.Errorf("open the local file: %v", err)

internal/tools/go.mod

@@ -135,7 +135,7 @@ require (
 	google.golang.org/protobuf v1.26.0-rc.1 // indirect
 	gopkg.in/ini.v1 v1.51.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
-	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+	gopkg.in/yaml.v3 v3.0.0 // indirect
 	honnef.co/go/tools v0.2.0 // indirect
 	mvdan.cc/gofumpt v0.1.1 // indirect
 	mvdan.cc/interfacer v0.0.0-20180901003855-c20040233aed // indirect

internal/tools/go.sum

@@ -1057,8 +1057,9 @@ gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0 h1:hjy8E9ON/egN1tAYqKb61G10WtihqetD4sz2H+8nIeA=
+gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=