Merge branch 'release-1.12' into cherry-pick-6917

changelogs/unreleased/7054-kaovilai

@@ -0,0 +1 @@
+restore: Use warning when Create IsAlreadyExist and Get error

changelogs/unreleased/7058-ywk253100

@@ -0,0 +1 @@
+Truncate the credential file to avoid the change of secret content messing it up

changelogs/unreleased/7060-Lyndon-Li

@@ -0,0 +1 @@
+Fix issue #7027, data mover backup exposer should not assume the first volume as the backup volume in backup pod

changelogs/unreleased/7087-blackpiglet

@@ -0,0 +1 @@
+Add DataUpload Result and CSI VolumeSnapshot check for restore PV.

changelogs/unreleased/7097-Lyndon-Li

@@ -0,0 +1 @@
+Fix issue #7094, fallback to full backup if previous snapshot is not found

changelogs/unreleased/7098-yanggangtony

@@ -0,0 +1 @@
+Fix the node-agent missing metrics-address defines.

internal/credentials/file_store.go

@@ -71,7 +71,7 @@ func (n *namespacedFileStore) Path(selector *corev1api.SecretKeySelector) (strin
 
 	keyFilePath := filepath.Join(n.fsRoot, fmt.Sprintf("%s-%s", selector.Name, selector.Key))
 
-	file, err := n.fs.OpenFile(keyFilePath, os.O_RDWR|os.O_CREATE, 0644)
+	file, err := n.fs.OpenFile(keyFilePath, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0644)
 	if err != nil {
 		return "", errors.Wrap(err, "unable to open credentials file for writing")
 	}

pkg/builder/volume_snapshot_builder.go

@@ -67,3 +67,8 @@ func (v *VolumeSnapshotBuilder) BoundVolumeSnapshotContentName(vscName string) *
 	v.object.Status.BoundVolumeSnapshotContentName = &vscName
 	return v
 }
+
+func (v *VolumeSnapshotBuilder) SourcePVC(name string) *VolumeSnapshotBuilder {
+	v.object.Spec.Source.PersistentVolumeClaimName = &name
+	return v
+}

pkg/cmd/cli/nodeagent/server.go

@@ -114,6 +114,7 @@ func NewServerCommand(f client.Factory) *cobra.Command {
 	command.Flags().Var(formatFlag, "log-format", fmt.Sprintf("The format for log output. Valid values are %s.", strings.Join(formatFlag.AllowedValues(), ", ")))
 	command.Flags().DurationVar(&config.resourceTimeout, "resource-timeout", config.resourceTimeout, "How long to wait for resource processes which are not covered by other specific timeout parameters. Default is 10 minutes.")
 	command.Flags().DurationVar(&config.dataMoverPrepareTimeout, "data-mover-prepare-timeout", config.dataMoverPrepareTimeout, "How long to wait for preparing a DataUpload/DataDownload. Default is 30 minutes.")
+	command.Flags().StringVar(&config.metricsAddress, "metrics-address", config.metricsAddress, "The address to expose prometheus metrics")
 
 	return command
 }
@@ -193,14 +194,15 @@ func newNodeAgentServer(logger logrus.FieldLogger, factory client.Factory, confi
 	}
 
 	s := &nodeAgentServer{
-		logger:     logger,
-		ctx:        ctx,
-		cancelFunc: cancelFunc,
-		fileSystem: filesystem.NewFileSystem(),
-		mgr:        mgr,
-		config:     config,
-		namespace:  factory.Namespace(),
-		nodeName:   nodeName,
+		logger:         logger,
+		ctx:            ctx,
+		cancelFunc:     cancelFunc,
+		fileSystem:     filesystem.NewFileSystem(),
+		mgr:            mgr,
+		config:         config,
+		namespace:      factory.Namespace(),
+		nodeName:       nodeName,
+		metricsAddress: config.metricsAddress,
 	}
 
 	// the cache isn't initialized yet when "validatePodVolumesHostPath" is called, the client returned by the manager cannot

pkg/controller/restore_controller.go

@@ -515,6 +515,11 @@ func (r *restoreReconciler) runValidatedRestore(restore *api.Restore, info backu
 		return errors.Wrap(err, "error fetching volume snapshots metadata")
 	}
 
