From 67923bca6f9f97967448fba7daadad83c1552462 Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Fri, 4 Sep 2026 14:17:28 +0800 Subject: [PATCH] uploader report incremental fallback Signed-off-by: Lyndon-Li --- pkg/datapath/data_path.go | 6 +++++- pkg/uploader/block/uploader.go | 8 ++++++++ pkg/uploader/kopia/snapshot.go | 25 +++++++++++++++---------- pkg/uploader/provider/kopia.go | 2 +- pkg/uploader/types.go | 5 +++-- 5 files changed, 32 insertions(+), 14 deletions(-) diff --git a/pkg/datapath/data_path.go b/pkg/datapath/data_path.go index 0513619be..2a4f4e18b 100644 --- a/pkg/datapath/data_path.go +++ b/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, + }) } } diff --git a/pkg/uploader/block/uploader.go b/pkg/uploader/block/uploader.go index e2d464872..4b69de59f 100644 --- a/pkg/uploader/block/uploader.go +++ b/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) diff --git a/pkg/uploader/kopia/snapshot.go b/pkg/uploader/kopia/snapshot.go index fae7a517c..682b0f314 100644 --- a/pkg/uploader/kopia/snapshot.go +++ b/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") diff --git a/pkg/uploader/provider/kopia.go b/pkg/uploader/provider/kopia.go index 1cd1ecd45..39abbbe8f 100644 --- a/pkg/uploader/provider/kopia.go +++ b/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 { diff --git a/pkg/uploader/types.go b/pkg/uploader/types.go index 9c700193f..a608add06 100644 --- a/pkg/uploader/types.go +++ b/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