Add changelog for v1.5.3

Signed-off-by: Bridget McErlean <bmcerlean@vmware.com>

changelogs/CHANGELOG-1.5.md

@@ -1,3 +1,45 @@
+## v1.5.3
+### 2021-01-14
+### Download
+https://github.com/vmware-tanzu/velero/releases/tag/v1.5.3
+
+### Container Image
+`velero/velero:v1.5.3`
+
+### Documentation
+https://velero.io/docs/v1.5/
+
+### Upgrading
+https://velero.io/docs/v1.5/upgrade-to-1.5/
+
+### All Changes
+  * Increased default Velero pod memory limit to 512Mi (#3234, @dsmithuchida)
+  * 🐛 BSLs with validation disabled should be validated at least once (#3084, @ashish-amarnath)
+  * Fixed an issue where the deletion of a backup would fail if the backup tarball couldn't be downloaded from object storage. Now the tarball is only downloaded if there are associated DeleteItemAction plugins and if downloading the tarball fails, the plugins are skipped. (#2993, @zubron)
+  * 🐛  ItemAction plugins for unresolvable types should not be run for all types (#3059, @ashish-amarnath)
+  * 🐛  Use namespace and name to match PVB to Pod restore (#3051, @ashish-amarnath)
+  * Allows the restic-wait container to exist in any order in the pod being restored. Prints a warning message in the case where the restic-wait container isn't the first container in the list of initialization containers. (#3011, @doughepi)
+
+## v1.5.2
+### 2020-10-20
+### Download
+https://github.com/vmware-tanzu/velero/releases/tag/v1.5.2
+
+### Container Image
+`velero/velero:v1.5.2`
+
+### Documentation
+https://velero.io/docs/v1.5/
+
+### Upgrading
+https://velero.io/docs/v1.5/upgrade-to-1.5/
+
+### All Changes
+  * Fix BSL controller to avoid invoking init() on all BSLs regardless of ValidationFrequency (#2992, @betta1)
+  * cli: allow creating multiple instances of Velero across two different namespaces (#2886, @alaypatel07)
+  * Restore CRD Resource name to fix CRD wait functionality. (#2949, @sseago)
+  * Ensure that bound PVCs and PVs remain bound on restore. (#3007, @nrb)
+
 ## v1.5.1
 ### 2020-09-16
 
@@ -80,3 +122,4 @@ Displays the Timestamps when issued a print or describe (#2748, @thejasbabu)
   * when creating new backup from schedule from cli, allow backup name to be automatically generated (#2569, @cblecker)
   * Convert manifests + BSL api client to kubebuilder (#2561, @carlisia)
   * backup/restore: reinstantiate backup store just before uploading artifacts to ensure credentials are up-to-date (#2550, @skriss)
+

internal/storage/storagelocation.go

@@ -51,24 +51,28 @@ func IsReadyToValidate(bslValidationFrequency *metav1.Duration, lastValidationTi
 		validationFrequency = bslValidationFrequency.Duration
 	}
 
-	if validationFrequency == 0 {
-		log.Debug("Validation period for this backup location is set to 0, skipping validation")
-		return false
-	}
-
 	if validationFrequency < 0 {
 		log.Debugf("Validation period must be non-negative, changing from %d to %d", validationFrequency, defaultLocationInfo.StoreValidationFrequency)
 		validationFrequency = defaultLocationInfo.StoreValidationFrequency
 	}
 
 	lastValidation := lastValidationTime
-	if lastValidation != nil { // always ready to validate the first time around, so only even do this check if this has happened before
-		nextValidation := lastValidation.Add(validationFrequency) // next validation time: last validation time + validation frequency
-		if time.Now().UTC().Before(nextValidation) {              // ready only when NOW is equal to or after the next validation time
-			return false
-		}
+	if lastValidation == nil {
+		// Regardless of validation frequency, we want to validate all BSLs at least once.
+		return true
 	}
 
+	if validationFrequency == 0 {
+		// Validation was disabled so return false.
+		log.Debug("Validation period for this backup location is set to 0, skipping validation")
+		return false
+	}
+
+	// We want to validate BSL only if the set validation frequency/ interval has elapsed.
+	nextValidation := lastValidation.Add(validationFrequency) // next validation time: last validation time + validation frequency
+	if time.Now().UTC().Before(nextValidation) {              // ready only when NOW is equal to or after the next validation time
+		return false
+	}
 	return true
 }
 

internal/storage/storagelocation_test.go

@@ -36,17 +36,23 @@ func TestIsReadyToValidate(t *testing.T) {
 		bslValidationFrequency *metav1.Duration
 		lastValidationTime     *metav1.Time
 		defaultLocationInfo    DefaultBackupLocationInfo
-
-		// serverDefaultValidationFrequency time.Duration
-		// backupLocation                   *velerov1api.BackupStorageLocation
-		ready bool
+		ready                  bool
 	}{
 		{
-			name:                   "don't validate, since frequency is set to zero",
+			name:                   "validate when true when validation frequency is zero and lastValidationTime is nil",
 			bslValidationFrequency: &metav1.Duration{Duration: 0},
 			defaultLocationInfo: DefaultBackupLocationInfo{
 				StoreValidationFrequency: 0,
 			},
+			ready: true,
+		},
+		{
+			name:                   "don't validate when false when validation is disabled and lastValidationTime is not nil",
+			bslValidationFrequency: &metav1.Duration{Duration: 0},
+			lastValidationTime:     &metav1.Time{Time: time.Now()},
+			defaultLocationInfo: DefaultBackupLocationInfo{
+				StoreValidationFrequency: 0,
+			},
 			ready: false,
 		},
 		{
@@ -63,7 +69,8 @@ func TestIsReadyToValidate(t *testing.T) {
 			defaultLocationInfo: DefaultBackupLocationInfo{
 				StoreValidationFrequency: 1,
 			},
-			ready: false,
+			lastValidationTime: &metav1.Time{Time: time.Now()},
+			ready:              false,
 		},
 		{
 			name: "validate as per default setting when location setting is not set",
@@ -77,7 +84,8 @@ func TestIsReadyToValidate(t *testing.T) {
 			defaultLocationInfo: DefaultBackupLocationInfo{
 				StoreValidationFrequency: 0,
 			},
-			ready: false,
+			lastValidationTime: &metav1.Time{Time: time.Now()},
+			ready:              false,
 		},
 		{
 			name:                   "don't validate when now is before the NEXT validation time (validation frequency + last validation time)",
@@ -112,8 +120,8 @@ func TestIsReadyToValidate(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			g := NewWithT(t)
 			log := velerotest.NewLogger()
-
-			g.Expect(IsReadyToValidate(tt.bslValidationFrequency, tt.lastValidationTime, tt.defaultLocationInfo, log)).To(BeIdenticalTo(tt.ready))
+			actual := IsReadyToValidate(tt.bslValidationFrequency, tt.lastValidationTime, tt.defaultLocationInfo, log)
+			g.Expect(actual).To(BeIdenticalTo(tt.ready))
 		})
 	}
 }

pkg/backup/backup.go

@@ -127,7 +127,7 @@ func resolveActions(actions []velero.BackupItemAction, helper discovery.Helper)
 			return nil, err
 		}
 
-		resources := getResourceIncludesExcludes(helper, resourceSelector.IncludedResources, resourceSelector.ExcludedResources)
+		resources := collections.GetResourceIncludesExcludes(helper, resourceSelector.IncludedResources, resourceSelector.ExcludedResources)
 		namespaces := collections.NewIncludesExcludes().Includes(resourceSelector.IncludedNamespaces...).Excludes(resourceSelector.ExcludedNamespaces...)
 
 		selector := labels.Everything()
@@ -150,30 +150,6 @@ func resolveActions(actions []velero.BackupItemAction, helper discovery.Helper)
 	return resolved, nil
 }
 
-// getResourceIncludesExcludes takes the lists of resources to include and exclude, uses the
-// discovery helper to resolve them to fully-qualified group-resource names, and returns an
-// IncludesExcludes list.
-func getResourceIncludesExcludes(helper discovery.Helper, includes, excludes []string) *collections.IncludesExcludes {
-	resources := collections.GenerateIncludesExcludes(
-		includes,
-		excludes,
-		func(item string) string {
-			gvr, _, err := helper.ResourceFor(schema.ParseGroupResource(item).WithVersion(""))
-			if err != nil {
-				// If we can't resolve it, return it as-is. This prevents the generated
-				// includes-excludes list from including *everything*, if none of the includes
-				// can be resolved. ref. https://github.com/vmware-tanzu/velero/issues/2461
-				return item
-			}
-
-			gr := gvr.GroupResource()
-			return gr.String()
-		},
-	)
-
-	return resources
-}
-
 // getNamespaceIncludesExcludes returns an IncludesExcludes list containing which namespaces to
 // include and exclude from the backup.
 func getNamespaceIncludesExcludes(backup *velerov1api.Backup) *collections.IncludesExcludes {
@@ -200,7 +176,7 @@ func getResourceHook(hookSpec velerov1api.BackupResourceHookSpec, discoveryHelpe
 		Name: hookSpec.Name,
 		Selector: hook.ResourceHookSelector{
 			Namespaces: collections.NewIncludesExcludes().Includes(hookSpec.IncludedNamespaces...).Excludes(hookSpec.ExcludedNamespaces...),
-			Resources:  getResourceIncludesExcludes(discoveryHelper, hookSpec.IncludedResources, hookSpec.ExcludedResources),
+			Resources:  collections.GetResourceIncludesExcludes(discoveryHelper, hookSpec.IncludedResources, hookSpec.ExcludedResources),
 		},
 		Pre:  hookSpec.PreHooks,
 		Post: hookSpec.PostHooks,
@@ -242,7 +218,7 @@ func (kb *kubernetesBackupper) Backup(log logrus.FieldLogger, backupRequest *Req
 	log.Infof("Including namespaces: %s", backupRequest.NamespaceIncludesExcludes.IncludesString())
 	log.Infof("Excluding namespaces: %s", backupRequest.NamespaceIncludesExcludes.ExcludesString())
 
-	backupRequest.ResourceIncludesExcludes = getResourceIncludesExcludes(kb.discoveryHelper, backupRequest.Spec.IncludedResources, backupRequest.Spec.ExcludedResources)
+	backupRequest.ResourceIncludesExcludes = collections.GetResourceIncludesExcludes(kb.discoveryHelper, backupRequest.Spec.IncludedResources, backupRequest.Spec.ExcludedResources)
 	log.Infof("Including resources: %s", backupRequest.ResourceIncludesExcludes.IncludesString())
 	log.Infof("Excluding resources: %s", backupRequest.ResourceIncludesExcludes.ExcludesString())
 	log.Infof("Backing up all pod volumes using restic: %t", *backupRequest.Backup.Spec.DefaultVolumesToRestic)

pkg/backup/remap_crd_version_action.go

@@ -31,7 +31,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 
 	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
-	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 )
 
@@ -111,7 +110,7 @@ func fetchV1beta1CRD(name string, betaCRDClient apiextv1beta1client.CustomResour
 	// See https://github.com/kubernetes/kubernetes/issues/3030. Unsure why this is happening here and not in main Velero;
 	// probably has to do with List calls and Dynamic client vs typed client
 	// Set these all the time, since they shouldn't ever be different, anyway
-	betaCRD.Kind = kuberesource.CustomResourceDefinitions.Resource
+	betaCRD.Kind = "CustomResourceDefinition"
 	betaCRD.APIVersion = apiextv1beta1.SchemeGroupVersion.String()
 
 	m, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&betaCRD)

pkg/builder/pod_volume_backup_builder.go

@@ -75,6 +75,12 @@ func (b *PodVolumeBackupBuilder) PodName(name string) *PodVolumeBackupBuilder {
 	return b
 }
 
+// PodNamespace sets the name of the pod associated with this PodVolumeBackup.
+func (b *PodVolumeBackupBuilder) PodNamespace(ns string) *PodVolumeBackupBuilder {
+	b.object.Spec.Pod.Namespace = ns
+	return b
+}
+
 // Volume sets the name of the volume associated with this PodVolumeBackup.
 func (b *PodVolumeBackupBuilder) Volume(volume string) *PodVolumeBackupBuilder {
 	b.object.Spec.Volume = volume

pkg/client/factory.go

@@ -155,6 +155,10 @@ func (f *factory) KubebuilderClient() (kbclient.Client, error) {
 		Scheme: scheme,
 	})
 
+	if err != nil {
+		return nil, err
+	}
+
 	return kubebuilderClient, nil
 }
 

pkg/controller/backup_controller.go

@@ -734,6 +734,10 @@ func persistBackup(backup *pkgbackup.Request,
 }
 
 func closeAndRemoveFile(file *os.File, log logrus.FieldLogger) {
+	if file == nil {
+		log.Debug("Skipping removal of file due to nil file pointer")
+		return
+	}
 	if err := file.Close(); err != nil {
 		log.WithError(err).WithField("file", file.Name()).Error("error closing file")
 	}

pkg/controller/backup_deletion_controller.go

@@ -295,13 +295,6 @@ func (c *backupDeletionController) processRequest(req *velerov1api.DeleteBackupR
 		errs = append(errs, err.Error())
 	}
 
-	// Download the tarball
-	backupFile, err := downloadToTempFile(backup.Name, backupStore, log)
-	if err != nil {
-		return errors.Wrap(err, "error downloading backup")
-	}
-	defer closeAndRemoveFile(backupFile, c.logger)
-
 	actions, err := pluginManager.GetDeleteItemActions()
 	log.Debugf("%d actions before invoking actions", len(actions))
 	if err != nil {
@@ -309,20 +302,30 @@ func (c *backupDeletionController) processRequest(req *velerov1api.DeleteBackupR
 	}
 	// don't defer CleanupClients here, since it was already called above.
 
-	ctx := &delete.Context{
-		Backup:          backup,
-		BackupReader:    backupFile,
-		Actions:         actions,
-		Log:             c.logger,
-		DiscoveryHelper: c.helper,
-		Filesystem:      filesystem.NewFileSystem(),
-	}
+	if len(actions) > 0 {
+		// Download the tarball
+		backupFile, err := downloadToTempFile(backup.Name, backupStore, log)
+		defer closeAndRemoveFile(backupFile, c.logger)
 
-	// Optimization: wrap in a gofunc? Would be useful for large backups with lots of objects.
-	// but what do we do with the error returned? We can't just swallow it as that may lead to dangling resources.
-	err = delete.InvokeDeleteActions(ctx)
-	if err != nil {
-		return errors.Wrap(err, "error invoking delete item actions")
+		if err != nil {
+			log.WithError(err).Errorf("Unable to download tarball for backup %s, skipping associated DeleteItemAction plugins", backup.Name)
+		} else {
+			ctx := &delete.Context{
+				Backup:          backup,
+				BackupReader:    backupFile,
+				Actions:         actions,
+				Log:             c.logger,
+				DiscoveryHelper: c.helper,
+				Filesystem:      filesystem.NewFileSystem(),
+			}
+
+			// Optimization: wrap in a gofunc? Would be useful for large backups with lots of objects.
+			// but what do we do with the error returned? We can't just swallow it as that may lead to dangling resources.
+			err = delete.InvokeDeleteActions(ctx)
+			if err != nil {
+				return errors.Wrap(err, "error invoking delete item actions")
+			}
+		}
 	}
 
 	if backupStore != nil {

pkg/controller/backup_deletion_controller_test.go

@@ -49,6 +49,8 @@ import (
 	persistencemocks "github.com/vmware-tanzu/velero/pkg/persistence/mocks"
 	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt"
 	pluginmocks "github.com/vmware-tanzu/velero/pkg/plugin/mocks"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero/mocks"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	"github.com/vmware-tanzu/velero/pkg/volume"
 )
@@ -739,6 +741,265 @@ func TestBackupDeletionControllerProcessRequest(t *testing.T) {
 		// Make sure snapshot was deleted
 		assert.Equal(t, 0, td.volumeSnapshotter.SnapshotsTaken.Len())
 	})
+
+	t.Run("backup is not downloaded when there are no DeleteItemAction plugins", func(t *testing.T) {
+		backup := builder.ForBackup(velerov1api.DefaultNamespace, "foo").Result()
+		backup.UID = "uid"
+		backup.Spec.StorageLocation = "primary"
+
+		td := setupBackupDeletionControllerTest(t, backup)
+
+		location := &velerov1api.BackupStorageLocation{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: backup.Namespace,
+				Name:      backup.Spec.StorageLocation,
+			},
+			Spec: velerov1api.BackupStorageLocationSpec{
+				Provider: "objStoreProvider",
+				StorageType: velerov1api.StorageType{
+					ObjectStorage: &velerov1api.ObjectStorageLocation{
+						Bucket: "bucket",
+					},
+				},
+			},
+		}
+
+		require.NoError(t, td.fakeClient.Create(context.Background(), location))
+
+		snapshotLocation := &velerov1api.VolumeSnapshotLocation{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: backup.Namespace,
+				Name:      "vsl-1",
+			},
+			Spec: velerov1api.VolumeSnapshotLocationSpec{
+				Provider: "provider-1",
+			},
+		}
+		require.NoError(t, td.sharedInformers.Velero().V1().VolumeSnapshotLocations().Informer().GetStore().Add(snapshotLocation))
+
+		// Clear out req labels to make sure the controller adds them and does not
+		// panic when encountering a nil Labels map
+		// (https://github.com/vmware-tanzu/velero/issues/1546)
+		td.req.Labels = nil
+
+		td.client.PrependReactor("get", "backups", func(action core.Action) (bool, runtime.Object, error) {
+			return true, backup, nil
+		})
+		td.volumeSnapshotter.SnapshotsTaken.Insert("snap-1")
+
+		td.client.PrependReactor("patch", "deletebackuprequests", func(action core.Action) (bool, runtime.Object, error) {
+			return true, td.req, nil
+		})
+
+		td.client.PrependReactor("patch", "backups", func(action core.Action) (bool, runtime.Object, error) {
+			return true, backup, nil
+		})
+
+		snapshots := []*volume.Snapshot{
+			{
+				Spec: volume.SnapshotSpec{
+					Location: "vsl-1",
+				},
+				Status: volume.SnapshotStatus{
+					ProviderSnapshotID: "snap-1",
+				},
+			},
+		}
+
+		pluginManager := &pluginmocks.Manager{}
+		pluginManager.On("GetVolumeSnapshotter", "provider-1").Return(td.volumeSnapshotter, nil)
+		pluginManager.On("GetDeleteItemActions").Return([]velero.DeleteItemAction{}, nil)
+		pluginManager.On("CleanupClients")
+		td.controller.newPluginManager = func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager }
+
+		td.backupStore.On("GetBackupVolumeSnapshots", td.req.Spec.BackupName).Return(snapshots, nil)
+		td.backupStore.On("DeleteBackup", td.req.Spec.BackupName).Return(nil)
+
+		err := td.controller.processRequest(td.req)
+		require.NoError(t, err)
+
+		td.backupStore.AssertNotCalled(t, "GetBackupContents", td.req.Spec.BackupName)
+
+		expectedActions := []core.Action{
+			core.NewPatchAction(
+				velerov1api.SchemeGroupVersion.WithResource("deletebackuprequests"),
+				td.req.Namespace,
+				td.req.Name,
+				types.MergePatchType,
+				[]byte(`{"metadata":{"labels":{"velero.io/backup-name":"foo"}},"status":{"phase":"InProgress"}}`),
+			),
+			core.NewGetAction(
+				velerov1api.SchemeGroupVersion.WithResource("backups"),
+				td.req.Namespace,
+				td.req.Spec.BackupName,
+			),
+			core.NewPatchAction(
+				velerov1api.SchemeGroupVersion.WithResource("deletebackuprequests"),
+				td.req.Namespace,
+				td.req.Name,
+				types.MergePatchType,
+				[]byte(`{"metadata":{"labels":{"velero.io/backup-uid":"uid"}}}`),
+			),
+			core.NewPatchAction(
+				velerov1api.SchemeGroupVersion.WithResource("backups"),
+				td.req.Namespace,
+				td.req.Spec.BackupName,
+				types.MergePatchType,
+				[]byte(`{"status":{"phase":"Deleting"}}`),
+			),
+			core.NewDeleteAction(
+				velerov1api.SchemeGroupVersion.WithResource("backups"),
+				td.req.Namespace,
+				td.req.Spec.BackupName,
+			),
+			core.NewPatchAction(
+				velerov1api.SchemeGroupVersion.WithResource("deletebackuprequests"),
+				td.req.Namespace,
+				td.req.Name,
+				types.MergePatchType,
+				[]byte(`{"status":{"phase":"Processed"}}`),
+			),
+			core.NewDeleteCollectionAction(
+				velerov1api.SchemeGroupVersion.WithResource("deletebackuprequests"),
+				td.req.Namespace,
+				pkgbackup.NewDeleteBackupRequestListOptions(td.req.Spec.BackupName, "uid"),
+			),
+		}
+
+		velerotest.CompareActions(t, expectedActions, td.client.Actions())
+
+		// Make sure snapshot was deleted
+		assert.Equal(t, 0, td.volumeSnapshotter.SnapshotsTaken.Len())
+	})
+
+	t.Run("backup is still deleted if downloading tarball fails for DeleteItemAction plugins", func(t *testing.T) {
+		backup := builder.ForBackup(velerov1api.DefaultNamespace, "foo").Result()
+		backup.UID = "uid"
+		backup.Spec.StorageLocation = "primary"
+
+		td := setupBackupDeletionControllerTest(t, backup)
+
+		location := &velerov1api.BackupStorageLocation{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: backup.Namespace,
+				Name:      backup.Spec.StorageLocation,
+			},
+			Spec: velerov1api.BackupStorageLocationSpec{
+				Provider: "objStoreProvider",
+				StorageType: velerov1api.StorageType{
+					ObjectStorage: &velerov1api.ObjectStorageLocation{
+						Bucket: "bucket",
+					},
+				},
+			},
+		}
+
+		require.NoError(t, td.fakeClient.Create(context.Background(), location))
+
+		snapshotLocation := &velerov1api.VolumeSnapshotLocation{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: backup.Namespace,
+				Name:      "vsl-1",
+			},
+			Spec: velerov1api.VolumeSnapshotLocationSpec{
+				Provider: "provider-1",
+			},
+		}
+		require.NoError(t, td.sharedInformers.Velero().V1().VolumeSnapshotLocations().Informer().GetStore().Add(snapshotLocation))
+
+		// Clear out req labels to make sure the controller adds them and does not
+		// panic when encountering a nil Labels map
+		// (https://github.com/vmware-tanzu/velero/issues/1546)
+		td.req.Labels = nil
+
+		td.client.PrependReactor("get", "backups", func(action core.Action) (bool, runtime.Object, error) {
+			return true, backup, nil
+		})
+		td.volumeSnapshotter.SnapshotsTaken.Insert("snap-1")
+
+		td.client.PrependReactor("patch", "deletebackuprequests", func(action core.Action) (bool, runtime.Object, error) {
+			return true, td.req, nil
+		})
+
+		td.client.PrependReactor("patch", "backups", func(action core.Action) (bool, runtime.Object, error) {
+			return true, backup, nil
+		})
+
+		snapshots := []*volume.Snapshot{
+			{
+				Spec: volume.SnapshotSpec{
+					Location: "vsl-1",
+				},
+				Status: volume.SnapshotStatus{
+					ProviderSnapshotID: "snap-1",
+				},
+			},
+		}
+
+		pluginManager := &pluginmocks.Manager{}
+		pluginManager.On("GetVolumeSnapshotter", "provider-1").Return(td.volumeSnapshotter, nil)
+		pluginManager.On("GetDeleteItemActions").Return([]velero.DeleteItemAction{new(mocks.DeleteItemAction)}, nil)
+		pluginManager.On("CleanupClients")
+		td.controller.newPluginManager = func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager }
+
+		td.backupStore.On("GetBackupVolumeSnapshots", td.req.Spec.BackupName).Return(snapshots, nil)
+		td.backupStore.On("GetBackupContents", td.req.Spec.BackupName).Return(nil, fmt.Errorf("error downloading tarball"))
+		td.backupStore.On("DeleteBackup", td.req.Spec.BackupName).Return(nil)
+
+		err := td.controller.processRequest(td.req)
+		require.NoError(t, err)
+
+		expectedActions := []core.Action{
+			core.NewPatchAction(
+				velerov1api.SchemeGroupVersion.WithResource("deletebackuprequests"),
+				td.req.Namespace,
+				td.req.Name,
+				types.MergePatchType,
+				[]byte(`{"metadata":{"labels":{"velero.io/backup-name":"foo"}},"status":{"phase":"InProgress"}}`),
+			),
+			core.NewGetAction(
+				velerov1api.SchemeGroupVersion.WithResource("backups"),
+				td.req.Namespace,
+				td.req.Spec.BackupName,
+			),
+			core.NewPatchAction(
+				velerov1api.SchemeGroupVersion.WithResource("deletebackuprequests"),
+				td.req.Namespace,
+				td.req.Name,
+				types.MergePatchType,
+				[]byte(`{"metadata":{"labels":{"velero.io/backup-uid":"uid"}}}`),
+			),
+			core.NewPatchAction(
+				velerov1api.SchemeGroupVersion.WithResource("backups"),
+				td.req.Namespace,
+				td.req.Spec.BackupName,
+				types.MergePatchType,
+				[]byte(`{"status":{"phase":"Deleting"}}`),
+			),
+			core.NewDeleteAction(
+				velerov1api.SchemeGroupVersion.WithResource("backups"),
+				td.req.Namespace,
+				td.req.Spec.BackupName,
+			),
+			core.NewPatchAction(
+				velerov1api.SchemeGroupVersion.WithResource("deletebackuprequests"),
+				td.req.Namespace,
+				td.req.Name,
+				types.MergePatchType,
+				[]byte(`{"status":{"phase":"Processed"}}`),
+			),
+			core.NewDeleteCollectionAction(
+				velerov1api.SchemeGroupVersion.WithResource("deletebackuprequests"),
+				td.req.Namespace,
+				pkgbackup.NewDeleteBackupRequestListOptions(td.req.Spec.BackupName, "uid"),
+			),
+		}
+
+		velerotest.CompareActions(t, expectedActions, td.client.Actions())
+
+		// Make sure snapshot was deleted
+		assert.Equal(t, 0, td.volumeSnapshotter.SnapshotsTaken.Len())
+	})
 }
 
 func TestBackupDeletionControllerDeleteExpiredRequests(t *testing.T) {

pkg/controller/backupstoragelocation_controller.go

@@ -77,14 +77,14 @@ func (r *BackupStorageLocationReconciler) Reconcile(req ctrl.Request) (ctrl.Resu
 			defaultFound = true
 		}
 
-		backupStore, err := r.NewBackupStore(location, pluginManager, log)
-		if err != nil {
-			log.WithError(err).Error("Error getting a backup store")
+		if !storage.IsReadyToValidate(location.Spec.ValidationFrequency, location.Status.LastValidationTime, r.DefaultBackupLocationInfo, log) {
+			log.Debug("Backup location not ready to be validated")
 			continue
 		}
 
-		if !storage.IsReadyToValidate(location.Spec.ValidationFrequency, location.Status.LastValidationTime, r.DefaultBackupLocationInfo, log) {
-			log.Debug("Backup location not ready to be validated")
+		backupStore, err := r.NewBackupStore(location, pluginManager, log)
+		if err != nil {
+			log.WithError(err).Error("Error getting a backup store")
 			continue
 		}
 

pkg/controller/backupstoragelocation_controller_test.go

@@ -120,13 +120,13 @@ var _ = Describe("Backup Storage Location Reconciler", func() {
 			wantErr        bool
 		}{
 			{
-				backupLocation: builder.ForBackupStorageLocation("ns-1", "location-1").ValidationFrequency(0).Result(),
+				backupLocation: builder.ForBackupStorageLocation("ns-1", "location-1").ValidationFrequency(0).LastValidationTime(time.Now()).Result(),
 				isValidError:   nil,
 				expectedPhase:  "",
 				wantErr:        false,
 			},
 			{
-				backupLocation: builder.ForBackupStorageLocation("ns-1", "location-2").ValidationFrequency(0).Result(),
+				backupLocation: builder.ForBackupStorageLocation("ns-1", "location-2").ValidationFrequency(0).LastValidationTime(time.Now()).Result(),
 				isValidError:   nil,
 				expectedPhase:  "",
 				wantErr:        false,

pkg/controller/pod_volume_restore_controller.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2018 the Velero contributors.
+Copyright 2020 the Velero contributors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -155,6 +155,12 @@ func (c *podVolumeRestoreController) pvrHandler(obj interface{}) {
 		return
 	}
 
+	resticInitContainerIndex := getResticInitContainerIndex(pod)
+	if resticInitContainerIndex > 0 {
+		log.Warnf(`Init containers before the %s container may cause issues
+		          if they interfere with volumes being restored: %s index %d`, restic.InitContainer, restic.InitContainer, resticInitContainerIndex)
+	}
+
 	log.Debug("Enqueueing")
 	c.enqueue(obj)
 }
@@ -174,6 +180,12 @@ func (c *podVolumeRestoreController) podHandler(obj interface{}) {
 		return
 	}
 
+	resticInitContainerIndex := getResticInitContainerIndex(pod)
+	if resticInitContainerIndex > 0 {
+		log.Warnf(`Init containers before the %s container may cause issues
+		          if they interfere with volumes being restored: %s index %d`, restic.InitContainer, restic.InitContainer, resticInitContainerIndex)
+	}
+
 	selector := labels.Set(map[string]string{
 		velerov1api.PodUIDLabel: string(pod.UID),
 	}).AsSelector()
@@ -208,18 +220,21 @@ func isPodOnNode(pod *corev1api.Pod, node string) bool {
 }
 
 func isResticInitContainerRunning(pod *corev1api.Pod) bool {
-	// no init containers, or the first one is not the velero restic one: return false
-	if len(pod.Spec.InitContainers) == 0 || pod.Spec.InitContainers[0].Name != restic.InitContainer {
-		return false
+
+	// Restic wait container can be anywhere in the list of init containers, but must be running.
+	i := getResticInitContainerIndex(pod)
+	return i >= 0 && pod.Status.InitContainerStatuses[i].State.Running != nil
+}
+
+func getResticInitContainerIndex(pod *corev1api.Pod) int {
+	// Restic wait container can be anywhere in the list of init containers so locate it.
+	for i, initContainer := range pod.Spec.InitContainers {
+		if initContainer.Name == restic.InitContainer {
+			return i
+		}
 	}
 
-	// status hasn't been created yet, or the first one is not yet running: return false
-	if len(pod.Status.InitContainerStatuses) == 0 || pod.Status.InitContainerStatuses[0].State.Running == nil {
-		return false
-	}
-
-	// else, it's running
-	return true
+	return -1
 }
 
 func (c *podVolumeRestoreController) processQueueItem(key string) error {

pkg/controller/pod_volume_restore_controller_test.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2018 the Velero contributors.
+Copyright 2020 the Velero contributors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -491,7 +491,7 @@ func TestIsResticContainerRunning(t *testing.T) {
 			expected: false,
 		},
 		{
-			name: "pod with running restic init container that's not first should return false",
+			name: "pod with running restic init container that's not first should still work",
 			pod: &corev1api.Pod{
 				ObjectMeta: metav1.ObjectMeta{
 					Namespace: "ns-1",
@@ -522,7 +522,7 @@ func TestIsResticContainerRunning(t *testing.T) {
 					},
 				},
 			},
-			expected: false,
+			expected: true,
 		},
 		{
 			name: "pod with restic init container as first initContainer that's not running should return false",
@@ -598,3 +598,105 @@ func TestIsResticContainerRunning(t *testing.T) {
 		})
 	}
 }
+
+func TestGetResticInitContainerIndex(t *testing.T) {
+	tests := []struct {
+		name     string
+		pod      *corev1api.Pod
+		expected int
+	}{
+		{
+			name: "init container is not present return -1",
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns-1",
+					Name:      "pod-1",
+				},
+			},
+			expected: -1,
+		},
+		{
+			name: "pod with no restic init container return -1",
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns-1",
+					Name:      "pod-1",
+				},
+				Spec: corev1api.PodSpec{
+					InitContainers: []corev1api.Container{
+						{
+							Name: "non-restic-init",
+						},
+					},
+				},
+			},
+			expected: -1,
+		},
+		{
+			name: "pod with restic container as second initContainern should return 1",
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns-1",
+					Name:      "pod-1",
+				},
+				Spec: corev1api.PodSpec{
+					InitContainers: []corev1api.Container{
+						{
+							Name: "non-restic-init",
+						},
+						{
+							Name: restic.InitContainer,
+						},
+					},
+				},
+			},
+			expected: 1,
+		},
+		{
+			name: "pod with restic init container as first initContainer should return 0",
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns-1",
+					Name:      "pod-1",
+				},
+				Spec: corev1api.PodSpec{
+					InitContainers: []corev1api.Container{
+						{
+							Name: restic.InitContainer,
+						},
+						{
+							Name: "non-restic-init",
+						},
+					},
+				},
+			},
+			expected: 0,
+		},
+		{
+			name: "pod with restic init container as first initContainer should return 0",
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns-1",
+					Name:      "pod-1",
+				},
+				Spec: corev1api.PodSpec{
+					InitContainers: []corev1api.Container{
+						{
+							Name: restic.InitContainer,
+						},
+						{
+							Name: "non-restic-init",
+						},
+					},
+				},
+			},
+			expected: 0,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			assert.Equal(t, test.expected, getResticInitContainerIndex(test.pod))
+		})
+	}
+}

pkg/install/resources.go

@@ -46,11 +46,12 @@ var (
 	DefaultVeleroPodCPURequest = "500m"
 	DefaultVeleroPodMemRequest = "128Mi"
 	DefaultVeleroPodCPULimit   = "1000m"
-	DefaultVeleroPodMemLimit   = "256Mi"
+	DefaultVeleroPodMemLimit   = "512Mi"
 	DefaultResticPodCPURequest = "500m"
 	DefaultResticPodMemRequest = "512Mi"
 	DefaultResticPodCPULimit   = "1000m"
 	DefaultResticPodMemLimit   = "1Gi"
+	DefaultVeleroNamespace     = "velero"
 )
 
 func labels() map[string]string {
@@ -105,8 +106,12 @@ func ServiceAccount(namespace string, annotations map[string]string) *corev1.Ser
 }
 
 func ClusterRoleBinding(namespace string) *rbacv1beta1.ClusterRoleBinding {
+	crbName := "velero"
+	if namespace != DefaultVeleroNamespace {
+		crbName = "velero-" + namespace
+	}
 	crb := &rbacv1beta1.ClusterRoleBinding{
-		ObjectMeta: objectMeta("", "velero"),
+		ObjectMeta: objectMeta("", crbName),
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "ClusterRoleBinding",
 			APIVersion: rbacv1beta1.SchemeGroupVersion.String(),

pkg/install/resources_test.go

@@ -23,7 +23,7 @@ import (
 )
 
 func TestResources(t *testing.T) {
-	bsl := BackupStorageLocation("velero", "test", "test", "", make(map[string]string), []byte("test"))
+	bsl := BackupStorageLocation(DefaultVeleroNamespace, "test", "test", "", make(map[string]string), []byte("test"))
 
 	assert.Equal(t, "velero", bsl.ObjectMeta.Namespace)
 	assert.Equal(t, "test", bsl.Spec.Provider)
@@ -31,7 +31,7 @@ func TestResources(t *testing.T) {
 	assert.Equal(t, make(map[string]string), bsl.Spec.Config)
 	assert.Equal(t, []byte("test"), bsl.Spec.ObjectStorage.CACert)
 
-	vsl := VolumeSnapshotLocation("velero", "test", make(map[string]string))
+	vsl := VolumeSnapshotLocation(DefaultVeleroNamespace, "test", make(map[string]string))
 
 	assert.Equal(t, "velero", vsl.ObjectMeta.Namespace)
 	assert.Equal(t, "test", vsl.Spec.Provider)
@@ -41,12 +41,19 @@ func TestResources(t *testing.T) {
 
 	assert.Equal(t, "velero", ns.Name)
 
-	crb := ClusterRoleBinding("velero")
+	crb := ClusterRoleBinding(DefaultVeleroNamespace)
 	// The CRB is a cluster-scoped resource
 	assert.Equal(t, "", crb.ObjectMeta.Namespace)
+	assert.Equal(t, "velero", crb.ObjectMeta.Name)
 	assert.Equal(t, "velero", crb.Subjects[0].Namespace)
 
-	sa := ServiceAccount("velero", map[string]string{"abcd": "cbd"})
+	customNamespaceCRB := ClusterRoleBinding("foo")
+	// The CRB is a cluster-scoped resource
+	assert.Equal(t, "", customNamespaceCRB.ObjectMeta.Namespace)
+	assert.Equal(t, "velero-foo", customNamespaceCRB.ObjectMeta.Name)
+	assert.Equal(t, "foo", customNamespaceCRB.Subjects[0].Namespace)
+
+	sa := ServiceAccount(DefaultVeleroNamespace, map[string]string{"abcd": "cbd"})
 	assert.Equal(t, "velero", sa.ObjectMeta.Namespace)
 	assert.Equal(t, "cbd", sa.ObjectMeta.Annotations["abcd"])
 }

pkg/kuberesource/kuberesource.go

@@ -23,7 +23,7 @@ import (
 var (
 	ClusterRoleBindings       = schema.GroupResource{Group: "rbac.authorization.k8s.io", Resource: "clusterrolebindings"}
 	ClusterRoles              = schema.GroupResource{Group: "rbac.authorization.k8s.io", Resource: "clusterroles"}
-	CustomResourceDefinitions = schema.GroupResource{Group: "apiextensions.k8s.io", Resource: "CustomResourceDefinition"}
+	CustomResourceDefinitions = schema.GroupResource{Group: "apiextensions.k8s.io", Resource: "customresourcedefinitions"}
 	Jobs                      = schema.GroupResource{Group: "batch", Resource: "jobs"}
 	Namespaces                = schema.GroupResource{Group: "", Resource: "namespaces"}
 	PersistentVolumeClaims    = schema.GroupResource{Group: "", Resource: "persistentvolumeclaims"}

pkg/restic/common.go

@@ -95,13 +95,17 @@ func getPodSnapshotAnnotations(obj metav1.Object) map[string]string {
 	return res
 }
 
+func isPVBMatchPod(pvb *velerov1api.PodVolumeBackup, pod metav1.Object) bool {
+	return pod.GetName() == pvb.Spec.Pod.Name && pod.GetNamespace() == pvb.Spec.Pod.Namespace
+}
+
 // GetVolumeBackupsForPod returns a map, of volume name -> snapshot id,
 // of the PodVolumeBackups that exist for the provided pod.
 func GetVolumeBackupsForPod(podVolumeBackups []*velerov1api.PodVolumeBackup, pod metav1.Object) map[string]string {
 	volumes := make(map[string]string)
 
 	for _, pvb := range podVolumeBackups {
-		if pod.GetName() != pvb.Spec.Pod.Name {
+		if !isPVBMatchPod(pvb, pod) {
 			continue
 		}
 

pkg/restic/common_test.go

@@ -564,6 +564,88 @@ func TestGetPodVolumesUsingRestic(t *testing.T) {
 	}
 }
 
+func TestIsPVBMatchPod(t *testing.T) {
+	testCases := []struct {
+		name     string
+		pod      metav1.Object
+		pvb      velerov1api.PodVolumeBackup
+		expected bool
+	}{
+		{
+			name: "should match PVB and pod",
+			pod: &metav1.ObjectMeta{
+				Name:      "matching-pod",
+				Namespace: "matching-namespace",
+			},
+			pvb: velerov1api.PodVolumeBackup{
+				Spec: velerov1api.PodVolumeBackupSpec{
+					Pod: corev1api.ObjectReference{
+						Name:      "matching-pod",
+						Namespace: "matching-namespace",
+					},
+				},
+			},
+			expected: true,
+		},
+		{
+			name: "should not match PVB and pod, pod name mismatch",
+			pod: &metav1.ObjectMeta{
+				Name:      "not-matching-pod",
+				Namespace: "matching-namespace",
+			},
+			pvb: velerov1api.PodVolumeBackup{
+				Spec: velerov1api.PodVolumeBackupSpec{
+					Pod: corev1api.ObjectReference{
+						Name:      "matching-pod",
+						Namespace: "matching-namespace",
+					},
+				},
+			},
+			expected: false,
+		},
+		{
+			name: "should not match PVB and pod, pod namespace mismatch",
+			pod: &metav1.ObjectMeta{
+				Name:      "matching-pod",
+				Namespace: "not-matching-namespace",
+			},
+			pvb: velerov1api.PodVolumeBackup{
+				Spec: velerov1api.PodVolumeBackupSpec{
+					Pod: corev1api.ObjectReference{
+						Name:      "matching-pod",
+						Namespace: "matching-namespace",
+					},
+				},
+			},
+			expected: false,
+		},
+		{
+			name: "should not match PVB and pod, pod name and namespace mismatch",
+			pod: &metav1.ObjectMeta{
+				Name:      "not-matching-pod",
+				Namespace: "not-matching-namespace",
+			},
+			pvb: velerov1api.PodVolumeBackup{
+				Spec: velerov1api.PodVolumeBackupSpec{
+					Pod: corev1api.ObjectReference{
+						Name:      "matching-pod",
+						Namespace: "matching-namespace",
+					},
+				},
+			},
+			expected: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			actual := isPVBMatchPod(&tc.pvb, tc.pod)
+			assert.Equal(t, tc.expected, actual)
+		})
+
+	}
+}
+
 func newFakeClient(t *testing.T, initObjs ...runtime.Object) client.Client {
 	err := velerov1api.AddToScheme(scheme.Scheme)
 	require.NoError(t, err)

pkg/restore/pv_restorer.go

@@ -47,20 +47,6 @@ func (r *pvRestorer) executePVAction(obj *unstructured.Unstructured) (*unstructu
 		return nil, errors.New("PersistentVolume is missing its name")
 	}
 
-	// It's simpler to just access the spec through the unstructured object than to convert
-	// to structured and back here, especially since the SetVolumeID(...) call below needs
-	// the unstructured representation (and does a conversion internally).
-	res, ok := obj.Object["spec"]
-	if !ok {
-		return nil, errors.New("spec not found")
-	}
-	spec, ok := res.(map[string]interface{})
-	if !ok {
-		return nil, errors.Errorf("spec was of type %T, expected map[string]interface{}", res)
-	}
-
-	delete(spec, "claimRef")
-
 	if boolptr.IsSetToFalse(r.snapshotVolumes) {
 		// The backup had snapshots disabled, so we can return early
 		return obj, nil

pkg/restore/pv_restorer_test.go

@@ -56,19 +56,6 @@ func TestExecutePVAction_NoSnapshotRestores(t *testing.T) {
 			restore:     builder.ForRestore(api.DefaultNamespace, "").Result(),
 			expectedErr: true,
 		},
-		{
-			name:        "no spec should error",
-			obj:         NewTestUnstructured().WithName("pv-1").Unstructured,
-			restore:     builder.ForRestore(api.DefaultNamespace, "").Result(),
-			expectedErr: true,
-		},
-		{
-			name:        "ensure spec.claimRef is deleted",
-			obj:         NewTestUnstructured().WithName("pv-1").WithAnnotations("a", "b").WithSpec("claimRef", "someOtherField").Unstructured,
-			restore:     builder.ForRestore(api.DefaultNamespace, "").RestorePVs(false).Result(),
-			backup:      defaultBackup().Phase(api.BackupPhaseInProgress).Result(),
-			expectedRes: NewTestUnstructured().WithAnnotations("a", "b").WithName("pv-1").WithSpec("someOtherField").Unstructured,
-		},
 		{
 			name:        "ensure spec.storageClassName is retained",
 			obj:         NewTestUnstructured().WithName("pv-1").WithAnnotations("a", "b").WithSpec("storageClassName", "someOtherField").Unstructured,
@@ -81,7 +68,7 @@ func TestExecutePVAction_NoSnapshotRestores(t *testing.T) {
 			obj:         NewTestUnstructured().WithName("pv-1").WithAnnotations("a", "b").WithSpec("claimRef", "storageClassName", "someOtherField").Unstructured,
 			restore:     builder.ForRestore(api.DefaultNamespace, "").RestorePVs(true).Result(),
 			backup:      defaultBackup().Phase(api.BackupPhaseInProgress).SnapshotVolumes(false).Result(),
-			expectedRes: NewTestUnstructured().WithName("pv-1").WithAnnotations("a", "b").WithSpec("storageClassName", "someOtherField").Unstructured,
+			expectedRes: NewTestUnstructured().WithName("pv-1").WithAnnotations("a", "b").WithSpec("claimRef", "storageClassName", "someOtherField").Unstructured,
 		},
 		{
 			name:    "restore.spec.restorePVs=false, return early",

pkg/restore/restic_restore_action_test.go

@@ -173,12 +173,14 @@ func TestResticRestoreActionExecute(t *testing.T) {
 			podVolumeBackups: []*velerov1api.PodVolumeBackup{
 				builder.ForPodVolumeBackup(veleroNs, "pvb-1").
 					PodName("my-pod").
+					PodNamespace("ns-1").
 					Volume("vol-1").
 					ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, backupName)).
 					SnapshotID("foo").
 					Result(),
 				builder.ForPodVolumeBackup(veleroNs, "pvb-2").
 					PodName("my-pod").
+					PodNamespace("ns-1").
 					Volume("vol-2").
 					ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, backupName)).
 					SnapshotID("foo").

pkg/restore/restore.go

@@ -62,6 +62,14 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/volume"
 )
 
+// These annotations are taken from the Kubernetes persistent volume/persistent volume claim controller.
+// They cannot be directly importing because they are part of the kubernetes/kubernetes package, and importing that package is unsupported.
+// Their values are well-known and slow changing. They're duplicated here as constants to provide compile-time checking.
+// Originals can be found in kubernetes/kubernetes/pkg/controller/volume/persistentvolume/util/util.go.
+const KubeAnnBindCompleted = "pv.kubernetes.io/bind-completed"
+const KubeAnnBoundByController = "pv.kubernetes.io/bound-by-controller"
+const KubeAnnDynamicallyProvisioned = "pv.kubernetes.io/provisioned-by"
+
 type VolumeSnapshotterGetter interface {
 	GetVolumeSnapshotter(name string) (velero.VolumeSnapshotter, error)
 }
@@ -882,6 +890,13 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 				return warnings, errs
 			}
 
+			// Check to see if the claimRef.namespace field needs to be remapped, and do so if necessary.
+			_, err = remapClaimRefNS(ctx, obj)
+			if err != nil {
+				errs.Add(namespace, err)
+				return warnings, errs
+			}
+
 			var shouldRestoreSnapshot bool
 			if !shouldRenamePV {
 				// Check if the PV exists in the cluster before attempting to create
@@ -899,6 +914,9 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 			}
 
 			if shouldRestoreSnapshot {
+				// reset the PV's binding status so that Kubernetes can properly associate it with the restored PVC.
+				obj = resetVolumeBindingInfo(obj)
+
 				// even if we're renaming the PV, obj still has the old name here, because the pvRestorer
 				// uses the original name to look up metadata about the snapshot.
 				ctx.log.Infof("Restoring persistent volume from snapshot.")
@@ -958,8 +976,9 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 		default:
 			ctx.log.Infof("Restoring persistent volume as-is because it doesn't have a snapshot and its reclaim policy is not Delete.")
 
-			// we call the pvRestorer here to clear out the PV's claimRef, so it can be re-claimed
-			// when its PVC is restored.
+			obj = resetVolumeBindingInfo(obj)
+			// we call the pvRestorer here to clear out the PV's claimRef.UID, so it can be re-claimed
+			// when its PVC is restored and gets a new UID.
 			updatedObj, err := ctx.pvRestorer.executePVAction(obj)
 			if err != nil {
 				errs.Add(namespace, fmt.Errorf("error executing PVAction for %s: %v", resourceID, err))
@@ -1052,17 +1071,16 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 			return warnings, errs
 		}
 
-		if pvc.Spec.VolumeName != "" && ctx.pvsToProvision.Has(pvc.Spec.VolumeName) {
-			ctx.log.Infof("Resetting PersistentVolumeClaim %s/%s for dynamic provisioning", namespace, name)
+		if pvc.Spec.VolumeName != "" {
+			// This used to only happen with restic volumes, but now always remove this binding metadata
+			obj = resetVolumeBindingInfo(obj)
 
-			// use the unstructured helpers here since we're only deleting and
-			// the unstructured converter will add back (empty) fields for metadata
-			// and status that we removed earlier.
-			unstructured.RemoveNestedField(obj.Object, "spec", "volumeName")
-			annotations := obj.GetAnnotations()
-			delete(annotations, "pv.kubernetes.io/bind-completed")
-			delete(annotations, "pv.kubernetes.io/bound-by-controller")
-			obj.SetAnnotations(annotations)
+			// This is the case for restic volumes, where we need to actually have an empty volume created instead of restoring one.
+			// The assumption is that any PV in pvsToProvision doesn't have an associated snapshot.
+			if ctx.pvsToProvision.Has(pvc.Spec.VolumeName) {
+				ctx.log.Infof("Resetting PersistentVolumeClaim %s/%s for dynamic provisioning", namespace, name)
+				unstructured.RemoveNestedField(obj.Object, "spec", "volumeName")
+			}
 		}
 
 		if newName, ok := ctx.renamedPVs[pvc.Spec.VolumeName]; ok {
@@ -1215,6 +1233,40 @@ func shouldRenamePV(ctx *restoreContext, obj *unstructured.Unstructured, client
 	return true, nil
 }
 
+// remapClaimRefNS remaps a PersistentVolume's claimRef.Namespace based on a restore's NamespaceMappings, if necessary.
+// Returns true if the namespace was remapped, false if it was not required.
+func remapClaimRefNS(ctx *restoreContext, obj *unstructured.Unstructured) (bool, error) {
+	if len(ctx.restore.Spec.NamespaceMapping) == 0 {
+		ctx.log.Debug("Persistent volume does not need to have the claimRef.namespace remapped because restore is not remapping any namespaces")
+		return false, nil
+	}
+
+	// Conversion to the real type here is more readable than all the error checking involved with reading each field individually.
+	pv := new(v1.PersistentVolume)
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(obj.Object, pv); err != nil {
+		return false, errors.Wrapf(err, "error converting persistent volume to structured")
+	}
+
+	if pv.Spec.ClaimRef == nil {
+		ctx.log.Debugf("Persistent volume does not need to have the claimRef.namepace remapped because it's not claimed")
+		return false, nil
+	}
+
+	targetNS, ok := ctx.restore.Spec.NamespaceMapping[pv.Spec.ClaimRef.Namespace]
+
+	if !ok {
+		ctx.log.Debugf("Persistent volume does not need to have the claimRef.namespace remapped because it's not claimed by a PVC in a namespace that's being remapped")
+		return false, nil
+	}
+
+	err := unstructured.SetNestedField(obj.Object, targetNS, "spec", "claimRef", "namespace")
+	if err != nil {
+		return false, err
+	}
+	ctx.log.Debug("Persistent volume's namespace was updated")
+	return true, nil
+}
+
 // restorePodVolumeBackups restores the PodVolumeBackups for the given restored pod
 func restorePodVolumeBackups(ctx *restoreContext, createdObj *unstructured.Unstructured, originalNamespace string) {
 	if ctx.resticRestorer == nil {
@@ -1329,6 +1381,29 @@ func hasDeleteReclaimPolicy(obj map[string]interface{}) bool {
 	return policy == string(v1.PersistentVolumeReclaimDelete)
 }
 
+// resetVolumeBindingInfo clears any necessary metadata out of a PersistentVolume or PersistentVolumeClaim that would make it ineligible to be re-bound by Velero.
+func resetVolumeBindingInfo(obj *unstructured.Unstructured) *unstructured.Unstructured {
+	// Clean out ClaimRef UID and resourceVersion, since this information is highly unique.
+	unstructured.RemoveNestedField(obj.Object, "spec", "claimRef", "uid")
+	unstructured.RemoveNestedField(obj.Object, "spec", "claimRef", "resourceVersion")
+
+	// Clear out any annotations used by the Kubernetes PV controllers to track bindings.
+	annotations := obj.GetAnnotations()
+
+	// Upon restore, this new PV will look like a statically provisioned, manually-bound volume rather than one bound by the controller, so remove the annotation that signals that a controller bound it.
+	delete(annotations, KubeAnnBindCompleted)
+	// Remove the annotation that signals that the PV is already bound; we want the PV(C) controller to take the two objects and bind them again.
+	delete(annotations, KubeAnnBoundByController)
+
+	// Remove the provisioned-by annotation which signals that the persistent volume was dynamically provisioned; it is now statically provisioned.
+	delete(annotations, KubeAnnDynamicallyProvisioned)
+
+	// GetAnnotations returns a copy, so we have to set them again
+	obj.SetAnnotations(annotations)
+
+	return obj
+}
+
 func resetMetadataAndStatus(obj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
 	res, ok := obj.Object["metadata"]
 	if !ok {

pkg/restore/restore_test.go

@@ -1830,7 +1830,7 @@ func TestRestorePersistentVolumes(t *testing.T) {
 			},
 		},
 		{
-			name:    "when a PV with a reclaim policy of retain has no snapshot and does not exist in-cluster, it gets restored, without its claim ref",
+			name:    "when a PV with a reclaim policy of retain has no snapshot and does not exist in-cluster, it gets restored, with its claim ref",
 			restore: defaultRestore().Result(),
 			backup:  defaultBackup().Result(),
 			tarball: test.NewTarWriter(t).
@@ -1849,6 +1849,7 @@ func TestRestorePersistentVolumes(t *testing.T) {
 						ObjectMeta(
 							builder.WithLabels("velero.io/backup-name", "backup-1", "velero.io/restore-name", "restore-1"),
 						).
+						ClaimRef("ns-1", "pvc-1").
 						Result(),
 				),
 			},
@@ -2096,13 +2097,12 @@ func TestRestorePersistentVolumes(t *testing.T) {
 			want: []*test.APIResource{
 				test.PVs(
 					builder.ForPersistentVolume("source-pv").AWSEBSVolumeID("source-volume").ClaimRef("source-ns", "pvc-1").Result(),
-					// note that the renamed PV is not expected to have a claimRef in this test; that would be
-					// added after creation by the Kubernetes PV/PVC controller when it does a bind.
 					builder.ForPersistentVolume("renamed-source-pv").
 						ObjectMeta(
 							builder.WithAnnotations("velero.io/original-pv-name", "source-pv"),
 							builder.WithLabels("velero.io/backup-name", "backup-1", "velero.io/restore-name", "restore-1"),
-						).
+						// the namespace for this PV's claimRef should be the one that the PVC was remapped into.
+						).ClaimRef("target-ns", "pvc-1").
 						AWSEBSVolumeID("new-volume").
 						Result(),
 				),
@@ -2161,6 +2161,7 @@ func TestRestorePersistentVolumes(t *testing.T) {
 						ObjectMeta(
 							builder.WithLabels("velero.io/backup-name", "backup-1", "velero.io/restore-name", "restore-1"),
 						).
+						ClaimRef("target-ns", "pvc-1").
 						AWSEBSVolumeID("new-volume").
 						Result(),
 				),
@@ -2221,6 +2222,7 @@ func TestRestorePersistentVolumes(t *testing.T) {
 							builder.WithLabels("velero.io/backup-name", "backup-1", "velero.io/restore-name", "restore-1"),
 							builder.WithAnnotations("velero.io/original-pv-name", "source-pv"),
 						).
+						ClaimRef("target-ns", "pvc-1").
 						AWSEBSVolumeID("new-pvname").
 						Result(),
 				),
@@ -2340,13 +2342,12 @@ func TestRestorePersistentVolumes(t *testing.T) {
 			want: []*test.APIResource{
 				test.PVs(
 					builder.ForPersistentVolume("source-pv").AWSEBSVolumeID("source-volume").ClaimRef("source-ns", "pvc-1").Result(),
-					// note that the renamed PV is not expected to have a claimRef in this test; that would be
-					// added after creation by the Kubernetes PV/PVC controller when it does a bind.
 					builder.ForPersistentVolume("volumesnapshotter-renamed-source-pv").
 						ObjectMeta(
 							builder.WithAnnotations("velero.io/original-pv-name", "source-pv"),
 							builder.WithLabels("velero.io/backup-name", "backup-1", "velero.io/restore-name", "restore-1"),
 						).
+						ClaimRef("target-ns", "pvc-1").
 						AWSEBSVolumeID("new-volume").
 						Result(),
 				),
@@ -2434,14 +2435,14 @@ func TestRestoreWithRestic(t *testing.T) {
 		want                        map[*test.APIResource][]string
 	}{
 		{
-			name:         "a pod that exists in given backup and contains associated PVBs should have should have RestorePodVolumes called",
+			name:         "a pod that exists in given backup and contains associated PVBs should have RestorePodVolumes called",
 			restore:      defaultRestore().Result(),
 			backup:       defaultBackup().Result(),
 			apiResources: []*test.APIResource{test.Pods()},
 			podVolumeBackups: []*velerov1api.PodVolumeBackup{
 				builder.ForPodVolumeBackup("velero", "pvb-1").PodName("pod-1").SnapshotID("foo").Result(),
-				builder.ForPodVolumeBackup("velero", "pvb-2").PodName("pod-2").SnapshotID("foo").Result(),
-				builder.ForPodVolumeBackup("velero", "pvb-3").PodName("pod-4").SnapshotID("foo").Result(),
+				builder.ForPodVolumeBackup("velero", "pvb-2").PodName("pod-2").PodNamespace("ns-1").SnapshotID("foo").Result(),
+				builder.ForPodVolumeBackup("velero", "pvb-3").PodName("pod-4").PodNamespace("ns-2").SnapshotID("foo").Result(),
 			},
 			podWithPVBs: []*corev1api.Pod{
 				builder.ForPod("ns-1", "pod-2").
@@ -2846,3 +2847,49 @@ func (h *harness) AddItems(t *testing.T, resource *test.APIResource) {
 		require.NoError(t, err)
 	}
 }
+
+func Test_resetVolumeBindingInfo(t *testing.T) {
+	tests := []struct {
+		name     string
+		obj      *unstructured.Unstructured
+		expected *unstructured.Unstructured
+	}{
+		{
+			name: "PVs that are bound have their binding and dynamic provisioning annotations removed",
+			obj: NewTestUnstructured().WithMetadataField("kind", "persistentVolume").
+				WithName("pv-1").WithAnnotations(
+				KubeAnnBindCompleted,
+				KubeAnnBoundByController,
+				KubeAnnDynamicallyProvisioned,
+			).WithSpecField("claimRef", map[string]interface{}{
+				"namespace":       "ns-1",
+				"name":            "pvc-1",
+				"uid":             "abc",
+				"resourceVersion": "1"}).Unstructured,
+			expected: NewTestUnstructured().WithMetadataField("kind", "persistentVolume").
+				WithName("pv-1").
+				WithAnnotations().
+				WithSpecField("claimRef", map[string]interface{}{
+					"namespace": "ns-1", "name": "pvc-1"}).Unstructured,
+		},
+		{
+			name: "PVCs that are bound have their binding annotations removed, but the volume name stays",
+			obj: NewTestUnstructured().WithMetadataField("kind", "persistentVolumeClaim").
+				WithName("pvc-1").WithAnnotations(
+				KubeAnnBindCompleted,
+				KubeAnnBoundByController,
+				KubeAnnDynamicallyProvisioned,
+			).WithSpecField("volumeName", "pv-1").Unstructured,
+			expected: NewTestUnstructured().WithMetadataField("kind", "persistentVolumeClaim").
+				WithName("pvc-1").WithAnnotations().
+				WithSpecField("volumeName", "pv-1").Unstructured,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			actual := resetVolumeBindingInfo(tc.obj)
+			assert.Equal(t, tc.expected, actual)
+		})
+	}
+}

pkg/util/collections/includes_excludes.go

@@ -201,7 +201,10 @@ func GetResourceIncludesExcludes(helper discovery.Helper, includes, excludes []s
 		func(item string) string {
 			gvr, _, err := helper.ResourceFor(schema.ParseGroupResource(item).WithVersion(""))
 			if err != nil {
-				return ""
+				// If we can't resolve it, return it as-is. This prevents the generated
+				// includes-excludes list from including *everything*, if none of the includes
+				// can be resolved. ref. https://github.com/vmware-tanzu/velero/issues/2461
+				return item
 			}
 
 			gr := gvr.GroupResource()

site/content/docs/main/customize-installation.md

@@ -80,7 +80,7 @@ At installation, Velero sets default resource requests and limits for the Velero
 |CPU request|500m|500m|
 |Memory requests|128Mi|512Mi|
 |CPU limit|1000m (1 CPU)|1000m (1 CPU)|
-|Memory limit|256Mi|1024Mi|
+|Memory limit|512Mi|1024Mi|
 {{< /table >}}
 
 ### Install with custom resource requests and limits
@@ -111,7 +111,7 @@ Update the `spec.template.spec.containers.resources.limits` and `spec.template.s
 
 ```bash
 kubectl patch deployment velero -n velero --patch \
-'{"spec":{"template":{"spec":{"containers":[{"name": "velero", "resources": {"limits":{"cpu": "1", "memory": "256Mi"}, "requests": {"cpu": "1", "memory": "128Mi"}}}]}}}}'
+'{"spec":{"template":{"spec":{"containers":[{"name": "velero", "resources": {"limits":{"cpu": "1", "memory": "512Mi"}, "requests": {"cpu": "1", "memory": "128Mi"}}}]}}}}'
 ```
 
 **restic pod**

site/content/docs/v1.5/customize-installation.md

@@ -80,7 +80,7 @@ At installation, Velero sets default resource requests and limits for the Velero
 |CPU request|500m|500m|
 |Memory requests|128Mi|512Mi|
 |CPU limit|1000m (1 CPU)|1000m (1 CPU)|
-|Memory limit|256Mi|1024Mi|
+|Memory limit|512Mi|1024Mi|
 {{< /table >}}
 
 ### Install with custom resource requests and limits
@@ -111,7 +111,7 @@ Update the `spec.template.spec.containers.resources.limits` and `spec.template.s
 
 ```bash
 kubectl patch deployment velero -n velero --patch \
-'{"spec":{"template":{"spec":{"containers":[{"name": "velero", "resources": {"limits":{"cpu": "1", "memory": "256Mi"}, "requests": {"cpu": "1", "memory": "128Mi"}}}]}}}}'
+'{"spec":{"template":{"spec":{"containers":[{"name": "velero", "resources": {"limits":{"cpu": "1", "memory": "512Mi"}, "requests": {"cpu": "1", "memory": "128Mi"}}}]}}}}'
 ```
 
 **restic pod**