Merge pull request #10517 from Lyndon-Li/add-fallback-full-to-volume-info

Add fallback full to volume info

changelogs/unreleased/10517-Lyndon-Li

@@ -0,0 +1 @@
+Add fallback full to DU/DD/PVB/PVR CRs, volume info and backup/restore describe

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

@@ -204,6 +204,10 @@ spec:
                 format: date-time
                 nullable: true
                 type: string
+              fallbackFull:
+                description: FallbackFull indicates whether the incremental backup
+                  has fallen back to full backup
+                type: boolean
               incrementalBytes:
                 description: |-
                   IncrementalBytes holds the number of bytes new or changed since the last backup.

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

@@ -204,6 +204,10 @@ spec:
                 format: date-time
                 nullable: true
                 type: string
+              fallbackFull:
+                description: FallbackFull indicates whether the incremental restore
+                  has fallen back to full restore
+                type: boolean
               incrementalBytes:
                 description: IncrementalBytes holds the number of bytes restored incrementally
                 format: int64

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

@@ -213,6 +213,10 @@ spec:
                 format: date-time
                 nullable: true
                 type: string
+              fallbackFull:
+                description: FallbackFull indicates whether the incremental restore
+                  has fallen back to full restore
+                type: boolean
               incrementalBytes:
                 description: IncrementalBytes holds the number of bytes restored incrementally
                   since the last snapshot

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

@@ -195,6 +195,10 @@ spec:
                   as a result of the DataUpload.
                 nullable: true
                 type: object
+              fallbackFull:
+                description: FallbackFull indicates whether the incremental backup
+                  has fallen back to full backup
+                type: boolean
               incrementalBytes:
                 description: |-
                   IncrementalBytes holds the number of bytes new or changed since the last backup.

internal/volume/volumes_information.go

