Merge branch 'main' into cache-volume-for-exposer

changelogs/unreleased/9353-Lyndon-Li

@@ -0,0 +1 @@
+Add cache dir configuration for udmrepo

changelogs/unreleased/9354-Lyndon-Li

@@ -0,0 +1 @@
+Add snapshotSize for DataDownload, PodVolumeRestore

changelogs/unreleased/9357-sseago

@@ -0,0 +1 @@
+Add incrementalSize to DU/PVB for reporting new/changed size

config/crd/v1/bases/velero.io_podvolumebackups.yaml

@@ -33,6 +33,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 this backup should be
         stored
       jsonPath: .spec.backupStorageLocation
@@ -189,6 +195,11 @@ spec:
                 format: date-time
                 nullable: true
                 type: string
+              incrementalBytes:
+                description: IncrementalBytes holds the number of bytes new or changed
+                  since the last backup
+                format: int64
+                type: integer
               message:
                 description: Message is a message about the pod volume backup's status.
                 type: string

config/crd/v1/bases/velero.io_podvolumerestores.yaml

@@ -133,6 +133,10 @@ spec:
               snapshotID:
                 description: SnapshotID is the ID of the volume snapshot to be restored.
                 type: string
+              snapshotSize:
+                description: SnapshotSize is the logical size in Bytes of the snapshot.
+                format: int64
+                type: integer
               sourceNamespace:
                 description: SourceNamespace is the original namespace for namaspace
                   mapping.

config/crd/v2alpha1/bases/velero.io_datadownloads.yaml

@@ -108,6 +108,10 @@ spec:
                 description: SnapshotID is the ID of the Velero backup snapshot to
                   be restored from.
                 type: string
+              snapshotSize:
+                description: SnapshotSize is the logical size in Bytes of the snapshot.
+                format: int64
+                type: integer
               sourceNamespace:
                 description: |-
                   SourceNamespace is the original namespace where the volume is backed up from.

config/crd/v2alpha1/bases/velero.io_datauploads.yaml

@@ -33,6 +33,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 this backup should be
         stored
       jsonPath: .spec.backupStorageLocation
@@ -173,6 +179,11 @@ spec:
                   as a result of the DataUpload.
                 nullable: true
                 type: object
+              incrementalBytes:
+                description: IncrementalBytes holds the number of bytes new or changed
+                  since the last backup
+                format: int64
+                type: integer
               message:
                 description: Message is a message about the DataUpload's status.
                 type: string

internal/volume/volumes_information.go

@@ -170,6 +170,9 @@ type SnapshotDataMovementInfo struct {
 	// Moved snapshot data size.
 	Size int64 `json:"size"`
 
+	// Moved snapshot incremental size.
+	IncrementalSize int64 `json:"incrementalSize,omitempty"`
+
 	// The DataUpload's Status.Phase value
 	Phase velerov2alpha1.DataUploadPhase
 }