+	csiVolumeSnapshots, err := backupStore.GetCSIVolumeSnapshots(restore.Spec.BackupName)
+	if err != nil {
+		return errors.Wrap(err, "fail to fetch CSI VolumeSnapshots metadata")
+	}
+
 	restoreLog.Info("starting restore")
 
 	var podVolumeBackups []*api.PodVolumeBackup
@@ -531,6 +536,7 @@ func (r *restoreReconciler) runValidatedRestore(restore *api.Restore, info backu
 		BackupReader:         backupFile,
 		ResourceModifiers:    resourceModifiers,
 		DisableInformerCache: r.disableInformerCache,
+		CSIVolumeSnapshots:   csiVolumeSnapshots,
 	}
 	restoreWarnings, restoreErrors := r.restorer.RestoreWithResolvers(restoreReq, actionsResolver, pluginManager)
 

pkg/controller/restore_controller_test.go

@@ -23,6 +23,7 @@ import (
 	"testing"
 	"time"
 
+	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
@@ -471,6 +472,7 @@ func TestRestoreReconcile(t *testing.T) {
 			}
 			if test.expectedRestorerCall != nil {
 				backupStore.On("GetBackupContents", test.backup.Name).Return(io.NopCloser(bytes.NewReader([]byte("hello world"))), nil)
+				backupStore.On("GetCSIVolumeSnapshots", test.backup.Name).Return([]*snapshotv1api.VolumeSnapshot{}, nil)
 
 				restorer.On("RestoreWithResolvers", mock.Anything, mock.Anything, mock.Anything, mock.Anything,
 					mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(warnings, errors)

pkg/exposer/csi_snapshot.go

@@ -190,6 +190,7 @@ func (e *csiSnapshotExposer) GetExposed(ctx context.Context, ownerObject corev1.
 
 	backupPodName := ownerObject.Name
 	backupPVCName := ownerObject.Name
+	volumeName := string(ownerObject.UID)
 
 	curLog := e.log.WithFields(logrus.Fields{
 		"owner": ownerObject.Name,
@@ -218,7 +219,20 @@ func (e *csiSnapshotExposer) GetExposed(ctx context.Context, ownerObject corev1.
 
 	curLog.WithField("backup pvc", backupPVCName).Info("Backup PVC is bound")
 
-	return &ExposeResult{ByPod: ExposeByPod{HostingPod: pod, VolumeName: pod.Spec.Volumes[0].Name}}, nil
+	i := 0
+	for i = 0; i < len(pod.Spec.Volumes); i++ {
+		if pod.Spec.Volumes[i].Name == volumeName {
+			break
+		}
+	}
+
+	if i == len(pod.Spec.Volumes) {
+		return nil, errors.Errorf("backup pod %s doesn't have the expected backup volume", pod.Name)
+	}
+
+	curLog.WithField("pod", pod.Name).Infof("Backup volume is found in pod at index %v", i)
+
+	return &ExposeResult{ByPod: ExposeByPod{HostingPod: pod, VolumeName: volumeName}}, nil
 }
 
 func (e *csiSnapshotExposer) CleanUp(ctx context.Context, ownerObject corev1.ObjectReference, vsName string, sourceNamespace string) {

pkg/exposer/csi_snapshot_test.go

@@ -37,6 +37,8 @@ import (
 	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
+
+	clientFake "sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
 type reactor struct {
@@ -384,3 +386,180 @@ func TestExpose(t *testing.T) {
 		})
 	}
 }
+
+func TestGetExpose(t *testing.T) {
+	backup := &velerov1.Backup{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: velerov1.SchemeGroupVersion.String(),
+			Kind:       "Backup",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: velerov1.DefaultNamespace,
+			Name:      "fake-backup",
+			UID:       "fake-uid",
+		},
+	}
+
+	backupPod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: backup.Namespace,
+			Name:      backup.Name,
+		},
+		Spec: corev1.PodSpec{
+			Volumes: []corev1.Volume{
+				{
+					Name: "fake-volume",
+				},
+				{
+					Name: "fake-volume-2",
+				},
+				{
+					Name: string(backup.UID),
+				},
+			},
+		},
+	}
+
+	backupPodWithoutVolume := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: backup.Namespace,
+			Name:      backup.Name,
+		},
+		Spec: corev1.PodSpec{
+			Volumes: []corev1.Volume{
+				{
+					Name: "fake-volume-1",
+				},
+				{
+					Name: "fake-volume-2",
+				},
+			},
+		},
+	}
+
+	backupPVC := &corev1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: backup.Namespace,
+			Name:      backup.Name,
+		},
+		Spec: corev1.PersistentVolumeClaimSpec{
+			VolumeName: "fake-pv-name",
+		},
+	}
+
+	backupPV := &corev1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "fake-pv-name",
+		},
+	}
+
+	scheme := runtime.NewScheme()
+	corev1.AddToScheme(scheme)
+
+	tests := []struct {
+		name            string
+		kubeClientObj   []runtime.Object
+		ownerBackup     *velerov1.Backup
+		exposeWaitParam CSISnapshotExposeWaitParam
+		Timeout         time.Duration
+		err             string
+		expectedResult  *ExposeResult
+	}{
+		{
+			name:        "backup pod is not found",
+			ownerBackup: backup,
+			exposeWaitParam: CSISnapshotExposeWaitParam{
+				NodeName: "fake-node",
+			},
+		},
+		{
+			name:        "wait pvc bound fail",
+			ownerBackup: backup,
+			exposeWaitParam: CSISnapshotExposeWaitParam{
+				NodeName: "fake-node",
+			},
+			kubeClientObj: []runtime.Object{
+				backupPod,
+			},
+			Timeout: time.Second,
+			err:     "error to wait backup PVC bound, fake-backup: error to wait for rediness of PVC: error to get pvc velero/fake-backup: persistentvolumeclaims \"fake-backup\" not found",
+		},
+		{
+			name:        "backup volume not found in pod",
+			ownerBackup: backup,
+			exposeWaitParam: CSISnapshotExposeWaitParam{
+				NodeName: "fake-node",
+			},
+			kubeClientObj: []runtime.Object{
+				backupPodWithoutVolume,
+				backupPVC,
+				backupPV,
+			},
+			Timeout: time.Second,
+			err:     "backup pod fake-backup doesn't have the expected backup volume",
+		},
+		{
+			name:        "succeed",
+			ownerBackup: backup,
+			exposeWaitParam: CSISnapshotExposeWaitParam{
+				NodeName: "fake-node",
+			},
+			kubeClientObj: []runtime.Object{
+				backupPod,
+				backupPVC,
+				backupPV,
+			},
+			Timeout: time.Second,
+			expectedResult: &ExposeResult{
+				ByPod: ExposeByPod{
+					HostingPod: backupPod,
+					VolumeName: string(backup.UID),
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fakeKubeClient := fake.NewSimpleClientset(test.kubeClientObj...)
+
+			fakeClientBuilder := clientFake.NewClientBuilder()
+			fakeClientBuilder = fakeClientBuilder.WithScheme(scheme)
+
+			fakeClient := fakeClientBuilder.WithRuntimeObjects(test.kubeClientObj...).Build()
+
+			exposer := csiSnapshotExposer{
+				kubeClient: fakeKubeClient,
+				log:        velerotest.NewLogger(),
+			}
+
+			var ownerObject corev1.ObjectReference
+			if test.ownerBackup != nil {
+				ownerObject = corev1.ObjectReference{
+					Kind:       test.ownerBackup.Kind,
+					Namespace:  test.ownerBackup.Namespace,
+					Name:       test.ownerBackup.Name,
+					UID:        test.ownerBackup.UID,
+					APIVersion: test.ownerBackup.APIVersion,
+				}
+			}
+
+			test.exposeWaitParam.NodeClient = fakeClient
+
+			result, err := exposer.GetExposed(context.Background(), ownerObject, test.Timeout, &test.exposeWaitParam)
+			if test.err == "" {
+				assert.NoError(t, err)
+
+				if test.expectedResult == nil {
+					assert.Nil(t, result)
+				} else {
+					assert.NoError(t, err)
+					assert.Equal(t, test.expectedResult.ByPod.VolumeName, result.ByPod.VolumeName)
+					assert.Equal(t, test.expectedResult.ByPod.HostingPod.Name, result.ByPod.HostingPod.Name)
+				}
+			} else {
+				assert.EqualError(t, err, test.err)
+			}
+		})
+	}
+}

