PBM-1114: check backup files before done

pbm/backup/backup.go

@@ -367,6 +367,11 @@ func (b *Backup) Run(ctx context.Context, bcp *ctrl.BackupCmd, opid ctrl.OPID, l
 			return errors.Wrap(err, "dump metadata")
 		}
 
+		err = CheckBackupFiles(ctx, stg, bcp.Name)
+		if err != nil {
+			return errors.Wrap(err, "check backup files")
+		}
+
 		err = ChangeBackupStateWithUnix(b.leadConn, bcp.Name, defs.StatusDone, unix, "")
 		return errors.Wrapf(err, "check cluster for backup done: update backup meta with %s",
 			defs.StatusDone)

pbm/backup/storage.go

@@ -2,6 +2,7 @@ package backup
 
 import (
 	"context"
+	"encoding/json"
 	"path"
 	"runtime"
 	"sync"
@@ -16,12 +17,45 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/version"
 )
 
-func CheckBackupFiles(ctx context.Context, bcp *BackupMeta, stg storage.Storage) error {
-	// !!! TODO: Check physical files ?
-	if bcp.Type != defs.LogicalBackup {
-		return nil
+func CheckBackupFiles(ctx context.Context, stg storage.Storage, name string) error {
+	bcp, err := ReadMetadata(stg, name+defs.MetadataFileSuffix)
+	if err != nil {
+		return errors.Wrap(err, "read backup metadata")
+	}
+
+	return CheckBackupDataFiles(ctx, stg, bcp)
+}
+
+func ReadMetadata(stg storage.Storage, filename string) (*BackupMeta, error) {
+	rdr, err := stg.SourceReader(filename)
+	if err != nil {
+		return nil, errors.Wrap(err, "open")
+	}
+	defer rdr.Close()
+
+	var meta *BackupMeta
+	err = json.NewDecoder(rdr).Decode(&meta)
+	if err != nil {
+		return nil, errors.Wrap(err, "decode")
 	}
 
+	return meta, nil
+}
+
+func CheckBackupDataFiles(ctx context.Context, stg storage.Storage, bcp *BackupMeta) error {
+	switch bcp.Type {
+	case defs.LogicalBackup:
+		return checkLogicalBackupFiles(ctx, stg, bcp)
+	case defs.PhysicalBackup, defs.IncrementalBackup:
+		return checkPhysicalBackupFiles(ctx, stg, bcp)
+	case defs.ExternalBackup:
+		return nil // no files available
+	}
+
+	return errors.Errorf("unknown backup type %s", bcp.Type)
+}
+
+func checkLogicalBackupFiles(ctx context.Context, stg storage.Storage, bcp *BackupMeta) error {
 	legacy := version.IsLegacyArchive(bcp.PBMVersion)
 	eg, _ := errgroup.WithContext(ctx)
 	for _, rs := range bcp.Replsets {
@@ -74,6 +108,10 @@ func CheckBackupFiles(ctx context.Context, bcp *BackupMeta, stg storage.Storage)
 	return eg.Wait()
 }
 
+func checkPhysicalBackupFiles(ctx context.Context, stg storage.Storage, bcp *BackupMeta) error {
+	return nil
+}
+
 func ReadArchiveNamespaces(stg storage.Storage, metafile string) ([]*archive.Namespace, error) {
 	r, err := stg.SourceReader(metafile)
 	if err != nil {

pbm/resync/rsync.go

@@ -2,7 +2,6 @@ package resync
 
 import (
 	"context"
-	"encoding/json"
 	"runtime"
 	"strings"
 	"sync"
@@ -293,21 +292,13 @@ func getAllBackupMetaFromStorage(
 
 	backupMeta := make([]*backup.BackupMeta, 0, len(backupFiles))
 	for _, b := range backupFiles {
-		d, err := stg.SourceReader(b.Name)
+		meta, err := backup.ReadMetadata(stg, b.Name)
 		if err != nil {
-			l.Error("read meta for %v", b.Name)
+			l.Error("read metadata of backup %s: %v", b.Name, err)
 			continue
 		}
 
-		var meta *backup.BackupMeta
-		err = json.NewDecoder(d).Decode(&meta)
-		d.Close()
-		if err != nil {
-			l.Error("unmarshal backup meta [%s]", b.Name)
-			continue
-		}
-
-		err = backup.CheckBackupFiles(ctx, meta, stg)
+		err = backup.CheckBackupDataFiles(ctx, stg, meta)
 		if err != nil {
 			l.Warning("skip snapshot %s: %v", meta.Name, err)
 			meta.Status = defs.StatusError