mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-09-20 06:54:32 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

return fallback info from uploader

Browse Source 
Signed-off-by: Lyndon-Li <lyonghui@vmware.com>
This commit is contained in:
Lyndon-Li
2026-09-10 16:40:02 +08:00
parent 55c756ed0f
commit f584d76b32
9 changed files with 135 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/uploader/block/snapshot.go
+19 -14
View File
@@ -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) {
pkg/uploader/block/snapshot_test.go
+30 -16
View File
@@ -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)
pkg/uploader/kopia/snapshot.go
+28 -22
View File
@@ -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
}
pkg/uploader/kopia/snapshot_test.go
+11 -3
View File
@@ -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)
})
}
}
pkg/uploader/provider/block.go
+14 -14
View File
@@ -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
}
pkg/uploader/provider/block_test.go
+9 -8
View File
@@ -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,
pkg/uploader/provider/kopia.go
+13 -13
View File
@@ -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
}
pkg/uploader/provider/kopia_test.go
+10 -10
View File
@@ -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 {
pkg/uploader/types.go
+1
View File
@@ -55,6 +55,7 @@ type SnapshotInfo struct {
SnapshotSize int64
IncrementalSize int64
SourceSize int64
Fallback bool
}
// Progress which defined two variables to record progress