pkg/install/daemonset.go

@@ -105,6 +105,7 @@ func DaemonSet(namespace string, opts ...podTemplateOption) *appsv1.DaemonSet {
 						{
 							Name:            "node-agent",
 							Image:           c.image,
+							Ports:           containerPorts(),
 							ImagePullPolicy: pullPolicy,
 							Command: []string{
 								"/velero",

pkg/restore/request.go

@@ -21,6 +21,7 @@ import (
 	"io"
 	"sort"
 
+	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
 	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/runtime"
 
@@ -60,6 +61,7 @@ type Request struct {
 	itemOperationsList   *[]*itemoperation.RestoreOperation
 	ResourceModifiers    *resourcemodifiers.ResourceModifiers
 	DisableInformerCache bool
+	CSIVolumeSnapshots   []*snapshotv1api.VolumeSnapshot
 }
 
 type restoredItemStatus struct {

pkg/restore/restore.go

@@ -30,6 +30,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	v1 "k8s.io/api/core/v1"
@@ -298,6 +299,7 @@ func (kr *kubernetesRestorer) RestoreWithResolvers(
 		pvsToProvision:                 sets.NewString(),
 		pvRestorer:                     pvRestorer,
 		volumeSnapshots:                req.VolumeSnapshots,
+		csiVolumeSnapshots:             req.CSIVolumeSnapshots,
 		podVolumeBackups:               req.PodVolumeBackups,
 		resourceTerminatingTimeout:     kr.resourceTerminatingTimeout,
 		resourceTimeout:                kr.resourceTimeout,
@@ -347,6 +349,7 @@ type restoreContext struct {
 	pvsToProvision                 sets.String
 	pvRestorer                     PVRestorer
 	volumeSnapshots                []*volume.Snapshot
+	csiVolumeSnapshots             []*snapshotv1api.VolumeSnapshot
 	podVolumeBackups               []*velerov1api.PodVolumeBackup
 	resourceTerminatingTimeout     time.Duration
 	resourceTimeout                time.Duration
@@ -1287,7 +1290,35 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 			}
 
 		case hasPodVolumeBackup(obj, ctx):
-			ctx.log.Infof("Dynamically re-provisioning persistent volume because it has a pod volume backup to be restored.")
+			ctx.log.WithFields(logrus.Fields{
+				"namespace":     obj.GetNamespace(),
+				"name":          obj.GetName(),
+				"groupResource": groupResource.String(),
+			}).Infof("Dynamically re-provisioning persistent volume because it has a pod volume backup to be restored.")
+			ctx.pvsToProvision.Insert(name)
+
+			// Return early because we don't want to restore the PV itself, we
+			// want to dynamically re-provision it.
+			return warnings, errs, itemExists
+
+		case hasCSIVolumeSnapshot(ctx, obj):
+			ctx.log.WithFields(logrus.Fields{
+				"namespace":     obj.GetNamespace(),
+				"name":          obj.GetName(),
+				"groupResource": groupResource.String(),
+			}).Infof("Dynamically re-provisioning persistent volume because it has a related CSI VolumeSnapshot.")
+			ctx.pvsToProvision.Insert(name)
+
+			// Return early because we don't want to restore the PV itself, we
+			// want to dynamically re-provision it.
+			return warnings, errs, itemExists
+
+		case hasSnapshotDataUpload(ctx, obj):
+			ctx.log.WithFields(logrus.Fields{
+				"namespace":     obj.GetNamespace(),
+				"name":          obj.GetName(),
+				"groupResource": groupResource.String(),
+			}).Infof("Dynamically re-provisioning persistent volume because it has a related snapshot DataUpload.")
 			ctx.pvsToProvision.Insert(name)
 
 			// Return early because we don't want to restore the PV itself, we
@@ -1295,7 +1326,11 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 			return warnings, errs, itemExists
 
 		case hasDeleteReclaimPolicy(obj.Object):
-			ctx.log.Infof("Dynamically re-provisioning persistent volume because it doesn't have a snapshot and its reclaim policy is Delete.")
+			ctx.log.WithFields(logrus.Fields{
+				"namespace":     obj.GetNamespace(),
+				"name":          obj.GetName(),
+				"groupResource": groupResource.String(),
+			}).Infof("Dynamically re-provisioning persistent volume because it doesn't have a snapshot and its reclaim policy is Delete.")
 			ctx.pvsToProvision.Insert(name)
 
 			// Return early because we don't want to restore the PV itself, we
@@ -1303,7 +1338,11 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 			return warnings, errs, itemExists
 
 		default:
-			ctx.log.Infof("Restoring persistent volume as-is because it doesn't have a snapshot and its reclaim policy is not Delete.")
+			ctx.log.WithFields(logrus.Fields{
+				"namespace":     obj.GetNamespace(),
+				"name":          obj.GetName(),
+				"groupResource": groupResource.String(),
+			}).Infof("Restoring persistent volume as-is because it doesn't have a snapshot and its reclaim policy is not Delete.")
 
 			// Check to see if the claimRef.namespace field needs to be remapped, and do so if necessary.
 			_, err = remapClaimRefNS(ctx, obj)
@@ -1524,18 +1563,17 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 	}
 
 	if restoreErr != nil {
-		// check for the existence of the object in cluster, if no error then it implies that object exists
-		// and if err then we want to judge whether there is an existing error in the previous creation.
-		// if so, we will return the 'get' error.
-		// otherwise, we will return the original creation error.
+		// check for the existence of the object that failed creation due to alreadyExist in cluster, if no error then it implies that object exists.
+		// and if err then itemExists remains false as we were not able to confirm the existence of the object via Get call or creation call.
+		// We return the get error as a warning to notify the user that the object could exist in cluster and we were not able to confirm it.
 		if !ctx.disableInformerCache {
 			fromCluster, err = ctx.getResource(groupResource, obj, namespace, name)
 		} else {
 			fromCluster, err = resourceClient.Get(name, metav1.GetOptions{})
 		}
 		if err != nil && isAlreadyExistsError {
-			ctx.log.Errorf("Error retrieving in-cluster version of %s: %v", kube.NamespaceAndName(obj), err)
-			errs.Add(namespace, err)
+			ctx.log.Warnf("Unable to retrieve in-cluster version of %s: %v, object won't be restored by velero or have restore labels, and existing resource policy is not applied", kube.NamespaceAndName(obj), err)
+			warnings.Add(namespace, err)
 			return warnings, errs, itemExists
 		}
 	}
@@ -1930,6 +1968,55 @@ func hasSnapshot(pvName string, snapshots []*volume.Snapshot) bool {
 	return false
 }
 
+func hasCSIVolumeSnapshot(ctx *restoreContext, unstructuredPV *unstructured.Unstructured) bool {
+	pv := new(v1.PersistentVolume)
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredPV.Object, pv); err != nil {
+		ctx.log.WithError(err).Warnf("Unable to convert PV from unstructured to structured")
+		return false
+	}
+
+	for _, vs := range ctx.csiVolumeSnapshots {
+		if pv.Spec.ClaimRef.Name == *vs.Spec.Source.PersistentVolumeClaimName &&
+			pv.Spec.ClaimRef.Namespace == vs.Namespace {
+			return true
+		}
+	}
+	return false
+}
+
+func hasSnapshotDataUpload(ctx *restoreContext, unstructuredPV *unstructured.Unstructured) bool {
+	pv := new(v1.PersistentVolume)
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredPV.Object, pv); err != nil {
+		ctx.log.WithError(err).Warnf("Unable to convert PV from unstructured to structured")
+		return false
+	}
+
+	if pv.Spec.ClaimRef == nil {
+		return false
+	}
+
+	dataUploadResultList := new(v1.ConfigMapList)
+	err := ctx.kbClient.List(go_context.TODO(), dataUploadResultList, &crclient.ListOptions{
+		LabelSelector: labels.SelectorFromSet(map[string]string{
+			velerov1api.RestoreUIDLabel:       label.GetValidName(string(ctx.restore.GetUID())),
+			velerov1api.PVCNamespaceNameLabel: label.GetValidName(pv.Spec.ClaimRef.Namespace + "." + pv.Spec.ClaimRef.Name),
+			velerov1api.ResourceUsageLabel:    label.GetValidName(string(velerov1api.VeleroResourceUsageDataUploadResult)),
+		}),
+	})
+	if err != nil {
+		ctx.log.WithError(err).Warnf("Fail to list DataUpload result CM.")
+		return false
+	}
+
+	if len(dataUploadResultList.Items) != 1 {
+		ctx.log.WithError(fmt.Errorf("dataupload result number is not expected")).
+			Warnf("Got %d DataUpload result. Expect one.", len(dataUploadResultList.Items))
+		return false
+	}
+
+	return true
+}
+
 func hasPodVolumeBackup(unstructuredPV *unstructured.Unstructured, ctx *restoreContext) bool {
 	if len(ctx.podVolumeBackups) == 0 {
 		return false

pkg/restore/restore_test.go

@@ -25,6 +25,7 @@ import (
 	"testing"
 	"time"
 
+	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
@@ -2256,6 +2257,7 @@ func (*volumeSnapshotter) DeleteSnapshot(snapshotID string) error {
 // Verification is done by looking at the contents of the API and the metadata/spec/status of
 // the items in the API.
 func TestRestorePersistentVolumes(t *testing.T) {
+	testPVCName := "testPVC"
 	tests := []struct {
 		name                    string
 		restore                 *velerov1api.Restore
@@ -2265,6 +2267,8 @@ func TestRestorePersistentVolumes(t *testing.T) {
 		volumeSnapshots         []*volume.Snapshot
 		volumeSnapshotLocations []*velerov1api.VolumeSnapshotLocation
 		volumeSnapshotterGetter volumeSnapshotterGetter
+		csiVolumeSnapshots      []*snapshotv1api.VolumeSnapshot
+		dataUploadResult        *corev1api.ConfigMap
 		want                    []*test.APIResource
 		wantError               bool
 		wantWarning             bool
@@ -2923,6 +2927,77 @@ func TestRestorePersistentVolumes(t *testing.T) {
 				),
 			},
 		},
+		{
+			name:    "when a PV with a reclaim policy of retain has a CSI VolumeSnapshot and does not exist in-cluster, the PV is not restored",
+			restore: defaultRestore().Result(),
+			backup:  defaultBackup().Result(),
+			tarball: test.NewTarWriter(t).
+				AddItems("persistentvolumes",
+					builder.ForPersistentVolume("pv-1").
+						ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).
+						ClaimRef("velero", testPVCName).
+						Result(),
+				).
+				Done(),
+			apiResources: []*test.APIResource{
+				test.PVs(),
+				test.PVCs(),
+			},
+			csiVolumeSnapshots: []*snapshotv1api.VolumeSnapshot{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "velero",
+						Name:      "test",
+					},
+					Spec: snapshotv1api.VolumeSnapshotSpec{
+						Source: snapshotv1api.VolumeSnapshotSource{
+							PersistentVolumeClaimName: &testPVCName,
+						},
+					},
+				},
+			},
+			volumeSnapshotLocations: []*velerov1api.VolumeSnapshotLocation{
+				builder.ForVolumeSnapshotLocation(velerov1api.DefaultNamespace, "default").Provider("provider-1").Result(),
+			},
+			volumeSnapshotterGetter: map[string]vsv1.VolumeSnapshotter{
+				"provider-1": &volumeSnapshotter{
+					snapshotVolumes: map[string]string{"snapshot-1": "new-volume"},
+				},
+			},
+			want: []*test.APIResource{},
+		},
+		{
+			name:    "when a PV with a reclaim policy of retain has a DataUpload result CM and does not exist in-cluster, the PV is not restored",
+			restore: defaultRestore().ObjectMeta(builder.WithUID("fakeUID")).Result(),
+			backup:  defaultBackup().Result(),
+			tarball: test.NewTarWriter(t).
+				AddItems("persistentvolumes",
+					builder.ForPersistentVolume("pv-1").
+						ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).
+						ClaimRef("velero", testPVCName).
+						Result(),
+				).
+				Done(),
+			apiResources: []*test.APIResource{
+				test.PVs(),
+				test.PVCs(),
+				test.ConfigMaps(),
+			},
+			volumeSnapshotLocations: []*velerov1api.VolumeSnapshotLocation{
+				builder.ForVolumeSnapshotLocation(velerov1api.DefaultNamespace, "default").Provider("provider-1").Result(),
+			},
+			volumeSnapshotterGetter: map[string]vsv1.VolumeSnapshotter{
+				"provider-1": &volumeSnapshotter{
+					snapshotVolumes: map[string]string{"snapshot-1": "new-volume"},
+				},
+			},
+			dataUploadResult: builder.ForConfigMap("velero", "test").ObjectMeta(builder.WithLabelsMap(map[string]string{
+				velerov1api.RestoreUIDLabel:       "fakeUID",
+				velerov1api.PVCNamespaceNameLabel: "velero/testPVC",
+				velerov1api.ResourceUsageLabel:    string(velerov1api.VeleroResourceUsageDataUploadResult),
+			})).Result(),
+			want: []*test.APIResource{},
+		},
 	}
 
 	for _, tc := range tests {
@@ -2939,6 +3014,10 @@ func TestRestorePersistentVolumes(t *testing.T) {
 				require.NoError(t, h.restorer.kbClient.Create(context.Background(), vsl))
 			}
 
+			if tc.dataUploadResult != nil {
+				require.NoError(t, h.restorer.kbClient.Create(context.TODO(), tc.dataUploadResult))
+			}
+
 			for _, r := range tc.apiResources {
 				h.AddItems(t, r)
 			}
@@ -2955,11 +3034,12 @@ func TestRestorePersistentVolumes(t *testing.T) {
 			}
 
 			data := &Request{
-				Log:             h.log,
-				Restore:         tc.restore,
-				Backup:          tc.backup,
-				VolumeSnapshots: tc.volumeSnapshots,
-				BackupReader:    tc.tarball,
+				Log:                h.log,
+				Restore:            tc.restore,
+				Backup:             tc.backup,
+				VolumeSnapshots:    tc.volumeSnapshots,
+				BackupReader:       tc.tarball,
+				CSIVolumeSnapshots: tc.csiVolumeSnapshots,
 			}
 			warnings, errs := h.restorer.Restore(
 				data,
@@ -3652,3 +3732,175 @@ func TestIsAlreadyExistsError(t *testing.T) {
 		})
 	}
 }
+
+func TestHasCSIVolumeSnapshot(t *testing.T) {
+	tests := []struct {
+		name           string
+		vs             *snapshotv1api.VolumeSnapshot
+		obj            *unstructured.Unstructured
+		expectedResult bool
+	}{
+		{
+			name: "Invalid PV, expect false.",
+			obj: &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"kind": 1,
+				},
+			},
+			expectedResult: false,
+		},
+		{
+			name: "Cannot find VS, expect false",
+			obj: &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"kind":       "PersistentVolume",
+					"apiVersion": "v1",
+					"metadata": map[string]interface{}{
+						"namespace": "default",
+						"name":      "test",
+					},
+				},
+			},
+			expectedResult: false,
+		},
+		{
+			name: "Find VS, expect true.",
+			obj: &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"kind":       "PersistentVolume",
+					"apiVersion": "v1",
+					"metadata": map[string]interface{}{
+						"namespace": "velero",
+						"name":      "test",
+					},
+					"spec": map[string]interface{}{
+						"claimRef": map[string]interface{}{
+							"namespace": "velero",
+							"name":      "test",
+						},
+					},
+				},
+			},
+			vs:             builder.ForVolumeSnapshot("velero", "test").SourcePVC("test").Result(),
+			expectedResult: true,
+		},
+	}
+
+	for _, tc := range tests {
+		h := newHarness(t)
+
+		ctx := &restoreContext{
+			log: h.log,
+		}
+
+		if tc.vs != nil {
+			ctx.csiVolumeSnapshots = []*snapshotv1api.VolumeSnapshot{tc.vs}
+		}
+
+		t.Run(tc.name, func(t *testing.T) {
+			require.Equal(t, tc.expectedResult, hasCSIVolumeSnapshot(ctx, tc.obj))
+		})
+	}
+}
+
+func TestHasSnapshotDataUpload(t *testing.T) {
+	tests := []struct {
+		name           string
+		duResult       *corev1api.ConfigMap
+		obj            *unstructured.Unstructured
+		expectedResult bool
+		restore        *velerov1api.Restore
+	}{
+		{
+			name: "Invalid PV, expect false.",
+			obj: &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"kind": 1,
+				},
+			},
+			expectedResult: false,
+		},
+		{
+			name: "PV without ClaimRef, expect false",
+			obj: &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"kind":       "PersistentVolume",
+					"apiVersion": "v1",
+					"metadata": map[string]interface{}{
+						"namespace": "default",
+						"name":      "test",
+					},
+				},
+			},
+			duResult:       builder.ForConfigMap("velero", "test").Result(),
+			restore:        builder.ForRestore("velero", "test").ObjectMeta(builder.WithUID("fakeUID")).Result(),
+			expectedResult: false,
+		},
+		{
+			name: "Cannot find DataUploadResult CM, expect false",
+			obj: &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"kind":       "PersistentVolume",
+					"apiVersion": "v1",
+					"metadata": map[string]interface{}{
+						"namespace": "default",
+						"name":      "test",
+					},
+					"spec": map[string]interface{}{
+						"claimRef": map[string]interface{}{
+							"namespace": "velero",
+							"name":      "testPVC",
+						},
+					},
+				},
+			},
+			duResult:       builder.ForConfigMap("velero", "test").Result(),
+			restore:        builder.ForRestore("velero", "test").ObjectMeta(builder.WithUID("fakeUID")).Result(),
+			expectedResult: false,
+		},
+		{
+			name: "Find DataUploadResult CM, expect true",
+			obj: &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"kind":       "PersistentVolume",
+					"apiVersion": "v1",
+					"metadata": map[string]interface{}{
+						"namespace": "default",
+						"name":      "test",
+					},
+					"spec": map[string]interface{}{
+						"claimRef": map[string]interface{}{
+							"namespace": "velero",
+							"name":      "testPVC",
+						},
+					},
+				},
+			},
+			duResult: builder.ForConfigMap("velero", "test").ObjectMeta(builder.WithLabelsMap(map[string]string{
+				velerov1api.RestoreUIDLabel:       "fakeUID",
+				velerov1api.PVCNamespaceNameLabel: "velero/testPVC",
+				velerov1api.ResourceUsageLabel:    string(velerov1api.VeleroResourceUsageDataUploadResult),
+			})).Result(),
+			restore:        builder.ForRestore("velero", "test").ObjectMeta(builder.WithUID("fakeUID")).Result(),
+			expectedResult: false,
+		},
+	}
+
+	for _, tc := range tests {
+		h := newHarness(t)
+
+		ctx := &restoreContext{
+			log:      h.log,
+			kbClient: h.restorer.kbClient,
+			restore:  tc.restore,
+		}
+
+		if tc.duResult != nil {
+			require.NoError(t, ctx.kbClient.Create(context.TODO(), tc.duResult))
+		}
+
+		t.Run(tc.name, func(t *testing.T) {
+			require.Equal(t, tc.expectedResult, hasSnapshotDataUpload(ctx, tc.obj))
+		})
+	}
+}

