refactor CBT retrievement to report the concrete error

Signed-off-by: Yonghui Li <lyonghui@vmware.com>

pkg/uploader/block/snapshot.go

@@ -40,6 +40,10 @@ type parentBackupInfo struct {
 	volumeID     string
 }
 
+type backupInfo struct {
+	changeID string
+}
+
 // Backup backup specific sourcePath and update progress
 func Backup(ctx context.Context, blkUp Uploader, repoWriter udmrepo.BackupRepo, sourcePath string, realSource string, cbtSource cbtservice.SourceInfo,
 	forceFull bool, parentSnapshot string, cbtService cbtservice.Service, uploaderCfg map[string]string, tags map[string]string, log logrus.FieldLogger) (uploader.SnapshotInfo, bool, error) {
@@ -106,11 +110,17 @@ func snapshotSource(
 	log.Info("Start to snapshot...")
 	snapshotStartTime := time.Now()
 
-	parentBackup := getParentBackupInfo(ctx, rep, forceFull, parentSnapshot, cbtSource.VolumeID, source.realSource, snapshotTags, log)
+	bitmap := cbt.NewBitmap(blockSize, uint64(source.size), cbtSource.Snapshot, cbtSource.VolumeID)
 
-	bitmap := cbt.NewBitmap(blockSize, uint64(source.size), cbtSource.Snapshot, parentBackup.changeID, parentBackup.volumeID)
+	parentBackup, err := getParentBackupInfo(ctx, rep, forceFull, parentSnapshot, cbtSource.VolumeID, source.realSource, snapshotTags, log)
+	if err != nil {
+		log.WithError(err).Warn("Failed to get parent backup info, fallback to full backup")
+		bitmap.SetError(errors.Wrap(err, "error getting parent backup info, fallback to full backup"))
+	} else {
+		bitmap.SetChangeID(parentBackup.changeID)
+	}
 
-	err := cbt.SetBitmapOrFull(ctx, cbtService, bitmap, false)
+	err = cbt.SetBitmapOrFull(ctx, cbtService, bitmap, false)
 	if err != nil {
 		parentBackup.parentObject = ""
 		log.WithError(err).Warnf("Failed to create CBT with source %v", cbtSource)
@@ -147,61 +157,67 @@ func snapshotSource(
 	return string(snapID), backupSize, nil
 }
 
-func getParentBackupInfo(ctx context.Context, rep udmrepo.BackupRepo, forceFull bool, parentSnapshot string, volumeID string, realSource string, snapshotTags map[string]string, log logrus.FieldLogger) parentBackupInfo {
-	var previous *udmrepo.Snapshot
-
-	// parentID names whichever snapshot ended up being the parent. On the discovery
-	// branch the parentSnapshot parameter is empty by definition, so logging it there
-	// produces messages that describe a decision without naming the object it was about.
-	parentID := parentSnapshot
-
-	if !forceFull {
-		if parentSnapshot != "" {
-			snap, err := rep.GetSnapshot(ctx, udmrepo.ID(parentSnapshot))
-			if err != nil {
-				log.WithError(err).Warn("Failed to load previous snapshot, fallback to full backup")
-			} else {
-				previous = &snap
-				log.Infof("Using provided parent snapshot %s", parentSnapshot)
-			}
-		} else {
-			log.Infof("Searching for parent snapshot")
-
-			snap, err := findPreviousSnapshot(ctx, rep, realSource, snapshotTags, nil, log)
-			if err != nil {
-				log.WithError(err).Warn("Failed to search previous snapshot, fallback to full backup")
-			} else {
-				previous = &snap
-				parentID = string(snap.RootObject.ID)
-				log.Infof("Using previous snapshot %s", snap.RootObject.ID)
-			}
-		}
-	} else {
+func getParentBackupInfo(ctx context.Context, rep udmrepo.BackupRepo, forceFull bool, parentSnapshot string, volumeID string,
+	realSource string, snapshotTags map[string]string, log logrus.FieldLogger) (parentBackupInfo, error) {
+	if forceFull {
 		log.Info("Forcing full snapshot")
+		return parentBackupInfo{}, nil
 	}
 
-	parentInfo := parentBackupInfo{}
-	if previous != nil {
-		if previous.Tags == nil {
-			log.Warnf("No tag from parent snapshot %s, fallback to full backup", parentID)
-		} else if previous.Tags[uploader.CBTChangeIDTag] == "" {
-			log.Warnf("No ChangeID tag from parent snapshot %s, fallback to full backup", parentID)
-		} else if previous.Tags[uploader.CBTVolumeIDTag] == "" {
-			log.Warnf("No VolumeID tag from parent snapshot %s, fallback to full backup", parentID)
-		} else if previous.Tags[uploader.CBTVolumeIDTag] != volumeID {
-			log.Warnf("VolumeID %s from parent snapshot %s is not expected as %s, fallback to full backup", previous.Tags[uploader.CBTVolumeIDTag], parentID, volumeID)
-		} else if obj, err := loadObjectFromSnapshot(ctx, rep, previous); err != nil {
-			log.WithError(err).Warnf("Failed to load object from parent snapshot %s, fallback to full backup", parentID)
-		} else {
-			parentInfo.parentObject = obj
-			parentInfo.changeID = previous.Tags[uploader.CBTChangeIDTag]
-			parentInfo.volumeID = previous.Tags[uploader.CBTVolumeIDTag]
+	if volumeID == "" {
+		return parentBackupInfo{}, errors.New("volumeID is not provided from the volume snapshot")
+	}
 
-			log.Infof("Using parent snapshot %s, start time %v, end time %v, description %s", parentID, previous.StartTime, previous.EndTime, previous.Description)
+	var previous *udmrepo.Snapshot
+	if parentSnapshot != "" {
+		log.Infof("Loading provided parent snapshot %s", parentSnapshot)
+
+		snap, err := rep.GetSnapshot(ctx, udmrepo.ID(parentSnapshot))
+		if err != nil {
+			return parentBackupInfo{}, errors.Wrapf(err, "error loading previous snapshot")
 		}
+
+		previous = &snap
+
+	} else {
+		log.Infof("Searching for parent snapshot")
+
+		snap, err := findPreviousSnapshot(ctx, rep, realSource, snapshotTags, nil, log)
+		if err != nil {
+			return parentBackupInfo{}, errors.Wrapf(err, "error searching previous snapshot")
+		}
+
+		previous = &snap
 	}
 
-	return parentInfo
+	if previous.Tags == nil {
+		return parentBackupInfo{}, errors.Errorf("no tag from parent snapshot %s", previous.ID)
+	}
+
+	if previous.Tags[uploader.CBTChangeIDTag] == "" {
+		return parentBackupInfo{}, errors.Errorf("no ChangeID tag from parent snapshot %s", previous.ID)
+	}
+
+	if previous.Tags[uploader.CBTVolumeIDTag] == "" {
+		return parentBackupInfo{}, errors.Errorf("no VolumeID tag from parent snapshot %s", previous.ID)
+	}
+
+	if previous.Tags[uploader.CBTVolumeIDTag] != volumeID {
+		return parentBackupInfo{}, errors.Errorf("VolumeID %s from parent snapshot %s is not expected as %s", previous.Tags[uploader.CBTVolumeIDTag], previous.ID, volumeID)
+	}
+
+	obj, err := loadObjectFromSnapshot(ctx, rep, previous)
+	if err != nil {
+		return parentBackupInfo{}, errors.Errorf("error loading object from parent snapshot %s", previous.ID)
+	}
+
+	log.Infof("Using parent snapshot %s, start time %v, end time %v, description %s", previous.ID, previous.StartTime, previous.EndTime, previous.Description)
+
+	return parentBackupInfo{
+		parentObject: obj,
+		changeID:     previous.Tags[uploader.CBTChangeIDTag],
+		volumeID:     previous.Tags[uploader.CBTVolumeIDTag],
+	}, nil
 }
 
 // Restore restore specific sourcePath with given snapshotID and update progress
@@ -214,29 +230,19 @@ func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapsh
 	}
 	log.Infof("Restore from snapshot %s, incremental %v, cbt source %v, description %s, created time %v, tags %v", snapshotID, incremental, cbtSource, snapshot.Description, snapshot.EndTime, snapshot.Tags)
 
-	var volumeSnapshot, changeID, volumeID string
-	if incremental {
-		if snapshot.Tags == nil {
-			log.Warnf("No tag from snapshot %s, fallback to full restore", snapshotID)
-		} else if snapshot.Tags[uploader.CBTChangeIDTag] == "" {
-			log.Warnf("No ChangeID tag from snapshot %s, fallback to full restore", snapshotID)
-		} else if snapshot.Tags[uploader.CBTVolumeIDTag] == "" {
-			log.Warnf("No VolumeID tag from snapshot %s, fallback to full restore", snapshotID)
-		} else if cbtSource.VolumeID == "" {
-			log.Warnf("No VolumeID in cbt source %v, fallback to full restore", cbtSource)
-		} else if snapshot.Tags[uploader.CBTVolumeIDTag] != cbtSource.VolumeID {
-			log.Warnf("VolumeID %s from snapshot %s is not expected as %s, fallback to full restore", snapshot.Tags[uploader.CBTVolumeIDTag], snapshotID, cbtSource.VolumeID)
-		} else {
-			volumeSnapshot = cbtSource.Snapshot
-			changeID = snapshot.Tags[uploader.CBTChangeIDTag]
-			volumeID = snapshot.Tags[uploader.CBTVolumeIDTag]
-		}
-	}
+	bitmap := cbt.NewBitmap(blockSize, uint64(snapshot.TotalSize), cbtSource.Snapshot, cbtSource.VolumeID)
 
-	bitmap := cbt.NewBitmap(blockSize, uint64(snapshot.TotalSize), volumeSnapshot, changeID, volumeID)
 	if incremental {
-		if err = cbt.SetBitmapOrFull(ctx, cbtService, bitmap, true); err != nil {
-			log.WithError(err).Warnf("Failed to create CBT with source %v", cbtSource)
+		if bkInfo, err := getBackupInfo(snapshot, cbtSource.VolumeID); err != nil {
+			log.WithError(err).Warn("Failed to get backup info, fallback to full restore")
+
+			bitmap.SetError(errors.Wrap(err, "error getting backup info, fallback to full restore"))
+			bitmap.SetFull()
+		} else {
+			bitmap.SetChangeID(bkInfo.changeID)
+			if err = cbt.SetBitmapOrFull(ctx, cbtService, bitmap, true); err != nil {
+				log.WithError(err).Warnf("Failed to create CBT with source %v", cbtSource)
+			}
 		}
 	} else {
 		bitmap.SetFull()
@@ -274,6 +280,32 @@ func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapsh
 	return incrementalBytes, totalSize, nil
 }
 
+func getBackupInfo(snapshot udmrepo.Snapshot, volumeID string) (backupInfo, error) {
+	if snapshot.Tags == nil {
+		return backupInfo{}, errors.Errorf("no tag from snapshot %s", snapshot.ID)
+	}
+
+	if snapshot.Tags[uploader.CBTChangeIDTag] == "" {
+		return backupInfo{}, errors.Errorf("no ChangeID tag from snapshot %s", snapshot.ID)
+	}
+
+	if snapshot.Tags[uploader.CBTVolumeIDTag] == "" {
+		return backupInfo{}, errors.Errorf("no VolumeID tag from snapshot %s", snapshot.ID)
+	}
+
+	if volumeID == "" {
+		return backupInfo{}, errors.New("no VolumeID tag from the volume snapshot")
+	}
+
+	if snapshot.Tags[uploader.CBTVolumeIDTag] != volumeID {
+		return backupInfo{}, errors.Errorf("volumeID %s from snapshot %s is not expected as %s", snapshot.Tags[uploader.CBTVolumeIDTag], snapshot.ID, volumeID)
+	}
+
+	return backupInfo{
+		changeID: snapshot.Tags[uploader.CBTChangeIDTag],
+	}, nil
+}
+
 func findPreviousSnapshot(ctx context.Context, rep udmrepo.BackupRepo, path string, snapshotTags map[string]string, noLaterThan *time.Time, log logrus.FieldLogger) (udmrepo.Snapshot, error) {
 	snaps, err := rep.ListSnapshot(ctx, path)
 	if err != nil {

pkg/uploader/block/uploader.go

@@ -86,8 +86,10 @@ func (blkup *blockUploader) Backup(source sourceInfo, parentObject udmrepo.ID, b
 		return udmrepo.Snapshot{}, 0, errors.New("bitmap is not available")
 	}
 
-	if bitmap.Error() != nil {
-		blkup.progress.UpdateProgress(&uploader.Progress{BytesDone: -1, TotalBytes: -1, Message: bitmap.Error().Error()})
+	if bitmap.Errors() != nil {
+		for _, err := range bitmap.Errors() {
+			blkup.progress.UpdateProgress(&uploader.Progress{BytesDone: -1, TotalBytes: -1, Message: err.Error()})
+		}
 	}
 
 	backupMode := udmrepo.ObjectDataBackupModeInc
@@ -157,8 +159,10 @@ func (blkup *blockUploader) Restore(snapshot udmrepo.Snapshot, dest destInfo, bi
 		return 0, 0, errors.New("bitmap is not available")
 	}
 
-	if bitmap.Error() != nil {
-		blkup.progress.UpdateProgress(&uploader.Progress{BytesDone: -1, TotalBytes: -1, Message: bitmap.Error().Error()})
+	if bitmap.Errors() != nil {
+		for _, err := range bitmap.Errors() {
+			blkup.progress.UpdateProgress(&uploader.Progress{BytesDone: -1, TotalBytes: -1, Message: err.Error()})
+		}
 	}
 
 	meta, err := blkup.repoWriter.ReadMetadata(blkup.ctx, snapshot.RootObject.ID)

pkg/uploader/cbt/bitmap.go

@@ -36,7 +36,7 @@ type bitmapImpl struct {
 	snapshot     string
 	changeID     string
 	volumeID     string
-	cbtError     error
+	cbtErrors    []error
 }
 
 type bitmapIterator struct {
@@ -44,14 +44,13 @@ type bitmapIterator struct {
 	iterator roaring.IntPeekable
 }
 
-func NewBitmap(blockSize uint, length uint64, snapshot string, changeID string, volumeID string) types.Bitmap {
+func NewBitmap(blockSize uint, length uint64, snapshot string, volumeID string) types.Bitmap {
 	return &bitmapImpl{
 		bitmap:       roaring.New(),
 		blockSize:    blockSize,
 		blockSizeLog: bits.Len(blockSize) - 1,
 		length:       length,
 		snapshot:     snapshot,
-		changeID:     changeID,
 		volumeID:     volumeID,
 	}
 }
@@ -82,6 +81,10 @@ func (c *bitmapImpl) Snapshot() string {
 	return c.snapshot
 }
 
+func (c *bitmapImpl) SetChangeID(id string) {
+	c.changeID = id
+}
+
 func (c *bitmapImpl) ChangeID() string {
 	return c.changeID
 }
@@ -91,11 +94,11 @@ func (c *bitmapImpl) VolumeID() string {
 }
 
 func (c *bitmapImpl) SetError(err error) {
-	c.cbtError = err
+	c.cbtErrors = append(c.cbtErrors, err)
 }
 
-func (c *bitmapImpl) Error() error {
-	return c.cbtError
+func (c *bitmapImpl) Errors() []error {
+	return c.cbtErrors
 }
 
 func (c *bitmapImpl) Iterator() types.Iterator {
@@ -125,6 +128,6 @@ func (c *bitmapIterator) BlockSize() uint {
 	return c.blockSize
 }
 
-func (c *bitmapIterator) Error() error {
-	return c.cbtError
+func (c *bitmapIterator) Errors() []error {
+	return c.cbtErrors
 }

pkg/uploader/cbt/types/types.go

@@ -39,8 +39,11 @@ type Bitmap interface {
 	// SetError sets CBT error when preparing this bitmap
 	SetError(error)
 
-	// Error returns the CBT error when preparing this bitmap
-	Error() error
+	// Errors returns the CBT errors when preparing this bitmap
+	Errors() []error
+
+	// SetChangeID sets the changeID of the bitmap
+	SetChangeID(string)
 }
 
 // Iterator defines the methods to iterate the CBT bitmap and query the associated information
@@ -63,6 +66,6 @@ type Iterator interface {
 	// Next returns the offset of the next set block and whether it comes to the end of the iteration
 	Next() (uint64, bool)
 
-	// Error returns the CBT error when preparing this bitmap
-	Error() error
+	// Errors returns the CBT errors when preparing this bitmap
+	Errors() []error
 }