diff --git a/pkg/uploader/block/snapshot.go b/pkg/uploader/block/snapshot.go index cefd422cd..cff152805 100644 --- a/pkg/uploader/block/snapshot.go +++ b/pkg/uploader/block/snapshot.go @@ -83,12 +83,13 @@ func Backup(ctx context.Context, blkUp Uploader, repoWriter udmrepo.BackupRepo, return uploader.SnapshotInfo{}, false, errors.Wrapf(err, "error reset pos of block device %s", source) } - snapID, backupSize, snapshotSize, err := snapshotSource(ctx, repoWriter, blkUp, sourceInfo, forceFull, parentSnapshot, cbtSource, cbtService, tags, uploaderCfg, log, "Block Uploader") + snapID, backupSize, snapshotSize, fallback, err := snapshotSource(ctx, repoWriter, blkUp, sourceInfo, forceFull, parentSnapshot, cbtSource, cbtService, tags, uploaderCfg, log, "Block Uploader") snapshotInfo := uploader.SnapshotInfo{ ID: snapID, SnapshotSize: snapshotSize, IncrementalSize: backupSize, SourceSize: sourceInfo.size, + Fallback: fallback, } return snapshotInfo, false, err @@ -107,7 +108,7 @@ func snapshotSource( uploaderCfg map[string]string, log logrus.FieldLogger, description string, -) (string, int64, int64, error) { +) (string, int64, int64, bool, error) { log.Info("Start to snapshot...") snapshotStartTime := time.Now() @@ -127,9 +128,11 @@ func snapshotSource( log.WithError(err).Warnf("Failed to create CBT with source %v", cbtSource) } + fallback := (len(bitmap.Errors()) > 0) + snap, backupSize, err := u.Backup(source, parentBackup.parentObject, bitmap.Iterator(), uploaderCfg) if err != nil { - return "", 0, 0, errors.Wrapf(err, "Failed to run uploader backup for si %v", source) + return "", 0, 0, fallback, errors.Wrapf(err, "Failed to run uploader backup for si %v", source) } if snap.Tags == nil { @@ -146,16 +149,16 @@ func snapshotSource( snapID, err := rep.SaveSnapshot(ctx, snap) if err != nil { - return "", 0, 0, errors.Wrapf(err, "Failed to save snapshot %v", snap) + return "", 0, 0, fallback, errors.Wrapf(err, "Failed to save snapshot %v", snap) } if err = rep.Flush(ctx); err != nil { - return "", 0, 0, errors.Wrapf(err, "Failed to flush repository") + return "", 0, 0, fallback, errors.Wrapf(err, "Failed to flush repository") } log.Infof("Created snapshot with root %v and ID %v in %v", snap.RootObject, snapID, time.Since(snapshotStartTime).Truncate(time.Second)) - return string(snapID), backupSize, snap.TotalSize, nil + return string(snapID), backupSize, snap.TotalSize, fallback, nil } func getParentBackupInfo(ctx context.Context, rep udmrepo.BackupRepo, forceFull bool, parentSnapshot string, volumeID string, @@ -221,12 +224,12 @@ func getParentBackupInfo(ctx context.Context, rep udmrepo.BackupRepo, forceFull } // Restore restore specific sourcePath with given snapshotID and update progress -func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapshotID, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, int64, error) { +func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapshotID, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, int64, bool, error) { log.Info("Start to restore...") snapshot, err := rep.GetSnapshot(ctx, udmrepo.ID(snapshotID)) if err != nil { - return 0, 0, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID) + return 0, 0, false, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID) } 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) @@ -248,36 +251,38 @@ func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapsh bitmap.SetFull() } + fallback := (len(bitmap.Errors()) > 0) + destPath, err := filepath.Abs(dest) if err != nil { - return 0, 0, errors.Wrapf(err, "invalid dest path '%s'", dest) + return 0, 0, fallback, errors.Wrapf(err, "invalid dest path '%s'", dest) } destPath = filepath.Clean(destPath) destDev, err := openBlockDeviceFunc(destPath, false) if err != nil { - return 0, 0, errors.Wrapf(err, "error opening block device '%s'", destPath) + return 0, 0, fallback, errors.Wrapf(err, "error opening block device '%s'", destPath) } defer destDev.Close() destSize, err := destDev.Seek(0, io.SeekEnd) if err != nil { - return 0, 0, errors.Wrapf(err, "error getting length of block device %s", dest) + return 0, 0, fallback, errors.Wrapf(err, "error getting length of block device %s", dest) } _, err = destDev.Seek(0, io.SeekStart) if err != nil { - return 0, 0, errors.Wrapf(err, "error reset pos of block device %s", dest) + return 0, 0, fallback, errors.Wrapf(err, "error reset pos of block device %s", dest) } incrementalBytes, totalSize, err := blkUp.Restore(snapshot, destInfo{dev: destDev, path: destPath, size: destSize}, bitmap.Iterator(), uploaderCfg) if err != nil { - return 0, 0, errors.Wrapf(err, "error restoring to block dev %s", destPath) + return 0, 0, fallback, errors.Wrapf(err, "error restoring to block dev %s", destPath) } - return incrementalBytes, totalSize, nil + return incrementalBytes, totalSize, fallback, nil } func getBackupInfo(snapshot udmrepo.Snapshot, volumeID string) (backupInfo, error) { diff --git a/pkg/uploader/block/snapshot_test.go b/pkg/uploader/block/snapshot_test.go index 71631163a..14f1b5a75 100644 --- a/pkg/uploader/block/snapshot_test.go +++ b/pkg/uploader/block/snapshot_test.go @@ -215,6 +215,7 @@ func TestSnapshotSource(t *testing.T) { expectedSnapID string expectedSize int64 expectedSnapshotSize int64 + expectedFallback bool cbtService func(t *testing.T) cbtservice.Service }{ { @@ -259,6 +260,7 @@ func TestSnapshotSource(t *testing.T) { expectedSnapID: "snap-success", expectedSize: 512, expectedSnapshotSize: 2048, + expectedFallback: true, }, { name: "tags from cbtSource and snapshotTags are merged onto snapshot", @@ -276,6 +278,7 @@ func TestSnapshotSource(t *testing.T) { expectedSnapID: "snap-tags", expectedSize: 256, expectedSnapshotSize: 4096, + expectedFallback: true, }, { name: "success with cbtService getting allocated blocks", @@ -298,6 +301,7 @@ func TestSnapshotSource(t *testing.T) { expectedSnapID: "snap-cbt-alloc", expectedSize: 1024, expectedSnapshotSize: 8192, + expectedFallback: false, }, { name: "cbtService error falls back to full", @@ -318,6 +322,7 @@ func TestSnapshotSource(t *testing.T) { expectedSnapID: "snap-cbt-fallback", expectedSize: 1024, expectedSnapshotSize: 1024, + expectedFallback: true, }, } @@ -337,7 +342,7 @@ func TestSnapshotSource(t *testing.T) { cbtSvc = tc.cbtService(t) } - snapID, size, snapshotSize, err := snapshotSource( + snapID, size, snapshotSize, fallback, err := snapshotSource( ctx, mockRepo, mockBlkup, baseSource, true, "", @@ -354,6 +359,7 @@ func TestSnapshotSource(t *testing.T) { assert.Equal(t, tc.expectedSnapID, snapID) assert.Equal(t, tc.expectedSize, size) assert.Equal(t, tc.expectedSnapshotSize, snapshotSize) + assert.Equal(t, tc.expectedFallback, fallback) } mockBlkup.AssertExpectations(t) @@ -834,14 +840,15 @@ func TestRestore(t *testing.T) { storedSnap := udmrepo.Snapshot{Description: "test snapshot"} testCases := []struct { - name string - incremental bool - cbtSource cbtservice.SourceInfo - cbtService func(t *testing.T) cbtservice.Service - setupMocks func(blkup *mockUploader, repo *udmrepomocks.BackupRepo) - setupOpenDev func(t *testing.T) *os.File - expectedErrStr string - expectedSize int64 + name string + incremental bool + cbtSource cbtservice.SourceInfo + cbtService func(t *testing.T) cbtservice.Service + setupMocks func(blkup *mockUploader, repo *udmrepomocks.BackupRepo) + setupOpenDev func(t *testing.T) *os.File + expectedErrStr string + expectedSize int64 + expectedFallback bool }{ { name: "GetSnapshot error", @@ -885,7 +892,8 @@ func TestRestore(t *testing.T) { t.Helper() return tempFile(t, "") }, - expectedSize: 4096, + expectedSize: 4096, + expectedFallback: false, }, { name: "incremental restore success", @@ -917,7 +925,8 @@ func TestRestore(t *testing.T) { t.Helper() return tempFile(t, "") }, - expectedSize: 512, + expectedSize: 512, + expectedFallback: false, }, { name: "incremental restore fallback - missing tags", @@ -931,7 +940,8 @@ func TestRestore(t *testing.T) { t.Helper() return tempFile(t, "") }, - expectedSize: 4096, + expectedSize: 4096, + expectedFallback: true, }, { name: "incremental restore fallback - empty cbtSource VolumeID", @@ -952,7 +962,8 @@ func TestRestore(t *testing.T) { t.Helper() return tempFile(t, "") }, - expectedSize: 4096, + expectedSize: 4096, + expectedFallback: true, }, { name: "incremental restore fallback - VolumeID mismatch", @@ -973,7 +984,8 @@ func TestRestore(t *testing.T) { t.Helper() return tempFile(t, "") }, - expectedSize: 4096, + expectedSize: 4096, + expectedFallback: true, }, { name: "incremental restore fallback - CBT service error", @@ -1002,7 +1014,8 @@ func TestRestore(t *testing.T) { t.Helper() return tempFile(t, "") }, - expectedSize: 1024, + expectedSize: 1024, + expectedFallback: true, }, } @@ -1030,7 +1043,7 @@ func TestRestore(t *testing.T) { cbtSvc = tc.cbtService(t) } - _, size, err := Restore(ctx, mockBlkup, mockRepo, "snap-001", "/dev/sdb", tc.incremental, tc.cbtSource, cbtSvc, map[string]string{}, testLog()) + _, size, fallback, err := Restore(ctx, mockBlkup, mockRepo, "snap-001", "/dev/sdb", tc.incremental, tc.cbtSource, cbtSvc, map[string]string{}, testLog()) if tc.expectedErrStr != "" { require.Error(t, err) @@ -1039,6 +1052,7 @@ func TestRestore(t *testing.T) { } else { require.NoError(t, err) assert.Equal(t, tc.expectedSize, size) + assert.Equal(t, tc.expectedFallback, fallback) } mockBlkup.AssertExpectations(t) diff --git a/pkg/uploader/kopia/snapshot.go b/pkg/uploader/kopia/snapshot.go index d07241388..783c1f9e4 100644 --- a/pkg/uploader/kopia/snapshot.go +++ b/pkg/uploader/kopia/snapshot.go @@ -190,11 +190,12 @@ 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, updater, log, "Kopia Uploader") + snapID, snapshotSize, fallback, err := SnapshotSource(kopiaCtx, repoWriter, fsUploader, sourceInfo, sourceEntry, forceFull, parentSnapshot, tags, uploaderCfg, updater, log, "Kopia Uploader") snapshotInfo := &uploader.SnapshotInfo{ ID: snapID, SnapshotSize: snapshotSize, SourceSize: snapshotSize, + Fallback: fallback, } return snapshotInfo, false, err @@ -242,11 +243,12 @@ func SnapshotSource( updater uploader.ProgressUpdater, log logrus.FieldLogger, description string, -) (string, int64, error) { +) (string, int64, bool, error) { log.Info("Start to snapshot...") snapshotStartTime := time.Now() var previous []*snapshot.Manifest + fallback := false if !forceFull { if parentSnapshot != "" { log.Infof("Using provided parent snapshot %s", parentSnapshot) @@ -258,6 +260,8 @@ func SnapshotSource( TotalBytes: -1, Message: fmt.Sprintf("Failed to load previous snapshot %v, fallback to full backup. Err: %v", parentSnapshot, err), }) + + fallback = true } else { previous = append(previous, mani) } @@ -271,6 +275,8 @@ func SnapshotSource( TotalBytes: -1, Message: fmt.Sprintf("Failed to find previous snapshots, fallback to full backup. Err: %v", err), }) + + fallback = true } else { previous = pre } @@ -285,12 +291,12 @@ func SnapshotSource( policyTree, err := setupPolicy(ctx, rep, sourceInfo, uploaderCfg) if err != nil { - return "", 0, errors.Wrapf(err, "unable to set policy for si %v", sourceInfo) + return "", 0, fallback, errors.Wrapf(err, "unable to set policy for si %v", sourceInfo) } manifest, err := u.Upload(ctx, rootDir, policyTree, sourceInfo, previous...) if err != nil { - return "", 0, errors.Wrapf(err, "Failed to upload the kopia snapshot for si %v", sourceInfo) + return "", 0, fallback, errors.Wrapf(err, "Failed to upload the kopia snapshot for si %v", sourceInfo) } manifest.Tags = snapshotTags @@ -299,22 +305,22 @@ func SnapshotSource( manifest.Pins = []string{"velero-pin"} if _, err = saveSnapshotFunc(ctx, rep, manifest); err != nil { - return "", 0, errors.Wrapf(err, "Failed to save kopia manifest %v", manifest.ID) + return "", 0, fallback, errors.Wrapf(err, "Failed to save kopia manifest %v", manifest.ID) } _, err = applyRetentionPolicyFunc(ctx, rep, sourceInfo, true) if err != nil { - return "", 0, errors.Wrapf(err, "Failed to apply kopia retention policy for si %v", sourceInfo) + return "", 0, fallback, errors.Wrapf(err, "Failed to apply kopia retention policy for si %v", sourceInfo) } if err = rep.Flush(ctx); err != nil { - return "", 0, errors.Wrapf(err, "Failed to flush kopia repository") + return "", 0, fallback, errors.Wrapf(err, "Failed to flush kopia repository") } log.Infof("Created snapshot with root %v and ID %v in %v", manifest.RootObjectID(), manifest.ID, time.Since(snapshotStartTime).Truncate(time.Second)) - return reportSnapshotStatus(manifest, policyTree) + return reportSnapshotStatus(manifest, policyTree, fallback) } -func reportSnapshotStatus(manifest *snapshot.Manifest, policyTree *policy.Tree) (string, int64, error) { +func reportSnapshotStatus(manifest *snapshot.Manifest, policyTree *policy.Tree, fallback bool) (string, int64, bool, error) { manifestID := manifest.ID snapSize := manifest.Stats.TotalFileSize @@ -333,10 +339,10 @@ func reportSnapshotStatus(manifest *snapshot.Manifest, policyTree *policy.Tree) } if len(errs) != 0 { - return string(manifestID), snapSize, errors.New(strings.Join(errs, "\n")) + return string(manifestID), snapSize, fallback, errors.New(strings.Join(errs, "\n")) } - return string(manifestID), snapSize, nil + return string(manifestID), snapSize, fallback, nil } // findPreviousSnapshotManifest returns the list of previous snapshots for a given source, including @@ -401,26 +407,26 @@ func (o *fileSystemRestoreOutput) Terminate() error { // Restore restore specific sourcePath with given snapshotID and update progress func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, - log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) { + log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, bool, error) { log.Info("Start to restore...") kopiaCtx := kopia.SetupKopiaLog(ctx, log) snapshot, err := snapshot.LoadSnapshot(kopiaCtx, rep, manifest.ID(snapshotID)) if err != nil { - return 0, 0, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID) + return 0, 0, false, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID) } log.Infof("Restore from snapshot %s, description %s, created time %v, tags %v", snapshotID, snapshot.Description, snapshot.EndTime.ToTime(), snapshot.Tags) rootEntry, err := filesystemEntryFunc(kopiaCtx, rep, snapshotID, false) if err != nil { - return 0, 0, errors.Wrapf(err, "Unable to get filesystem entry for snapshot %v", snapshotID) + return 0, 0, false, errors.Wrapf(err, "Unable to get filesystem entry for snapshot %v", snapshotID) } path, err := filepath.Abs(dest) if err != nil { - return 0, 0, errors.Wrapf(err, "Unable to resolve path %v", dest) + return 0, 0, false, errors.Wrapf(err, "Unable to resolve path %v", dest) } fsOutput := &restore.FilesystemOutput{ @@ -436,7 +442,7 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, if len(uploaderCfg) > 0 { writeSparseFiles, err := uploaderutil.GetWriteSparseFiles(uploaderCfg) if err != nil { - return 0, 0, errors.Wrap(err, "failed to get uploader config") + return 0, 0, false, errors.Wrap(err, "failed to get uploader config") } if writeSparseFiles { fsOutput.WriteSparseFiles = true @@ -444,7 +450,7 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, concurrency, err := uploaderutil.GetRestoreConcurrency(uploaderCfg) if err != nil { - return 0, 0, errors.Wrap(err, "failed to get parallel restore uploader config") + return 0, 0, false, errors.Wrap(err, "failed to get parallel restore uploader config") } if concurrency > 0 { restoreConcurrency = concurrency @@ -452,7 +458,7 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, deleteExtra, err = uploaderutil.GetDeleteExtraFiles(uploaderCfg) if err != nil { - return 0, 0, errors.Wrap(err, "failed to get delete extra files config") + return 0, 0, false, errors.Wrap(err, "failed to get delete extra files config") } } @@ -460,7 +466,7 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, err = fsOutput.Init(ctx) if err != nil { - return 0, 0, errors.Wrap(err, "error to init output") + return 0, 0, false, errors.Wrap(err, "error to init output") } var output RestoreOutput @@ -500,18 +506,18 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, }) if err != nil { - return 0, 0, errors.Wrapf(err, "Failed to copy snapshot data to the target") + return 0, 0, false, errors.Wrapf(err, "Failed to copy snapshot data to the target") } if err := output.Flush(); err != nil { if err == errFlushUnsupported { log.Warnf("Skip flushing data for %v under the current OS %v", path, runtime.GOOS) } else { - return 0, 0, errors.Wrapf(err, "Failed to flush data to target") + return 0, 0, false, errors.Wrapf(err, "Failed to flush data to target") } } else { log.Infof("Flush done for volume dir %v", path) } - return stat.RestoredTotalFileSize, stat.RestoredFileCount, nil + return stat.RestoredTotalFileSize, stat.RestoredFileCount, false, nil } diff --git a/pkg/uploader/kopia/snapshot_test.go b/pkg/uploader/kopia/snapshot_test.go index 08c34befd..657bf298a 100644 --- a/pkg/uploader/kopia/snapshot_test.go +++ b/pkg/uploader/kopia/snapshot_test.go @@ -200,7 +200,7 @@ func TestSnapshotSource(t *testing.T) { t.Run(tc.name, func(t *testing.T) { s := injectSnapshotFuncs() MockFuncs(s, tc.args) - _, _, err = SnapshotSource(ctx, s.repoWriterMock, s.uploderMock, sourceInfo, rootDir, false, "/", nil, tc.uploaderCfg, &fakeProgressUpdater{}, log, "TestSnapshotSource") + _, _, _, err = SnapshotSource(ctx, s.repoWriterMock, s.uploderMock, sourceInfo, rootDir, false, "/", nil, tc.uploaderCfg, &fakeProgressUpdater{}, log, "TestSnapshotSource") if tc.notError { assert.NoError(t, err) } else { @@ -212,6 +212,7 @@ func TestSnapshotSource(t *testing.T) { func TestReportSnapshotStatus(t *testing.T) { testCases := []struct { + fallback bool shouldError bool expectedResult string expectedSize int64 @@ -219,6 +220,7 @@ func TestReportSnapshotStatus(t *testing.T) { expectedErrors []string }{ { + fallback: false, shouldError: false, expectedResult: "sample-manifest-id", expectedSize: 1024, @@ -227,6 +229,7 @@ func TestReportSnapshotStatus(t *testing.T) { }, }, { + fallback: true, shouldError: true, expectedResult: "sample-manifest-id", expectedSize: 1024, @@ -253,7 +256,7 @@ func TestReportSnapshotStatus(t *testing.T) { }, } - result, size, err := reportSnapshotStatus(manifest, policy.BuildTree(nil, getDefaultPolicy())) + result, size, fallback, err := reportSnapshotStatus(manifest, policy.BuildTree(nil, getDefaultPolicy()), tc.fallback) switch { case tc.shouldError && err == nil: @@ -274,6 +277,10 @@ func TestReportSnapshotStatus(t *testing.T) { if size != tc.expectedSize { t.Errorf("unexpected size: got %v, want %v", size, tc.expectedSize) } + + if fallback != tc.fallback { + t.Errorf("unexpected fallback: got %v, want %v", fallback, tc.fallback) + } } } @@ -819,7 +826,7 @@ func TestRestore(t *testing.T) { repoWriterMock.On("OpenObject", mock.Anything, mock.Anything).Return(em, nil) progress := new(Progress) - bytesRestored, fileCount, err := Restore(t.Context(), repoWriterMock, progress, tc.snapshotID, tc.dest, tc.incremental, tc.volMode, map[string]string{}, logrus.New(), nil) + bytesRestored, fileCount, fallback, err := Restore(t.Context(), repoWriterMock, progress, tc.snapshotID, tc.dest, tc.incremental, tc.volMode, map[string]string{}, logrus.New(), nil) // Check if the returned error matches the expected error if tc.expectedError != nil { @@ -833,6 +840,7 @@ func TestRestore(t *testing.T) { // Check the number of files restored assert.Equal(t, tc.expectedCount, fileCount) + assert.False(t, fallback) }) } } diff --git a/pkg/uploader/provider/block.go b/pkg/uploader/provider/block.go index 73c90a798..36ab80820 100644 --- a/pkg/uploader/provider/block.go +++ b/pkg/uploader/provider/block.go @@ -103,13 +103,13 @@ func (bp *blockProvider) RunBackup( cbtParam CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, - updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, error) { + updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, bool, error) { if updater == nil { - return "", false, 0, 0, 0, errors.New("backup progress updater is invalid") + return "", false, 0, 0, 0, false, errors.New("backup progress updater is invalid") } if path == "" { - return "", false, 0, 0, 0, errors.New("path is empty") + return "", false, 0, 0, 0, false, errors.New("path is empty") } log := bp.log.WithFields(logrus.Fields{ @@ -140,11 +140,11 @@ func (bp *blockProvider) RunBackup( // equality check never matches and cancellation gets reported as a failure. if errors.Is(err, block.ErrCanceled) { log.Warn("Block backup is canceled") - return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, ErrorCanceled + return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, snapshotInfo.Fallback, ErrorCanceled } if err != nil { - return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, errors.Wrapf(err, "Failed to run block backup") + return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, snapshotInfo.Fallback, errors.Wrapf(err, "Failed to run block backup") } updater.UpdateProgress( @@ -154,9 +154,9 @@ func (bp *blockProvider) RunBackup( }, ) - log.Infof("Block backup finished, snapshot ID %s, backup size %v, incremental size %v, source size %v", snapshotInfo.ID, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize) + log.Infof("Block backup finished, snapshot ID %s, backup size %v, incremental size %v, source size %v, fallback %v", snapshotInfo.ID, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, snapshotInfo.Fallback) - return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, nil + return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, snapshotInfo.Fallback, nil } func (bp *blockProvider) RunRestore( @@ -167,9 +167,9 @@ func (bp *blockProvider) RunRestore( cbtParam CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, - updater uploader.ProgressUpdater) (int64, int64, error) { + updater uploader.ProgressUpdater) (int64, int64, bool, error) { if updater == nil { - return 0, 0, errors.New("restore progress updater is invalid") + return 0, 0, false, errors.New("restore progress updater is invalid") } log := bp.log.WithFields(logrus.Fields{ @@ -180,16 +180,16 @@ func (bp *blockProvider) RunRestore( blkUploader := block.NewUploader(ctx, bp.bkRepo, updater, log) - incrementalBytes, totalBytes, err := blockRestoreFunc(ctx, blkUploader, bp.bkRepo, snapshotID, volumePath, incremental, cbtParam.Source, cbtParam.Service, uploaderCfg, log) + incrementalBytes, totalBytes, fallback, err := blockRestoreFunc(ctx, blkUploader, bp.bkRepo, snapshotID, volumePath, incremental, cbtParam.Source, cbtParam.Service, uploaderCfg, log) // errors.Is, not ==: see the equivalent comment on the backup path above. if errors.Is(err, block.ErrCanceled) { log.Warn("Block restore is canceled") - return 0, 0, ErrorCanceled + return 0, 0, fallback, ErrorCanceled } if err != nil { - return 0, 0, errors.Wrapf(err, "Failed to run block restore") + return 0, 0, fallback, errors.Wrapf(err, "Failed to run block restore") } updater.UpdateProgress(&uploader.Progress{ @@ -197,7 +197,7 @@ func (bp *blockProvider) RunRestore( BytesDone: totalBytes, }) - log.Infof("Block restore finished, restore incremental size %v, total size %v", incrementalBytes, totalBytes) + log.Infof("Block restore finished, restore incremental size %v, total size %v, fallback %v", incrementalBytes, totalBytes, fallback) - return incrementalBytes, totalBytes, nil + return incrementalBytes, totalBytes, fallback, nil } diff --git a/pkg/uploader/provider/block_test.go b/pkg/uploader/provider/block_test.go index b0067e7f4..bd8cc1ea1 100644 --- a/pkg/uploader/provider/block_test.go +++ b/pkg/uploader/provider/block_test.go @@ -337,7 +337,7 @@ func TestBlockProviderRunBackup(t *testing.T) { log: logrus.New(), } - snapshotID, isEmpty, size, incrSize, sourceSize, err := bp.RunBackup( + snapshotID, isEmpty, size, incrSize, sourceSize, fallback, err := bp.RunBackup( t.Context(), tc.path, tc.realSource, @@ -354,6 +354,7 @@ func TestBlockProviderRunBackup(t *testing.T) { assert.Equal(t, tc.expectedSize, size) assert.Equal(t, tc.expectedIncrSize, incrSize) assert.Equal(t, tc.expectedSourceSize, sourceSize) + assert.Equal(t, tc.mockBackupResult.Fallback, fallback) if tc.expectError { require.Error(t, err) @@ -404,7 +405,7 @@ func TestBlockProviderCancelThroughWrappedError(t *testing.T) { log: logrus.New(), } - _, _, _, _, _, err := bp.RunBackup( + _, _, _, _, _, _, err := bp.RunBackup( t.Context(), "/dev/sda", "ns/pvc", map[string]string{}, false, "", CBTParam{}, uploader.PersistentVolumeBlock, map[string]string{}, &FakeBackupProgressUpdater{}, @@ -418,8 +419,8 @@ func TestBlockProviderCancelThroughWrappedError(t *testing.T) { t.Run("restore", func(t *testing.T) { orig := blockRestoreFunc defer func() { blockRestoreFunc = orig }() - blockRestoreFunc = func(_ context.Context, _ block.Uploader, _ udmrepo.BackupRepo, _ string, _ string, _ bool, _ cbtservice.SourceInfo, _ cbtservice.Service, _ map[string]string, _ logrus.FieldLogger) (int64, int64, error) { - return 0, 0, errors.Wrap(block.ErrCanceled, "error restoring bdev") + blockRestoreFunc = func(_ context.Context, _ block.Uploader, _ udmrepo.BackupRepo, _ string, _ string, _ bool, _ cbtservice.SourceInfo, _ cbtservice.Service, _ map[string]string, _ logrus.FieldLogger) (int64, int64, bool, error) { + return 0, 0, false, errors.Wrap(block.ErrCanceled, "error restoring bdev") } bp := &blockProvider{ @@ -428,7 +429,7 @@ func TestBlockProviderCancelThroughWrappedError(t *testing.T) { log: logrus.New(), } - _, _, err := bp.RunRestore(t.Context(), "snap-1", "/dev/sda", false, CBTParam{}, + _, _, _, err := bp.RunRestore(t.Context(), "snap-1", "/dev/sda", false, CBTParam{}, uploader.PersistentVolumeBlock, map[string]string{}, &blockMockProgressUpdater{}) require.ErrorIs(t, err, ErrorCanceled) @@ -502,10 +503,10 @@ func TestBlockProviderRunRestore(t *testing.T) { var capturedSnapshotID string var capturedVolumePath string - blockRestoreFunc = func(ctx context.Context, blkUp block.Uploader, rep udmrepo.BackupRepo, snapshotID string, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, int64, error) { + blockRestoreFunc = func(ctx context.Context, blkUp block.Uploader, rep udmrepo.BackupRepo, snapshotID string, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, int64, bool, error) { capturedSnapshotID = snapshotID capturedVolumePath = dest - return tc.mockRestoreSize, tc.mockRestoreSize, tc.mockRestoreErr + return tc.mockRestoreSize, tc.mockRestoreSize, false, tc.mockRestoreErr } bp := &blockProvider{ @@ -513,7 +514,7 @@ func TestBlockProviderRunRestore(t *testing.T) { log: logrus.New(), } - _, size, err := bp.RunRestore( + _, size, _, err := bp.RunRestore( t.Context(), tc.snapshotID, tc.volumePath, diff --git a/pkg/uploader/provider/kopia.go b/pkg/uploader/provider/kopia.go index b5be51434..7b2325db1 100644 --- a/pkg/uploader/provider/kopia.go +++ b/pkg/uploader/provider/kopia.go @@ -121,13 +121,13 @@ func (kp *kopiaProvider) RunBackup( volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater, -) (string, bool, int64, int64, int64, error) { +) (string, bool, int64, int64, int64, bool, error) { if updater == nil { - return "", false, 0, 0, 0, errors.New("Need to initial backup progress updater first") + return "", false, 0, 0, 0, false, errors.New("Need to initial backup progress updater first") } if path == "" { - return "", false, 0, 0, 0, errors.New("path is empty") + return "", false, 0, 0, 0, false, errors.New("path is empty") } log := kp.log.WithFields(logrus.Fields{ @@ -177,9 +177,9 @@ func (kp *kopiaProvider) RunBackup( if kpUploader.IsCanceled() { log.Warn("Kopia backup is canceled") - return snapshotID, false, 0, 0, 0, ErrorCanceled + return snapshotID, false, 0, 0, 0, snapshotInfo.Fallback, ErrorCanceled } - return snapshotID, false, 0, 0, 0, errors.Wrapf(err, "Failed to run kopia backup") + return snapshotID, false, 0, 0, 0, snapshotInfo.Fallback, errors.Wrapf(err, "Failed to run kopia backup") } // which ensure that the statistic data of TotalBytes equal to BytesDone when finished @@ -190,8 +190,8 @@ func (kp *kopiaProvider) RunBackup( }, ) - log.Debugf("Kopia backup finished, snapshot ID %s, backup size %d", snapshotInfo.ID, snapshotInfo.SnapshotSize) - return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, progress.GetIncrementalSize(), snapshotInfo.SourceSize, nil + log.Debugf("Kopia backup finished, snapshot ID %s, backup size %d, fallback %v", snapshotInfo.ID, snapshotInfo.SnapshotSize, snapshotInfo.Fallback) + return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, progress.GetIncrementalSize(), snapshotInfo.SourceSize, snapshotInfo.Fallback, nil } func (kp *kopiaProvider) GetPassword(param any) (string, error) { @@ -215,7 +215,7 @@ func (kp *kopiaProvider) RunRestore( _ CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, - updater uploader.ProgressUpdater) (int64, int64, error) { + updater uploader.ProgressUpdater) (int64, int64, bool, error) { log := kp.log.WithFields(logrus.Fields{ "snapshotID": snapshotID, "volumePath": volumePath, @@ -236,15 +236,15 @@ func (kp *kopiaProvider) RunRestore( // We use the cancel channel to control the restore cancel, so don't pass a context with cancel to Kopia restore. // Otherwise, Kopia restore will not response to the cancel control but return an arbitrary error. // Kopia restore cancel is not designed as well as Kopia backup which uses the context to control backup cancel all the way. - size, fileCount, err := kopiaRestoreFunc(context.Background(), repoWriter, progress, snapshotID, volumePath, incremental, volMode, uploaderCfg, log, restoreCancel) + size, fileCount, fallback, err := kopiaRestoreFunc(context.Background(), repoWriter, progress, snapshotID, volumePath, incremental, volMode, uploaderCfg, log, restoreCancel) if err != nil { - return 0, 0, errors.Wrapf(err, "Failed to run kopia restore") + return 0, 0, fallback, errors.Wrapf(err, "Failed to run kopia restore") } if atomic.LoadInt32(&kp.canceling) == 1 { log.Error("Kopia restore is canceled") - return 0, 0, ErrorCanceled + return 0, 0, fallback, ErrorCanceled } // which ensure that the statistic data of TotalBytes equal to BytesDone when finished @@ -253,10 +253,10 @@ func (kp *kopiaProvider) RunRestore( BytesDone: size, }) - output := fmt.Sprintf("Kopia restore finished, restore size %d, file count %d", size, fileCount) + output := fmt.Sprintf("Kopia restore finished, restore size %d, file count %d, fallback %v", size, fileCount, fallback) log.Info(output) // the incremental bytes is the same as the total bytes because total bytes is the size of actual data Kopia writes - return size, size, nil + return size, size, fallback, nil } diff --git a/pkg/uploader/provider/kopia_test.go b/pkg/uploader/provider/kopia_test.go index 8b6918d40..e2ca6eaea 100644 --- a/pkg/uploader/provider/kopia_test.go +++ b/pkg/uploader/provider/kopia_test.go @@ -106,7 +106,7 @@ func TestRunBackup(t *testing.T) { tc.volMode = uploader.PersistentVolumeFilesystem } kopiaBackupFunc = tc.hookBackupFunc - _, _, _, _, _, err := kp.RunBackup(t.Context(), "var", "", nil, false, "", CBTParam{}, tc.volMode, map[string]string{}, &updater) + _, _, _, _, _, _, err := kp.RunBackup(t.Context(), "var", "", nil, false, "", CBTParam{}, tc.volMode, map[string]string{}, &updater) if tc.notError { assert.NoError(t, err) } else { @@ -119,30 +119,30 @@ func TestRunBackup(t *testing.T) { func TestRunRestore(t *testing.T) { testCases := []struct { name string - hookRestoreFunc func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) + hookRestoreFunc func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, bool, error) notError bool volMode uploader.PersistentVolumeMode incremental bool }{ { name: "normal restore", - hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) { - return 0, 0, nil + hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, bool, error) { + return 0, 0, false, nil }, notError: true, }, { name: "normal block mode restore", - hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) { - return 0, 0, nil + hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, bool, error) { + return 0, 0, false, nil }, volMode: uploader.PersistentVolumeBlock, notError: true, }, { name: "failed to restore", - hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) { - return 0, 0, errors.New("failed to restore") + hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, bool, error) { + return 0, 0, false, errors.New("failed to restore") }, notError: false, }, @@ -157,10 +157,10 @@ func TestRunRestore(t *testing.T) { if tc.volMode == "" { tc.volMode = uploader.PersistentVolumeFilesystem } - kopiaRestoreFunc = func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) { + kopiaRestoreFunc = func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, bool, error) { return tc.hookRestoreFunc(ctx, rep, progress, snapshotID, dest, incremental, volMode, uploaderCfg, log, cancleCh) } - _, _, err := kp.RunRestore(t.Context(), "", "/var", tc.incremental, CBTParam{}, tc.volMode, map[string]string{}, &updater) + _, _, _, err := kp.RunRestore(t.Context(), "", "/var", tc.incremental, CBTParam{}, tc.volMode, map[string]string{}, &updater) if tc.notError { assert.NoError(t, err) } else { diff --git a/pkg/uploader/types.go b/pkg/uploader/types.go index 82b6d8393..f4afded38 100644 --- a/pkg/uploader/types.go +++ b/pkg/uploader/types.go @@ -55,6 +55,7 @@ type SnapshotInfo struct { SnapshotSize int64 IncrementalSize int64 SourceSize int64 + Fallback bool } // Progress which defined two variables to record progress