pkg/test/resources.go

@@ -142,6 +142,17 @@ func ServiceAccounts(items ...metav1.Object) *APIResource {
 	}
 }
 
+func ConfigMaps(items ...metav1.Object) *APIResource {
+	return &APIResource{
+		Group:      "",
+		Version:    "v1",
+		Name:       "configmaps",
+		ShortName:  "cm",
+		Namespaced: true,
+		Items:      items,
+	}
+}
+
 func CRDs(items ...metav1.Object) *APIResource {
 	return &APIResource{
 		Group:      "apiextensions.k8s.io",

pkg/uploader/kopia/snapshot.go

@@ -236,10 +236,10 @@ func SnapshotSource(
 
 			mani, err := loadSnapshotFunc(ctx, rep, manifest.ID(parentSnapshot))
 			if err != nil {
-				return "", 0, errors.Wrapf(err, "Failed to load previous snapshot %v from kopia", parentSnapshot)
+				log.WithError(err).Warnf("Failed to load previous snapshot %v from kopia, fallback to full backup", parentSnapshot)
+			} else {
+				previous = append(previous, mani)
 			}
-
-			previous = append(previous, mani)
 		} else {
 			log.Infof("Searching for parent snapshot")
 

pkg/uploader/kopia/snapshot_test.go

@@ -112,7 +112,7 @@ func TestSnapshotSource(t *testing.T) {
 			notError: true,
 		},
 		{
-			name: "failed to load snapshot",
+			name: "failed to load snapshot, should fallback to full backup and not error",
 			args: []mockArgs{
 				{methodName: "LoadSnapshot", returns: []interface{}{manifest, errors.New("failed to load snapshot")}},
 				{methodName: "SaveSnapshot", returns: []interface{}{manifest.ID, nil}},
@@ -122,7 +122,7 @@ func TestSnapshotSource(t *testing.T) {
 				{methodName: "Upload", returns: []interface{}{manifest, nil}},
 				{methodName: "Flush", returns: []interface{}{nil}},
 			},
-			notError: false,
+			notError: true,
 		},
 		{
 			name: "failed to save snapshot",