Add --restore-point flag to Greenplum backup-fetch (#1221)

* Add --restore-point flag to the greenplum backup-fetch

* Add tests

* Fix problems

cmd/gp/backup_fetch.go

@@ -14,6 +14,7 @@ import (
 const (
 	backupFetchShortDescription  = "Fetches a backup from storage"
 	targetUserDataDescription    = "Fetch storage backup which has the specified user data"
+	restorePointDescription      = "Fetch storage backup w/ restore point specified by name"
 	restoreConfigPathDescription = "Path to the cluster restore configuration"
 	fetchContentIdsDescription   = "If set, WAL-G will fetch only the specified segments"
 	fetchModeDescription         = "Backup fetch mode. default: do the backup unpacking " +
@@ -22,13 +23,14 @@ const (
 )
 
 var fetchTargetUserData string
+var restorePoint string
 var restoreConfigPath string
 var fetchContentIds *[]int
 var fetchModeStr string
 var inPlaceRestore bool
 
 var backupFetchCmd = &cobra.Command{
-	Use:   "backup-fetch [backup_name | --target-user-data <data>]",
+	Use:   "backup-fetch [backup_name | --target-user-data <data> | --restore-point <name>]",
 	Short: backupFetchShortDescription, // TODO : improve description
 	Args:  cobra.RangeArgs(0, 1),
 	Run: func(cmd *cobra.Command, args []string) {
@@ -40,7 +42,7 @@ var backupFetchCmd = &cobra.Command{
 		if fetchTargetUserData == "" {
 			fetchTargetUserData = viper.GetString(internal.FetchTargetUserDataSetting)
 		}
-		targetBackupSelector, err := createTargetFetchBackupSelector(cmd, args, fetchTargetUserData)
+		targetBackupSelector, err := createTargetFetchBackupSelector(cmd, args, fetchTargetUserData, restorePoint)
 		tracelog.ErrorLogger.FatalOnError(err)
 
 		folder, err := internal.ConfigureFolder()
@@ -56,18 +58,26 @@ var backupFetchCmd = &cobra.Command{
 		tracelog.ErrorLogger.FatalOnError(err)
 
 		internal.HandleBackupFetch(folder, targetBackupSelector,
-			greenplum.NewGreenplumBackupFetcher(restoreConfigPath, inPlaceRestore, logsDir, *fetchContentIds, fetchMode))
+			greenplum.NewGreenplumBackupFetcher(restoreConfigPath, inPlaceRestore, logsDir, *fetchContentIds, fetchMode, restorePoint))
 	},
 }
 
 // create the BackupSelector to select the backup to fetch
 func createTargetFetchBackupSelector(cmd *cobra.Command,
-	args []string, targetUserData string) (internal.BackupSelector, error) {
+	args []string, targetUserData, restorePoint string) (internal.BackupSelector, error) {
 	targetName := ""
 	if len(args) >= 1 {
 		targetName = args[0]
 	}
 
+	// if target restore point is provided without the backup name, then
+	// choose the latest backup up to the specified restore point name
+	if restorePoint != "" && targetUserData == "" && targetName == "" {
+		tracelog.InfoLogger.Printf("Restore point %s is specified without the backup name or target user data, "+
+			"will search for a matching backup", restorePoint)
+		return greenplum.NewRestorePointBackupSelector(restorePoint), nil
+	}
+
 	backupSelector, err := internal.NewTargetBackupSelector(targetUserData, targetName, greenplum.NewGenericMetaFetcher())
 	if err != nil {
 		fmt.Println(cmd.UsageString())
@@ -79,6 +89,7 @@ func createTargetFetchBackupSelector(cmd *cobra.Command,
 func init() {
 	backupFetchCmd.Flags().StringVar(&fetchTargetUserData, "target-user-data",
 		"", targetUserDataDescription)
+	backupFetchCmd.Flags().StringVar(&restorePoint, "restore-point", "", restorePointDescription)
 	backupFetchCmd.Flags().StringVar(&restoreConfigPath, "restore-config",
 		"", restoreConfigPathDescription)
 	backupFetchCmd.Flags().BoolVar(&inPlaceRestore, "in-place", false, inPlaceFlagDescription)

docker/gp_tests/scripts/tests/create_restore_point.sh → docker/gp_tests/scripts/tests/restore_point.sh

@@ -38,5 +38,31 @@ if [ "$EXIT_STATUS" -eq 0 ] ; then
     exit 1
 fi
 
+wal-g backup-push --config=${TMP_CONFIG}
+
+wal-g create-restore-point after_backup --config=${TMP_CONFIG}
+stop_and_delete_cluster_dir
+
+wal-g backup-fetch LATEST --restore-point rp1 --in-place --config=${TMP_CONFIG} && EXIT_STATUS=$? || EXIT_STATUS=$?
+if [ "$EXIT_STATUS" -eq 0 ] ; then
+    echo "Error: backup fetched with restore point in the past"
+    exit 1
+fi
+
+wal-g backup-fetch --restore-point rp1 --in-place --config=${TMP_CONFIG} && EXIT_STATUS=$? || EXIT_STATUS=$?
+if [ "$EXIT_STATUS" -eq 0 ] ; then
+    echo "Error: backup fetched with restore point in the past"
+    exit 1
+fi
+
+# should not fail
+wal-g backup-fetch LATEST --restore-point after_backup --in-place --config=${TMP_CONFIG}
+delete_cluster_dirs
+
+# should not fail
+wal-g backup-fetch --restore-point after_backup --in-place --config=${TMP_CONFIG}
+
+start_cluster
+
 cleanup
 rm ${TMP_CONFIG}

docker/gp_tests/scripts/tests/test_functions/util.sh

@@ -59,11 +59,15 @@ setup_wal_archiving() {
   start_cluster
 }
 
-stop_and_delete_cluster_dir() {
-  stop_cluster
+delete_cluster_dirs() {
   #remove the database files
   for elem in "${SEGMENTS_DIRS[@]}"; do
     read -a arr <<< "$elem"
     rm -rf "${arr[1]}"
   done
+}
+
+stop_and_delete_cluster_dir() {
+  stop_cluster
+  delete_cluster_dirs
 }

internal/backup.go

@@ -2,7 +2,6 @@ package internal
 
 import (
 	"fmt"
-	"io"
 
 	"github.com/pkg/errors"
 	"github.com/wal-g/tracelog"
@@ -65,17 +64,26 @@ func (backup *Backup) SentinelExists() (bool, error) {
 
 // TODO : unit tests
 func (backup *Backup) FetchSentinel(sentinelDto interface{}) error {
-	return backup.FetchDto(sentinelDto, backup.getStopSentinelPath())
+	return FetchDto(backup.Folder, sentinelDto, backup.getStopSentinelPath())
 }
 
 // TODO : unit tests
 func (backup *Backup) FetchMetadata(metadataDto interface{}) error {
-	return backup.FetchDto(metadataDto, backup.getMetadataPath())
+	return FetchDto(backup.Folder, metadataDto, backup.getMetadataPath())
+}
+
+func (backup *Backup) UploadMetadata(metadataDto interface{}) error {
+	return UploadDto(backup.Folder, metadataDto, backup.getMetadataPath())
+}
+
+func (backup *Backup) UploadSentinel(sentinelDto interface{}) error {
+	return UploadDto(backup.Folder, sentinelDto, backup.getStopSentinelPath())
 }
 
 // FetchDto gets data from path and de-serializes it to given object
-func (backup *Backup) FetchDto(dto interface{}, path string) error {
-	reader, err := backup.fetchStorageStream(path)
+func FetchDto(folder storage.Folder, dto interface{}, path string) error {
+	backupReaderMaker := NewStorageReaderMaker(folder, path)
+	reader, err := backupReaderMaker.Reader()
 	if err != nil {
 		return err
 	}
@@ -86,19 +94,6 @@ func (backup *Backup) FetchDto(dto interface{}, path string) error {
 	return errors.Wrap(unmarshaller.Unmarshal(reader, dto), fmt.Sprintf("failed to fetch dto from %s", path))
 }
 
-func (backup *Backup) fetchStorageStream(path string) (io.ReadCloser, error) {
-	backupReaderMaker := NewStorageReaderMaker(backup.Folder, path)
-	return backupReaderMaker.Reader()
-}
-
-func (backup *Backup) UploadMetadata(metadataDto interface{}) error {
-	return UploadDto(backup.Folder, metadataDto, backup.getMetadataPath())
-}
-
-func (backup *Backup) UploadSentinel(sentinelDto interface{}) error {
-	return UploadDto(backup.Folder, sentinelDto, backup.getStopSentinelPath())
-}
-
 // UploadDto serializes given object to JSON and puts it to path
 func UploadDto(folder storage.Folder, dto interface{}, path string) error {
 	marshaller, err := NewDtoSerializer()

internal/databases/greenplum/backup_fetch_handler.go

@@ -53,13 +53,15 @@ type FetchHandler struct {
 	backup              internal.Backup
 	contentIDsToFetch   map[int]bool
 	fetchMode           BackupFetchMode
+	restorePoint        string
 }
 
 // nolint:gocritic
 func NewFetchHandler(
 	backup internal.Backup, sentinel BackupSentinelDto,
 	segCfgMaker SegConfigMaker, logsDir string,
 	fetchContentIds []int, mode BackupFetchMode,
+	restorePoint string,
 ) *FetchHandler {
 	backupIDByContentID := make(map[int]string)
 	segmentConfigs := make([]cluster.SegConfig, 0)
@@ -90,6 +92,7 @@ func NewFetchHandler(
 		backup:              backup,
 		contentIDsToFetch:   prepareContentIDsToFetch(fetchContentIds, segmentConfigs),
 		fetchMode:           mode,
+		restorePoint:        restorePoint,
 	}
 }
 
@@ -189,7 +192,12 @@ func (fh *FetchHandler) createPgHbaOnSegments() error {
 // files to each segment instance (including master) so they can recover correctly
 // during the database startup
 func (fh *FetchHandler) createRecoveryConfigs() error {
-	restoreCfgMaker := NewRecoveryConfigMaker("/usr/bin/wal-g", internal.CfgFile, fh.backup.Name)
+	recoveryTarget := fh.backup.Name
+	if fh.restorePoint != "" {
+		recoveryTarget = fh.restorePoint
+	}
+	tracelog.InfoLogger.Printf("Recovery target is %s", recoveryTarget)
+	restoreCfgMaker := NewRecoveryConfigMaker("/usr/bin/wal-g", internal.CfgFile, recoveryTarget)
 
 	remoteOutput := fh.cluster.GenerateAndExecuteCommand("Creating recovery.conf on segments and master",
 		cluster.ON_SEGMENTS|cluster.EXCLUDE_MIRRORS|cluster.INCLUDE_MASTER,
@@ -245,17 +253,22 @@ func (fh *FetchHandler) buildFetchCommand(contentID int) string {
 	return cmdLine
 }
 
-func NewGreenplumBackupFetcher(restoreCfgPath string, inPlaceRestore bool, logsDir string, fetchContentIds []int, mode BackupFetchMode,
+func NewGreenplumBackupFetcher(restoreCfgPath string, inPlaceRestore bool, logsDir string,
+	fetchContentIds []int, mode BackupFetchMode, restorePoint string,
 ) func(folder storage.Folder, backup internal.Backup) {
 	return func(folder storage.Folder, backup internal.Backup) {
+		tracelog.InfoLogger.Printf("Starting backup-fetch for %s", backup.Name)
+		if restorePoint != "" {
+			tracelog.ErrorLogger.FatalOnError(ValidateMatch(folder, backup.Name, restorePoint))
+		}
 		var sentinel BackupSentinelDto
 		err := backup.FetchSentinel(&sentinel)
 		tracelog.ErrorLogger.FatalOnError(err)
 
 		segCfgMaker, err := NewSegConfigMaker(restoreCfgPath, inPlaceRestore)
 		tracelog.ErrorLogger.FatalOnError(err)
 
-		err = NewFetchHandler(backup, sentinel, segCfgMaker, logsDir, fetchContentIds, mode).Fetch()
+		err = NewFetchHandler(backup, sentinel, segCfgMaker, logsDir, fetchContentIds, mode, restorePoint).Fetch()
 		tracelog.ErrorLogger.FatalOnError(err)
 	}
 }

internal/databases/greenplum/backup_list_handler.go

@@ -0,0 +1,38 @@
+package greenplum
+
+import (
+	"fmt"
+	"sort"
+
+	"github.com/wal-g/wal-g/internal"
+	"github.com/wal-g/wal-g/pkg/storages/storage"
+	"github.com/wal-g/wal-g/utility"
+)
+
+//TODO: Implement backup-list handler
+
+// ListStorageBackups returns the list of storage backups sorted by finish time (in ascending order)
+func ListStorageBackups(folder storage.Folder) ([]Backup, error) {
+	backupObjects, err := internal.GetBackups(folder.GetSubFolder(utility.BaseBackupPath))
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch list of backups in storage: %w", err)
+	}
+
+	backups := make([]Backup, 0, len(backupObjects))
+	for _, b := range backupObjects {
+		backup := NewBackup(folder, b.BackupName)
+
+		_, err = backup.GetSentinel()
+		if err != nil {
+			return nil, fmt.Errorf("failed to load sentinel for backup %s: %w", b.BackupName, err)
+		}
+
+		backups = append(backups, backup)
+	}
+
+	sort.Slice(backups, func(i, j int) bool {
+		return backups[i].SentinelDto.FinishTime.Before(backups[j].SentinelDto.FinishTime)
+	})
+
+	return backups, nil
+}

internal/databases/greenplum/backup_selector.go

@@ -0,0 +1,38 @@
+package greenplum
+
+import (
+	"fmt"
+
+	"github.com/wal-g/wal-g/pkg/storages/storage"
+)
+
+type RestorePointBackupSelector struct {
+	restorePoint string
+}
+
+func NewRestorePointBackupSelector(restorePoint string) *RestorePointBackupSelector {
+	return &RestorePointBackupSelector{restorePoint: restorePoint}
+}
+
+func (s *RestorePointBackupSelector) Select(folder storage.Folder) (string, error) {
+	restorePoint, err := FetchRestorePointMetadata(folder, s.restorePoint)
+	if err != nil {
+		return "", err
+	}
+
+	backups, err := ListStorageBackups(folder)
+	if err != nil {
+		return "", err
+	}
+
+	// pick the latest (closest) backup to the restore point
+	for i := len(backups) - 1; i >= 0; i-- {
+		if backups[i].SentinelDto.FinishTime.Before(restorePoint.FinishTime) {
+			return backups[i].Name, nil
+		}
+	}
+
+	return "", fmt.Errorf(
+		"failed to find matching backup (earlier than the finish time %s of the restore point %s)",
+		restorePoint.Name, restorePoint.FinishTime)
+}

internal/databases/greenplum/restore_point_creator.go → internal/databases/greenplum/restore_point.go

@@ -6,6 +6,8 @@ import (
 	"os"
 	"time"
 
+	"github.com/wal-g/wal-g/pkg/storages/storage"
+
 	"github.com/spf13/viper"
 
 	"github.com/blang/semver"
@@ -41,6 +43,40 @@ func RestorePointMetadataFileName(pointName string) string {
 	return pointName + RestorePointSuffix
 }
 
+func FetchRestorePointMetadata(folder storage.Folder, pointName string) (RestorePointMetadata, error) {
+	var restorePoint RestorePointMetadata
+	err := internal.FetchDto(folder.GetSubFolder(utility.BaseBackupPath),
+		&restorePoint, RestorePointMetadataFileName(pointName))
+	if err != nil {
+		return RestorePointMetadata{}, fmt.Errorf("failed to fetch metadata for restore point %s: %w", pointName, err)
+	}
+
+	return restorePoint, nil
+}
+
+// ValidateMatch checks that restore point is reachable from the provided backup
+func ValidateMatch(folder storage.Folder, backupName string, restorePoint string) error {
+	backup := NewBackup(folder, backupName)
+	bSentinel, err := backup.GetSentinel()
+	if err != nil {
+		return fmt.Errorf("failed to fetch %s sentinel: %w", backupName, err)
+	}
+
+	rpMeta, err := FetchRestorePointMetadata(folder, restorePoint)
+	if err != nil {
+		tracelog.WarningLogger.Printf(
+			"failed to fetch restore point %s metadata, will skip the validation check: %v", restorePoint, err)
+		return nil
+	}
+
+	if bSentinel.FinishTime.After(rpMeta.FinishTime) {
+		return fmt.Errorf("%s backup finish time (%s) is after the %s provided restore point finish time (%s)",
+			backupName, bSentinel.FinishTime, restorePoint, rpMeta.FinishTime)
+	}
+
+	return nil
+}
+
 type RestorePointCreator struct {
 	pointName        string
 	startTime        time.Time

internal/databases/postgres/backup.go

@@ -133,7 +133,7 @@ func (backup *Backup) GetSentinelAndFilesMetadata() (BackupSentinelDto, FilesMet
 		return sentinel, filesMetadata, nil
 	}
 
-	err = backup.FetchDto(&filesMetadata, getFilesMetadataPath(backup.Name))
+	err = internal.FetchDto(backup.Folder, &filesMetadata, getFilesMetadataPath(backup.Name))
 	if err != nil {
 		// double-check that this is not V2 backup
 		sentinelV2, err2 := backup.getSentinelV2()

internal/stream_metadata.go

@@ -23,7 +23,7 @@ type BackupStreamMetadata struct {
 
 func GetBackupStreamFetcher(backup Backup) (StreamFetcher, error) {
 	var metadata BackupStreamMetadata
-	err := backup.FetchDto(&metadata, StreamMetadataNameFromBackup(backup.Name))
+	err := FetchDto(backup.Folder, &metadata, StreamMetadataNameFromBackup(backup.Name))
 	var test storage.ObjectNotFoundError
 	if errors.As(err, &test) {
 		return DownloadAndDecompressStream, nil