@@ -93,6 +93,9 @@ type BackupVolumeInfo struct {
 	// How volume data is backed up, with possible values including Full and Incremental.
 	BackupType velerov1api.BackupType `json:"backupType,omitempty"`
 
+	// FallbackFull indicates whether the incremental backup has fallen back to full backup
+	FallbackFull bool `json:"fallbackFull,omitempty"`
+
 	CSISnapshotInfo          *CSISnapshotInfo                `json:"csiSnapshotInfo,omitempty"`
 	SnapshotDataMovementInfo *BackupSnapshotDataMovementInfo `json:"snapshotDataMovementInfo,omitempty"`
 	NativeSnapshotInfo       *NativeSnapshotInfo             `json:"nativeSnapshotInfo,omitempty"`
@@ -126,6 +129,12 @@ type RestoreVolumeInfo struct {
 	// Whether the volume's data are restored via data movement
 	SnapshotDataMoved bool `json:"snapshotDataMoved"`
 
+	// Indicates the type of the restore, incremental or full.
+	RestoreType string `json:"restoreType,omitempty"`
+
+	// FallbackFull indicates whether the incremental restore has fallen back to full restore
+	FallbackFull bool `json:"fallbackFull,omitempty"`
+
 	CSISnapshotInfo          *CSISnapshotInfo                 `json:"csiSnapshotInfo,omitempty"`
 	SnapshotDataMovementInfo *RestoreSnapshotDataMovementInfo `json:"snapshotDataMovementInfo,omitempty"`
 	NativeSnapshotInfo       *NativeSnapshotInfo              `json:"nativeSnapshotInfo,omitempty"`
@@ -185,12 +194,6 @@ type BackupSnapshotDataMovementInfo struct {
 	// incremental and must stay distinguishable from "unknown".
 	IncrementalSize *int64 `json:"incrementalSize,omitempty"`
 
-	// ParentSnapshot specifies the parent snapshot that current backup is based on.
-	// If its value is "" or "auto", the data mover finds the recent backup of the same volume as parent.
-	// If its value is "none", the data mover will do a full backup
-	// If its value is a specific snapshotID, the data mover finds the specific snapshot as parent.
-	ParentSnapshot string `json:"parentSnapshot,omitempty"`
-
 	// The size of source volume, for backup only
 	SourceSize int64 `json:"sourceSize,omitempty"`
 
@@ -228,9 +231,6 @@ type RestoreSnapshotDataMovementInfo struct {
 
 	// The DataDownload's Status.Phase value
 	Phase velerov2alpha1.DataDownloadPhase `json:"phase"`
-
-	// Indicates the type of the restore, incremental or full.
-	RestoreType string `json:"restoreType,omitempty"`
 }
 
 // NativeSnapshotInfo is used for displaying the Velero native snapshot status.
@@ -335,9 +335,6 @@ type PodVolumeRestoreInfo struct {
 
 	// The PVR's Status.Phase value
 	Phase velerov1api.PodVolumeRestorePhase `json:"phase,omitempty"`
-
-	// Indicates the type of the restore, incremental or full.
-	RestoreType string `json:"restoreType,omitempty"`
 }
 
 func newPodVolumeInfoFromPVB(pvb *velerov1api.PodVolumeBackup) *PodVolumeBackupInfo {
@@ -364,7 +361,6 @@ func newPodVolumeInfoFromPVR(pvr *velerov1api.PodVolumeRestore) *PodVolumeRestor
 		PodName:         pvr.Spec.Pod.Name,
 		PodNamespace:    pvr.Spec.Pod.Namespace,
 		Phase:           pvr.Status.Phase,
-		RestoreType:     pvr.Spec.RestoreType,
 		IncrementalSize: pvr.Status.IncrementalBytes,
 	}
 }
@@ -617,6 +613,7 @@ func (v *BackupVolumesInformation) generateVolumeInfoFromPVB() {
 			CompletionTimestamp: pvb.Status.CompletionTimestamp,
 			PVBInfo:             newPodVolumeInfoFromPVB(pvb),
 			BackupType:          velerov1api.BackupTypeIncremental,
+			FallbackFull:        pvb.Status.FallbackFull,
 		}
 
 		if pvb.Spec.ParentSnapshot == veleroshared.ParentSnapshotNone {
@@ -749,7 +746,6 @@ func (v *BackupVolumesInformation) generateVolumeInfoFromDataUpload() {
 					UploaderType:   datamover.GetUploaderType(dataMover),
 					OperationID:    operation.Spec.OperationID,
 					Phase:          dataUpload.Status.Phase,
-					ParentSnapshot: dataUpload.Spec.ParentSnapshot,
 					Size:           dataUpload.Status.Progress.TotalBytes,
 					SnapshotHandle: dataUpload.Status.SnapshotID,
 				},
@@ -905,6 +901,8 @@ func (t *RestoreVolumeInfoTracker) Result() []*RestoreVolumeInfo {
 			SnapshotDataMoved: false,
 			PVRInfo:           newPodVolumeInfoFromPVR(pvr),
 			RestoreMethod:     PodVolumeRestore,
+			RestoreType:       pvr.Spec.RestoreType,
+			FallbackFull:      pvr.Status.FallbackFull,
 		}
 		pvcName, err := pvcByPodvolume(context.TODO(), t.client, pvr.Spec.Pod.Name, pvr.Spec.Pod.Namespace, pvr.Spec.Volume)
 		if err != nil {
@@ -1002,12 +1000,12 @@ func (t *RestoreVolumeInfoTracker) Result() []*RestoreVolumeInfo {
 			// the datadownload was initiated in CSI plugin
 			// For the same reason, no CSI snapshot info will be populated into volumeInfo
 			RestoreMethod: CSISnapshot,
+			RestoreType:   dd.Spec.RestoreType,
 			SnapshotDataMovementInfo: &RestoreSnapshotDataMovementInfo{
 				DataMover:       dataMover,
 				UploaderType:    datamover.GetUploaderType(dataMover),
 				SnapshotHandle:  dd.Spec.SnapshotID,
 				OperationID:     operationID,
-				RestoreType:     dd.Spec.RestoreType,
 				Size:            dd.Status.Progress.TotalBytes,
 				IncrementalSize: dd.Status.IncrementalBytes,
 				Phase:           dd.Status.Phase,

internal/volume/volumes_information_test.go

@@ -730,6 +730,79 @@ func TestGenerateVolumeInfoFromPVB(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "PVB's volume has a PVC with fallback to full",
+			pvMap: map[string]pvcPvInfo{
+				"testPV": {
+					PVCName:      "testPVC",
+					PVCNamespace: "velero",
+					PV: corev1api.PersistentVolume{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:   "testPV",
+							Labels: map[string]string{"a": "b"},
+						},
+						Spec: corev1api.PersistentVolumeSpec{
+							PersistentVolumeReclaimPolicy: corev1api.PersistentVolumeReclaimDelete,
+						},
+					},
+				},
+			},
+			pvb: func() *velerov1api.PodVolumeBackup {
+				pvb := builder.ForPodVolumeBackup("velero", "testPVB").
+					PodName("testPod").
+					PodNamespace("velero").
+					StartTimestamp(&now).
+					CompletionTimestamp(&now).
+					Phase(velerov1api.PodVolumeBackupPhaseCompleted).
+					TotalBytes(1024).
+					IncrementalBytes(512).
+					Result()
+				pvb.Status.FallbackFull = true
+				return pvb
+			}(),
+			pod: builder.ForPod("velero", "testPod").Containers(&corev1api.Container{
+				Name: "test",
+				VolumeMounts: []corev1api.VolumeMount{
+					{
+						Name:      "testVolume",
+						MountPath: "/data",
+					},
+				},
+			}).Volumes(
+				&corev1api.Volume{
+					Name: "",
+					VolumeSource: corev1api.VolumeSource{
+						PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+							ClaimName: "testPVC",
+						},
+					},
+				},
+			).Result(),
+			expectedVolumeInfos: []*BackupVolumeInfo{
+				{
+					PVCName:             "testPVC",
+					PVCNamespace:        "velero",
+					PVName:              "testPV",
+					BackupMethod:        PodVolumeBackup,
+					BackupType:          velerov1api.BackupTypeIncremental,
+					FallbackFull:        true,
+					StartTimestamp:      &now,
+					CompletionTimestamp: &now,
+					Result:              VolumeResultSucceeded,
+					PVBInfo: &PodVolumeBackupInfo{
+						PodName:         "testPod",
+						PodNamespace:    "velero",
+						Phase:           velerov1api.PodVolumeBackupPhaseCompleted,
+						Size:            1024,
+						IncrementalSize: ptr.To(int64(512)),
+					},
+					PVInfo: &PVInfo{
+						ReclaimPolicy: string(corev1api.PersistentVolumeReclaimDelete),
+						Labels:        map[string]string{"a": "b"},
+					},
+				},
+			},
+		},
 	}
 
 	for _, tc := range tests {
@@ -1095,6 +1168,7 @@ func TestRestoreVolumeInfoResult(t *testing.T) {
 					PVName:            "testPV2",
 					RestoreMethod:     PodVolumeRestore,
 					SnapshotDataMoved: false,
+					RestoreType:       "Incremental",
 					PVRInfo: &PodVolumeRestoreInfo{
 						SnapshotHandle:  "pvr-snap-001",
 						PodName:         "testPod",
@@ -1102,7 +1176,6 @@ func TestRestoreVolumeInfoResult(t *testing.T) {
 						UploaderType:    "kopia",
 						VolumeName:      "data-volume-1",
 						Phase:           velerov1api.PodVolumeRestorePhaseCompleted,
-						RestoreType:     "Incremental",
 						Size:            1024,
 						IncrementalSize: ptr.To(int64(512)),
 					},
@@ -1251,6 +1324,7 @@ func TestRestoreVolumeInfoResult(t *testing.T) {
 					PVName:            "testPV",
 					RestoreMethod:     CSISnapshot,
 					SnapshotDataMoved: true,
+					RestoreType:       "Incremental",
 					SnapshotDataMovementInfo: &RestoreSnapshotDataMovementInfo{
 						DataMover:       "velero",
 						UploaderType:    velerov1api.BackupRepositoryTypeKopia,
@@ -1259,7 +1333,6 @@ func TestRestoreVolumeInfoResult(t *testing.T) {
 						Phase:           velerov2alpha1.DataDownloadPhaseCompleted,
 						Size:            2048,
 						IncrementalSize: ptr.To(int64(512)),
-						RestoreType:     "Incremental",
 					},
 				},
 				{
@@ -1268,6 +1341,7 @@ func TestRestoreVolumeInfoResult(t *testing.T) {
 					PVName:            "testPV2",
 					RestoreMethod:     CSISnapshot,
 					SnapshotDataMoved: true,
+					RestoreType:       "Full",
 					SnapshotDataMovementInfo: &RestoreSnapshotDataMovementInfo{
 						DataMover:      "velero",
 						UploaderType:   velerov1api.BackupRepositoryTypeKopia,
@@ -1275,7 +1349,6 @@ func TestRestoreVolumeInfoResult(t *testing.T) {
 						OperationID:    "dd-operation-002",
 						Phase:          velerov2alpha1.DataDownloadPhaseCompleted,
 						Size:           4096,
-						RestoreType:    "Full",
 					},
 				},
 			},
@@ -1330,6 +1403,7 @@ func TestBackupVolumeInfoJSONRoundTrip(t *testing.T) {
 		Skipped:               false,
 		Result:                VolumeResultSucceeded,
 		BackupType:            velerov1api.BackupTypeIncremental,
+		FallbackFull:          true,
 		CSISnapshotInfo: &CSISnapshotInfo{
 			SnapshotHandle:            "csi-snap-1",
 			Size:                      2000,
@@ -1346,7 +1420,6 @@ func TestBackupVolumeInfoJSONRoundTrip(t *testing.T) {
 			OperationID:      "op-1",
 			Size:             1000,
 			IncrementalSize:  int64Ptr(200),
-			ParentSnapshot:   "parent-1",
 			Phase:            velerov2alpha1.DataUploadPhaseCompleted,
 		},
 		NativeSnapshotInfo: &NativeSnapshotInfo{
@@ -1391,7 +1464,7 @@ func TestBackupVolumeInfoJSONRoundTrip(t *testing.T) {
 	assert.Contains(t, jsonStr, `"operationID":"op-1"`)
 	assert.Contains(t, jsonStr, `"size":1000`)
 	assert.Contains(t, jsonStr, `"incrementalSize":200`)
-	assert.Contains(t, jsonStr, `"parentSnapshot":"parent-1"`)
+	assert.Contains(t, jsonStr, `"fallbackFull":true`)
 	assert.Contains(t, jsonStr, `"phase":"Completed"`)
 	assert.Contains(t, jsonStr, `"pvbInfo":{`)
 	assert.Contains(t, jsonStr, `"podName":"pod-1"`)
@@ -1421,6 +1494,8 @@ func TestRestoreVolumeInfoJSONRoundTrip(t *testing.T) {
 		PVName:            "pv-2",
 		RestoreMethod:     CSISnapshot,
 		SnapshotDataMoved: true,
+		RestoreType:       "Incremental",
+		FallbackFull:      true,
 		SnapshotDataMovementInfo: &RestoreSnapshotDataMovementInfo{
 			DataMover:        "velero",
 			UploaderType:     "kopia",
@@ -1430,7 +1505,6 @@ func TestRestoreVolumeInfoJSONRoundTrip(t *testing.T) {
 			Size:             3000,
 			IncrementalSize:  int64Ptr(300),
 			Phase:            velerov2alpha1.DataDownloadPhaseCompleted,
-			RestoreType:      "Incremental",
 		},
 		PVRInfo: &PodVolumeRestoreInfo{
 			SnapshotHandle:  "pvr-snap-1",
@@ -1442,7 +1516,6 @@ func TestRestoreVolumeInfoJSONRoundTrip(t *testing.T) {
 			PodNamespace:    "ns-2",
 			NodeName:        "node-2",
 			Phase:           velerov1api.PodVolumeRestorePhaseCompleted,
-			RestoreType:     "Incremental",
 		},
 		CSISnapshotInfo: &CSISnapshotInfo{
 			SnapshotHandle: "csi-snap-2",
@@ -1477,6 +1550,7 @@ func TestRestoreVolumeInfoJSONRoundTrip(t *testing.T) {
 	assert.Contains(t, jsonStr, `"incrementalSize":300`)
 	assert.Contains(t, jsonStr, `"phase":"Completed"`)
 	assert.Contains(t, jsonStr, `"restoreType":"Incremental"`)
+	assert.Contains(t, jsonStr, `"fallbackFull":true`)
 	assert.Contains(t, jsonStr, `"pvrInfo":{`)
 	assert.Contains(t, jsonStr, `"podName":"pod-2"`)
 	assert.Contains(t, jsonStr, `"podNamespace":"ns-2"`)
@@ -1519,7 +1593,6 @@ func TestNewPodVolumeInfoFromPVR(t *testing.T) {
 				PodName:         "pod-1",
 				PodNamespace:    "ns-1",
 				Phase:           velerov1api.PodVolumeRestorePhaseCompleted,
-				RestoreType:     "Incremental",
 			},
 		},
 		{
@@ -1558,17 +1631,21 @@ func TestNewPodVolumeInfoFromPVB(t *testing.T) {
 	}{
 		{
 			name: "all fields populated including incremental bytes",
-			pvb: builder.ForPodVolumeBackup("velero", "pvb-1").
-				SnapshotID("snap-1").
-				Volume("vol-1").
-				PodName("pod-1").
-				PodNamespace("ns-1").
-				Node("node-1").
-				UploaderType("kopia").
-				Phase(velerov1api.PodVolumeBackupPhaseCompleted).
-				TotalBytes(2048).
-				IncrementalBytes(512).
-				Result(),
+			pvb: func() *velerov1api.PodVolumeBackup {
+				pvb := builder.ForPodVolumeBackup("velero", "pvb-1").
+					SnapshotID("snap-1").
+					Volume("vol-1").
+					PodName("pod-1").
+					PodNamespace("ns-1").
+					Node("node-1").
+					UploaderType("kopia").
+					Phase(velerov1api.PodVolumeBackupPhaseCompleted).
+					TotalBytes(2048).
+					IncrementalBytes(512).
+					Result()
+				pvb.Status.FallbackFull = true
+				return pvb
+			}(),
 			expected: &PodVolumeBackupInfo{
 				SnapshotHandle:  "snap-1",
 				Size:            2048,

pkg/apis/velero/v1/pod_volume_backup_types.go

@@ -141,6 +141,9 @@ type PodVolumeBackupStatus struct {
 	// +optional
 	// +nullable
 	AcceptedTimestamp *metav1.Time `json:"acceptedTimestamp,omitempty"`
+
+	// FallbackFull indicates whether the incremental backup has fallen back to full backup
+	FallbackFull bool `json:"fallbackFull,omitempty"`
 }
 
 // TODO(2.0) After converting all resources to use the runttime-controller client,

pkg/apis/velero/v1/pod_volume_restore_type.go

@@ -124,6 +124,9 @@ type PodVolumeRestoreStatus struct {
 	// Node is name of the node where the pod volume restore is processed.
 	// +optional
 	Node string `json:"node,omitempty"`
+
+	// FallbackFull indicates whether the incremental restore has fallen back to full restore
+	FallbackFull bool `json:"fallbackFull,omitempty"`
 }
 
 // TODO(2.0) After converting all resources to use the runtime-controller client, the genclient and k8s:deepcopy markers will no longer be needed and should be removed.

pkg/apis/velero/v2alpha1/data_download_types.go

@@ -149,6 +149,9 @@ type DataDownloadStatus struct {
 	// +optional
 	// +nullable
 	AcceptedTimestamp *metav1.Time `json:"acceptedTimestamp,omitempty"`
+
+	// FallbackFull indicates whether the incremental restore has fallen back to full restore
+	FallbackFull bool `json:"fallbackFull,omitempty"`
 }
 
 // TODO(2.0) After converting all resources to use the runtime-controller client, the genclient and k8s:deepcopy markers will no longer be needed and should be removed.

pkg/apis/velero/v2alpha1/data_upload_types.go

@@ -198,6 +198,9 @@ type DataUploadStatus struct {
 	// +optional
 	// +nullable
 	AcceptedTimestamp *metav1.Time `json:"acceptedTimestamp,omitempty"`
+
+	// FallbackFull indicates whether the incremental backup has fallen back to full backup
+	FallbackFull bool `json:"fallbackFull,omitempty"`
 }
 
 // TODO(2.0) After converting all resources to use the runttime-controller client,

pkg/backup/backup.go

@@ -1330,6 +1330,7 @@ func updateVolumeInfos(
 				volumeInfos[index].SnapshotDataMovementInfo.IncrementalSize = dataUpload.Status.IncrementalBytes
 				volumeInfos[index].SnapshotDataMovementInfo.SourceSize = dataUpload.Status.SourceSize
 				volumeInfos[index].SnapshotDataMovementInfo.Phase = dataUpload.Status.Phase
+				volumeInfos[index].FallbackFull = dataUpload.Status.FallbackFull
 
 				if dataUpload.Status.Phase == velerov2alpha1.DataUploadPhaseCompleted {
 					volumeInfos[index].Result = volume.VolumeResultSucceeded

pkg/builder/data_download_builder.go

@@ -160,6 +160,12 @@ func (d *DataDownloadBuilder) IncrementalBytes(incrementalBytes int64) *DataDown
 	return d
 }
 
+// FallbackFull sets the DataDownload's FallbackFull status.
+func (d *DataDownloadBuilder) FallbackFull(fallbackFull bool) *DataDownloadBuilder {
+	d.object.Status.FallbackFull = fallbackFull
+	return d
+}
+
 // Node sets the DataDownload's Node.
 func (d *DataDownloadBuilder) Node(node string) *DataDownloadBuilder {
 	d.object.Status.Node = node

pkg/builder/volume_builder_test.go

@@ -36,6 +36,18 @@ func TestDataDownloadBuilder_Bytes(t *testing.T) {
 	assert.Equal(t, int64(512), *dd.Status.IncrementalBytes)
 }
 
+func TestDataDownloadBuilder_FallbackFull(t *testing.T) {
+	dd1 := ForDataDownload("velero", "dd-1").
+		FallbackFull(true).
+		Result()
+	assert.True(t, dd1.Status.FallbackFull)
+
+	dd2 := ForDataDownload("velero", "dd-2").
+		FallbackFull(false).
+		Result()
+	assert.False(t, dd2.Status.FallbackFull)
+}
+
 func TestPodVolumeBackupBuilder_ProgressAndBytes(t *testing.T) {
 	pvb1 := ForPodVolumeBackup("velero", "pvb-1").
 		Progress(shared.DataMoveOperationProgress{

pkg/cmd/util/output/backup_describer.go

@@ -741,6 +741,16 @@ func describeDataMovement(d *Describer, details bool, info *volume.BackupVolumeI
 			dataMover = info.SnapshotDataMovementInfo.DataMover
 		}
 		d.Printf("\t\t\t\tData Mover: %s\n", dataMover)
+
+		if info.BackupType != "" {
+			backupType := string(info.BackupType)
+			if info.FallbackFull {
+				backupType += " (fallen back to Full)"
+			}
+
+			d.Printf("\t\t\t\tBackup Type: %s\n", backupType)
+		}
+
 		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)
 		// Print whenever the uploader measured a figure, including zero. A zero-delta
@@ -750,9 +760,6 @@ func describeDataMovement(d *Describer, details bool, info *volume.BackupVolumeI
 		if info.SnapshotDataMovementInfo.IncrementalSize != nil {
 			d.Printf("\t\t\t\tIncremental data Size (bytes): %d\n", *info.SnapshotDataMovementInfo.IncrementalSize)
 		}
-		if info.SnapshotDataMovementInfo.ParentSnapshot != "" {
-			d.Printf("\t\t\t\tParent Snapshot: %s\n", info.SnapshotDataMovementInfo.ParentSnapshot)
-		}
 
 		d.Printf("\t\t\t\tResult: %s\n", info.Result)
 	} else {

pkg/cmd/util/output/backup_describer_test.go

@@ -635,7 +635,6 @@ func TestCSISnapshots(t *testing.T) {
 						Size:            100,
 						IncrementalSize: ptr.To(int64(50)),
 						Phase:           velerov2alpha1.DataUploadPhaseFailed,
-						ParentSnapshot:  "fake-parent-snapshot",
 					},
 				},
 			},
@@ -645,11 +644,46 @@ func TestCSISnapshots(t *testing.T) {
       Data Movement:
         Operation ID: fake-operation-5
         Data Mover: velero
+        Backup Type: Incremental
         Uploader Type: fake-uploader
         Moved data Size (bytes): 100
         Incremental data Size (bytes): 50
-        Parent Snapshot: fake-parent-snapshot
         Result: failed
+`,
+		},
+		{
+			name: "details, data movement, incremental fallback to full",
+			volumeInfo: []*volume.BackupVolumeInfo{
+				{
+					BackupMethod:      volume.CSISnapshot,
+					PVCNamespace:      "pvc-ns-6",
+					PVCName:           "pvc-6",
+					Result:            volume.VolumeResultSucceeded,
+					SnapshotDataMoved: true,
+					BackupType:        velerov1api.BackupTypeIncremental,
+					FallbackFull:      true,
+					SnapshotDataMovementInfo: &volume.BackupSnapshotDataMovementInfo{
+						DataMover:       "velero",
+						UploaderType:    "fake-uploader",
+						SnapshotHandle:  "fake-repo-id-6",
+						OperationID:     "fake-operation-6",
+						Size:            200,
+						IncrementalSize: ptr.To(int64(200)),
+						Phase:           velerov2alpha1.DataUploadPhaseCompleted,
+					},
+				},
+			},
+			inputDetails: true,
+			expect: `  CSI Snapshots:
+    pvc-ns-6/pvc-6:
+      Data Movement:
+        Operation ID: fake-operation-6
+        Data Mover: velero
+        Backup Type: Incremental (fallen back to Full)
+        Uploader Type: fake-uploader
+        Moved data Size (bytes): 200
+        Incremental data Size (bytes): 200
+        Result: succeeded
 `,
 		},
 	}

pkg/cmd/util/output/backup_structured_describer.go

@@ -462,6 +462,15 @@ func describeDataMovementInSF(details bool, info *volume.BackupVolumeInfo, snaps
 		dataMovement := make(map[string]any)
 		dataMovement["operationID"] = info.SnapshotDataMovementInfo.OperationID
 
+		if info.BackupType != "" {
+			backupType := string(info.BackupType)
+			if info.FallbackFull {
+				backupType += " (fallen back to Full)"
+			}
+
+			dataMovement["backupType"] = backupType
+		}
+
 		dataMover := "velero"
 		if info.SnapshotDataMovementInfo.DataMover != "" {
 			dataMover = info.SnapshotDataMovementInfo.DataMover
@@ -478,9 +487,6 @@ func describeDataMovementInSF(details bool, info *volume.BackupVolumeInfo, snaps
 		if info.SnapshotDataMovementInfo.IncrementalSize != nil {
 			dataMovement["incrementalSize"] = *info.SnapshotDataMovementInfo.IncrementalSize
 		}
-		if info.SnapshotDataMovementInfo.ParentSnapshot != "" {
-			dataMovement["parentSnapshot"] = info.SnapshotDataMovementInfo.ParentSnapshot
-		}
 
 		snapshotDetail["dataMovement"] = dataMovement
 	} else {

pkg/cmd/util/output/backup_structured_describer_test.go

@@ -585,7 +585,6 @@ func TestDescribeCSISnapshotsInSF(t *testing.T) {
 						Size:            100,
 						IncrementalSize: ptr.To(int64(50)),
 						Phase:           velerov2alpha1.DataUploadPhaseFailed,
-						ParentSnapshot:  "fake-parent-snapshot",
 					},
 				},
 			},
@@ -596,11 +595,50 @@ func TestDescribeCSISnapshotsInSF(t *testing.T) {
 						"dataMovement": map[string]any{
 							"operationID":     "fake-operation-4",
 							"dataMover":       "velero",
+							"backupType":      "Incremental",
 							"uploaderType":    "fake-uploader",
 							"size":            int64(100),
 							"incrementalSize": int64(50),
 							"result":          "failed",
-							"parentSnapshot":  "fake-parent-snapshot",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "details, data movement, incremental fallback to full",
+			volumeInfo: []*volume.BackupVolumeInfo{
+				{
+					BackupMethod:      volume.CSISnapshot,
+					PVCNamespace:      "pvc-ns-5",
+					Result:            volume.VolumeResultSucceeded,
+					PVCName:           "pvc-5",
+					SnapshotDataMoved: true,
+					BackupType:        velerov1api.BackupTypeIncremental,
+					FallbackFull:      true,
+					SnapshotDataMovementInfo: &volume.BackupSnapshotDataMovementInfo{
+						DataMover:       "velero",
+						UploaderType:    "fake-uploader",
+						SnapshotHandle:  "fake-repo-id-5",
+						OperationID:     "fake-operation-5",
+						Size:            200,
+						IncrementalSize: ptr.To(int64(200)),
+						Phase:           velerov2alpha1.DataUploadPhaseCompleted,
+					},
+				},
+			},
+			inputDetails: true,
+			expect: map[string]any{
+				"csiSnapshots": map[string]any{
+					"pvc-ns-5/pvc-5": map[string]any{
+						"dataMovement": map[string]any{
+							"operationID":     "fake-operation-5",
+							"dataMover":       "velero",
+							"backupType":      "Incremental (fallen back to Full)",
+							"uploaderType":    "fake-uploader",
+							"size":            int64(200),
+							"incrementalSize": int64(200),
+							"result":          "succeeded",
 						},
 					},
 				},

pkg/cmd/util/output/restore_describer.go

@@ -473,8 +473,13 @@ func describeCSISnapshotsRestores(d *Describer, restoreVolInfo []volume.RestoreV
 			d.Printf("\t\t\tOperation ID: %s\n", info.SnapshotDataMovementInfo.OperationID)
 			d.Printf("\t\t\tData Mover: %s\n", info.SnapshotDataMovementInfo.DataMover)
 			d.Printf("\t\t\tUploader Type: %s\n", info.SnapshotDataMovementInfo.UploaderType)
-			if info.SnapshotDataMovementInfo.RestoreType != "" {
-				d.Printf("\t\t\tRestore Type: %s\n", info.SnapshotDataMovementInfo.RestoreType)
+			if info.RestoreType != "" {
+				restoreType := info.RestoreType
+				if info.FallbackFull {
+					restoreType += " (fallen back to Full)"
+				}
+
+				d.Printf("\t\t\tRestore Type: %s\n", restoreType)
 			}
 			if info.SnapshotDataMovementInfo.Size > 0 {
 				d.Printf("\t\t\tRestored data Size (bytes): %d\n", info.SnapshotDataMovementInfo.Size)

pkg/cmd/util/output/restore_describer_test.go

@@ -312,13 +312,13 @@ CSI Snapshot Restores:
 					PVName:            "pv-3",
 					RestoreMethod:     volume.CSISnapshot,
 					SnapshotDataMoved: true,
+					RestoreType:       "Incremental",
 					SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{
 						OperationID:     "op-3",
 						DataMover:       "velero",
 						UploaderType:    "kopia",
 						Size:            1234,
 						IncrementalSize: ptr.To(int64(500)),
-						RestoreType:     "Incremental",
 					},
 				},
 			},
@@ -333,6 +333,39 @@ CSI Snapshot Restores:
       Restore Type: Incremental
       Restored data Size (bytes): 1234
       Incremental data Size (bytes): 500
+`,
+		},
+		{
+			name: "CSI restore with data movement, detailed, fallback to full",
+			inputVolInfoList: []volume.RestoreVolumeInfo{
+				{
+					PVCName:           "pvc-4",
+					PVCNamespace:      "ns-4",
+					PVName:            "pv-4",
+					RestoreMethod:     volume.CSISnapshot,
+					SnapshotDataMoved: true,
+					RestoreType:       "Incremental",
+					FallbackFull:      true,
+					SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{
+						OperationID:     "op-4",
+						DataMover:       "velero",
+						UploaderType:    "kopia",
+						Size:            2345,
+						IncrementalSize: ptr.To(int64(600)),
+					},
+				},
+			},
+			inputDetail: true,
+			expect: `
+CSI Snapshot Restores:
+  ns-4/pvc-4:
+    Data Movement:
+      Operation ID: op-4
+      Data Mover: velero
+      Uploader Type: kopia
+      Restore Type: Incremental (fallen back to Full)
+      Restored data Size (bytes): 2345
+      Incremental data Size (bytes): 600
 `,
 		},
 		{

pkg/cmd/util/output/restore_structured_describer.go

@@ -372,8 +372,13 @@ func describeCSISnapshotsRestoresInSF(d *StructuredDescriber, restoreVolInfo []v
 				"dataMover":    info.SnapshotDataMovementInfo.DataMover,
 				"uploaderType": info.SnapshotDataMovementInfo.UploaderType,
 			}
-			if info.SnapshotDataMovementInfo.RestoreType != "" {
-				dmInfo["restoreType"] = info.SnapshotDataMovementInfo.RestoreType
+			if info.RestoreType != "" {
+				restoreType := info.RestoreType
+				if info.FallbackFull {
+					restoreType += " (fallen back to Full)"
+				}
+
+				dmInfo["restoreType"] = restoreType
 			}
 			if info.SnapshotDataMovementInfo.Size > 0 {
 				dmInfo["size"] = info.SnapshotDataMovementInfo.Size

pkg/cmd/util/output/restore_structured_describer_test.go

@@ -473,13 +473,13 @@ func TestDescribeRestoreCSISnapshotsInSF_NoData(t *testing.T) {
 					SnapshotDataMoved: true,
 					PVCName:           "pvc-3",
 					PVCNamespace:      "ns-3",
+					RestoreType:       "Incremental",
 					SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{
 						OperationID:     "op-3",
 						DataMover:       "velero",
 						UploaderType:    "kopia",
 						Size:            1234,
 						IncrementalSize: ptr.To(int64(500)),
-						RestoreType:     "Incremental",
 					},
 				},
 			},
@@ -499,6 +499,41 @@ func TestDescribeRestoreCSISnapshotsInSF_NoData(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "data movement entry, with details, fallback to full",
+			inputVolInfoList: []volume.RestoreVolumeInfo{
+				{
+					RestoreMethod:     volume.CSISnapshot,
+					SnapshotDataMoved: true,
+					PVCName:           "pvc-4",
+					PVCNamespace:      "ns-4",
+					RestoreType:       "Incremental",
+					FallbackFull:      true,
+					SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{
+						OperationID:     "op-4",
+						DataMover:       "velero",
+						UploaderType:    "kopia",
+						Size:            2345,
+						IncrementalSize: ptr.To(int64(600)),
+					},
+				},
+			},
+			details: true,
+			expect: map[string]any{
+				"csiSnapshotRestores": map[string]any{
+					"ns-4/pvc-4": map[string]any{
+						"dataMovement": map[string]any{
+							"operationID":     "op-4",
+							"dataMover":       "velero",
+							"uploaderType":    "kopia",
+							"size":            int64(2345),
+							"incrementalSize": int64(600),
+							"restoreType":     "Incremental (fallen back to Full)",
+						},
+					},
+				},
+			},
+		},
 		{
 			name: "data movement entry, no details",
 			inputVolInfoList: []volume.RestoreVolumeInfo{

pkg/controller/data_download_controller.go

@@ -502,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.FallbackFull = result.Restore.FallbackFull
 		dd.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
 
 		delete(dd.Labels, exposer.ExposeOnGoingLabel)

pkg/controller/data_upload_controller.go

@@ -514,6 +514,7 @@ func (r *DataUploadReconciler) OnDataUploadCompleted(ctx context.Context, namesp
 		du.Status.SnapshotID = result.Backup.SnapshotID
 		du.Status.IncrementalBytes = result.Backup.IncrementalBytes
 		du.Status.SourceSize = result.Backup.SourceSize
+		du.Status.FallbackFull = result.Backup.FallbackFull
 
 		du.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
 		if result.Backup.EmptySnapshot {

pkg/controller/pod_volume_backup_controller.go

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

pkg/controller/pod_volume_restore_controller.go

@@ -838,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)
+		pvr.Status.FallbackFull = result.Restore.FallbackFull
 
 		delete(pvr.Labels, exposer.ExposeOnGoingLabel)
 

pkg/controller/restore_finalizer_controller.go

@@ -603,6 +603,7 @@ func (ctx *finalizerContext) updateVolumeInfos() (errs results.Result) {
 				ctx.restoreVolumeInfos[index].SnapshotDataMovementInfo != nil {
 				ctx.restoreVolumeInfos[index].SnapshotDataMovementInfo.Size = dataDownload.Status.Progress.TotalBytes
 				ctx.restoreVolumeInfos[index].SnapshotDataMovementInfo.IncrementalSize = dataDownload.Status.IncrementalBytes
+				ctx.restoreVolumeInfos[index].FallbackFull = dataDownload.Status.FallbackFull
 				ctx.restoreVolumeInfos[index].SnapshotDataMovementInfo.Phase = dataDownload.Status.Phase
 			}
 		}

pkg/controller/restore_finalizer_controller_test.go

@@ -1199,17 +1199,18 @@ func TestCleanupStubVGSC(t *testing.T) {
 
 func TestUpdateVolumeInfos(t *testing.T) {
 	tests := []struct {
-		name               string
-		restore            *velerov1api.Restore
-		restoreVolumeInfos []*volume.RestoreVolumeInfo
-		dataDownloads      []*velerov2alpha1.DataDownload
-		listErr            error
-		putErr             error
-		expectedSize       int64
-		expectedIncrSize   *int64
-		expectedPhase      velerov2alpha1.DataDownloadPhase
-		expectErrs         bool
-		expectErrMsg       string
+		name                 string
+		restore              *velerov1api.Restore
+		restoreVolumeInfos   []*volume.RestoreVolumeInfo
+		dataDownloads        []*velerov2alpha1.DataDownload
+		listErr              error
+		putErr               error
+		expectedSize         int64
+		expectedIncrSize     *int64
+		expectedPhase        velerov2alpha1.DataDownloadPhase
+		expectedFallbackFull bool
+		expectErrs           bool
+		expectErrMsg         string
 	}{
 		{
 			name:    "successful update of restore volume infos from data downloads",
@@ -1238,6 +1239,16 @@ func TestUpdateVolumeInfos(t *testing.T) {
 						Phase:     velerov2alpha1.DataDownloadPhaseCompleted,
 					},
 				},
+				{
+					PVCName:      "pvc-4",
+					PVCNamespace: "ns-4",
+					FallbackFull: true,
+					SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{
+						DataMover: "velero",
+						Size:      0,
+						Phase:     "",
+					},
+				},
 			},
 			dataDownloads: []*velerov2alpha1.DataDownload{
 				builder.ForDataDownload("velero", "dd-1").
@@ -1246,6 +1257,7 @@ func TestUpdateVolumeInfos(t *testing.T) {
 					TotalBytes(4096).
 					IncrementalBytes(1024).
 					Phase(velerov2alpha1.DataDownloadPhaseCompleted).
+					FallbackFull(true).
 					Result(),
 				builder.ForDataDownload("velero", "dd-2").
 					ObjectMeta(builder.WithLabelsMap(map[string]string{velerov1api.RestoreNameLabel: "restore-1"})).
@@ -1253,6 +1265,7 @@ func TestUpdateVolumeInfos(t *testing.T) {
 					TotalBytes(2048).
 					IncrementalBytes(512).
 					Phase(velerov2alpha1.DataDownloadPhaseCompleted).
+					FallbackFull(true).
 					Result(),
 				builder.ForDataDownload("velero", "dd-other-restore").
 					ObjectMeta(builder.WithLabelsMap(map[string]string{velerov1api.RestoreNameLabel: "restore-other"})).
@@ -1260,12 +1273,22 @@ func TestUpdateVolumeInfos(t *testing.T) {
 					TotalBytes(9999).
 					IncrementalBytes(8888).
 					Phase(velerov2alpha1.DataDownloadPhaseFailed).
+					FallbackFull(true).
+					Result(),
+				builder.ForDataDownload("velero", "dd-4").
+					ObjectMeta(builder.WithLabelsMap(map[string]string{velerov1api.RestoreNameLabel: "restore-1"})).
+					TargetVolume(velerov2alpha1.TargetVolumeSpec{PVC: "pvc-4", Namespace: "ns-4"}).
+					TotalBytes(1024).
+					IncrementalBytes(256).
+					Phase(velerov2alpha1.DataDownloadPhaseCompleted).
+					FallbackFull(false).
 					Result(),
 			},
-			expectedSize:     4096,
-			expectedIncrSize: ptr.To(int64(1024)),
-			expectedPhase:    velerov2alpha1.DataDownloadPhaseCompleted,
-			expectErrs:       false,
+			expectedSize:         4096,
+			expectedIncrSize:     ptr.To(int64(1024)),
+			expectedPhase:        velerov2alpha1.DataDownloadPhaseCompleted,
+			expectedFallbackFull: true,
+			expectErrs:           false,
 		},
 		{
 			name:    "failed to list data downloads",
@@ -1350,10 +1373,18 @@ func TestUpdateVolumeInfos(t *testing.T) {
 				assert.Equal(t, tc.expectedSize, ctx.restoreVolumeInfos[0].SnapshotDataMovementInfo.Size)
 				assert.Equal(t, tc.expectedIncrSize, ctx.restoreVolumeInfos[0].SnapshotDataMovementInfo.IncrementalSize)
 				assert.Equal(t, tc.expectedPhase, ctx.restoreVolumeInfos[0].SnapshotDataMovementInfo.Phase)
-				// pvc-2 had nil SnapshotDataMovementInfo and should remain nil
+				assert.Equal(t, tc.expectedFallbackFull, ctx.restoreVolumeInfos[0].FallbackFull)
+				// pvc-2 had nil SnapshotDataMovementInfo and should remain nil, FallbackFull should remain false
 				assert.Nil(t, ctx.restoreVolumeInfos[1].SnapshotDataMovementInfo)
+				assert.False(t, ctx.restoreVolumeInfos[1].FallbackFull)
 				// pvc-3 belonged to another restore and should be untouched
 				assert.Equal(t, int64(100), ctx.restoreVolumeInfos[2].SnapshotDataMovementInfo.Size)
+				assert.False(t, ctx.restoreVolumeInfos[2].FallbackFull)
+				// pvc-4 had FallbackFull updated to false from data download
+				assert.Equal(t, int64(1024), ctx.restoreVolumeInfos[3].SnapshotDataMovementInfo.Size)
+				assert.Equal(t, ptr.To(int64(256)), ctx.restoreVolumeInfos[3].SnapshotDataMovementInfo.IncrementalSize)
+				assert.Equal(t, velerov2alpha1.DataDownloadPhaseCompleted, ctx.restoreVolumeInfos[3].SnapshotDataMovementInfo.Phase)
+				assert.False(t, ctx.restoreVolumeInfos[3].FallbackFull)
 
 				// Verify the content uploaded to backup store can be decoded and matches
 				require.NotEmpty(t, uploadedData)

pkg/datamover/backup_micro_service_test.go

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

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 0}: fake-marshal-error",
+			expectedErr: "Failed to marshal restore result {{ } 0 0 false}: fake-marshal-error",
 		},
 		{
 			name:                "succeed",

pkg/datapath/data_path.go

@@ -197,7 +197,7 @@ func (dp *generalDataPath) StartBackup(source AccessPoint, uploaderConfig map[st
 			dp.wgDataPath.Done()
 		}()
 
-		snapshotID, emptySnapshot, totalBytes, incrementalBytes, sourceSize, err := dp.uploaderProv.RunBackup(
+		snapshotID, emptySnapshot, totalBytes, incrementalBytes, sourceSize, fallback, err := dp.uploaderProv.RunBackup(
 			dp.ctx,
 			source.ByPath,
 			backupParam.RealSource,
@@ -233,6 +233,7 @@ func (dp *generalDataPath) StartBackup(source AccessPoint, uploaderConfig map[st
 				TotalBytes:       totalBytes,
 				IncrementalBytes: ptr.To(incrementalBytes),
 				SourceSize:       sourceSize,
+				FallbackFull:     fallback,
 			}})
 		}
 	}()
@@ -257,7 +258,7 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u
 			dp.wgDataPath.Done()
 		}()
 
-		incrementalBytes, totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental,
+		incrementalBytes, totalBytes, fallback, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental,
 			provider.CBTParam{
 				Source: cbtservice.SourceInfo{
 					Snapshot: restoreParam.VolumeSnapshotName,
@@ -275,7 +276,11 @@ 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, IncrementalBytes: incrementalBytes}})
+			dp.callbacks.OnCompleted(context.Background(), dp.namespace, dp.jobName, Result{Restore: RestoreResult{Target: target,
+				TotalBytes:       totalBytes,
+				IncrementalBytes: incrementalBytes,
+				FallbackFull:     fallback,
+			}})
 		}
 	}()
 

pkg/datapath/data_path_test.go

@@ -89,6 +89,7 @@ func TestAsyncBackup(t *testing.T) {
 					TotalBytes:       3000,
 					IncrementalBytes: ptr.To(int64(200)),
 					SourceSize:       2000,
+					FallbackFull:     true,
 				},
 			},
 			path: "fake-path",
@@ -112,6 +113,7 @@ func TestAsyncBackup(t *testing.T) {
 					TotalBytes:       2000,
 					IncrementalBytes: ptr.To(int64(200)),
 					SourceSize:       2000,
+					FallbackFull:     false,
 				},
 			},
 			path: "fake-path",
@@ -126,7 +128,7 @@ func TestAsyncBackup(t *testing.T) {
 			if test.result.Backup.IncrementalBytes != nil {
 				incrementalBytes = *test.result.Backup.IncrementalBytes
 			}
-			mockProvider.On("RunBackup", mock.Anything, 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, incrementalBytes, test.result.Backup.SourceSize, test.err)
+			mockProvider.On("RunBackup", mock.Anything, 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, incrementalBytes, test.result.Backup.SourceSize, test.result.Backup.FallbackFull, test.err)
 			mockProvider.On("Close", mock.Anything).Return(nil)
 			dp.uploaderProv = mockProvider
 			dp.initialized = true
@@ -201,8 +203,10 @@ func TestAsyncRestore(t *testing.T) {
 			},
 			result: Result{
 				Restore: RestoreResult{
-					Target:     AccessPoint{ByPath: "fake-path"},
-					TotalBytes: 1000,
+					Target:           AccessPoint{ByPath: "fake-path"},
+					TotalBytes:       1000,
+					IncrementalBytes: 500,
+					FallbackFull:     true,
 				},
 			},
 			path:     "fake-path",
@@ -214,7 +218,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.IncrementalBytes, 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.result.Restore.FallbackFull, test.err)
 			mockProvider.On("Close", mock.Anything).Return(nil)
 			dp.uploaderProv = mockProvider
 			dp.initialized = true

pkg/datapath/types.go

@@ -40,6 +40,7 @@ type BackupResult struct {
 	// unmarshals to a non-nil zero, distinguishing "measured zero" from "not measured".
 	IncrementalBytes *int64 `json:"incrementalBytes,omitempty"`
 	SourceSize       int64  `json:"sourceSize,omitempty"`
+	FallbackFull     bool   `json:"fallbackFull,omitempty"`
 }
 
 // RestoreResult represents the result of a restore
@@ -47,6 +48,7 @@ type RestoreResult struct {
 	Target           AccessPoint `json:"target,omitempty"`
 	TotalBytes       int64       `json:"totalBytes,omitempty"`
 	IncrementalBytes int64       `json:"incrementalBytes,omitempty"`
+	FallbackFull     bool        `json:"fallbackFull,omitempty"`
 }
 
 // Callbacks defines the collection of callbacks during backup/restore

pkg/podvolume/backup_micro_service_test.go

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

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 0}: fake-marshal-error",
+			expectedErr: "error marshaling restore result {{ } 0 0 false}: fake-marshal-error",
 		},
 		{
 			name:                "succeed",

pkg/uploader/block/snapshot.go

@@ -83,12 +83,13 @@ func Backup(ctx context.Context, blkUp Uploader, repoWriter udmrepo.BackupRepo,
 		return uploader.SnapshotInfo{}, false, errors.Wrapf(err, "error reset pos of block device %s", source)
 	}
 
-	snapID, backupSize, snapshotSize, err := snapshotSource(ctx, repoWriter, blkUp, sourceInfo, forceFull, parentSnapshot, cbtSource, cbtService, tags, uploaderCfg, log, "Block Uploader")
+	snapID, backupSize, snapshotSize, fallback, err := snapshotSource(ctx, repoWriter, blkUp, sourceInfo, forceFull, parentSnapshot, cbtSource, cbtService, tags, uploaderCfg, log, "Block Uploader")
 	snapshotInfo := uploader.SnapshotInfo{
 		ID:              snapID,
 		SnapshotSize:    snapshotSize,
 		IncrementalSize: backupSize,
 		SourceSize:      sourceInfo.size,
+		Fallback:        fallback,
 	}
 
 	return snapshotInfo, false, err
@@ -107,7 +108,7 @@ func snapshotSource(
 	uploaderCfg map[string]string,
 	log logrus.FieldLogger,
 	description string,
-) (string, int64, int64, error) {
+) (string, int64, int64, bool, error) {
 	log.Info("Start to snapshot...")
 	snapshotStartTime := time.Now()
 
@@ -127,9 +128,14 @@ func snapshotSource(
 		log.WithError(err).Warnf("Failed to create CBT with source %v", cbtSource)
 	}
 
+	fallback := false
+	if !forceFull {
+		fallback = (len(bitmap.Errors()) > 0)
+	}
+
 	snap, backupSize, err := u.Backup(source, parentBackup.parentObject, bitmap.Iterator(), uploaderCfg)
 	if err != nil {
-		return "", 0, 0, errors.Wrapf(err, "Failed to run uploader backup for si %v", source)
+		return "", 0, 0, fallback, errors.Wrapf(err, "Failed to run uploader backup for si %v", source)
 	}
 
 	if snap.Tags == nil {
@@ -146,16 +152,16 @@ func snapshotSource(
 
 	snapID, err := rep.SaveSnapshot(ctx, snap)
 	if err != nil {
-		return "", 0, 0, errors.Wrapf(err, "Failed to save snapshot %v", snap)
+		return "", 0, 0, fallback, errors.Wrapf(err, "Failed to save snapshot %v", snap)
 	}
 
 	if err = rep.Flush(ctx); err != nil {
-		return "", 0, 0, errors.Wrapf(err, "Failed to flush repository")
+		return "", 0, 0, fallback, errors.Wrapf(err, "Failed to flush repository")
 	}
 
 	log.Infof("Created snapshot with root %v and ID %v in %v", snap.RootObject, snapID, time.Since(snapshotStartTime).Truncate(time.Second))
 
-	return string(snapID), backupSize, snap.TotalSize, nil
+	return string(snapID), backupSize, snap.TotalSize, fallback, nil
 }
 
 func getParentBackupInfo(ctx context.Context, rep udmrepo.BackupRepo, forceFull bool, parentSnapshot string, volumeID string,
@@ -221,12 +227,12 @@ func getParentBackupInfo(ctx context.Context, rep udmrepo.BackupRepo, forceFull
 }
 
 // Restore restore specific sourcePath with given snapshotID and update progress
-func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapshotID, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, int64, error) {
+func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapshotID, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, int64, bool, error) {
 	log.Info("Start to restore...")
 
 	snapshot, err := rep.GetSnapshot(ctx, udmrepo.ID(snapshotID))
 	if err != nil {
-		return 0, 0, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID)
+		return 0, 0, false, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID)
 	}
 	log.Infof("Restore from snapshot %s, incremental %v, cbt source %v, description %s, created time %v, tags %v", snapshotID, incremental, cbtSource, snapshot.Description, snapshot.EndTime, snapshot.Tags)
 
@@ -248,36 +254,41 @@ func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapsh
 		bitmap.SetFull()
 	}
 
+	fallback := false
+	if incremental {
+		fallback = (len(bitmap.Errors()) > 0)
+	}
+
 	destPath, err := filepath.Abs(dest)
 	if err != nil {
-		return 0, 0, errors.Wrapf(err, "invalid dest path '%s'", dest)
+		return 0, 0, fallback, errors.Wrapf(err, "invalid dest path '%s'", dest)
 	}
 
 	destPath = filepath.Clean(destPath)
 
 	destDev, err := openBlockDeviceFunc(destPath, false)
 	if err != nil {
-		return 0, 0, errors.Wrapf(err, "error opening block device '%s'", destPath)
+		return 0, 0, fallback, errors.Wrapf(err, "error opening block device '%s'", destPath)
 	}
 
 	defer destDev.Close()
 
 	destSize, err := destDev.Seek(0, io.SeekEnd)
 	if err != nil {
-		return 0, 0, errors.Wrapf(err, "error getting length of block device %s", dest)
+		return 0, 0, fallback, errors.Wrapf(err, "error getting length of block device %s", dest)
 	}
 
 	_, err = destDev.Seek(0, io.SeekStart)
 	if err != nil {
-		return 0, 0, errors.Wrapf(err, "error reset pos of block device %s", dest)
+		return 0, 0, fallback, errors.Wrapf(err, "error reset pos of block device %s", dest)
 	}
 
 	incrementalBytes, totalSize, err := blkUp.Restore(snapshot, destInfo{dev: destDev, path: destPath, size: destSize}, bitmap.Iterator(), uploaderCfg)
 	if err != nil {
-		return 0, 0, errors.Wrapf(err, "error restoring to block dev %s", destPath)
+		return 0, 0, fallback, errors.Wrapf(err, "error restoring to block dev %s", destPath)
 	}
 
-	return incrementalBytes, totalSize, nil
+	return incrementalBytes, totalSize, fallback, nil
 }
 
 func getBackupInfo(snapshot udmrepo.Snapshot, volumeID string) (backupInfo, error) {

pkg/uploader/block/snapshot_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
+	"k8s.io/utils/ptr"
 
 	"github.com/vmware-tanzu/velero/pkg/cbtservice"
 	cbtservicemocks "github.com/vmware-tanzu/velero/pkg/cbtservice/mocks"
@@ -126,6 +127,7 @@ func TestBackup(t *testing.T) {
 				assert.Equal(t, int64(8), info.IncrementalSize)
 				assert.Equal(t, int64(2048), info.SnapshotSize)
 				assert.Equal(t, int64(len("test-block-data")), info.SourceSize)
+				assert.False(t, info.Fallback)
 			},
 		},
 		{
@@ -149,6 +151,7 @@ func TestBackup(t *testing.T) {
 				assert.Equal(t, int64(8), info.IncrementalSize)
 				assert.Equal(t, int64(len("test-block-data")), info.SnapshotSize)
 				assert.Equal(t, int64(len("test-block-data")), info.SourceSize)
+				assert.False(t, info.Fallback)
 			},
 		},
 	}
@@ -210,11 +213,14 @@ func TestSnapshotSource(t *testing.T) {
 
 	testCases := []struct {
 		name                 string
+		forceFull            *bool
+		parentSnapshot       string
 		setupMocks           func(blkup *mockUploader, repo *udmrepomocks.BackupRepo)
 		expectedErrStr       string
 		expectedSnapID       string
 		expectedSize         int64
 		expectedSnapshotSize int64
+		expectedFallback     bool
 		cbtService           func(t *testing.T) cbtservice.Service
 	}{
 		{
@@ -259,6 +265,7 @@ func TestSnapshotSource(t *testing.T) {
 			expectedSnapID:       "snap-success",
 			expectedSize:         512,
 			expectedSnapshotSize: 2048,
+			expectedFallback:     false,
 		},
 		{
 			name: "tags from cbtSource and snapshotTags are merged onto snapshot",
@@ -276,6 +283,7 @@ func TestSnapshotSource(t *testing.T) {
 			expectedSnapID:       "snap-tags",
 			expectedSize:         256,
 			expectedSnapshotSize: 4096,
+			expectedFallback:     false,
 		},
 		{
 			name: "success with cbtService getting allocated blocks",
@@ -298,6 +306,7 @@ func TestSnapshotSource(t *testing.T) {
 			expectedSnapID:       "snap-cbt-alloc",
 			expectedSize:         1024,
 			expectedSnapshotSize: 8192,
+			expectedFallback:     false,
 		},
 		{
 			name: "cbtService error falls back to full",
@@ -318,6 +327,24 @@ func TestSnapshotSource(t *testing.T) {
 			expectedSnapID:       "snap-cbt-fallback",
 			expectedSize:         1024,
 			expectedSnapshotSize: 1024,
+			expectedFallback:     false,
+		},
+		{
+			name:           "incremental backup falls back to full when parent snapshot load fails",
+			forceFull:      ptr.To(false),
+			parentSnapshot: "snap-parent",
+			setupMocks: func(blkup *mockUploader, repo *udmrepomocks.BackupRepo) {
+				repo.On("GetSnapshot", mock.Anything, udmrepo.ID("snap-parent")).
+					Return(udmrepo.Snapshot{}, errors.New("parent not found"))
+				blkup.On("Backup", mock.Anything, udmrepo.ID(""), mock.Anything, mock.Anything).
+					Return(udmrepo.Snapshot{TotalSize: 1024}, int64(1024), nil)
+				repo.On("SaveSnapshot", mock.Anything, mock.Anything).Return(udmrepo.ID("snap-inc-fallback"), nil)
+				repo.On("Flush", mock.Anything).Return(nil)
+			},
+			expectedSnapID:       "snap-inc-fallback",
+			expectedSize:         1024,
+			expectedSnapshotSize: 1024,
+			expectedFallback:     true,
 		},
 	}
 
@@ -337,10 +364,15 @@ func TestSnapshotSource(t *testing.T) {
 				cbtSvc = tc.cbtService(t)
 			}
 
-			snapID, size, snapshotSize, err := snapshotSource(
+			forceFull := true
+			if tc.forceFull != nil {
+				forceFull = *tc.forceFull
+			}
+
+			snapID, size, snapshotSize, fallback, err := snapshotSource(
 				ctx, mockRepo, mockBlkup,
 				baseSource,
-				true, "",
+				forceFull, tc.parentSnapshot,
 				cbtSrc, cbtSvc,
 				snapshotTags, map[string]string{},
 				testLog(), "Block Uploader",
@@ -354,6 +386,7 @@ func TestSnapshotSource(t *testing.T) {
 				assert.Equal(t, tc.expectedSnapID, snapID)
 				assert.Equal(t, tc.expectedSize, size)
 				assert.Equal(t, tc.expectedSnapshotSize, snapshotSize)
+				assert.Equal(t, tc.expectedFallback, fallback)
 			}
 
 			mockBlkup.AssertExpectations(t)
@@ -834,14 +867,15 @@ func TestRestore(t *testing.T) {
 	storedSnap := udmrepo.Snapshot{Description: "test snapshot"}
 
 	testCases := []struct {
-		name           string
-		incremental    bool
-		cbtSource      cbtservice.SourceInfo
-		cbtService     func(t *testing.T) cbtservice.Service
-		setupMocks     func(blkup *mockUploader, repo *udmrepomocks.BackupRepo)
-		setupOpenDev   func(t *testing.T) *os.File
-		expectedErrStr string
-		expectedSize   int64
+		name             string
+		incremental      bool
+		cbtSource        cbtservice.SourceInfo
+		cbtService       func(t *testing.T) cbtservice.Service
+		setupMocks       func(blkup *mockUploader, repo *udmrepomocks.BackupRepo)
+		setupOpenDev     func(t *testing.T) *os.File
+		expectedErrStr   string
+		expectedSize     int64
+		expectedFallback bool
 	}{
 		{
 			name: "GetSnapshot error",
@@ -885,7 +919,8 @@ func TestRestore(t *testing.T) {
 				t.Helper()
 				return tempFile(t, "")
 			},
-			expectedSize: 4096,
+			expectedSize:     4096,
+			expectedFallback: false,
 		},
 		{
 			name:        "incremental restore success",
@@ -917,7 +952,8 @@ func TestRestore(t *testing.T) {
 				t.Helper()
 				return tempFile(t, "")
 			},
-			expectedSize: 512,
+			expectedSize:     512,
+			expectedFallback: false,
 		},
 		{
 			name:        "incremental restore fallback - missing tags",
@@ -931,7 +967,8 @@ func TestRestore(t *testing.T) {
 				t.Helper()
 				return tempFile(t, "")
 			},
-			expectedSize: 4096,
+			expectedSize:     4096,
+			expectedFallback: true,
 		},
 		{
 			name:        "incremental restore fallback - empty cbtSource VolumeID",
@@ -952,7 +989,8 @@ func TestRestore(t *testing.T) {
 				t.Helper()
 				return tempFile(t, "")
 			},
-			expectedSize: 4096,
+			expectedSize:     4096,
+			expectedFallback: true,
 		},
 		{
 			name:        "incremental restore fallback - VolumeID mismatch",
@@ -973,7 +1011,8 @@ func TestRestore(t *testing.T) {
 				t.Helper()
 				return tempFile(t, "")
 			},
-			expectedSize: 4096,
+			expectedSize:     4096,
+			expectedFallback: true,
 		},
 		{
 			name:        "incremental restore fallback - CBT service error",
@@ -1002,7 +1041,8 @@ func TestRestore(t *testing.T) {
 				t.Helper()
 				return tempFile(t, "")
 			},
-			expectedSize: 1024,
+			expectedSize:     1024,
+			expectedFallback: true,
 		},
 	}
 
@@ -1030,7 +1070,7 @@ func TestRestore(t *testing.T) {
 				cbtSvc = tc.cbtService(t)
 			}
 
-			_, size, err := Restore(ctx, mockBlkup, mockRepo, "snap-001", "/dev/sdb", tc.incremental, tc.cbtSource, cbtSvc, map[string]string{}, testLog())
+			_, size, fallback, err := Restore(ctx, mockBlkup, mockRepo, "snap-001", "/dev/sdb", tc.incremental, tc.cbtSource, cbtSvc, map[string]string{}, testLog())
 
 			if tc.expectedErrStr != "" {
 				require.Error(t, err)
@@ -1039,6 +1079,7 @@ func TestRestore(t *testing.T) {
 			} else {
 				require.NoError(t, err)
 				assert.Equal(t, tc.expectedSize, size)
+				assert.Equal(t, tc.expectedFallback, fallback)
 			}
 
 			mockBlkup.AssertExpectations(t)

pkg/uploader/kopia/snapshot.go

@@ -190,11 +190,12 @@ func Backup(ctx context.Context, fsUploader SnapshotUploader, repoWriter repo.Re
 
 	kopiaCtx := kopia.SetupKopiaLog(ctx, log)
 
-	snapID, snapshotSize, err := SnapshotSource(kopiaCtx, repoWriter, fsUploader, sourceInfo, sourceEntry, forceFull, parentSnapshot, tags, uploaderCfg, updater, log, "Kopia Uploader")
+	snapID, snapshotSize, fallback, err := SnapshotSource(kopiaCtx, repoWriter, fsUploader, sourceInfo, sourceEntry, forceFull, parentSnapshot, tags, uploaderCfg, updater, log, "Kopia Uploader")
 	snapshotInfo := &uploader.SnapshotInfo{
 		ID:           snapID,
 		SnapshotSize: snapshotSize,
 		SourceSize:   snapshotSize,
+		Fallback:     fallback,
 	}
 
 	return snapshotInfo, false, err
@@ -242,11 +243,12 @@ func SnapshotSource(
 	updater uploader.ProgressUpdater,
 	log logrus.FieldLogger,
 	description string,
-) (string, int64, error) {
+) (string, int64, bool, error) {
 	log.Info("Start to snapshot...")
 	snapshotStartTime := time.Now()
 
 	var previous []*snapshot.Manifest
+	fallback := false
 	if !forceFull {
 		if parentSnapshot != "" {
 			log.Infof("Using provided parent snapshot %s", parentSnapshot)
@@ -258,6 +260,8 @@ func SnapshotSource(
 					TotalBytes: -1,
 					Message:    fmt.Sprintf("Failed to load previous snapshot %v, fallback to full backup. Err: %v", parentSnapshot, err),
 				})
+
+				fallback = true
 			} else {
 				previous = append(previous, mani)
 			}
@@ -271,6 +275,8 @@ func SnapshotSource(
 					TotalBytes: -1,
 					Message:    fmt.Sprintf("Failed to find previous snapshots, fallback to full backup. Err: %v", err),
 				})
+
+				fallback = true
 			} else {
 				previous = pre
 			}
@@ -285,12 +291,12 @@ func SnapshotSource(
 
 	policyTree, err := setupPolicy(ctx, rep, sourceInfo, uploaderCfg)
 	if err != nil {
-		return "", 0, errors.Wrapf(err, "unable to set policy for si %v", sourceInfo)
+		return "", 0, fallback, errors.Wrapf(err, "unable to set policy for si %v", sourceInfo)
 	}
 
 	manifest, err := u.Upload(ctx, rootDir, policyTree, sourceInfo, previous...)
 	if err != nil {
-		return "", 0, errors.Wrapf(err, "Failed to upload the kopia snapshot for si %v", sourceInfo)
+		return "", 0, fallback, errors.Wrapf(err, "Failed to upload the kopia snapshot for si %v", sourceInfo)
 	}
 
 	manifest.Tags = snapshotTags
@@ -299,22 +305,22 @@ func SnapshotSource(
 	manifest.Pins = []string{"velero-pin"}
 
 	if _, err = saveSnapshotFunc(ctx, rep, manifest); err != nil {
-		return "", 0, errors.Wrapf(err, "Failed to save kopia manifest %v", manifest.ID)
+		return "", 0, fallback, errors.Wrapf(err, "Failed to save kopia manifest %v", manifest.ID)
 	}
 
 	_, err = applyRetentionPolicyFunc(ctx, rep, sourceInfo, true)
 	if err != nil {
-		return "", 0, errors.Wrapf(err, "Failed to apply kopia retention policy for si %v", sourceInfo)
+		return "", 0, fallback, errors.Wrapf(err, "Failed to apply kopia retention policy for si %v", sourceInfo)
 	}
 
 	if err = rep.Flush(ctx); err != nil {
-		return "", 0, errors.Wrapf(err, "Failed to flush kopia repository")
+		return "", 0, fallback, errors.Wrapf(err, "Failed to flush kopia repository")
 	}
 	log.Infof("Created snapshot with root %v and ID %v in %v", manifest.RootObjectID(), manifest.ID, time.Since(snapshotStartTime).Truncate(time.Second))
-	return reportSnapshotStatus(manifest, policyTree)
+	return reportSnapshotStatus(manifest, policyTree, fallback)
 }
 
-func reportSnapshotStatus(manifest *snapshot.Manifest, policyTree *policy.Tree) (string, int64, error) {
+func reportSnapshotStatus(manifest *snapshot.Manifest, policyTree *policy.Tree, fallback bool) (string, int64, bool, error) {
 	manifestID := manifest.ID
 	snapSize := manifest.Stats.TotalFileSize
 
@@ -333,10 +339,10 @@ func reportSnapshotStatus(manifest *snapshot.Manifest, policyTree *policy.Tree)
 	}
 
 	if len(errs) != 0 {
-		return string(manifestID), snapSize, errors.New(strings.Join(errs, "\n"))
+		return string(manifestID), snapSize, fallback, errors.New(strings.Join(errs, "\n"))
 	}
 
-	return string(manifestID), snapSize, nil
+	return string(manifestID), snapSize, fallback, nil
 }
 
 // findPreviousSnapshotManifest returns the list of previous snapshots for a given source, including
@@ -401,26 +407,26 @@ func (o *fileSystemRestoreOutput) Terminate() error {
 
 // Restore restore specific sourcePath with given snapshotID and update progress
 func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string,
-	log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
+	log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, bool, error) {
 	log.Info("Start to restore...")
 
 	kopiaCtx := kopia.SetupKopiaLog(ctx, log)
 
 	snapshot, err := snapshot.LoadSnapshot(kopiaCtx, rep, manifest.ID(snapshotID))
 	if err != nil {
-		return 0, 0, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID)
+		return 0, 0, false, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID)
 	}
 
 	log.Infof("Restore from snapshot %s, description %s, created time %v, tags %v", snapshotID, snapshot.Description, snapshot.EndTime.ToTime(), snapshot.Tags)
 
 	rootEntry, err := filesystemEntryFunc(kopiaCtx, rep, snapshotID, false)
 	if err != nil {
-		return 0, 0, errors.Wrapf(err, "Unable to get filesystem entry for snapshot %v", snapshotID)
+		return 0, 0, false, errors.Wrapf(err, "Unable to get filesystem entry for snapshot %v", snapshotID)
 	}
 
 	path, err := filepath.Abs(dest)
 	if err != nil {
-		return 0, 0, errors.Wrapf(err, "Unable to resolve path %v", dest)
+		return 0, 0, false, errors.Wrapf(err, "Unable to resolve path %v", dest)
 	}
 
 	fsOutput := &restore.FilesystemOutput{
@@ -436,7 +442,7 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 	if len(uploaderCfg) > 0 {
 		writeSparseFiles, err := uploaderutil.GetWriteSparseFiles(uploaderCfg)
 		if err != nil {
-			return 0, 0, errors.Wrap(err, "failed to get uploader config")
+			return 0, 0, false, errors.Wrap(err, "failed to get uploader config")
 		}
 		if writeSparseFiles {
 			fsOutput.WriteSparseFiles = true
@@ -444,7 +450,7 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 
 		concurrency, err := uploaderutil.GetRestoreConcurrency(uploaderCfg)
 		if err != nil {
-			return 0, 0, errors.Wrap(err, "failed to get parallel restore uploader config")
+			return 0, 0, false, errors.Wrap(err, "failed to get parallel restore uploader config")
 		}
 		if concurrency > 0 {
 			restoreConcurrency = concurrency
@@ -452,7 +458,7 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 
 		deleteExtra, err = uploaderutil.GetDeleteExtraFiles(uploaderCfg)
 		if err != nil {
-			return 0, 0, errors.Wrap(err, "failed to get delete extra files config")
+			return 0, 0, false, errors.Wrap(err, "failed to get delete extra files config")
 		}
 	}
 
@@ -460,7 +466,7 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 
 	err = fsOutput.Init(ctx)
 	if err != nil {
-		return 0, 0, errors.Wrap(err, "error to init output")
+		return 0, 0, false, errors.Wrap(err, "error to init output")
 	}
 
 	var output RestoreOutput
@@ -500,18 +506,18 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 	})
 
 	if err != nil {
-		return 0, 0, errors.Wrapf(err, "Failed to copy snapshot data to the target")
+		return 0, 0, false, errors.Wrapf(err, "Failed to copy snapshot data to the target")
 	}
 
 	if err := output.Flush(); err != nil {
 		if err == errFlushUnsupported {
 			log.Warnf("Skip flushing data for %v under the current OS %v", path, runtime.GOOS)
 		} else {
-			return 0, 0, errors.Wrapf(err, "Failed to flush data to target")
+			return 0, 0, false, errors.Wrapf(err, "Failed to flush data to target")
 		}
 	} else {
 		log.Infof("Flush done for volume dir %v", path)
 	}
 
-	return stat.RestoredTotalFileSize, stat.RestoredFileCount, nil
+	return stat.RestoredTotalFileSize, stat.RestoredFileCount, false, nil
 }

pkg/uploader/kopia/snapshot_test.go

@@ -200,7 +200,7 @@ func TestSnapshotSource(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			s := injectSnapshotFuncs()
 			MockFuncs(s, tc.args)
-			_, _, err = SnapshotSource(ctx, s.repoWriterMock, s.uploderMock, sourceInfo, rootDir, false, "/", nil, tc.uploaderCfg, &fakeProgressUpdater{}, log, "TestSnapshotSource")
+			_, _, _, err = SnapshotSource(ctx, s.repoWriterMock, s.uploderMock, sourceInfo, rootDir, false, "/", nil, tc.uploaderCfg, &fakeProgressUpdater{}, log, "TestSnapshotSource")
 			if tc.notError {
 				assert.NoError(t, err)
 			} else {
@@ -212,6 +212,7 @@ func TestSnapshotSource(t *testing.T) {
 
 func TestReportSnapshotStatus(t *testing.T) {
 	testCases := []struct {
+		fallback         bool
 		shouldError      bool
 		expectedResult   string
 		expectedSize     int64
@@ -219,6 +220,7 @@ func TestReportSnapshotStatus(t *testing.T) {
 		expectedErrors   []string
 	}{
 		{
+			fallback:       false,
 			shouldError:    false,
 			expectedResult: "sample-manifest-id",
 			expectedSize:   1024,
@@ -227,6 +229,7 @@ func TestReportSnapshotStatus(t *testing.T) {
 			},
 		},
 		{
+			fallback:       true,
 			shouldError:    true,
 			expectedResult: "sample-manifest-id",
 			expectedSize:   1024,
@@ -253,7 +256,7 @@ func TestReportSnapshotStatus(t *testing.T) {
 			},
 		}
 
-		result, size, err := reportSnapshotStatus(manifest, policy.BuildTree(nil, getDefaultPolicy()))
+		result, size, fallback, err := reportSnapshotStatus(manifest, policy.BuildTree(nil, getDefaultPolicy()), tc.fallback)
 
 		switch {
 		case tc.shouldError && err == nil:
@@ -274,6 +277,10 @@ func TestReportSnapshotStatus(t *testing.T) {
 		if size != tc.expectedSize {
 			t.Errorf("unexpected size: got %v, want %v", size, tc.expectedSize)
 		}
+
+		if fallback != tc.fallback {
+			t.Errorf("unexpected fallback: got %v, want %v", fallback, tc.fallback)
+		}
 	}
 }
 
@@ -819,7 +826,7 @@ func TestRestore(t *testing.T) {
 			repoWriterMock.On("OpenObject", mock.Anything, mock.Anything).Return(em, nil)
 
 			progress := new(Progress)
-			bytesRestored, fileCount, err := Restore(t.Context(), repoWriterMock, progress, tc.snapshotID, tc.dest, tc.incremental, tc.volMode, map[string]string{}, logrus.New(), nil)
+			bytesRestored, fileCount, fallback, err := Restore(t.Context(), repoWriterMock, progress, tc.snapshotID, tc.dest, tc.incremental, tc.volMode, map[string]string{}, logrus.New(), nil)
 
 			// Check if the returned error matches the expected error
 			if tc.expectedError != nil {
@@ -833,6 +840,7 @@ func TestRestore(t *testing.T) {
 
 			// Check the number of files restored
 			assert.Equal(t, tc.expectedCount, fileCount)
+			assert.False(t, fallback)
 		})
 	}
 }

pkg/uploader/provider/block.go

@@ -103,13 +103,13 @@ func (bp *blockProvider) RunBackup(
 	cbtParam CBTParam,
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
-	updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, error) {
+	updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, bool, error) {
 	if updater == nil {
-		return "", false, 0, 0, 0, errors.New("backup progress updater is invalid")
+		return "", false, 0, 0, 0, false, errors.New("backup progress updater is invalid")
 	}
 
 	if path == "" {
-		return "", false, 0, 0, 0, errors.New("path is empty")
+		return "", false, 0, 0, 0, false, errors.New("path is empty")
 	}
 
 	log := bp.log.WithFields(logrus.Fields{
@@ -140,11 +140,11 @@ func (bp *blockProvider) RunBackup(
 	// equality check never matches and cancellation gets reported as a failure.
 	if errors.Is(err, block.ErrCanceled) {
 		log.Warn("Block backup is canceled")
-		return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, ErrorCanceled
+		return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, snapshotInfo.Fallback, ErrorCanceled
 	}
 
 	if err != nil {
-		return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, errors.Wrapf(err, "Failed to run block backup")
+		return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, snapshotInfo.Fallback, errors.Wrapf(err, "Failed to run block backup")
 	}
 
 	updater.UpdateProgress(
@@ -154,9 +154,9 @@ func (bp *blockProvider) RunBackup(
 		},
 	)
 
-	log.Infof("Block backup finished, snapshot ID %s, backup size %v, incremental size %v, source size %v", snapshotInfo.ID, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize)
+	log.Infof("Block backup finished, snapshot ID %s, backup size %v, incremental size %v, source size %v, fallback %v", snapshotInfo.ID, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, snapshotInfo.Fallback)
 
-	return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, nil
+	return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, snapshotInfo.IncrementalSize, snapshotInfo.SourceSize, snapshotInfo.Fallback, nil
 }
 
 func (bp *blockProvider) RunRestore(
@@ -167,9 +167,9 @@ func (bp *blockProvider) RunRestore(
 	cbtParam CBTParam,
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
-	updater uploader.ProgressUpdater) (int64, int64, error) {
+	updater uploader.ProgressUpdater) (int64, int64, bool, error) {
 	if updater == nil {
-		return 0, 0, errors.New("restore progress updater is invalid")
+		return 0, 0, false, errors.New("restore progress updater is invalid")
 	}
 
 	log := bp.log.WithFields(logrus.Fields{
@@ -180,16 +180,16 @@ func (bp *blockProvider) RunRestore(
 
 	blkUploader := block.NewUploader(ctx, bp.bkRepo, updater, log)
 
-	incrementalBytes, totalBytes, err := blockRestoreFunc(ctx, blkUploader, bp.bkRepo, snapshotID, volumePath, incremental, cbtParam.Source, cbtParam.Service, uploaderCfg, log)
+	incrementalBytes, totalBytes, fallback, err := blockRestoreFunc(ctx, blkUploader, bp.bkRepo, snapshotID, volumePath, incremental, cbtParam.Source, cbtParam.Service, uploaderCfg, log)
 
 	// errors.Is, not ==: see the equivalent comment on the backup path above.
 	if errors.Is(err, block.ErrCanceled) {
 		log.Warn("Block restore is canceled")
-		return 0, 0, ErrorCanceled
+		return 0, 0, fallback, ErrorCanceled
 	}
 
 	if err != nil {
-		return 0, 0, errors.Wrapf(err, "Failed to run block restore")
+		return 0, 0, fallback, errors.Wrapf(err, "Failed to run block restore")
 	}
 
 	updater.UpdateProgress(&uploader.Progress{
@@ -197,7 +197,7 @@ func (bp *blockProvider) RunRestore(
 		BytesDone:  totalBytes,
 	})
 
-	log.Infof("Block restore finished, restore incremental size %v, total size %v", incrementalBytes, totalBytes)
+	log.Infof("Block restore finished, restore incremental size %v, total size %v, fallback %v", incrementalBytes, totalBytes, fallback)
 
-	return incrementalBytes, totalBytes, nil
+	return incrementalBytes, totalBytes, fallback, nil
 }

pkg/uploader/provider/block_test.go

@@ -337,7 +337,7 @@ func TestBlockProviderRunBackup(t *testing.T) {
 				log:           logrus.New(),
 			}
 
-			snapshotID, isEmpty, size, incrSize, sourceSize, err := bp.RunBackup(
+			snapshotID, isEmpty, size, incrSize, sourceSize, fallback, err := bp.RunBackup(
 				t.Context(),
 				tc.path,
 				tc.realSource,
@@ -354,6 +354,7 @@ func TestBlockProviderRunBackup(t *testing.T) {
 			assert.Equal(t, tc.expectedSize, size)
 			assert.Equal(t, tc.expectedIncrSize, incrSize)
 			assert.Equal(t, tc.expectedSourceSize, sourceSize)
+			assert.Equal(t, tc.mockBackupResult.Fallback, fallback)
 
 			if tc.expectError {
 				require.Error(t, err)
@@ -404,7 +405,7 @@ func TestBlockProviderCancelThroughWrappedError(t *testing.T) {
 			log:           logrus.New(),
 		}
 
-		_, _, _, _, _, err := bp.RunBackup(
+		_, _, _, _, _, _, err := bp.RunBackup(
 			t.Context(), "/dev/sda", "ns/pvc", map[string]string{}, false, "",
 			CBTParam{}, uploader.PersistentVolumeBlock, map[string]string{},
 			&FakeBackupProgressUpdater{},
@@ -418,8 +419,8 @@ func TestBlockProviderCancelThroughWrappedError(t *testing.T) {
 	t.Run("restore", func(t *testing.T) {
 		orig := blockRestoreFunc
 		defer func() { blockRestoreFunc = orig }()
-		blockRestoreFunc = func(_ context.Context, _ block.Uploader, _ udmrepo.BackupRepo, _ string, _ string, _ bool, _ cbtservice.SourceInfo, _ cbtservice.Service, _ map[string]string, _ logrus.FieldLogger) (int64, int64, error) {
-			return 0, 0, errors.Wrap(block.ErrCanceled, "error restoring bdev")
+		blockRestoreFunc = func(_ context.Context, _ block.Uploader, _ udmrepo.BackupRepo, _ string, _ string, _ bool, _ cbtservice.SourceInfo, _ cbtservice.Service, _ map[string]string, _ logrus.FieldLogger) (int64, int64, bool, error) {
+			return 0, 0, false, errors.Wrap(block.ErrCanceled, "error restoring bdev")
 		}
 
 		bp := &blockProvider{
@@ -428,7 +429,7 @@ func TestBlockProviderCancelThroughWrappedError(t *testing.T) {
 			log:           logrus.New(),
 		}
 
-		_, _, err := bp.RunRestore(t.Context(), "snap-1", "/dev/sda", false, CBTParam{},
+		_, _, _, err := bp.RunRestore(t.Context(), "snap-1", "/dev/sda", false, CBTParam{},
 			uploader.PersistentVolumeBlock, map[string]string{}, &blockMockProgressUpdater{})
 
 		require.ErrorIs(t, err, ErrorCanceled)
@@ -502,10 +503,10 @@ func TestBlockProviderRunRestore(t *testing.T) {
 			var capturedSnapshotID string
 			var capturedVolumePath string
 
-			blockRestoreFunc = func(ctx context.Context, blkUp block.Uploader, rep udmrepo.BackupRepo, snapshotID string, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, int64, error) {
+			blockRestoreFunc = func(ctx context.Context, blkUp block.Uploader, rep udmrepo.BackupRepo, snapshotID string, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, int64, bool, error) {
 				capturedSnapshotID = snapshotID
 				capturedVolumePath = dest
-				return tc.mockRestoreSize, tc.mockRestoreSize, tc.mockRestoreErr
+				return tc.mockRestoreSize, tc.mockRestoreSize, false, tc.mockRestoreErr
 			}
 
 			bp := &blockProvider{
@@ -513,7 +514,7 @@ func TestBlockProviderRunRestore(t *testing.T) {
 				log:    logrus.New(),
 			}
 
-			_, size, err := bp.RunRestore(
+			_, size, _, err := bp.RunRestore(
 				t.Context(),
 				tc.snapshotID,
 				tc.volumePath,

pkg/uploader/provider/kopia.go

@@ -121,13 +121,13 @@ func (kp *kopiaProvider) RunBackup(
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
 	updater uploader.ProgressUpdater,
-) (string, bool, int64, int64, int64, error) {
+) (string, bool, int64, int64, int64, bool, error) {
 	if updater == nil {
-		return "", false, 0, 0, 0, errors.New("Need to initial backup progress updater first")
+		return "", false, 0, 0, 0, false, errors.New("Need to initial backup progress updater first")
 	}
 
 	if path == "" {
-		return "", false, 0, 0, 0, errors.New("path is empty")
+		return "", false, 0, 0, 0, false, errors.New("path is empty")
 	}
 
 	log := kp.log.WithFields(logrus.Fields{
@@ -177,9 +177,9 @@ func (kp *kopiaProvider) RunBackup(
 
 		if kpUploader.IsCanceled() {
 			log.Warn("Kopia backup is canceled")
-			return snapshotID, false, 0, 0, 0, ErrorCanceled
+			return snapshotID, false, 0, 0, 0, snapshotInfo.Fallback, ErrorCanceled
 		}
-		return snapshotID, false, 0, 0, 0, errors.Wrapf(err, "Failed to run kopia backup")
+		return snapshotID, false, 0, 0, 0, snapshotInfo.Fallback, errors.Wrapf(err, "Failed to run kopia backup")
 	}
 
 	// which ensure that the statistic data of TotalBytes equal to BytesDone when finished
@@ -190,8 +190,8 @@ func (kp *kopiaProvider) RunBackup(
 		},
 	)
 
-	log.Debugf("Kopia backup finished, snapshot ID %s, backup size %d", snapshotInfo.ID, snapshotInfo.SnapshotSize)
-	return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, progress.GetIncrementalSize(), snapshotInfo.SourceSize, nil
+	log.Debugf("Kopia backup finished, snapshot ID %s, backup size %d, fallback %v", snapshotInfo.ID, snapshotInfo.SnapshotSize, snapshotInfo.Fallback)
+	return snapshotInfo.ID, false, snapshotInfo.SnapshotSize, progress.GetIncrementalSize(), snapshotInfo.SourceSize, snapshotInfo.Fallback, nil
 }
 
 func (kp *kopiaProvider) GetPassword(param any) (string, error) {
@@ -215,7 +215,7 @@ func (kp *kopiaProvider) RunRestore(
 	_ CBTParam,
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
-	updater uploader.ProgressUpdater) (int64, int64, error) {
+	updater uploader.ProgressUpdater) (int64, int64, bool, error) {
 	log := kp.log.WithFields(logrus.Fields{
 		"snapshotID": snapshotID,
 		"volumePath": volumePath,
@@ -236,15 +236,15 @@ func (kp *kopiaProvider) RunRestore(
 	// We use the cancel channel to control the restore cancel, so don't pass a context with cancel to Kopia restore.
 	// Otherwise, Kopia restore will not response to the cancel control but return an arbitrary error.
 	// Kopia restore cancel is not designed as well as Kopia backup which uses the context to control backup cancel all the way.
-	size, fileCount, err := kopiaRestoreFunc(context.Background(), repoWriter, progress, snapshotID, volumePath, incremental, volMode, uploaderCfg, log, restoreCancel)
+	size, fileCount, fallback, err := kopiaRestoreFunc(context.Background(), repoWriter, progress, snapshotID, volumePath, incremental, volMode, uploaderCfg, log, restoreCancel)
 
 	if err != nil {
-		return 0, 0, errors.Wrapf(err, "Failed to run kopia restore")
+		return 0, 0, fallback, errors.Wrapf(err, "Failed to run kopia restore")
 	}
 
 	if atomic.LoadInt32(&kp.canceling) == 1 {
 		log.Error("Kopia restore is canceled")
-		return 0, 0, ErrorCanceled
+		return 0, 0, fallback, ErrorCanceled
 	}
 
 	// which ensure that the statistic data of TotalBytes equal to BytesDone when finished
@@ -253,10 +253,10 @@ func (kp *kopiaProvider) RunRestore(
 		BytesDone:  size,
 	})
 
-	output := fmt.Sprintf("Kopia restore finished, restore size %d, file count %d", size, fileCount)
+	output := fmt.Sprintf("Kopia restore finished, restore size %d, file count %d, fallback %v", size, fileCount, fallback)
 
 	log.Info(output)
 
 	// the incremental bytes is the same as the total bytes because total bytes is the size of actual data Kopia writes
-	return size, size, nil
+	return size, size, fallback, nil
 }

pkg/uploader/provider/kopia_test.go

@@ -106,7 +106,7 @@ func TestRunBackup(t *testing.T) {
 				tc.volMode = uploader.PersistentVolumeFilesystem
 			}
 			kopiaBackupFunc = tc.hookBackupFunc
-			_, _, _, _, _, err := kp.RunBackup(t.Context(), "var", "", nil, false, "", CBTParam{}, tc.volMode, map[string]string{}, &updater)
+			_, _, _, _, _, _, err := kp.RunBackup(t.Context(), "var", "", nil, false, "", CBTParam{}, tc.volMode, map[string]string{}, &updater)
 			if tc.notError {
 				assert.NoError(t, err)
 			} else {
@@ -119,30 +119,30 @@ func TestRunBackup(t *testing.T) {
 func TestRunRestore(t *testing.T) {
 	testCases := []struct {
 		name            string
-		hookRestoreFunc func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error)
+		hookRestoreFunc func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, bool, error)
 		notError        bool
 		volMode         uploader.PersistentVolumeMode
 		incremental     bool
 	}{
 		{
 			name: "normal restore",
-			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
-				return 0, 0, nil
+			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, bool, error) {
+				return 0, 0, false, nil
 			},
 			notError: true,
 		},
 		{
 			name: "normal block mode restore",
-			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
-				return 0, 0, nil
+			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, bool, error) {
+				return 0, 0, false, nil
 			},
 			volMode:  uploader.PersistentVolumeBlock,
 			notError: true,
 		},
 		{
 			name: "failed to restore",
-			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
-				return 0, 0, errors.New("failed to restore")
+			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, bool, error) {
+				return 0, 0, false, errors.New("failed to restore")
 			},
 			notError: false,
 		},
@@ -157,10 +157,10 @@ func TestRunRestore(t *testing.T) {
 			if tc.volMode == "" {
 				tc.volMode = uploader.PersistentVolumeFilesystem
 			}
-			kopiaRestoreFunc = func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
+			kopiaRestoreFunc = func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, bool, error) {
 				return tc.hookRestoreFunc(ctx, rep, progress, snapshotID, dest, incremental, volMode, uploaderCfg, log, cancleCh)
 			}
-			_, _, err := kp.RunRestore(t.Context(), "", "/var", tc.incremental, CBTParam{}, tc.volMode, map[string]string{}, &updater)
+			_, _, _, err := kp.RunRestore(t.Context(), "", "/var", tc.incremental, CBTParam{}, tc.volMode, map[string]string{}, &updater)
 			if tc.notError {
 				assert.NoError(t, err)
 			} else {

pkg/uploader/provider/mocks/Provider.go

@@ -91,7 +91,7 @@ func (_c *Provider_Close_Call) RunAndReturn(run func(ctx context.Context) error)
 }
 
 // RunBackup provides a mock function for the type Provider
-func (_mock *Provider) RunBackup(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, error) {
+func (_mock *Provider) RunBackup(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, bool, error) {
 	ret := _mock.Called(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
 
 	if len(ret) == 0 {
@@ -103,8 +103,9 @@ func (_mock *Provider) RunBackup(ctx context.Context, path string, realSource st
 	var r2 int64
 	var r3 int64
 	var r4 int64
-	var r5 error
-	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (string, bool, int64, int64, int64, error)); ok {
+	var r5 bool
+	var r6 error
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (string, bool, int64, int64, int64, bool, error)); ok {
 		return returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
 	}
 	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) string); ok {
@@ -132,12 +133,17 @@ func (_mock *Provider) RunBackup(ctx context.Context, path string, realSource st
 	} else {
 		r4 = ret.Get(4).(int64)
 	}
-	if returnFunc, ok := ret.Get(5).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+	if returnFunc, ok := ret.Get(5).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) bool); ok {
 		r5 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
 	} else {
-		r5 = ret.Error(5)
+		r5 = ret.Get(5).(bool)
 	}
-	return r0, r1, r2, r3, r4, r5
+	if returnFunc, ok := ret.Get(6).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+		r6 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+	} else {
+		r6 = ret.Error(6)
+	}
+	return r0, r1, r2, r3, r4, r5, r6
 }
 
 // Provider_RunBackup_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RunBackup'
@@ -218,18 +224,18 @@ func (_c *Provider_RunBackup_Call) Run(run func(ctx context.Context, path string
 	return _c
 }
 
-func (_c *Provider_RunBackup_Call) Return(_a0 string, _a1 bool, _a2 int64, _a3 int64, _a4 int64, _a5 error) *Provider_RunBackup_Call {
-	_c.Call.Return(_a0, _a1, _a2, _a3, _a4, _a5)
+func (_c *Provider_RunBackup_Call) Return(_a0 string, _a1 bool, _a2 int64, _a3 int64, _a4 int64, _a5 bool, _a6 error) *Provider_RunBackup_Call {
+	_c.Call.Return(_a0, _a1, _a2, _a3, _a4, _a5, _a6)
 	return _c
 }
 
-func (_c *Provider_RunBackup_Call) RunAndReturn(run func(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, error)) *Provider_RunBackup_Call {
+func (_c *Provider_RunBackup_Call) RunAndReturn(run func(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, bool, error)) *Provider_RunBackup_Call {
 	_c.Call.Return(run)
 	return _c
 }
 
 // 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, 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, bool, error) {
 	ret := _mock.Called(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)
 
 	if len(ret) == 0 {
@@ -238,8 +244,9 @@ func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volume
 
 	var r0 int64
 	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 {
+	var r2 bool
+	var r3 error
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, int64, bool, 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 {
@@ -254,13 +261,19 @@ func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volume
 		r1 = ret.Get(1).(int64)
 	}
 
-	if returnFunc, ok := ret.Get(2).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+	if returnFunc, ok := ret.Get(2).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) bool); ok {
 		r2 = returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)
 	} else {
-		r2 = ret.Error(2)
+		r2 = ret.Get(2).(bool)
 	}
 
-	return r0, r1, r2
+	if returnFunc, ok := ret.Get(3).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+		r3 = returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)
+	} else {
+		r3 = ret.Error(3)
+	}
+
+	return r0, r1, r2, r3
 }
 
 // Provider_RunRestore_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RunRestore'
@@ -329,12 +342,12 @@ func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID
 	return _c
 }
 
-func (_c *Provider_RunRestore_Call) Return(_a0 int64, _a1 int64, _a2 error) *Provider_RunRestore_Call {
-	_c.Call.Return(_a0, _a1, _a2)
+func (_c *Provider_RunRestore_Call) Return(_a0 int64, _a1 int64, _a2 bool, _a3 error) *Provider_RunRestore_Call {
+	_c.Call.Return(_a0, _a1, _a2, _a3)
 	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, 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, bool, error)) *Provider_RunRestore_Call {
 	_c.Call.Return(run)
 	return _c
 }

pkg/uploader/provider/provider.go

@@ -57,7 +57,7 @@ type Provider interface {
 		cbtParam CBTParam,
 		volMode uploader.PersistentVolumeMode,
 		uploaderCfg map[string]string,
-		updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, error)
+		updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, bool, error)
 	// 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(
@@ -68,7 +68,7 @@ type Provider interface {
 		cbtParam CBTParam,
 		volMode uploader.PersistentVolumeMode,
 		uploaderConfig map[string]string,
-		updater uploader.ProgressUpdater) (int64, int64, error)
+		updater uploader.ProgressUpdater) (int64, int64, bool, error)
 	// Close which will close related repository
 	Close(ctx context.Context) error
 }

pkg/uploader/types.go

@@ -55,6 +55,7 @@ type SnapshotInfo struct {
 	SnapshotSize    int64
 	IncrementalSize int64
 	SourceSize      int64
+	Fallback        bool
 }
 
 // Progress which defined two variables to record progress