Add "IncrementalBytes" field to status of DataDownload and PVR to indicate data transferred by the incremental restore (#10421)

Add "IncrementalBytes" field to status of DataDownload and PVR to indicate data transferred by the incremental restore

Signed-off-by: Wenkai Yin(尹文开) <yinw@vmware.com>

changelogs/unreleased/10421-ywk253100

@@ -0,0 +1 @@
+Add "IncrementalBytes" field to status of DataDownload and PVR to indicate data transferred by the incremental restore

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

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

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"

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()

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"

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()

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)

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)
 

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",

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}})
 		}
 	}()
 

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

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

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",

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) {

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)

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
 }

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,

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
 }

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 {

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
 }

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
 }