uploader report incremental fallback

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

pkg/datapath/data_path.go

@@ -278,7 +278,11 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u
 // UpdateProgress which implement ProgressUpdater interface to update progress status
 func (dp *generalDataPath) UpdateProgress(p *uploader.Progress) {
 	if dp.callbacks.OnProgress != nil {
-		dp.callbacks.OnProgress(context.Background(), dp.namespace, dp.jobName, &uploader.Progress{TotalBytes: p.TotalBytes, BytesDone: p.BytesDone})
+		dp.callbacks.OnProgress(context.Background(), dp.namespace, dp.jobName, &uploader.Progress{
+			TotalBytes: p.TotalBytes,
+			BytesDone:  p.BytesDone,
+			Message:    p.Message,
+		})
 	}
 }
 

pkg/uploader/block/uploader.go

@@ -86,6 +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()})
+	}
+
 	backupMode := udmrepo.ObjectDataBackupModeInc
 	if parentObject == "" {
 		backupMode = udmrepo.ObjectDataBackupModeFull
@@ -153,6 +157,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()})
+	}
+
 	meta, err := blkup.repoWriter.ReadMetadata(blkup.ctx, snapshot.RootObject.ID)
 	if err != nil {
 		return 0, 0, errors.Wrapf(err, "error reading snapshot metadata for %s", snapshot.Description)

pkg/uploader/kopia/snapshot.go

@@ -153,7 +153,8 @@ func setupPolicy(ctx context.Context, rep repo.RepositoryWriter, sourceInfo snap
 
 // Backup backup specific sourcePath and update progress
 func Backup(ctx context.Context, fsUploader SnapshotUploader, repoWriter repo.RepositoryWriter, sourcePath string, realSource string,
-	forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error) {
+	forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, tags map[string]string,
+	updater uploader.ProgressUpdater, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error) {
 	if fsUploader == nil {
 		return nil, false, errors.New("get empty kopia uploader")
 	}
@@ -189,7 +190,7 @@ func Backup(ctx context.Context, fsUploader SnapshotUploader, repoWriter repo.Re
 
 	kopiaCtx := kopia.SetupKopiaLog(ctx, log)
 
-	snapID, snapshotSize, err := SnapshotSource(kopiaCtx, repoWriter, fsUploader, sourceInfo, sourceEntry, forceFull, parentSnapshot, tags, uploaderCfg, log, "Kopia Uploader")
+	snapID, snapshotSize, err := SnapshotSource(kopiaCtx, repoWriter, fsUploader, sourceInfo, sourceEntry, forceFull, parentSnapshot, tags, uploaderCfg, updater, log, "Kopia Uploader")
 	snapshotInfo := &uploader.SnapshotInfo{
 		ID:   snapID,
 		Size: snapshotSize,
@@ -237,6 +238,7 @@ func SnapshotSource(
 	parentSnapshot string,
 	snapshotTags map[string]string,
 	uploaderCfg map[string]string,
+	updater uploader.ProgressUpdater,
 	log logrus.FieldLogger,
 	description string,
 ) (string, int64, error) {
@@ -248,21 +250,24 @@ func SnapshotSource(
 		if parentSnapshot != "" {
 			log.Infof("Using provided parent snapshot %s", parentSnapshot)
 
-			mani, err := loadSnapshotFunc(ctx, rep, manifest.ID(parentSnapshot))
-			if err != nil {
-				log.WithError(err).Warnf("Failed to load previous snapshot %v from kopia, fallback to full backup", parentSnapshot)
+			if mani, err := loadSnapshotFunc(ctx, rep, manifest.ID(parentSnapshot)); err != nil {
+				msg := fmt.Sprintf("Failed to load previous snapshot %v from kopia, fallback to full backup", parentSnapshot)
+				log.WithError(err).Warn(msg)
+				updater.UpdateProgress(&uploader.Progress{BytesDone: -1, TotalBytes: -1, Message: msg})
 			} else {
 				previous = append(previous, mani)
 			}
 		} else {
 			log.Infof("Searching for parent snapshot")
 
-			pre, err := findPreviousSnapshotManifest(ctx, rep, sourceInfo, snapshotTags, nil, log)
-			if err != nil {
-				return "", 0, errors.Wrapf(err, "Failed to find previous kopia snapshot manifests for si %v", sourceInfo)
-			}
+			if pre, err := findPreviousSnapshotManifest(ctx, rep, sourceInfo, snapshotTags, nil, log); err != nil {
+				log.WithError(err).Warnf("Failed to find previous kopia snapshot manifests for si %v, fallback to full backup", sourceInfo)
 
-			previous = pre
+				msg := fmt.Sprint("Failed to find previous kopia snapshot manifests, fallback to full backup")
+				updater.UpdateProgress(&uploader.Progress{BytesDone: -1, TotalBytes: -1, Message: msg})
+			} else {
+				previous = pre
+			}
 		}
 	} else {
 		log.Info("Forcing full snapshot")

pkg/uploader/provider/kopia.go

@@ -166,7 +166,7 @@ func (kp *kopiaProvider) RunBackup(
 		uploaderCfg[kopia.UploaderConfigMultipartKey] = "true"
 	}
 
-	snapshotInfo, _, err := kopiaBackupFunc(ctx, kpUploader, repoWriter, path, realSource, forceFull, parentSnapshot, volMode, uploaderCfg, tags, log)
+	snapshotInfo, _, err := kopiaBackupFunc(ctx, kpUploader, repoWriter, path, realSource, forceFull, parentSnapshot, volMode, uploaderCfg, tags, updater, log)
 	if err != nil {
 		snapshotID := ""
 		if snapshotInfo != nil {

pkg/uploader/types.go

@@ -58,8 +58,9 @@ type SnapshotInfo struct {
 
 // Progress which defined two variables to record progress
 type Progress struct {
-	TotalBytes int64 `json:"totalBytes,omitempty"`
-	BytesDone  int64 `json:"doneBytes,omitempty"`
+	TotalBytes int64  `json:"totalBytes,omitempty"`
+	BytesDone  int64  `json:"doneBytes,omitempty"`
+	Message    string `json:"message,omitempty"`
 }
 
 // UploaderProgress which defined generic interface to update progress