Add Volume's snapshot data mover result check.

Signed-off-by: Xun Jiang <jxun@vmware.com>

pkg/backup/item_backupper.go

@@ -525,6 +525,11 @@ func (ib *itemBackupper) takePVSnapshot(obj runtime.Unstructured, log logrus.Fie
 	// After that, this warning can be removed.
 	if boolptr.IsSetToTrue(ib.backupRequest.Spec.SnapshotMoveData) {
 		log.Warnf("VolumeSnapshotter plugin doesn't support data movement.")
+
+		if features.IsEnabled(velerov1api.CSIFeatureFlag) && pv.Spec.CSI == nil {
+			log.Warn("Cannot use CSI data mover to handle PV, because PV doesn't contain CSI in spec.",
+				"Fall back to Velero native snapshot.")
+		}
 	}
 
 	if ib.backupRequest.ResPolicies != nil {

pkg/restore/restore.go

@@ -1326,6 +1326,10 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 		}
 
 		if pvc.Spec.VolumeName != "" {
+			if err := ctx.checkVolumeRestoreByDataMover(pvc); err != nil {
+				errs.Add(namespace, err)
+			}
+
 			// This used to only happen with PVB volumes, but now always remove this binding metadata
 			obj = resetVolumeBindingInfo(obj)
 
@@ -2206,3 +2210,32 @@ func (ctx *restoreContext) processUpdateResourcePolicy(fromCluster, fromClusterW
 	}
 	return warnings, errs
 }
+
+// isVolumeRestoreByDataMover checks whether the passed PVC is restored by CSI snapshot data
+// mover plugin, or it is restored by the fall-backed Velero native snapshot.
+// If it's not restored by either of the mentioned method, return a error.
+func (ctx *restoreContext) checkVolumeRestoreByDataMover(pvc *v1.PersistentVolumeClaim) error {
+	if boolptr.IsSetToTrue(ctx.backup.Spec.SnapshotMoveData) == true {
+		dataMoverOperationFound := false
+		for _, operation := range *ctx.itemOperationsList {
+			if operation.Spec.RestoreUID == string(pvc.UID) && strings.Contains(operation.Spec.OperationID, string(velerov1api.AsyncOperationIDPrefixDataDownload)) {
+				ctx.log.Debug("PVC has Async operation. It's handled by CSI snapshot data mover.")
+				return nil
+			}
+		}
+
+		if dataMoverOperationFound == false {
+			for k, v := range ctx.restoredItems {
+				if strings.EqualFold(strings.ToLower(k.resource), "v1/persistentvolume") &&
+					k.name == pvc.Spec.VolumeName && v.action == itemRestoreResultCreated {
+					ctx.log.Debug("PVC's PV is restored by fall-backed Velero native snapshot.")
+					return nil
+				}
+			}
+		}
+
+		ctx.log.Errorf("PVC %s/%s is not handled by both snapshot data mover and Velero native snapshot", pvc.Namespace, pvc.Name)
+		return fmt.Errorf("pvc %s/%s is not restored by both snapshot data mover and Velero native snapshot", pvc.Namespace, pvc.Name)
+	}
+	return nil
+}

pkg/restore/restore_test.go

@@ -3630,3 +3630,69 @@ func TestIsAlreadyExistsError(t *testing.T) {
 		})
 	}
 }
+
+func TestCheckVolumeRestoreByDataMover(t *testing.T) {
+	tests := []struct {
+		name          string
+		backup        velerov1api.Backup
+		operation     *itemoperation.RestoreOperation
+		restoredItems map[itemKey]restoredItemStatus
+		pvc           *corev1api.PersistentVolumeClaim
+		wantErr       bool
+	}{
+		{
+			name:          "backup with snapshotMoveData set to false",
+			backup:        *builder.ForBackup("velero", "backup").SnapshotMoveData(false).Result(),
+			operation:     nil,
+			restoredItems: nil,
+			pvc:           builder.ForPersistentVolumeClaim("velero", "pvc1").Result(),
+			wantErr:       false,
+		},
+		{
+			name:          "no operation and restoredItems",
+			backup:        *builder.ForBackup("velero", "backup").SnapshotMoveData(true).Result(),
+			operation:     nil,
+			restoredItems: nil,
+			pvc:           builder.ForPersistentVolumeClaim("velero", "pvc1").Result(),
+			wantErr:       true,
+		},
+		{
+			name:          "has operation",
+			backup:        *builder.ForBackup("velero", "backup").SnapshotMoveData(true).Result(),
+			operation:     &itemoperation.RestoreOperation{Spec: itemoperation.RestoreOperationSpec{OperationID: "dd-abcdef", RestoreUID: "pvc-uid"}},
+			restoredItems: nil,
+			pvc:           builder.ForPersistentVolumeClaim("velero", "pvc1").ObjectMeta(builder.WithUID("pvc-uid")).Result(),
+			wantErr:       false,
+		},
+		{
+			name:          "has restoredItems",
+			backup:        *builder.ForBackup("velero", "backup").SnapshotMoveData(true).Result(),
+			operation:     nil,
+			restoredItems: map[itemKey]restoredItemStatus{{resource: "v1/PersistentVolume", name: "pv-name"}: {action: itemRestoreResultCreated}},
+			pvc:           builder.ForPersistentVolumeClaim("velero", "pvc1").VolumeName("pv-name").ObjectMeta(builder.WithUID("pvc-uid")).Result(),
+			wantErr:       false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			h := newHarness(t)
+
+			ctx := &restoreContext{
+				log:           h.log,
+				backup:        &tc.backup,
+				restoredItems: tc.restoredItems,
+			}
+
+			list := []*itemoperation.RestoreOperation{}
+			if tc.operation != nil {
+				list = append(list, tc.operation)
+			}
+			ctx.itemOperationsList = &list
+
+			if tc.wantErr {
+				require.Error(t, ctx.checkVolumeRestoreByDataMover(tc.pvc))
+			}
+		})
+	}
+}