From 07768e7b336d6b5b4cceae1a1a68e09acefd788d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Wenkai=20Yin=28=E5=B0=B9=E6=96=87=E5=BC=80=29?= Date: Tue, 1 Sep 2026 15:43:53 +0800 Subject: [PATCH] Add "IncrementalBytes" field to status of DataDownload and PVR to indicate data transferred by the incremental restore (#10421) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add "IncrementalBytes" field to status of DataDownload and PVR to indicate data transferred by the incremental restore Signed-off-by: Wenkai Yin(尹文开) --- changelogs/unreleased/10421-ywk253100 | 1 + .../v1/bases/velero.io_podvolumerestores.yaml | 15 +++++++++++ .../bases/velero.io_datadownloads.yaml | 16 +++++++++++ pkg/apis/velero/v1/pod_volume_restore_type.go | 8 ++++++ pkg/apis/velero/v1/zz_generated.deepcopy.go | 5 ++++ .../velero/v2alpha1/data_download_types.go | 8 ++++++ .../velero/v2alpha1/zz_generated.deepcopy.go | 5 ++++ pkg/controller/data_download_controller.go | 2 ++ .../pod_volume_restore_controller.go | 2 ++ pkg/datamover/restore_micro_service_test.go | 2 +- pkg/datapath/data_path.go | 4 +-- pkg/datapath/data_path_test.go | 2 +- pkg/datapath/types.go | 5 ++-- pkg/podvolume/restore_micro_service_test.go | 2 +- pkg/uploader/block/snapshot.go | 18 ++++++------- pkg/uploader/block/snapshot_test.go | 2 +- pkg/uploader/provider/block.go | 18 ++++++------- pkg/uploader/provider/block_test.go | 12 ++++----- pkg/uploader/provider/kopia.go | 9 ++++--- pkg/uploader/provider/kopia_test.go | 6 +++-- pkg/uploader/provider/mocks/Provider.go | 27 ++++++++++++------- pkg/uploader/provider/provider.go | 2 +- 22 files changed, 123 insertions(+), 48 deletions(-) create mode 100644 changelogs/unreleased/10421-ywk253100 diff --git a/changelogs/unreleased/10421-ywk253100 b/changelogs/unreleased/10421-ywk253100 new file mode 100644 index 000000000..20e5525a6 --- /dev/null +++ b/changelogs/unreleased/10421-ywk253100 @@ -0,0 +1 @@ +Add "IncrementalBytes" field to status of DataDownload and PVR to indicate data transferred by the incremental restore \ No newline at end of file diff --git a/config/crd/v1/bases/velero.io_podvolumerestores.yaml b/config/crd/v1/bases/velero.io_podvolumerestores.yaml index 2eea696c2..c8ddb9c87 100644 --- a/config/crd/v1/bases/velero.io_podvolumerestores.yaml +++ b/config/crd/v1/bases/velero.io_podvolumerestores.yaml @@ -21,6 +21,11 @@ spec: jsonPath: .status.phase name: Status type: string + - description: Restore type such as Full/Incremental + jsonPath: .spec.restoreType + name: Restore Type + priority: 10 + type: string - description: Time duration since this PodVolumeRestore was started jsonPath: .status.startTimestamp name: Started @@ -35,6 +40,12 @@ spec: jsonPath: .status.progress.totalBytes name: Total Bytes type: integer + - description: Incremental bytes + format: int64 + jsonPath: .status.incrementalBytes + name: Incremental Bytes + priority: 10 + type: integer - description: Name of the Backup Storage Location where the backup data is stored jsonPath: .spec.backupStorageLocation name: Storage Location @@ -193,6 +204,10 @@ spec: format: date-time nullable: true type: string + incrementalBytes: + description: IncrementalBytes holds the number of bytes restored incrementally + format: int64 + type: integer message: description: Message is a message about the pod volume restore's status. type: string diff --git a/config/crd/v2alpha1/bases/velero.io_datadownloads.yaml b/config/crd/v2alpha1/bases/velero.io_datadownloads.yaml index 71e662fe8..88fe710e4 100644 --- a/config/crd/v2alpha1/bases/velero.io_datadownloads.yaml +++ b/config/crd/v2alpha1/bases/velero.io_datadownloads.yaml @@ -21,6 +21,11 @@ spec: jsonPath: .status.phase name: Status type: string + - description: Restore type such as Full/Incremental + jsonPath: .spec.restoreType + name: Restore Type + priority: 10 + type: string - description: Time duration since this DataDownload was started jsonPath: .status.startTimestamp name: Started @@ -35,6 +40,12 @@ spec: jsonPath: .status.progress.totalBytes name: Total Bytes type: integer + - description: Incremental bytes + format: int64 + jsonPath: .status.incrementalBytes + name: Incremental Bytes + priority: 10 + type: integer - description: Name of the Backup Storage Location where the backup data is stored jsonPath: .spec.backupStorageLocation name: Storage Location @@ -202,6 +213,11 @@ spec: format: date-time nullable: true type: string + incrementalBytes: + description: IncrementalBytes holds the number of bytes restored incrementally + since the last snapshot + format: int64 + type: integer message: description: Message is a message about the DataDownload's status. type: string diff --git a/pkg/apis/velero/v1/pod_volume_restore_type.go b/pkg/apis/velero/v1/pod_volume_restore_type.go index 5ded78175..725630a83 100644 --- a/pkg/apis/velero/v1/pod_volume_restore_type.go +++ b/pkg/apis/velero/v1/pod_volume_restore_type.go @@ -111,6 +111,10 @@ type PodVolumeRestoreStatus struct { // +optional Progress shared.DataMoveOperationProgress `json:"progress,omitempty"` + // IncrementalBytes holds the number of bytes restored incrementally + // +optional + IncrementalBytes *int64 `json:"incrementalBytes,omitempty"` + // AcceptedTimestamp records the time the pod volume restore is to be prepared. // The server's time is used for AcceptedTimestamp // +optional @@ -129,9 +133,13 @@ type PodVolumeRestoreStatus struct { // +kubebuilder:object:root=true // +kubebuilder:storageversion // +kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.phase",description="PodVolumeRestore status such as New/InProgress" +// The "Restore Type" column is hidden by default to align with PVB. +// +kubebuilder:printcolumn:name="Restore Type",type="string",JSONPath=".spec.restoreType",description="Restore type such as Full/Incremental",priority=10 // +kubebuilder:printcolumn:name="Started",type="date",JSONPath=".status.startTimestamp",description="Time duration since this PodVolumeRestore was started" // +kubebuilder:printcolumn:name="Bytes Done",type="integer",format="int64",JSONPath=".status.progress.bytesDone",description="Completed bytes" // +kubebuilder:printcolumn:name="Total Bytes",type="integer",format="int64",JSONPath=".status.progress.totalBytes",description="Total bytes" +// The "Incremental Bytes" column is hidden by default to align with PVB. +// +kubebuilder:printcolumn:name="Incremental Bytes",type="integer",format="int64",JSONPath=".status.incrementalBytes",description="Incremental bytes",priority=10 // +kubebuilder:printcolumn:name="Storage Location",type="string",JSONPath=".spec.backupStorageLocation",description="Name of the Backup Storage Location where the backup data is stored" // +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp",description="Time duration since this PodVolumeRestore was created" // +kubebuilder:printcolumn:name="Node",type="string",JSONPath=".status.node",description="Name of the node where the PodVolumeRestore is processed" diff --git a/pkg/apis/velero/v1/zz_generated.deepcopy.go b/pkg/apis/velero/v1/zz_generated.deepcopy.go index f4dc8a79a..9c6c84dda 100644 --- a/pkg/apis/velero/v1/zz_generated.deepcopy.go +++ b/pkg/apis/velero/v1/zz_generated.deepcopy.go @@ -1170,6 +1170,11 @@ func (in *PodVolumeRestoreStatus) DeepCopyInto(out *PodVolumeRestoreStatus) { *out = (*in).DeepCopy() } out.Progress = in.Progress + if in.IncrementalBytes != nil { + in, out := &in.IncrementalBytes, &out.IncrementalBytes + *out = new(int64) + **out = **in + } if in.AcceptedTimestamp != nil { in, out := &in.AcceptedTimestamp, &out.AcceptedTimestamp *out = (*in).DeepCopy() diff --git a/pkg/apis/velero/v2alpha1/data_download_types.go b/pkg/apis/velero/v2alpha1/data_download_types.go index 57827b97d..caaeceba7 100644 --- a/pkg/apis/velero/v2alpha1/data_download_types.go +++ b/pkg/apis/velero/v2alpha1/data_download_types.go @@ -132,6 +132,10 @@ type DataDownloadStatus struct { // +optional Progress shared.DataMoveOperationProgress `json:"progress,omitempty"` + // IncrementalBytes holds the number of bytes restored incrementally since the last snapshot + // +optional + IncrementalBytes *int64 `json:"incrementalBytes,omitempty"` + // Node is name of the node where the DataDownload is processed. // +optional Node string `json:"node,omitempty"` @@ -154,9 +158,13 @@ type DataDownloadStatus struct { // +kubebuilder:object:root=true // +kubebuilder:storageversion // +kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.phase",description="DataDownload status such as New/InProgress" +// The "Restore Type" column is hidden by default to align with DataUpload. +// +kubebuilder:printcolumn:name="Restore Type",type="string",JSONPath=".spec.restoreType",description="Restore type such as Full/Incremental",priority=10 // +kubebuilder:printcolumn:name="Started",type="date",JSONPath=".status.startTimestamp",description="Time duration since this DataDownload was started" // +kubebuilder:printcolumn:name="Bytes Done",type="integer",format="int64",JSONPath=".status.progress.bytesDone",description="Completed bytes" // +kubebuilder:printcolumn:name="Total Bytes",type="integer",format="int64",JSONPath=".status.progress.totalBytes",description="Total bytes" +// The "Incremental Bytes" column is hidden by default to align with DataUpload. +// +kubebuilder:printcolumn:name="Incremental Bytes",type="integer",format="int64",JSONPath=".status.incrementalBytes",description="Incremental bytes",priority=10 // +kubebuilder:printcolumn:name="Storage Location",type="string",JSONPath=".spec.backupStorageLocation",description="Name of the Backup Storage Location where the backup data is stored" // +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp",description="Time duration since this DataDownload was created" // +kubebuilder:printcolumn:name="Node",type="string",JSONPath=".status.node",description="Name of the node where the DataDownload is processed" diff --git a/pkg/apis/velero/v2alpha1/zz_generated.deepcopy.go b/pkg/apis/velero/v2alpha1/zz_generated.deepcopy.go index 927dc531c..d8186605f 100644 --- a/pkg/apis/velero/v2alpha1/zz_generated.deepcopy.go +++ b/pkg/apis/velero/v2alpha1/zz_generated.deepcopy.go @@ -123,6 +123,11 @@ func (in *DataDownloadStatus) DeepCopyInto(out *DataDownloadStatus) { *out = (*in).DeepCopy() } out.Progress = in.Progress + if in.IncrementalBytes != nil { + in, out := &in.IncrementalBytes, &out.IncrementalBytes + *out = new(int64) + **out = **in + } if in.AcceptedTimestamp != nil { in, out := &in.AcceptedTimestamp, &out.AcceptedTimestamp *out = (*in).DeepCopy() diff --git a/pkg/controller/data_download_controller.go b/pkg/controller/data_download_controller.go index d8062bc72..e2a62830e 100644 --- a/pkg/controller/data_download_controller.go +++ b/pkg/controller/data_download_controller.go @@ -32,6 +32,7 @@ import ( "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/kubernetes" "k8s.io/utils/clock" + "k8s.io/utils/ptr" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" @@ -501,6 +502,7 @@ func (r *DataDownloadReconciler) OnDataDownloadCompleted(ctx context.Context, na } dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseCompleted + dd.Status.IncrementalBytes = ptr.To(result.Restore.IncrementalBytes) dd.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()} delete(dd.Labels, exposer.ExposeOnGoingLabel) diff --git a/pkg/controller/pod_volume_restore_controller.go b/pkg/controller/pod_volume_restore_controller.go index b6d4985fa..4fe9bdaa3 100644 --- a/pkg/controller/pod_volume_restore_controller.go +++ b/pkg/controller/pod_volume_restore_controller.go @@ -33,6 +33,7 @@ import ( "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/kubernetes" clocks "k8s.io/utils/clock" + "k8s.io/utils/ptr" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" @@ -837,6 +838,7 @@ func (r *PodVolumeRestoreReconciler) OnDataPathCompleted(ctx context.Context, na pvr.Status.Phase = velerov1api.PodVolumeRestorePhaseCompleted pvr.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()} + pvr.Status.IncrementalBytes = ptr.To(result.Restore.IncrementalBytes) delete(pvr.Labels, exposer.ExposeOnGoingLabel) diff --git a/pkg/datamover/restore_micro_service_test.go b/pkg/datamover/restore_micro_service_test.go index 311c015a7..63a3e6a5e 100644 --- a/pkg/datamover/restore_micro_service_test.go +++ b/pkg/datamover/restore_micro_service_test.go @@ -102,7 +102,7 @@ func TestOnDataDownloadCompleted(t *testing.T) { { name: "marshal fail", marshalErr: errors.New("fake-marshal-error"), - expectedErr: "Failed to marshal restore result {{ } 0}: fake-marshal-error", + expectedErr: "Failed to marshal restore result {{ } 0 0}: fake-marshal-error", }, { name: "succeed", diff --git a/pkg/datapath/data_path.go b/pkg/datapath/data_path.go index 1e7ae948e..0513619be 100644 --- a/pkg/datapath/data_path.go +++ b/pkg/datapath/data_path.go @@ -250,7 +250,7 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u dp.wgDataPath.Done() }() - totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental, + incrementalBytes, totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental, provider.CBTParam{ Source: cbtservice.SourceInfo{ Snapshot: restoreParam.VolumeSnapshotName, @@ -268,7 +268,7 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u } dp.callbacks.OnFailed(context.Background(), dp.namespace, dp.jobName, dataPathErr) } else { - dp.callbacks.OnCompleted(context.Background(), dp.namespace, dp.jobName, Result{Restore: RestoreResult{Target: target, TotalBytes: totalBytes}}) + dp.callbacks.OnCompleted(context.Background(), dp.namespace, dp.jobName, Result{Restore: RestoreResult{Target: target, TotalBytes: totalBytes, IncrementalBytes: incrementalBytes}}) } }() diff --git a/pkg/datapath/data_path_test.go b/pkg/datapath/data_path_test.go index 34f989517..495c949f9 100644 --- a/pkg/datapath/data_path_test.go +++ b/pkg/datapath/data_path_test.go @@ -190,7 +190,7 @@ func TestAsyncRestore(t *testing.T) { t.Run(test.name, func(t *testing.T) { dp := newGeneralDataPath("job-1", "test", nil, "velero", Callbacks{}, velerotest.NewLogger()).(*generalDataPath) mockProvider := providerMock.NewProvider(t) - mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Restore.TotalBytes, test.err) + mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Restore.IncrementalBytes, test.result.Restore.TotalBytes, test.err) mockProvider.On("Close", mock.Anything).Return(nil) dp.uploaderProv = mockProvider dp.initialized = true diff --git a/pkg/datapath/types.go b/pkg/datapath/types.go index 339aa6ca4..bc7d4a9a9 100644 --- a/pkg/datapath/types.go +++ b/pkg/datapath/types.go @@ -43,8 +43,9 @@ type BackupResult struct { // RestoreResult represents the result of a restore type RestoreResult struct { - Target AccessPoint `json:"target,omitempty"` - TotalBytes int64 `json:"totalBytes,omitempty"` + Target AccessPoint `json:"target,omitempty"` + TotalBytes int64 `json:"totalBytes,omitempty"` + IncrementalBytes int64 `json:"incrementalBytes,omitempty"` } // Callbacks defines the collection of callbacks during backup/restore diff --git a/pkg/podvolume/restore_micro_service_test.go b/pkg/podvolume/restore_micro_service_test.go index 1964d5035..46c8813b6 100644 --- a/pkg/podvolume/restore_micro_service_test.go +++ b/pkg/podvolume/restore_micro_service_test.go @@ -165,7 +165,7 @@ func TestOnPvrCompleted(t *testing.T) { { name: "marshal fail", marshalErr: errors.New("fake-marshal-error"), - expectedErr: "error marshaling restore result {{ } 0}: fake-marshal-error", + expectedErr: "error marshaling restore result {{ } 0 0}: fake-marshal-error", }, { name: "succeed", diff --git a/pkg/uploader/block/snapshot.go b/pkg/uploader/block/snapshot.go index ff844d197..595a80fd5 100644 --- a/pkg/uploader/block/snapshot.go +++ b/pkg/uploader/block/snapshot.go @@ -205,12 +205,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, 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, error) { log.Info("Start to restore...") snapshot, err := rep.GetSnapshot(ctx, udmrepo.ID(snapshotID)) if err != nil { - return 0, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID) + return 0, 0, 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) @@ -246,34 +246,34 @@ func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapsh destPath, err := filepath.Abs(dest) if err != nil { - return 0, errors.Wrapf(err, "invalid dest path '%s'", dest) + return 0, 0, errors.Wrapf(err, "invalid dest path '%s'", dest) } destPath = filepath.Clean(destPath) destDev, err := openBlockDeviceFunc(destPath, false) if err != nil { - return 0, errors.Wrapf(err, "error opening block device '%s'", destPath) + return 0, 0, errors.Wrapf(err, "error opening block device '%s'", destPath) } defer destDev.Close() destSize, err := destDev.Seek(0, io.SeekEnd) if err != nil { - return 0, errors.Wrapf(err, "error getting length of block device %s", dest) + return 0, 0, errors.Wrapf(err, "error getting length of block device %s", dest) } _, err = destDev.Seek(0, io.SeekStart) if err != nil { - return 0, errors.Wrapf(err, "error reset pos of block device %s", dest) + return 0, 0, errors.Wrapf(err, "error reset pos of block device %s", dest) } - _, totalSize, err := blkUp.Restore(snapshot, destInfo{dev: destDev, path: destPath, size: destSize}, bitmap.Iterator(), uploaderCfg) + incrementalBytes, totalSize, err := blkUp.Restore(snapshot, destInfo{dev: destDev, path: destPath, size: destSize}, bitmap.Iterator(), uploaderCfg) if err != nil { - return 0, errors.Wrapf(err, "error restoring to block dev %s", destPath) + return 0, 0, errors.Wrapf(err, "error restoring to block dev %s", destPath) } - return totalSize, nil + return incrementalBytes, totalSize, nil } func findPreviousSnapshot(ctx context.Context, rep udmrepo.BackupRepo, path string, snapshotTags map[string]string, noLaterThan *time.Time, log logrus.FieldLogger) (udmrepo.Snapshot, error) { diff --git a/pkg/uploader/block/snapshot_test.go b/pkg/uploader/block/snapshot_test.go index d7e7d2ee2..5a18c376e 100644 --- a/pkg/uploader/block/snapshot_test.go +++ b/pkg/uploader/block/snapshot_test.go @@ -842,7 +842,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, 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) diff --git a/pkg/uploader/provider/block.go b/pkg/uploader/provider/block.go index 2b5ad275f..3d51e275b 100644 --- a/pkg/uploader/provider/block.go +++ b/pkg/uploader/provider/block.go @@ -167,9 +167,9 @@ func (bp *blockProvider) RunRestore( cbtParam CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, - updater uploader.ProgressUpdater) (int64, error) { + updater uploader.ProgressUpdater) (int64, int64, error) { if updater == nil { - return 0, errors.New("restore progress updater is invalid") + return 0, 0, errors.New("restore progress updater is invalid") } log := bp.log.WithFields(logrus.Fields{ @@ -180,24 +180,24 @@ func (bp *blockProvider) RunRestore( blkUploader := block.NewUploader(ctx, bp.bkRepo, updater, log) - size, err := blockRestoreFunc(ctx, blkUploader, bp.bkRepo, snapshotID, volumePath, incremental, cbtParam.Source, cbtParam.Service, uploaderCfg, log) + incrementalBytes, totalBytes, 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, ErrorCanceled + return 0, 0, ErrorCanceled } if err != nil { - return 0, errors.Wrapf(err, "Failed to run block restore") + return 0, 0, errors.Wrapf(err, "Failed to run block restore") } updater.UpdateProgress(&uploader.Progress{ - TotalBytes: size, - BytesDone: size, + TotalBytes: totalBytes, + BytesDone: totalBytes, }) - log.Infof("Block restore finished, restore size %v", size) + log.Infof("Block restore finished, restore incremental size %v, total size %v", incrementalBytes, totalBytes) - return size, nil + return incrementalBytes, totalBytes, nil } diff --git a/pkg/uploader/provider/block_test.go b/pkg/uploader/provider/block_test.go index 970fc7cf6..e38de4985 100644 --- a/pkg/uploader/provider/block_test.go +++ b/pkg/uploader/provider/block_test.go @@ -412,8 +412,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, error) { - return 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, error) { + return 0, 0, errors.Wrap(block.ErrCanceled, "error restoring bdev") } bp := &blockProvider{ @@ -422,7 +422,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) @@ -496,10 +496,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, 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, error) { capturedSnapshotID = snapshotID capturedVolumePath = dest - return tc.mockRestoreSize, tc.mockRestoreErr + return tc.mockRestoreSize, tc.mockRestoreSize, tc.mockRestoreErr } bp := &blockProvider{ @@ -507,7 +507,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 c9d9948bf..1cd1ecd45 100644 --- a/pkg/uploader/provider/kopia.go +++ b/pkg/uploader/provider/kopia.go @@ -215,7 +215,7 @@ func (kp *kopiaProvider) RunRestore( _ CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, - updater uploader.ProgressUpdater) (int64, error) { + updater uploader.ProgressUpdater) (int64, int64, error) { log := kp.log.WithFields(logrus.Fields{ "snapshotID": snapshotID, "volumePath": volumePath, @@ -239,12 +239,12 @@ func (kp *kopiaProvider) RunRestore( size, fileCount, err := kopiaRestoreFunc(context.Background(), repoWriter, progress, snapshotID, volumePath, incremental, volMode, uploaderCfg, log, restoreCancel) if err != nil { - return 0, errors.Wrapf(err, "Failed to run kopia restore") + return 0, 0, errors.Wrapf(err, "Failed to run kopia restore") } if atomic.LoadInt32(&kp.canceling) == 1 { log.Error("Kopia restore is canceled") - return 0, ErrorCanceled + return 0, 0, ErrorCanceled } // which ensure that the statistic data of TotalBytes equal to BytesDone when finished @@ -257,5 +257,6 @@ func (kp *kopiaProvider) RunRestore( log.Info(output) - return size, nil + // 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 } diff --git a/pkg/uploader/provider/kopia_test.go b/pkg/uploader/provider/kopia_test.go index a29a3c424..ca1cf8f5a 100644 --- a/pkg/uploader/provider/kopia_test.go +++ b/pkg/uploader/provider/kopia_test.go @@ -157,8 +157,10 @@ func TestRunRestore(t *testing.T) { if tc.volMode == "" { tc.volMode = uploader.PersistentVolumeFilesystem } - kopiaRestoreFunc = tc.hookRestoreFunc - _, err := kp.RunRestore(t.Context(), "", "/var", tc.incremental, CBTParam{}, tc.volMode, map[string]string{}, &updater) + 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) { + 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) if tc.notError { assert.NoError(t, err) } else { diff --git a/pkg/uploader/provider/mocks/Provider.go b/pkg/uploader/provider/mocks/Provider.go index 5bd3dda54..7fad73e59 100644 --- a/pkg/uploader/provider/mocks/Provider.go +++ b/pkg/uploader/provider/mocks/Provider.go @@ -223,7 +223,7 @@ func (_c *Provider_RunBackup_Call) RunAndReturn(run func(ctx context.Context, pa } // RunRestore provides a mock function for the type Provider -func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error) { +func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, int64, error) { ret := _mock.Called(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater) if len(ret) == 0 { @@ -231,8 +231,9 @@ func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volume } var r0 int64 - var r1 error - if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, error)); ok { + var r1 int64 + var r2 error + if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, int64, error)); ok { return returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater) } if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok { @@ -240,12 +241,20 @@ func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volume } else { r0 = ret.Get(0).(int64) } - if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok { + + if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok { r1 = returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater) } else { - r1 = ret.Error(1) + r1 = ret.Get(1).(int64) } - return r0, r1 + + if returnFunc, ok := ret.Get(2).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok { + r2 = returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater) + } else { + r2 = ret.Error(2) + } + + return r0, r1, r2 } // Provider_RunRestore_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RunRestore' @@ -314,12 +323,12 @@ func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID return _c } -func (_c *Provider_RunRestore_Call) Return(n int64, err error) *Provider_RunRestore_Call { - _c.Call.Return(n, err) +func (_c *Provider_RunRestore_Call) Return(_a0 int64, _a1 int64, _a2 error) *Provider_RunRestore_Call { + _c.Call.Return(_a0, _a1, _a2) return _c } -func (_c *Provider_RunRestore_Call) RunAndReturn(run func(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error)) *Provider_RunRestore_Call { +func (_c *Provider_RunRestore_Call) RunAndReturn(run func(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, int64, error)) *Provider_RunRestore_Call { _c.Call.Return(run) return _c } diff --git a/pkg/uploader/provider/provider.go b/pkg/uploader/provider/provider.go index 9d06578d8..7f003989d 100644 --- a/pkg/uploader/provider/provider.go +++ b/pkg/uploader/provider/provider.go @@ -68,7 +68,7 @@ type Provider interface { cbtParam CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, - updater uploader.ProgressUpdater) (int64, error) + updater uploader.ProgressUpdater) (int64, int64, error) // Close which will close related repository Close(ctx context.Context) error }