@@ -217,6 +220,9 @@ type PodVolumeInfo struct {
 	// The snapshot corresponding volume size.
 	Size int64 `json:"size,omitempty"`
 
+	// The incremental snapshot size.
+	IncrementalSize int64 `json:"incrementalSize,omitempty"`
+
 	// The type of the uploader that uploads the data. The valid values are `kopia` and `restic`.
 	UploaderType string `json:"uploaderType"`
 
@@ -240,14 +246,15 @@ type PodVolumeInfo struct {
 
 func newPodVolumeInfoFromPVB(pvb *velerov1api.PodVolumeBackup) *PodVolumeInfo {
 	return &PodVolumeInfo{
-		SnapshotHandle: pvb.Status.SnapshotID,
-		Size:           pvb.Status.Progress.TotalBytes,
-		UploaderType:   pvb.Spec.UploaderType,
-		VolumeName:     pvb.Spec.Volume,
-		PodName:        pvb.Spec.Pod.Name,
-		PodNamespace:   pvb.Spec.Pod.Namespace,
-		NodeName:       pvb.Spec.Node,
-		Phase:          pvb.Status.Phase,
+		SnapshotHandle:  pvb.Status.SnapshotID,
+		Size:            pvb.Status.Progress.TotalBytes,
+		IncrementalSize: pvb.Status.IncrementalBytes,
+		UploaderType:    pvb.Spec.UploaderType,
+		VolumeName:      pvb.Spec.Volume,
+		PodName:         pvb.Spec.Pod.Name,
+		PodNamespace:    pvb.Spec.Pod.Namespace,
+		NodeName:        pvb.Spec.Node,
+		Phase:           pvb.Status.Phase,
 	}
 }
 

pkg/apis/velero/v1/pod_volume_backup_types.go

@@ -118,6 +118,10 @@ type PodVolumeBackupStatus struct {
 	// +optional
 	Progress shared.DataMoveOperationProgress `json:"progress,omitempty"`
 
+	// IncrementalBytes holds the number of bytes new or changed since the last backup
+	// +optional
+	IncrementalBytes int64 `json:"incrementalBytes,omitempty"`
+
 	// AcceptedTimestamp records the time the pod volume backup is to be prepared.
 	// The server's time is used for AcceptedTimestamp
 	// +optional
@@ -134,6 +138,7 @@ type PodVolumeBackupStatus struct {
 // +kubebuilder:printcolumn:name="Started",type="date",JSONPath=".status.startTimestamp",description="Time duration since this PodVolumeBackup 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"
+// +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 this backup should be stored"
 // +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp",description="Time duration since this PodVolumeBackup was created"
 // +kubebuilder:printcolumn:name="Node",type="string",JSONPath=".status.node",description="Name of the node where the PodVolumeBackup is processed"

pkg/apis/velero/v1/pod_volume_restore_type.go

@@ -58,6 +58,10 @@ type PodVolumeRestoreSpec struct {
 	// Cancel indicates request to cancel the ongoing PodVolumeRestore. It can be set
 	// when the PodVolumeRestore is in InProgress phase
 	Cancel bool `json:"cancel,omitempty"`
+
+	// SnapshotSize is the logical size in Bytes of the snapshot.
+	// +optional
+	SnapshotSize int64 `json:"snapshotSize,omitempty"`
 }
 
 // PodVolumeRestorePhase represents the lifecycle phase of a PodVolumeRestore.

pkg/apis/velero/v2alpha1/data_download_types.go

@@ -58,6 +58,10 @@ type DataDownloadSpec struct {
 	// NodeOS is OS of the node where the DataDownload is processed.
 	// +optional
 	NodeOS NodeOS `json:"nodeOS,omitempty"`
+
+	// SnapshotSize is the logical size in Bytes of the snapshot.
+	// +optional
+	SnapshotSize int64 `json:"snapshotSize,omitempty"`
 }
 
 // TargetVolumeSpec is the specification for a target PVC.

pkg/apis/velero/v2alpha1/data_upload_types.go

@@ -155,6 +155,10 @@ type DataUploadStatus struct {
 	// +optional
 	Progress shared.DataMoveOperationProgress `json:"progress,omitempty"`
 
+	// IncrementalBytes holds the number of bytes new or changed since the last backup
+	// +optional
+	IncrementalBytes int64 `json:"incrementalBytes,omitempty"`
+
 	// Node is name of the node where the DataUpload is processed.
 	// +optional
 	Node string `json:"node,omitempty"`
@@ -185,6 +189,7 @@ type DataUploadStatus struct {
 // +kubebuilder:printcolumn:name="Started",type="date",JSONPath=".status.startTimestamp",description="Time duration since this DataUpload 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"
+// +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 this backup should be stored"
 // +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp",description="Time duration since this DataUpload was created"
 // +kubebuilder:printcolumn:name="Node",type="string",JSONPath=".status.node",description="Name of the node where the DataUpload is processed"
@@ -244,4 +249,8 @@ type DataUploadResult struct {
 	// NodeOS is OS of the node where the DataUpload is processed.
 	// +optional
 	NodeOS NodeOS `json:"nodeOS,omitempty"`
+
+	// SnapshotSize is the logical size in Bytes of the snapshot.
+	// +optional
+	SnapshotSize int64 `json:"snapshotSize,omitempty"`
 }

pkg/backup/backup.go

@@ -1198,6 +1198,7 @@ func updateVolumeInfos(
 				volumeInfos[index].SnapshotDataMovementInfo.SnapshotHandle = dataUpload.Status.SnapshotID
 				volumeInfos[index].SnapshotDataMovementInfo.RetainedSnapshot = dataUpload.Spec.CSISnapshot.VolumeSnapshot
 				volumeInfos[index].SnapshotDataMovementInfo.Size = dataUpload.Status.Progress.TotalBytes
+				volumeInfos[index].SnapshotDataMovementInfo.IncrementalSize = dataUpload.Status.IncrementalBytes
 				volumeInfos[index].SnapshotDataMovementInfo.Phase = dataUpload.Status.Phase
 
 				if dataUpload.Status.Phase == velerov2alpha1.DataUploadPhaseCompleted {

pkg/backup/backup_test.go

@@ -5578,6 +5578,7 @@ func TestUpdateVolumeInfos(t *testing.T) {
 				CSISnapshot(&velerov2alpha1.CSISnapshotSpec{VolumeSnapshot: "vs-1"}).
 				SnapshotID("snapshot-id").
 				Progress(shared.DataMoveOperationProgress{TotalBytes: 1000}).
+				IncrementalBytes(500).
 				Phase(velerov2alpha1.DataUploadPhaseFailed).
 				SourceNamespace("ns-1").
 				SourcePVC("pvc-1").
@@ -5603,6 +5604,7 @@ func TestUpdateVolumeInfos(t *testing.T) {
 						RetainedSnapshot: "vs-1",
 						SnapshotHandle:   "snapshot-id",
 						Size:             1000,
+						IncrementalSize:  500,
 						Phase:            velerov2alpha1.DataUploadPhaseFailed,
 					},
 				},
@@ -5616,6 +5618,7 @@ func TestUpdateVolumeInfos(t *testing.T) {
 				CSISnapshot(&velerov2alpha1.CSISnapshotSpec{VolumeSnapshot: "vs-1"}).
 				SnapshotID("snapshot-id").
 				Progress(shared.DataMoveOperationProgress{TotalBytes: 1000}).
+				IncrementalBytes(500).
 				Phase(velerov2alpha1.DataUploadPhaseCompleted).
 				SourceNamespace("ns-1").
 				SourcePVC("pvc-1").
@@ -5641,6 +5644,7 @@ func TestUpdateVolumeInfos(t *testing.T) {
 						RetainedSnapshot: "vs-1",
 						SnapshotHandle:   "snapshot-id",
 						Size:             1000,
+						IncrementalSize:  500,
 						Phase:            velerov2alpha1.DataUploadPhaseCompleted,
 					},
 				},
@@ -5655,6 +5659,7 @@ func TestUpdateVolumeInfos(t *testing.T) {
 				CSISnapshot(&velerov2alpha1.CSISnapshotSpec{VolumeSnapshot: "vs-1"}).
 				SnapshotID("snapshot-id").
 				Progress(shared.DataMoveOperationProgress{TotalBytes: 1000}).
+				IncrementalBytes(500).
 				Phase(velerov2alpha1.DataUploadPhaseCompleted).
 				SourceNamespace("ns-1").
 				SourcePVC("pvc-1").

pkg/builder/data_upload_builder.go

@@ -145,6 +145,12 @@ func (d *DataUploadBuilder) Progress(progress shared.DataMoveOperationProgress)
 	return d
 }
 
+// IncrementalBytes sets the DataUpload's IncrementalBytes.
+func (d *DataUploadBuilder) IncrementalBytes(incrementalBytes int64) *DataUploadBuilder {
+	d.object.Status.IncrementalBytes = incrementalBytes
+	return d
+}
+
 // Node sets the DataUpload's Node.
 func (d *DataUploadBuilder) Node(node string) *DataUploadBuilder {
 	d.object.Status.Node = node
@@ -180,3 +186,9 @@ func (d *DataUploadBuilder) Message(msg string) *DataUploadBuilder {
 	d.object.Status.Message = msg
 	return d
 }
+
+// TotalBytes sets the DataUpload's TotalBytes.
+func (d *DataUploadBuilder) TotalBytes(size int64) *DataUploadBuilder {
+	d.object.Status.Progress.TotalBytes = size
+	return d
+}

pkg/cmd/util/output/backup_describer.go

@@ -713,6 +713,9 @@ func describeDataMovement(d *Describer, details bool, info *volume.BackupVolumeI
 		d.Printf("\t\t\t\tData Mover: %s\n", dataMover)
 		d.Printf("\t\t\t\tUploader Type: %s\n", info.SnapshotDataMovementInfo.UploaderType)
 		d.Printf("\t\t\t\tMoved data Size (bytes): %d\n", info.SnapshotDataMovementInfo.Size)
+		if info.SnapshotDataMovementInfo.IncrementalSize > 0 {
+			d.Printf("\t\t\t\tIncremental data Size (bytes): %d\n", info.SnapshotDataMovementInfo.IncrementalSize)
+		}
 		d.Printf("\t\t\t\tResult: %s\n", info.Result)
 	} else {
 		d.Printf("\t\t\tData Movement: %s\n", "included, specify --details for more information")
@@ -835,7 +838,7 @@ func describePodVolumeBackups(d *Describer, details bool, podVolumeBackups []vel
 		backupsByPod := new(volumesByPod)
 
 		for _, backup := range backupsByPhase[phase] {
-			backupsByPod.Add(backup.Spec.Pod.Namespace, backup.Spec.Pod.Name, backup.Spec.Volume, phase, backup.Status.Progress)
+			backupsByPod.Add(backup.Spec.Pod.Namespace, backup.Spec.Pod.Name, backup.Spec.Volume, phase, backup.Status.Progress, backup.Status.IncrementalBytes)
 		}
 
 		d.Printf("\t\t%s:\n", phase)
@@ -885,7 +888,8 @@ type volumesByPod struct {
 
 // Add adds a pod volume with the specified pod namespace, name
 // and volume to the appropriate group.
-func (v *volumesByPod) Add(namespace, name, volume, phase string, progress veleroapishared.DataMoveOperationProgress) {
+// Used for both backup and restore
+func (v *volumesByPod) Add(namespace, name, volume, phase string, progress veleroapishared.DataMoveOperationProgress, incrementalBytes int64) {
 	if v.volumesByPodMap == nil {
 		v.volumesByPodMap = make(map[string]*podVolumeGroup)
 	}
@@ -895,6 +899,12 @@ func (v *volumesByPod) Add(namespace, name, volume, phase string, progress veler
 	// append backup progress percentage if backup is in progress
 	if phase == "In Progress" && progress.TotalBytes != 0 {
 		volume = fmt.Sprintf("%s (%.2f%%)", volume, float64(progress.BytesDone)/float64(progress.TotalBytes)*100)
+	} else if phase == string(velerov1api.PodVolumeBackupPhaseCompleted) && incrementalBytes > 0 {
+		volume = fmt.Sprintf("%s (size: %v, incremental size: %v)", volume, progress.TotalBytes, incrementalBytes)
+	} else if (phase == string(velerov1api.PodVolumeBackupPhaseCompleted) ||
+		phase == string(velerov1api.PodVolumeRestorePhaseCompleted)) &&
+		progress.TotalBytes > 0 {
+		volume = fmt.Sprintf("%s (size: %v)", volume, progress.TotalBytes)
 	}
 
 	if group, ok := v.volumesByPodMap[key]; !ok {

pkg/cmd/util/output/backup_describer_test.go

@@ -597,11 +597,12 @@ func TestCSISnapshots(t *testing.T) {
 					Result:            volume.VolumeResultFailed,
 					SnapshotDataMoved: true,
 					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
-						UploaderType:   "fake-uploader",
-						SnapshotHandle: "fake-repo-id-5",
-						OperationID:    "fake-operation-5",
-						Size:           100,
-						Phase:          velerov2alpha1.DataUploadPhaseFailed,
+						UploaderType:    "fake-uploader",
+						SnapshotHandle:  "fake-repo-id-5",
+						OperationID:     "fake-operation-5",
+						Size:            100,
+						IncrementalSize: 50,
+						Phase:           velerov2alpha1.DataUploadPhaseFailed,
 					},
 				},
 			},
@@ -613,6 +614,7 @@ func TestCSISnapshots(t *testing.T) {
         Data Mover: velero
         Uploader Type: fake-uploader
         Moved data Size (bytes): 100
+        Incremental data Size (bytes): 50
         Result: failed
 `,
 		},

pkg/cmd/util/output/backup_structured_describer.go

@@ -464,6 +464,10 @@ func describeDataMovementInSF(details bool, info *volume.BackupVolumeInfo, snaps
 
 		dataMovement["uploaderType"] = info.SnapshotDataMovementInfo.UploaderType
 		dataMovement["result"] = string(info.Result)
+		if info.SnapshotDataMovementInfo.Size > 0 || info.SnapshotDataMovementInfo.IncrementalSize > 0 {
+			dataMovement["size"] = info.SnapshotDataMovementInfo.Size
+			dataMovement["incrementalSize"] = info.SnapshotDataMovementInfo.IncrementalSize
+		}
 
 		snapshotDetail["dataMovement"] = dataMovement
 	} else {
@@ -534,7 +538,7 @@ func describePodVolumeBackupsInSF(backups []velerov1api.PodVolumeBackup, details
 		// group the backups in the current phase by pod (i.e. "ns/name")
 		backupsByPod := new(volumesByPod)
 		for _, backup := range backupsByPhase[phase] {
-			backupsByPod.Add(backup.Spec.Pod.Namespace, backup.Spec.Pod.Name, backup.Spec.Volume, phase, backup.Status.Progress)
+			backupsByPod.Add(backup.Spec.Pod.Namespace, backup.Spec.Pod.Name, backup.Spec.Volume, phase, backup.Status.Progress, backup.Status.IncrementalBytes)
 		}
 
 		backupsByPods := make([]map[string]string, 0)

pkg/cmd/util/output/restore_describer.go

@@ -408,7 +408,7 @@ func describePodVolumeRestores(d *Describer, restores []velerov1api.PodVolumeRes
 		restoresByPod := new(volumesByPod)
 
 		for _, restore := range restoresByPhase[phase] {
-			restoresByPod.Add(restore.Spec.Pod.Namespace, restore.Spec.Pod.Name, restore.Spec.Volume, phase, restore.Status.Progress)
+			restoresByPod.Add(restore.Spec.Pod.Namespace, restore.Spec.Pod.Name, restore.Spec.Volume, phase, restore.Status.Progress, 0)
 		}
 
 		d.Printf("\t%s:\n", phase)

pkg/controller/backup_repository_controller.go

@@ -608,7 +608,7 @@ func getBackupRepositoryConfig(ctx context.Context, ctrlClient client.Client, co
 
 	jsonData, found := loc.Data[repoType]
 	if !found {
-		log.Info("No data for repo type %s in config map %s", repoType, configName)
+		log.Infof("No data for repo type %s in config map %s", repoType, configName)
 		return nil, nil
 	}
 

pkg/controller/data_upload_controller.go

@@ -493,6 +493,7 @@ func (r *DataUploadReconciler) OnDataUploadCompleted(ctx context.Context, namesp
 		du.Status.Path = result.Backup.Source.ByPath
 		du.Status.Phase = velerov2alpha1api.DataUploadPhaseCompleted
 		du.Status.SnapshotID = result.Backup.SnapshotID
+		du.Status.IncrementalBytes = result.Backup.IncrementalBytes
 		du.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
 		if result.Backup.EmptySnapshot {
 			du.Status.Message = "volume was empty so no data was upload"

pkg/controller/data_upload_controller_test.go

@@ -850,11 +850,20 @@ func TestOnDataUploadCompleted(t *testing.T) {
 	// Add the DataUpload object to the fake client
 	require.NoError(t, r.client.Create(ctx, du))
 	r.snapshotExposerList = map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer{velerov2alpha1api.SnapshotTypeCSI: exposer.NewCSISnapshotExposer(r.kubeClient, r.csiSnapshotClient, velerotest.NewLogger())}
-	r.OnDataUploadCompleted(ctx, namespace, duName, datapath.Result{})
+	r.OnDataUploadCompleted(ctx, namespace, duName, datapath.Result{
+		Backup: datapath.BackupResult{
+			SnapshotID: "fake-id",
+			Source: datapath.AccessPoint{
+				ByPath: "fake-path",
+			},
+		},
+	})
 	updatedDu := &velerov2alpha1api.DataUpload{}
 	require.NoError(t, r.client.Get(ctx, types.NamespacedName{Name: duName, Namespace: namespace}, updatedDu))
 	assert.Equal(t, velerov2alpha1api.DataUploadPhaseCompleted, updatedDu.Status.Phase)
 	assert.False(t, updatedDu.Status.CompletionTimestamp.IsZero())
+	assert.Equal(t, "fake-id", updatedDu.Status.SnapshotID)
+	assert.Equal(t, "fake-path", updatedDu.Status.Path)
 }
 
 func TestFindDataUploadForPod(t *testing.T) {

pkg/controller/pod_volume_backup_controller.go

@@ -526,6 +526,7 @@ func (r *PodVolumeBackupReconciler) OnDataPathCompleted(ctx context.Context, nam
 		pvb.Status.Phase = velerov1api.PodVolumeBackupPhaseCompleted
 		pvb.Status.SnapshotID = result.Backup.SnapshotID
 		pvb.Status.CompletionTimestamp = &completionTime
+		pvb.Status.IncrementalBytes = result.Backup.IncrementalBytes
 		if result.Backup.EmptySnapshot {
 			pvb.Status.Message = "volume was empty so no snapshot was taken"
 		}

pkg/datamover/backup_micro_service_test.go

@@ -156,7 +156,7 @@ func TestOnDataUploadCompleted(t *testing.T) {
 		{
 			name:        "marshal fail",
 			marshalErr:  errors.New("fake-marshal-error"),
-			expectedErr: "Failed to marshal backup result { false { } 0}: fake-marshal-error",
+			expectedErr: "Failed to marshal backup result { false { } 0 0}: fake-marshal-error",
 		},
 		{
 			name:                "succeed",

pkg/datapath/file_system.go

@@ -182,7 +182,7 @@ func (fs *fileSystemBR) StartBackup(source AccessPoint, uploaderConfig map[strin
 			fs.wgDataPath.Done()
 		}()
 
-		snapshotID, emptySnapshot, totalBytes, err := fs.uploaderProv.RunBackup(fs.ctx, source.ByPath, backupParam.RealSource, backupParam.Tags, backupParam.ForceFull,
+		snapshotID, emptySnapshot, totalBytes, incrementalBytes, err := fs.uploaderProv.RunBackup(fs.ctx, source.ByPath, backupParam.RealSource, backupParam.Tags, backupParam.ForceFull,
 			backupParam.ParentSnapshot, source.VolMode, uploaderConfig, fs)
 
 		if err == provider.ErrorCanceled {
@@ -194,7 +194,7 @@ func (fs *fileSystemBR) StartBackup(source AccessPoint, uploaderConfig map[strin
 			}
 			fs.callbacks.OnFailed(context.Background(), fs.namespace, fs.jobName, dataPathErr)
 		} else {
-			fs.callbacks.OnCompleted(context.Background(), fs.namespace, fs.jobName, Result{Backup: BackupResult{snapshotID, emptySnapshot, source, totalBytes}})
+			fs.callbacks.OnCompleted(context.Background(), fs.namespace, fs.jobName, Result{Backup: BackupResult{snapshotID, emptySnapshot, source, totalBytes, incrementalBytes}})
 		}
 	}()
 

pkg/datapath/file_system_test.go

@@ -96,7 +96,7 @@ func TestAsyncBackup(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			fs := newFileSystemBR("job-1", "test", nil, "velero", Callbacks{}, velerotest.NewLogger()).(*fileSystemBR)
 			mockProvider := providerMock.NewProvider(t)
-			mockProvider.On("RunBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Backup.SnapshotID, test.result.Backup.EmptySnapshot, test.result.Backup.TotalBytes, test.err)
+			mockProvider.On("RunBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Backup.SnapshotID, test.result.Backup.EmptySnapshot, test.result.Backup.TotalBytes, test.result.Backup.IncrementalBytes, test.err)
 			mockProvider.On("Close", mock.Anything).Return(nil)
 			fs.uploaderProv = mockProvider
 			fs.initialized = true

pkg/datapath/types.go

@@ -30,10 +30,11 @@ type Result struct {
 
 // BackupResult represents the result of a backup
 type BackupResult struct {
-	SnapshotID    string      `json:"snapshotID"`
-	EmptySnapshot bool        `json:"emptySnapshot"`
-	Source        AccessPoint `json:"source,omitempty"`
-	TotalBytes    int64       `json:"totalBytes,omitempty"`
+	SnapshotID       string      `json:"snapshotID"`
+	EmptySnapshot    bool        `json:"emptySnapshot"`
+	Source           AccessPoint `json:"source,omitempty"`
+	TotalBytes       int64       `json:"totalBytes,omitempty"`
+	IncrementalBytes int64       `json:"incrementalBytes,omitempty"`
 }
 
 // RestoreResult represents the result of a restore

pkg/podvolume/backup_micro_service_test.go

@@ -155,7 +155,7 @@ func TestOnDataPathCompleted(t *testing.T) {
 		{
 			name:        "marshal fail",
 			marshalErr:  errors.New("fake-marshal-error"),
-			expectedErr: "Failed to marshal backup result { false { } 0}: fake-marshal-error",
+			expectedErr: "Failed to marshal backup result { false { } 0 0}: fake-marshal-error",
 		},
 		{
 			name:                "succeed",

pkg/podvolume/restorer.go

@@ -179,7 +179,7 @@ func (r *restorer) RestorePodVolumes(data RestoreData, tracker *volume.RestoreVo
 			}
 		}
 
-		volumeRestore := newPodVolumeRestore(data.Restore, data.Pod, data.BackupLocation, volume, backupInfo.snapshotID, repoIdentifier, backupInfo.uploaderType, data.SourceNamespace, pvc)
+		volumeRestore := newPodVolumeRestore(data.Restore, data.Pod, data.BackupLocation, volume, backupInfo.snapshotID, backupInfo.snapshotSize, repoIdentifier, backupInfo.uploaderType, data.SourceNamespace, pvc)
 		if err := veleroclient.CreateRetryGenerateName(r.crClient, r.ctx, volumeRestore); err != nil {
 			errs = append(errs, errors.WithStack(err))
 			continue
@@ -252,7 +252,7 @@ ForEachVolume:
 	return errs
 }
 
-func newPodVolumeRestore(restore *velerov1api.Restore, pod *corev1api.Pod, backupLocation, volume, snapshot, repoIdentifier, uploaderType, sourceNamespace string, pvc *corev1api.PersistentVolumeClaim) *velerov1api.PodVolumeRestore {
+func newPodVolumeRestore(restore *velerov1api.Restore, pod *corev1api.Pod, backupLocation, volume, snapshot string, size int64, repoIdentifier, uploaderType, sourceNamespace string, pvc *corev1api.PersistentVolumeClaim) *velerov1api.PodVolumeRestore {
 	pvr := &velerov1api.PodVolumeRestore{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace:    restore.Namespace,
@@ -281,6 +281,7 @@ func newPodVolumeRestore(restore *velerov1api.Restore, pod *corev1api.Pod, backu
 			},
 			Volume:                volume,
 			SnapshotID:            snapshot,
+			SnapshotSize:          size,
 			BackupStorageLocation: backupLocation,
 			RepoIdentifier:        repoIdentifier,
 			UploaderType:          uploaderType,

pkg/podvolume/restorer_test.go

@@ -57,34 +57,34 @@ func TestGetVolumesRepositoryType(t *testing.T) {
 		{
 			name: "empty repository type, first one",
 			volumes: map[string]volumeBackupInfo{
-				"volume1": {"fake-snapshot-id-1", "fake-uploader-1", ""},
-				"volume2": {"", "", "fake-type"},
+				"volume1": {"fake-snapshot-id-1", 0, "fake-uploader-1", ""},
+				"volume2": {"", 0, "", "fake-type"},
 			},
 			expectedErr: "empty repository type found among volume snapshots, snapshot ID fake-snapshot-id-1, uploader fake-uploader-1",
 		},
 		{
 			name: "empty repository type, last one",
 			volumes: map[string]volumeBackupInfo{
-				"volume1": {"", "", "fake-type"},
-				"volume2": {"", "", "fake-type"},
-				"volume3": {"fake-snapshot-id-3", "fake-uploader-3", ""},
+				"volume1": {"", 0, "", "fake-type"},
+				"volume2": {"", 0, "", "fake-type"},
+				"volume3": {"fake-snapshot-id-3", 0, "fake-uploader-3", ""},
 			},
 			expectedErr: "empty repository type found among volume snapshots, snapshot ID fake-snapshot-id-3, uploader fake-uploader-3",
 		},
 		{
 			name: "empty repository type, middle one",
 			volumes: map[string]volumeBackupInfo{
-				"volume1": {"", "", "fake-type"},
-				"volume2": {"fake-snapshot-id-2", "fake-uploader-2", ""},
-				"volume3": {"", "", "fake-type"},
+				"volume1": {"", 0, "", "fake-type"},
+				"volume2": {"fake-snapshot-id-2", 0, "fake-uploader-2", ""},
+				"volume3": {"", 0, "", "fake-type"},
 			},
 			expectedErr: "empty repository type found among volume snapshots, snapshot ID fake-snapshot-id-2, uploader fake-uploader-2",
 		},
 		{
 			name: "mismatch repository type",
 			volumes: map[string]volumeBackupInfo{
-				"volume1": {"", "", "fake-type1"},
-				"volume2": {"fake-snapshot-id-2", "fake-uploader-2", "fake-type2"},
+				"volume1": {"", 0, "", "fake-type1"},
+				"volume2": {"fake-snapshot-id-2", 0, "fake-uploader-2", "fake-type2"},
 			},
 			prefixOnly:  true,
 			expectedErr: "multiple repository type in one backup",
@@ -92,9 +92,9 @@ func TestGetVolumesRepositoryType(t *testing.T) {
 		{
 			name: "success",
 			volumes: map[string]volumeBackupInfo{
-				"volume1": {"", "", "fake-type"},
-				"volume2": {"", "", "fake-type"},
-				"volume3": {"", "", "fake-type"},
+				"volume1": {"", 0, "", "fake-type"},
+				"volume2": {"", 0, "", "fake-type"},
+				"volume3": {"", 0, "", "fake-type"},
 			},
 			expected: "fake-type",
 		},

pkg/podvolume/util.go

@@ -39,6 +39,7 @@ const (
 // volumeBackupInfo describes the backup info of a volume backed up by PodVolumeBackups
 type volumeBackupInfo struct {
 	snapshotID     string
+	snapshotSize   int64
 	uploaderType   string
 	repositoryType string
 }
@@ -94,6 +95,7 @@ func getVolumeBackupInfoForPod(podVolumeBackups []*velerov1api.PodVolumeBackup,
 
 		volumes[pvb.Spec.Volume] = volumeBackupInfo{
 			snapshotID:     pvb.Status.SnapshotID,
+			snapshotSize:   pvb.Status.Progress.TotalBytes,
 			uploaderType:   getUploaderTypeOrDefault(pvb.Spec.UploaderType),
 			repositoryType: getRepositoryType(pvb.Spec.UploaderType),
 		}
@@ -109,7 +111,7 @@ func getVolumeBackupInfoForPod(podVolumeBackups []*velerov1api.PodVolumeBackup,
 	}
 
 	for k, v := range fromAnnntation {
-		volumes[k] = volumeBackupInfo{v, uploader.ResticType, velerov1api.BackupRepositoryTypeRestic}
+		volumes[k] = volumeBackupInfo{v, 0, uploader.ResticType, velerov1api.BackupRepositoryTypeRestic}
 	}
 
 	return volumes

pkg/repository/provider/unified_repo.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/base64"
 	"fmt"
+	"maps"
 	"net/url"
 	"path"
 	"strconv"
@@ -80,11 +81,16 @@ func NewUnifiedRepoProvider(
 		log:              log,
 	}
 
-	repo.repoService = createRepoService(log)
+	repo.repoService = createRepoService(repoBackend, log)
 
 	return &repo
 }
 
+func GetUnifiedRepoClientSideCacheLimit(repoOption map[string]string, repoBackend string, log logrus.FieldLogger) int64 {
+	repoService := createRepoService(repoBackend, log)
+	return repoService.ClientSideCacheLimit(repoOption)
+}
+
 func (urp *unifiedRepoProvider) InitRepo(ctx context.Context, param RepoParam) error {
 	log := urp.log.WithFields(logrus.Fields{
 		"BSL name":  param.BackupLocation.Name,
@@ -413,12 +419,11 @@ func (urp *unifiedRepoProvider) GetStoreOptions(param any) (map[string]string, e
 	}
 
 	storeOptions := make(map[string]string)
-	for k, v := range storeVar {
-		storeOptions[k] = v
-	}
+	maps.Copy(storeOptions, storeVar)
+	maps.Copy(storeOptions, storeCred)
 
-	for k, v := range storeCred {
-		storeOptions[k] = v
+	if repoParam.CacheDir != "" {
+		storeOptions[udmrepo.StoreOptionCacheDir] = repoParam.CacheDir
 	}
 
 	return storeOptions, nil
@@ -594,6 +599,6 @@ func getStorageVariables(backupLocation *velerov1api.BackupStorageLocation, repo
 	return result, nil
 }
 
-func createRepoService(log logrus.FieldLogger) udmrepo.BackupRepoService {
-	return reposervice.Create(log)
+func createRepoService(repoBackend string, log logrus.FieldLogger) udmrepo.BackupRepoService {
+	return reposervice.Create(repoBackend, log)
 }

pkg/repository/udmrepo/kopialib/backend/common.go

@@ -33,7 +33,7 @@ import (
 )
 
 const (
-	defaultCacheLimitMB    = 5000
+	DefaultCacheLimitMB    = 5000
 	maxCacheDurationSecond = 30
 )
 
@@ -66,7 +66,8 @@ func SetupNewRepositoryOptions(ctx context.Context, flags map[string]string) rep
 
 // SetupConnectOptions setups the options when connecting to an existing Kopia repository
 func SetupConnectOptions(ctx context.Context, repoOptions udmrepo.RepoOptions) repo.ConnectOptions {
-	cacheLimit := optionalHaveIntWithDefault(ctx, udmrepo.StoreOptionCacheLimit, repoOptions.StorageOptions, defaultCacheLimitMB) << 20
+	cacheLimit := optionalHaveIntWithDefault(ctx, udmrepo.StoreOptionCacheLimit, repoOptions.StorageOptions, DefaultCacheLimitMB) << 20
+	cacheDir := optionalHaveString(udmrepo.StoreOptionCacheDir, repoOptions.StorageOptions)
 
 	// 80% for data cache and 20% for metadata cache and align to KB
 	dataCacheLimit := (cacheLimit / 5 * 4) >> 10
@@ -74,6 +75,7 @@ func SetupConnectOptions(ctx context.Context, repoOptions udmrepo.RepoOptions) r
 
 	return repo.ConnectOptions{
 		CachingOptions: content.CachingOptions{
+			CacheDirectory: cacheDir,
 			// softLimit 80%
 			ContentCacheSizeBytes:  (dataCacheLimit / 5 * 4) << 10,
 			MetadataCacheSizeBytes: (metadataCacheLimit / 5 * 4) << 10,

pkg/repository/udmrepo/kopialib/lib_repo.go

@@ -18,6 +18,7 @@ package kopialib
 
 import (
 	"context"
+	"encoding/json"
 	"os"
 	"strings"
 	"sync/atomic"
@@ -35,6 +36,7 @@ import (
 
 	"github.com/vmware-tanzu/velero/pkg/kopia"
 	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo/kopialib/backend"
 )
 
 type kopiaRepoService struct {
@@ -79,6 +81,7 @@ const (
 	defaultMaintainCheckPeriod     = time.Hour
 	overwriteFullMaintainInterval  = time.Duration(0)
 	overwriteQuickMaintainInterval = time.Duration(0)
+	repoBackend                    = "kopia"
 )
 
 var kopiaRepoOpen = repo.Open
@@ -220,6 +223,34 @@ func (ks *kopiaRepoService) DefaultMaintenanceFrequency() time.Duration {
 	return defaultMaintainCheckPeriod
 }
 
+func (ks *kopiaRepoService) ClientSideCacheLimit(repoOption map[string]string) int64 {
+	defaultLimit := int64(backend.DefaultCacheLimitMB << 20)
+	if repoOption == nil {
+		return defaultLimit
+	}
+
+	if v, found := repoOption[repoBackend]; found {
+		var configs map[string]any
+		if err := json.Unmarshal([]byte(v), &configs); err != nil {
+			ks.logger.WithError(err).Warnf("error unmarshalling config data from data %v", v)
+			return defaultLimit
+		}
+
+		limit := defaultLimit
+		if v, found := configs[udmrepo.StoreOptionCacheLimit]; found {
+			if iv, ok := v.(float64); ok {
+				limit = int64(iv) << 20
+			} else {
+				ks.logger.Warnf("ignore cache limit from data %v", v)
+			}
+		}
+
+		return limit
+	}
+
+	return defaultLimit
+}
+
 func (km *kopiaMaintenance) runMaintenance(ctx context.Context, rep repo.DirectRepositoryWriter) error {
 	err := snapshotmaintenance.Run(kopia.SetupKopiaLog(ctx, km.logger), rep, km.mode, false, maintenance.SafetyFull)
 	if err != nil {

pkg/repository/udmrepo/kopialib/lib_repo_test.go

@@ -1345,3 +1345,63 @@ func TestMaintainProgress(t *testing.T) {
 		})
 	}
 }
+
+func TestClientSideCacheLimit(t *testing.T) {
+	testCases := []struct {
+		name       string
+		repoOption map[string]string
+		expected   int64
+	}{
+		{
+			name:     "nil option",
+			expected: 5000 << 20,
+		},
+		{
+			name: "no option",
+			repoOption: map[string]string{
+				"other-repo": "\"enableCompression\": true",
+			},
+			expected: 5000 << 20,
+		},
+		{
+			name: "unmarshall fails",
+			repoOption: map[string]string{
+				"kopia": "wrong-json",
+			},
+			expected: 5000 << 20,
+		},
+		{
+			name: "no cache limit",
+			repoOption: map[string]string{
+				"kopia": "{\"enableCompression\": true}",
+			},
+			expected: 5000 << 20,
+		},
+		{
+			name: "wrong cache value type",
+			repoOption: map[string]string{
+				"kopia": "{\"cacheLimitMB\": \"abcd\",\"enableCompression\": true}",
+			},
+			expected: 5000 << 20,
+		},
+		{
+			name: "succeed",
+			repoOption: map[string]string{
+				"kopia": "{\"cacheLimitMB\": 1,\"enableCompression\": true}",
+			},
+			expected: 1048576,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ks := &kopiaRepoService{
+				logger: velerotest.NewLogger(),
+			}
+
+			limit := ks.ClientSideCacheLimit(tc.repoOption)
+
+			assert.Equal(t, tc.expected, limit)
+		})
+	}
+}

pkg/repository/udmrepo/mocks/BackupRepoService.go

@@ -16,6 +16,24 @@ type BackupRepoService struct {
 	mock.Mock
 }
 
+// ClientSideCacheLimit provides a mock function with given fields: repoOption
+func (_m *BackupRepoService) ClientSideCacheLimit(repoOption map[string]string) int64 {
+	ret := _m.Called(repoOption)
+
+	if len(ret) == 0 {
+		panic("no return value specified for ClientSideCacheLimit")
+	}
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(map[string]string) int64); ok {
+		r0 = rf(repoOption)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	return r0
+}
+
 // Connect provides a mock function with given fields: ctx, repoOption
 func (_m *BackupRepoService) Connect(ctx context.Context, repoOption udmrepo.RepoOptions) error {
 	ret := _m.Called(ctx, repoOption)

pkg/repository/udmrepo/repo.go

@@ -100,6 +100,9 @@ type BackupRepoService interface {
 	// DefaultMaintenanceFrequency returns the defgault frequency of maintenance, callers refer this
 	// frequency to maintain the backup repository to get the best maintenance performance
 	DefaultMaintenanceFrequency() time.Duration
+
+	// ClientSideCacheLimit returns the max cache size required on client side
+	ClientSideCacheLimit(repoOption map[string]string) int64
 }
 
 // BackupRepo provides the access to the backup repository

pkg/repository/udmrepo/repo_options.go

@@ -17,6 +17,7 @@ limitations under the License.
 package udmrepo
 
 import (
+	"maps"
 	"os"
 	"path/filepath"
 	"strings"
@@ -65,6 +66,7 @@ const (
 	StoreOptionGenReadOnly        = "readOnly"
 
 	StoreOptionCacheLimit = "cacheLimitMB"
+	StoreOptionCacheDir   = "cacheDir"
 
 	ThrottleOptionReadOps       = "readOPS"
 	ThrottleOptionWriteOps      = "writeOPS"
@@ -184,9 +186,7 @@ func WithStoreOptions(getter StoreOptionsGetter, param any) func(*RepoOptions) e
 
 		options.StorageType = storeType
 
-		for k, v := range storeOptions {
-			options.StorageOptions[k] = v
-		}
+		maps.Copy(options.StorageOptions, storeOptions)
 
 		return nil
 	}

pkg/repository/udmrepo/service/service.go

@@ -24,6 +24,6 @@ import (
 )
 
 // Create creates an instance of BackupRepoService
-func Create(logger logrus.FieldLogger) udmrepo.BackupRepoService {
+func Create(repoBackend string, logger logrus.FieldLogger) udmrepo.BackupRepoService {
 	return kopialib.NewKopiaRepoService(logger)
 }

pkg/restore/actions/csi/pvc_action.go

@@ -431,6 +431,7 @@ func newDataDownload(
 			BackupStorageLocation: dataUploadResult.BackupStorageLocation,
 			DataMover:             dataUploadResult.DataMover,
 			SnapshotID:            dataUploadResult.SnapshotID,
+			SnapshotSize:          dataUploadResult.SnapshotSize,
 			SourceNamespace:       dataUploadResult.SourceNamespace,
 			OperationTimeout:      backup.Spec.CSISnapshotTimeout,
 			NodeOS:                dataUploadResult.NodeOS,

pkg/restore/actions/dataupload_retrieve_action.go

@@ -76,6 +76,7 @@ func (d *DataUploadRetrieveAction) Execute(input *velero.RestoreItemActionExecut
 		BackupStorageLocation: backup.Spec.StorageLocation,
 		DataMover:             dataUpload.Spec.DataMover,
 		SnapshotID:            dataUpload.Status.SnapshotID,
+		SnapshotSize:          dataUpload.Status.Progress.TotalBytes,
 		SourceNamespace:       dataUpload.Spec.SourceNamespace,
 		DataMoverResult:       dataUpload.Status.DataMoverResult,
 		NodeOS:                dataUpload.Status.NodeOS,

pkg/restore/actions/dataupload_retrieve_action_test.go

@@ -58,13 +58,13 @@ func TestDataUploadRetrieveActionExectue(t *testing.T) {
 		},
 		{
 			name:          "DataUploadRetrieve Action test",
-			dataUpload:    builder.ForDataUpload("velero", "testDU").SourceNamespace("testNamespace").SourcePVC("testPVC").Result(),
+			dataUpload:    builder.ForDataUpload("velero", "testDU").SourceNamespace("testNamespace").SourcePVC("testPVC").SnapshotID("fake-id").TotalBytes(1000).Result(),
 			restore:       builder.ForRestore("velero", "testRestore").ObjectMeta(builder.WithUID("testingUID")).Backup("testBackup").Result(),
 			runtimeScheme: scheme,
 			veleroObjs: []runtime.Object{
 				builder.ForBackup("velero", "testBackup").StorageLocation("testLocation").Result(),
 			},
-			expectedDataUploadResult: builder.ForConfigMap("velero", "").ObjectMeta(builder.WithGenerateName("testDU-"), builder.WithLabels(velerov1.PVCNamespaceNameLabel, "testNamespace.testPVC", velerov1.RestoreUIDLabel, "testingUID", velerov1.ResourceUsageLabel, string(velerov1.VeleroResourceUsageDataUploadResult))).Data("testingUID", `{"backupStorageLocation":"testLocation","sourceNamespace":"testNamespace"}`).Result(),
+			expectedDataUploadResult: builder.ForConfigMap("velero", "").ObjectMeta(builder.WithGenerateName("testDU-"), builder.WithLabels(velerov1.PVCNamespaceNameLabel, "testNamespace.testPVC", velerov1.RestoreUIDLabel, "testingUID", velerov1.ResourceUsageLabel, string(velerov1.VeleroResourceUsageDataUploadResult))).Data("testingUID", `{"backupStorageLocation":"testLocation","snapshotID":"fake-id","sourceNamespace":"testNamespace","snapshotSize":1000}`).Result(),
 		},
 		{
 			name:          "Long source namespace and PVC name should also work",

pkg/uploader/kopia/progress.go

@@ -177,3 +177,7 @@ func (p *Progress) EstimationParameters() upload.EstimationParameters {
 func (p *Progress) Enabled() bool {
 	return true
 }
+
+func (p *Progress) GetIncrementalSize() int64 {
+	return p.estimatedTotalBytes - p.cachedBytes
+}

pkg/uploader/provider/kopia.go

@@ -74,7 +74,7 @@ func NewKopiaUploaderProvider(
 		return nil, errors.Wrapf(err, "error to get repo options")
 	}
 
-	repoSvc := BackupRepoServiceCreateFunc(log)
+	repoSvc := BackupRepoServiceCreateFunc(backupRepo.Spec.RepositoryType, log)
 	log.WithField("repoUID", repoUID).Info("Opening backup repo")
 
 	kp.bkRepo, err = repoSvc.Open(ctx, *repoOpt)
@@ -120,13 +120,13 @@ func (kp *kopiaProvider) RunBackup(
 	parentSnapshot string,
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
-	updater uploader.ProgressUpdater) (string, bool, int64, error) {
+	updater uploader.ProgressUpdater) (string, bool, int64, int64, error) {
 	if updater == nil {
-		return "", false, 0, errors.New("Need to initial backup progress updater first")
+		return "", false, 0, 0, errors.New("Need to initial backup progress updater first")
 	}
 
 	if path == "" {
-		return "", false, 0, errors.New("path is empty")
+		return "", false, 0, 0, errors.New("path is empty")
 	}
 
 	log := kp.log.WithFields(logrus.Fields{
@@ -136,7 +136,8 @@ func (kp *kopiaProvider) RunBackup(
 	})
 	repoWriter := kopia.NewShimRepo(kp.bkRepo)
 	kpUploader := upload.NewUploader(repoWriter)
-	kpUploader.Progress = kopia.NewProgress(updater, backupProgressCheckInterval, log)
+	progress := kopia.NewProgress(updater, backupProgressCheckInterval, log)
+	kpUploader.Progress = progress
 	kpUploader.FailFast = true
 	quit := make(chan struct{})
 	log.Info("Starting backup")
@@ -175,9 +176,9 @@ func (kp *kopiaProvider) RunBackup(
 
 		if kpUploader.IsCanceled() {
 			log.Warn("Kopia backup is canceled")
-			return snapshotID, false, 0, ErrorCanceled
+			return snapshotID, false, 0, 0, ErrorCanceled
 		}
-		return snapshotID, false, 0, errors.Wrapf(err, "Failed to run kopia backup")
+		return snapshotID, false, 0, 0, errors.Wrapf(err, "Failed to run kopia backup")
 	}
 
 	// which ensure that the statistic data of TotalBytes equal to BytesDone when finished
@@ -189,7 +190,7 @@ func (kp *kopiaProvider) RunBackup(
 	)
 
 	log.Debugf("Kopia backup finished, snapshot ID %s, backup size %d", snapshotInfo.ID, snapshotInfo.Size)
-	return snapshotInfo.ID, false, snapshotInfo.Size, nil
+	return snapshotInfo.ID, false, snapshotInfo.Size, progress.GetIncrementalSize(), nil
 }
 
 func (kp *kopiaProvider) GetPassword(param any) (string, error) {

pkg/uploader/provider/kopia_test.go

@@ -106,7 +106,7 @@ func TestRunBackup(t *testing.T) {
 				tc.volMode = uploader.PersistentVolumeFilesystem
 			}
 			BackupFunc = tc.hookBackupFunc
-			_, _, _, err := kp.RunBackup(t.Context(), "var", "", nil, false, "", tc.volMode, map[string]string{}, &updater)
+			_, _, _, _, err := kp.RunBackup(t.Context(), "var", "", nil, false, "", tc.volMode, map[string]string{}, &updater)
 			if tc.notError {
 				assert.NoError(t, err)
 			} else {
@@ -373,7 +373,7 @@ func TestNewKopiaUploaderProvider(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			credGetter := &credentials.CredentialGetter{FromSecret: tc.mockCredGetter}
-			BackupRepoServiceCreateFunc = func(logger logrus.FieldLogger) udmrepo.BackupRepoService {
+			BackupRepoServiceCreateFunc = func(string, logrus.FieldLogger) udmrepo.BackupRepoService {
 				return tc.mockBackupRepoService
 			}
 			// Call the function being tested.

pkg/uploader/provider/mocks/Provider.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v2.39.1. DO NOT EDIT.
+// Code generated by mockery v2.53.5. DO NOT EDIT.
 
 package mocks
 
@@ -34,7 +34,7 @@ func (_m *Provider) Close(ctx context.Context) error {
 }
 
 // RunBackup provides a mock function with given fields: ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater
-func (_m *Provider) RunBackup(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, error) {
+func (_m *Provider) RunBackup(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, error) {
 	ret := _m.Called(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
 
 	if len(ret) == 0 {
@@ -44,8 +44,9 @@ func (_m *Provider) RunBackup(ctx context.Context, path string, realSource strin
 	var r0 string
 	var r1 bool
 	var r2 int64
-	var r3 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (string, bool, int64, error)); ok {
+	var r3 int64
+	var r4 error
+	if rf, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (string, bool, int64, int64, error)); ok {
 		return rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
 	}
 	if rf, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) string); ok {
@@ -66,13 +67,19 @@ func (_m *Provider) RunBackup(ctx context.Context, path string, realSource strin
 		r2 = ret.Get(2).(int64)
 	}
 
-	if rf, ok := ret.Get(3).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+	if rf, ok := ret.Get(3).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
 		r3 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
 	} else {
-		r3 = ret.Error(3)
+		r3 = ret.Get(3).(int64)
 	}
 
-	return r0, r1, r2, r3
+	if rf, ok := ret.Get(4).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+		r4 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
+	} else {
+		r4 = ret.Error(4)
+	}
+
+	return r0, r1, r2, r3, r4
 }
 
 // RunRestore provides a mock function with given fields: ctx, snapshotID, volumePath, volMode, uploaderConfig, updater

pkg/uploader/provider/provider.go

@@ -50,7 +50,7 @@ type Provider interface {
 		parentSnapshot string,
 		volMode uploader.PersistentVolumeMode,
 		uploaderCfg map[string]string,
-		updater uploader.ProgressUpdater) (string, bool, int64, error)
+		updater uploader.ProgressUpdater) (string, bool, int64, int64, error)
 	// RunRestore which will do restore for one specific volume with given snapshot id and return error
 	// updater is used for updating backup progress which implement by third-party
 	RunRestore(

pkg/uploader/provider/restic.go

@@ -124,21 +124,21 @@ func (rp *resticProvider) RunBackup(
 	parentSnapshot string,
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
-	updater uploader.ProgressUpdater) (string, bool, int64, error) {
+	updater uploader.ProgressUpdater) (string, bool, int64, int64, error) {
 	if updater == nil {
-		return "", false, 0, errors.New("Need to initial backup progress updater first")
+		return "", false, 0, 0, errors.New("Need to initial backup progress updater first")
 	}
 
 	if path == "" {
-		return "", false, 0, errors.New("path is empty")
+		return "", false, 0, 0, errors.New("path is empty")
 	}
 
 	if realSource != "" {
-		return "", false, 0, errors.New("real source is not empty, this is not supported by restic uploader")
+		return "", false, 0, 0, errors.New("real source is not empty, this is not supported by restic uploader")
 	}
 
 	if volMode == uploader.PersistentVolumeBlock {
-		return "", false, 0, errors.New("unable to support block mode")
+		return "", false, 0, 0, errors.New("unable to support block mode")
 	}
 
 	log := rp.log.WithFields(logrus.Fields{
@@ -149,7 +149,7 @@ func (rp *resticProvider) RunBackup(
 	if len(uploaderCfg) > 0 {
 		parallelFilesUpload, err := uploaderutil.GetParallelFilesUpload(uploaderCfg)
 		if err != nil {
-			return "", false, 0, errors.Wrap(err, "failed to get uploader config")
+			return "", false, 0, 0, errors.Wrap(err, "failed to get uploader config")
 		}
 		if parallelFilesUpload > 0 {
 			log.Warnf("ParallelFilesUpload is set to %d, but restic does not support parallel file uploads. Ignoring.", parallelFilesUpload)
@@ -171,9 +171,9 @@ func (rp *resticProvider) RunBackup(
 	if err != nil {
 		if strings.Contains(stderrBuf, "snapshot is empty") {
 			log.Debugf("Restic backup got empty dir with %s path", path)
-			return "", true, 0, nil
+			return "", true, 0, 0, nil
 		}
-		return "", false, 0, errors.WithStack(fmt.Errorf("error running restic backup command %s with error: %v stderr: %v", backupCmd.String(), err, stderrBuf))
+		return "", false, 0, 0, errors.WithStack(fmt.Errorf("error running restic backup command %s with error: %v stderr: %v", backupCmd.String(), err, stderrBuf))
 	}
 	// GetSnapshotID
 	snapshotIDCmd := resticGetSnapshotFunc(rp.repoIdentifier, rp.credentialsFile, tags)
@@ -184,10 +184,10 @@ func (rp *resticProvider) RunBackup(
 	}
 	snapshotID, err := resticGetSnapshotIDFunc(snapshotIDCmd)
 	if err != nil {
-		return "", false, 0, errors.WithStack(fmt.Errorf("error getting snapshot id with error: %v", err))
+		return "", false, 0, 0, errors.WithStack(fmt.Errorf("error getting snapshot id with error: %v", err))
 	}
 	log.Infof("Run command=%s, stdout=%s, stderr=%s", backupCmd.String(), summary, stderrBuf)
-	return snapshotID, false, 0, nil
+	return snapshotID, false, 0, 0, nil
 }
 
 // RunRestore runs a `restore` command and monitors the volume size to

pkg/uploader/provider/restic_test.go

@@ -149,9 +149,9 @@ func TestResticRunBackup(t *testing.T) {
 			}
 			if !tc.nilUpdater {
 				updater := FakeBackupProgressUpdater{PodVolumeBackup: &velerov1api.PodVolumeBackup{}, Log: tc.rp.log, Ctx: t.Context(), Cli: fake.NewClientBuilder().WithScheme(util.VeleroScheme).Build()}
-				_, _, _, err = tc.rp.RunBackup(t.Context(), "var", "", map[string]string{}, false, parentSnapshot, tc.volMode, map[string]string{}, &updater)
+				_, _, _, _, err = tc.rp.RunBackup(t.Context(), "var", "", map[string]string{}, false, parentSnapshot, tc.volMode, map[string]string{}, &updater)
 			} else {
-				_, _, _, err = tc.rp.RunBackup(t.Context(), "var", "", map[string]string{}, false, parentSnapshot, tc.volMode, map[string]string{}, nil)
+				_, _, _, _, err = tc.rp.RunBackup(t.Context(), "var", "", map[string]string{}, false, parentSnapshot, tc.volMode, map[string]string{}, nil)
 			}
 
 			tc.rp.log.Infof("test name %v error %v", tc.name, err)

site/content/docs/main/csi-snapshot-data-movement.md

@@ -116,12 +116,17 @@ velero backup create NAME --snapshot-move-data --data-mover DATA-MOVER-NAME OPTI
 When the backup starts, you will see the `VolumeSnapshot` and `VolumeSnapshotContent` objects created, but after the backup finishes, the objects will disappear.  
 After snapshots are created, you will see one or more `DataUpload` CRs created.  
 You may also see some intermediate objects (i.e., pods, PVCs, PVs) created in Velero namespace or the cluster scope, they are to help data movers to move data. And they will be removed after the backup completes.  
-The phase of a `DataUpload` CR changes several times during the backup process and finally goes to one of the terminal status, `Completed`, `Failed` or `Cancelled`. You can see the phase changes as well as the data upload progress by watching the `DataUpload` CRs:  
+The phase of a `DataUpload` CR changes several times during the backup process and finally goes to one of the terminal status, `Completed`, `Failed` or `Cancelled`. You can see the phase changes as well as the data upload progress by watching the `DataUpload` CRs. While the `DataUpload` is being processed, progress is shown with `BYTES DONE` representing the amount of volume data that has been processed so far and `TOTAL BYTES` representing the estimated total volume data. Upon completion, these two numbers will be the same. In addition, once the `DataUpload` is done, `INCREMENTAL BYTES` will be filled in with the amount of data which is new or changed since the last backup of this volume. Note that the actual uploaded content may be smaller than `INCREMENTAL BYTES` due to kopia deduplication, compression, etc.  
 
 ```bash
 kubectl -n velero get datauploads -l velero.io/backup-name=YOUR_BACKUP_NAME -w
 ```
 
+By default, `INCREMENTAL BYTES` is not displayed in the `kubectl get` output. To see this extended field, the `-o wide` arg is needed:
+```bash
+kubectl -n velero get datauploads -o wide -l velero.io/backup-name=YOUR_BACKUP_NAME -w
+```
+
 When the backup completes, you can view information about the backups:
 
 ```bash