Merge branch 'main' of https://github.com/qiuming-best/velero into restic-repo-tmp

2026-09-04 07:07:11 +00:00 · 2022-05-06 14:43:00 +08:00
parent 06d3d731ed d6d9a0ec08
commit 752b28166f
47 changed files with 1738 additions and 1863 deletions
@@ -25,11 +25,15 @@ import (
 	"io"
 	"io/ioutil"
 	"os"
+	"sync"
 	"time"

+	"github.com/apex/log"
 	jsonpatch "github.com/evanphx/json-patch"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
+	"golang.org/x/sync/errgroup"
+	v1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -37,11 +41,13 @@ import (
 	"k8s.io/apimachinery/pkg/util/clock"
 	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"

 	"github.com/vmware-tanzu/velero/pkg/util/csi"

 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
+	snapshotterClientSet "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned"
 	snapshotv1listers "github.com/kubernetes-csi/external-snapshotter/client/v4/listers/volumesnapshot/v1"

 	"github.com/vmware-tanzu/velero/internal/storage"
@@ -64,6 +70,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/util/logging"
 	"github.com/vmware-tanzu/velero/pkg/volume"

+	"sigs.k8s.io/cluster-api/util/patch"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
 )

@@ -87,6 +94,7 @@ type backupController struct {
 	backupStoreGetter           persistence.ObjectBackupStoreGetter
 	formatFlag                  logging.Format
 	volumeSnapshotLister        snapshotv1listers.VolumeSnapshotLister
+	volumeSnapshotClient        *snapshotterClientSet.Clientset
 	volumeSnapshotContentLister snapshotv1listers.VolumeSnapshotContentLister
 	volumeSnapshotClassLister   snapshotv1listers.VolumeSnapshotClassLister
 }
@@ -109,6 +117,7 @@ func NewBackupController(
 	metrics *metrics.ServerMetrics,
 	formatFlag logging.Format,
 	volumeSnapshotLister snapshotv1listers.VolumeSnapshotLister,
+	volumeSnapshotClient *snapshotterClientSet.Clientset,
 	volumeSnapshotContentLister snapshotv1listers.VolumeSnapshotContentLister,
 	volumesnapshotClassLister snapshotv1listers.VolumeSnapshotClassLister,
 	backupStoreGetter persistence.ObjectBackupStoreGetter,
@@ -132,6 +141,7 @@ func NewBackupController(
 		metrics:                     metrics,
 		formatFlag:                  formatFlag,
 		volumeSnapshotLister:        volumeSnapshotLister,
+		volumeSnapshotClient:        volumeSnapshotClient,
 		volumeSnapshotContentLister: volumeSnapshotContentLister,
 		volumeSnapshotClassLister:   volumesnapshotClassLister,
 		backupStoreGetter:           backupStoreGetter,
@@ -147,13 +157,12 @@ func NewBackupController(
 				backup := obj.(*velerov1api.Backup)

 				switch backup.Status.Phase {
-				case "", velerov1api.BackupPhaseNew:
-					// only process new backups
+				case "", velerov1api.BackupPhaseNew, velerov1api.BackupPhaseInProgress:
 				default:
 					c.logger.WithFields(logrus.Fields{
 						"backup": kubeutil.NamespaceAndName(backup),
 						"phase":  backup.Status.Phase,
-					}).Debug("Backup is not new, skipping")
+					}).Debug("Backup is not new or in-progress, skipping")
 					return
 				}

@@ -241,7 +250,22 @@ func (c *backupController) processBackup(key string) error {
 	// this key (even though it was a no-op).
 	switch original.Status.Phase {
 	case "", velerov1api.BackupPhaseNew:
-		// only process new backups
+	case velerov1api.BackupPhaseInProgress:
+		// A backup may stay in-progress forever because of
+		// 1) the controller restarts during the processing of a backup
+		// 2) the backup with in-progress status isn't updated to completed or failed status successfully
+		// So we try to mark such Backups as failed to avoid it
+		updated := original.DeepCopy()
+		updated.Status.Phase = velerov1api.BackupPhaseFailed
+		updated.Status.FailureReason = fmt.Sprintf("got a Backup with unexpected status %q, this may be due to a restart of the controller during the backing up, mark it as %q",
+			velerov1api.BackupPhaseInProgress, updated.Status.Phase)
+		updated.Status.CompletionTimestamp = &metav1.Time{Time: c.clock.Now()}
+		_, err = patchBackup(original, updated, c.client)
+		if err != nil {
+			return errors.Wrapf(err, "error updating Backup status to %s", updated.Status.Phase)
+		}
+		log.Warn(updated.Status.FailureReason)
+		return nil
 	default:
 		return nil
 	}
@@ -261,6 +285,7 @@ func (c *backupController) processBackup(key string) error {
 	if err != nil {
 		return errors.Wrapf(err, "error updating Backup status to %s", request.Status.Phase)
 	}
+
 	// store ref to just-updated item for creating patch
 	original = updatedBackup
 	request.Backup = updatedBackup.DeepCopy()
@@ -287,6 +312,7 @@ func (c *backupController) processBackup(key string) error {
 		// result in the backup being Failed.
 		log.WithError(err).Error("backup failed")
 		request.Status.Phase = velerov1api.BackupPhaseFailed
+		request.Status.FailureReason = err.Error()
 	}

 	switch request.Status.Phase {
@@ -621,6 +647,13 @@ func (c *backupController) runBackup(backup *pkgbackup.Request) error {
 			if err != nil {
 				backupLog.Error(err)
 			}
+
+			err = c.checkVolumeSnapshotReadyToUse(context.Background(), volumeSnapshots)
+			if err != nil {
+				backupLog.Errorf("fail to wait VolumeSnapshot change to Ready: %s", err.Error())
+			}
+
+			backup.CSISnapshots = volumeSnapshots
 		}

 		if c.volumeSnapshotContentLister != nil {
@@ -646,6 +679,10 @@ func (c *backupController) runBackup(backup *pkgbackup.Request) error {
 				backupLog.Error(err)
 			}
 		}
+
+		// Delete the VolumeSnapshots created in the backup, when CSI feature is enabled.
+		c.deleteVolumeSnapshot(volumeSnapshots, volumeSnapshotContents, *backup, backupLog)
+
 	}

 	// Mark completion timestamp before serializing and uploading.
@@ -661,7 +698,7 @@ func (c *backupController) runBackup(backup *pkgbackup.Request) error {

 	backup.Status.CSIVolumeSnapshotsAttempted = len(backup.CSISnapshots)
 	for _, vs := range backup.CSISnapshots {
-		if *vs.Status.ReadyToUse {
+		if vs.Status != nil && boolptr.IsSetToTrue(vs.Status.ReadyToUse) {
 			backup.Status.CSIVolumeSnapshotsCompleted++
 		}
 	}
@@ -847,3 +884,170 @@ func encodeToJSONGzip(data interface{}, desc string) (*bytes.Buffer, []error) {

 	return buf, nil
 }
+
+// Waiting for VolumeSnapshot ReadyTosue to true is time consuming. Try to make the process parallel by
+// using goroutine here instead of waiting in CSI plugin, because it's not easy to make BackupItemAction
+// parallel by now. After BackupItemAction parallel is implemented, this logic should be moved to CSI plugin
+// as https://github.com/vmware-tanzu/velero-plugin-for-csi/pull/100
+func (c *backupController) checkVolumeSnapshotReadyToUse(ctx context.Context, volumesnapshots []*snapshotv1api.VolumeSnapshot) error {
+	eg, _ := errgroup.WithContext(ctx)
+	timeout := 10 * time.Minute
+	interval := 5 * time.Second
+
+	for _, vs := range volumesnapshots {
+		volumeSnapshot := vs
+		eg.Go(func() error {
+			err := wait.PollImmediate(interval, timeout, func() (bool, error) {
+				tmpVS, err := c.volumeSnapshotClient.SnapshotV1().VolumeSnapshots(volumeSnapshot.Namespace).Get(ctx, volumeSnapshot.Name, metav1.GetOptions{})
+				if err != nil {
+					return false, errors.Wrapf(err, fmt.Sprintf("failed to get volumesnapshot %s/%s", volumeSnapshot.Namespace, volumeSnapshot.Name))
+				}
+				if tmpVS.Status == nil || tmpVS.Status.BoundVolumeSnapshotContentName == nil || !boolptr.IsSetToTrue(tmpVS.Status.ReadyToUse) {
+					log.Infof("Waiting for CSI driver to reconcile volumesnapshot %s/%s. Retrying in %ds", volumeSnapshot.Namespace, volumeSnapshot.Name, interval/time.Second)
+					return false, nil
+				}
+
+				return true, nil
+			})
+			if err == wait.ErrWaitTimeout {
+				log.Errorf("Timed out awaiting reconciliation of volumesnapshot %s/%s", volumeSnapshot.Namespace, volumeSnapshot.Name)
+			}
+			return err
+		})
+	}
+	return eg.Wait()
+}
+
+// deleteVolumeSnapshot delete VolumeSnapshot created during backup.
+// This is used to avoid deleting namespace in cluster triggers the VolumeSnapshot deletion,
+// which will cause snapshot deletion on cloud provider, then backup cannot restore the PV.
+// If DeletionPolicy is Retain, just delete it. If DeletionPolicy is Delete, need to
+// change DeletionPolicy to Retain before deleting VS, then change DeletionPolicy back to Delete.
+func (c *backupController) deleteVolumeSnapshot(volumeSnapshots []*snapshotv1api.VolumeSnapshot,
+	volumeSnapshotContents []*snapshotv1api.VolumeSnapshotContent,
+	backup pkgbackup.Request, logger logrus.FieldLogger) {
+	var wg sync.WaitGroup
+	vscMap := make(map[string]*snapshotv1api.VolumeSnapshotContent)
+	for _, vsc := range volumeSnapshotContents {
+		vscMap[vsc.Name] = vsc
+	}
+
+	for _, vs := range volumeSnapshots {
+		wg.Add(1)
+		go func(vs *snapshotv1api.VolumeSnapshot) {
+			defer wg.Done()
+			var vsc *snapshotv1api.VolumeSnapshotContent
+			modifyVSCFlag := false
+			if vs.Status.BoundVolumeSnapshotContentName != nil &&
+				len(*vs.Status.BoundVolumeSnapshotContentName) > 0 {
+				vsc = vscMap[*vs.Status.BoundVolumeSnapshotContentName]
+				if vsc.Spec.DeletionPolicy == snapshotv1api.VolumeSnapshotContentDelete {
+					modifyVSCFlag = true
+				}
+			}
+
+			// Change VolumeSnapshotContent's DeletionPolicy to Retain before deleting VolumeSnapshot,
+			// because VolumeSnapshotContent will be deleted by deleting VolumeSnapshot, when
+			// DeletionPolicy is set to Delete, but Velero needs VSC for cleaning snapshot on cloud
+			// in backup deletion.
+			if modifyVSCFlag {
+				logger.Debugf("Patching VolumeSnapshotContent %s", vsc.Name)
+				_, err := c.patchVolumeSnapshotContent(vsc, func(req *snapshotv1api.VolumeSnapshotContent) {
+					req.Spec.DeletionPolicy = snapshotv1api.VolumeSnapshotContentRetain
+				})
+				if err != nil {
+					logger.Errorf("fail to modify VolumeSnapshotContent %s DeletionPolicy to Retain: %s", vsc.Name, err.Error())
+					return
+				}
+
+				defer func() {
+					logger.Debugf("Start to recreate VolumeSnapshotContent %s", vsc.Name)
+					err := c.recreateVolumeSnapshotContent(vsc)
+					if err != nil {
+						logger.Errorf("fail to recreate VolumeSnapshotContent %s: %s", vsc.Name, err.Error())
+					}
+				}()
+			}
+
+			// Delete VolumeSnapshot from cluster
+			logger.Debugf("Deleting VolumeSnapshotContent %s", vsc.Name)
+			err := c.volumeSnapshotClient.SnapshotV1().VolumeSnapshots(vs.Namespace).Delete(context.TODO(), vs.Name, metav1.DeleteOptions{})
+			if err != nil {
+				logger.Errorf("fail to delete VolumeSnapshot %s/%s: %s", vs.Namespace, vs.Name, err.Error())
+			}
+		}(vs)
+	}
+
+	wg.Wait()
+}
+
+func (c *backupController) patchVolumeSnapshotContent(req *snapshotv1api.VolumeSnapshotContent, mutate func(*snapshotv1api.VolumeSnapshotContent)) (*snapshotv1api.VolumeSnapshotContent, error) {
+	patchHelper, err := patch.NewHelper(req, c.kbClient)
+	if err != nil {
+		return nil, errors.Wrap(err, "fail to get patch helper.")
+	}
+
+	// Mutate
+	mutate(req)
+
+	if err := patchHelper.Patch(context.TODO(), req); err != nil {
+		return nil, errors.Wrapf(err, "fail to patch VolumeSnapshotContent %s", req.Name)
+	}
+
+	return req, nil
+}
+
+// recreateVolumeSnapshotContent will delete then re-create VolumeSnapshotContent,
+// because some parameter in VolumeSnapshotContent Spec is immutable, e.g. VolumeSnapshotRef
+// and Source. Source is updated to let csi-controller thinks the VSC is statically provsisioned with VS.
+// Set VolumeSnapshotRef's UID to nil will let the csi-controller finds out the related VS is gone, then
+// VSC can be deleted.
+func (c *backupController) recreateVolumeSnapshotContent(vsc *snapshotv1api.VolumeSnapshotContent) error {
+	timeout := 1 * time.Minute
+	interval := 1 * time.Second
+
+	err := c.volumeSnapshotClient.SnapshotV1().VolumeSnapshotContents().Delete(context.TODO(), vsc.Name, metav1.DeleteOptions{})
+	if err != nil {
+		return errors.Wrapf(err, "fail to delete VolumeSnapshotContent: %s", vsc.Name)
+	}
+
+	// Check VolumeSnapshotContents is already deleted, before re-creating it.
+	err = wait.PollImmediate(interval, timeout, func() (bool, error) {
+		_, err := c.volumeSnapshotClient.SnapshotV1().VolumeSnapshotContents().Get(context.TODO(), vsc.Name, metav1.GetOptions{})
+		if err != nil {
+			if apierrors.IsNotFound(err) {
+				return true, nil
+			}
+			return false, errors.Wrapf(err, fmt.Sprintf("failed to get VolumeSnapshotContent %s", vsc.Name))
+		}
+		return false, nil
+	})
+	if err != nil {
+		return errors.Wrapf(err, "fail to retrieve VolumeSnapshotContent %s info", vsc.Name)
+	}
+
+	// Make the VolumeSnapshotContent static
+	vsc.Spec.Source = snapshotv1api.VolumeSnapshotContentSource{
+		SnapshotHandle: vsc.Status.SnapshotHandle,
+	}
+	// Set VolumeSnapshotRef to none exist one, because VolumeSnapshotContent
+	// validation webhook will check whether name and namespace are nil.
+	// external-snapshotter needs Source pointing to snapshot and VolumeSnapshot
+	// reference's UID to nil to determine the VolumeSnapshotContent is deletable.
+	vsc.Spec.VolumeSnapshotRef = v1.ObjectReference{
+		APIVersion: snapshotv1api.SchemeGroupVersion.String(),
+		Kind:       "VolumeSnapshot",
+		Namespace:  "ns-" + string(vsc.UID),
+		Name:       "name-" + string(vsc.UID),
+	}
+	// Revert DeletionPolicy to Delete
+	vsc.Spec.DeletionPolicy = snapshotv1api.VolumeSnapshotContentDelete
+	// ResourceVersion shouldn't exist for new creation.
+	vsc.ResourceVersion = ""
+	_, err = c.volumeSnapshotClient.SnapshotV1().VolumeSnapshotContents().Create(context.TODO(), vsc, metav1.CreateOptions{})
+	if err != nil {
+		return errors.Wrapf(err, "fail to create VolumeSnapshotContent %s", vsc.Name)
+	}
+
+	return nil
+}
@@ -94,11 +94,6 @@ func TestProcessBackupNonProcessedItems(t *testing.T) {
 			key:    "velero/backup-1",
 			backup: defaultBackup().Phase(velerov1api.BackupPhaseFailedValidation).Result(),
 		},
-		{
-			name:   "InProgress backup is not processed",
-			key:    "velero/backup-1",
-			backup: defaultBackup().Phase(velerov1api.BackupPhaseInProgress).Result(),
-		},
 		{
 			name:   "Completed backup is not processed",
 			key:    "velero/backup-1",
@@ -140,6 +135,28 @@ func TestProcessBackupNonProcessedItems(t *testing.T) {
 	}
 }

+func TestMarkInProgressBackupAsFailed(t *testing.T) {
+	backup := defaultBackup().Phase(velerov1api.BackupPhaseInProgress).Result()
+	clientset := fake.NewSimpleClientset(backup)
+	sharedInformers := informers.NewSharedInformerFactory(clientset, 0)
+	logger := logging.DefaultLogger(logrus.DebugLevel, logging.FormatText)
+
+	c := &backupController{
+		genericController: newGenericController("backup-test", logger),
+		client:            clientset.VeleroV1(),
+		lister:            sharedInformers.Velero().V1().Backups().Lister(),
+		clock:             &clock.RealClock{},
+	}
+	require.NoError(t, sharedInformers.Velero().V1().Backups().Informer().GetStore().Add(backup))
+
+	err := c.processBackup(fmt.Sprintf("%s/%s", backup.Namespace, backup.Name))
+	require.Nil(t, err)
+
+	res, err := clientset.VeleroV1().Backups(backup.Namespace).Get(context.TODO(), backup.Name, metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, velerov1api.BackupPhaseFailed, res.Status.Phase)
+}
+
 func TestProcessBackupValidationFailures(t *testing.T) {
 	defaultBackupLocation := builder.ForBackupStorageLocation("velero", "loc-1").Result()

@@ -729,6 +746,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:               velerov1api.BackupPhaseFailed,
+					FailureReason:       "backup already exists in object storage",
 					Version:             1,
 					FormatVersion:       "1.1.0",
 					StartTimestamp:      &timestamp,
@@ -766,6 +784,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:               velerov1api.BackupPhaseFailed,
+					FailureReason:       "error checking if backup already exists in object storage: Backup already exists in object storage",
 					Version:             1,
 					FormatVersion:       "1.1.0",
 					StartTimestamp:      &timestamp,
@@ -23,25 +23,19 @@ import (
 	"time"

 	jsonpatch "github.com/evanphx/json-patch"
-	snapshotterClientSet "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned"
-	snapshotv1listers "github.com/kubernetes-csi/external-snapshotter/client/v4/listers/volumesnapshot/v1"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/clock"
 	kubeerrs "k8s.io/apimachinery/pkg/util/errors"
-	"k8s.io/client-go/tools/cache"
+	"sigs.k8s.io/cluster-api/util/patch"
+	ctrl "sigs.k8s.io/controller-runtime"

 	"github.com/vmware-tanzu/velero/internal/delete"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
-	pkgbackup "github.com/vmware-tanzu/velero/pkg/backup"
 	"github.com/vmware-tanzu/velero/pkg/discovery"
-	velerov1client "github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned/typed/velero/v1"
-	velerov1informers "github.com/vmware-tanzu/velero/pkg/generated/informers/externalversions/velero/v1"
-	velerov1listers "github.com/vmware-tanzu/velero/pkg/generated/listers/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/label"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
 	"github.com/vmware-tanzu/velero/pkg/persistence"
@@ -54,250 +48,202 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )

-const resticTimeout = time.Minute
+const (
+	resticTimeout             = time.Minute
+	deleteBackupRequestMaxAge = 24 * time.Hour
+)

-type backupDeletionController struct {
-	*genericController
-
-	deleteBackupRequestClient velerov1client.DeleteBackupRequestsGetter
-	deleteBackupRequestLister velerov1listers.DeleteBackupRequestLister
-	backupClient              velerov1client.BackupsGetter
-	restoreLister             velerov1listers.RestoreLister
-	restoreClient             velerov1client.RestoresGetter
-	backupTracker             BackupTracker
-	resticMgr                 restic.RepositoryManager
-	podvolumeBackupLister     velerov1listers.PodVolumeBackupLister
-	kbClient                  client.Client
-	snapshotLocationLister    velerov1listers.VolumeSnapshotLocationLister
-	csiSnapshotLister         snapshotv1listers.VolumeSnapshotLister
-	csiSnapshotContentLister  snapshotv1listers.VolumeSnapshotContentLister
-	csiSnapshotClient         *snapshotterClientSet.Clientset
-	processRequestFunc        func(*velerov1api.DeleteBackupRequest) error
-	clock                     clock.Clock
-	newPluginManager          func(logrus.FieldLogger) clientmgmt.Manager
-	backupStoreGetter         persistence.ObjectBackupStoreGetter
-	metrics                   *metrics.ServerMetrics
-	helper                    discovery.Helper
+type backupDeletionReconciler struct {
+	client.Client
+	logger            logrus.FieldLogger
+	backupTracker     BackupTracker
+	resticMgr         restic.RepositoryManager
+	metrics           *metrics.ServerMetrics
+	clock             clock.Clock
+	discoveryHelper   discovery.Helper
+	newPluginManager  func(logrus.FieldLogger) clientmgmt.Manager
+	backupStoreGetter persistence.ObjectBackupStoreGetter
 }

-// NewBackupDeletionController creates a new backup deletion controller.
-func NewBackupDeletionController(
+// NewBackupDeletionReconciler creates a new backup deletion reconciler.
+func NewBackupDeletionReconciler(
 	logger logrus.FieldLogger,
-	deleteBackupRequestInformer velerov1informers.DeleteBackupRequestInformer,
-	deleteBackupRequestClient velerov1client.DeleteBackupRequestsGetter,
-	backupClient velerov1client.BackupsGetter,
-	restoreLister velerov1listers.RestoreLister,
-	restoreClient velerov1client.RestoresGetter,
+	client client.Client,
 	backupTracker BackupTracker,
 	resticMgr restic.RepositoryManager,
-	podvolumeBackupLister velerov1listers.PodVolumeBackupLister,
-	kbClient client.Client,
-	snapshotLocationLister velerov1listers.VolumeSnapshotLocationLister,
-	csiSnapshotLister snapshotv1listers.VolumeSnapshotLister,
-	csiSnapshotContentLister snapshotv1listers.VolumeSnapshotContentLister,
-	csiSnapshotClient *snapshotterClientSet.Clientset,
-	newPluginManager func(logrus.FieldLogger) clientmgmt.Manager,
-	backupStoreGetter persistence.ObjectBackupStoreGetter,
 	metrics *metrics.ServerMetrics,
 	helper discovery.Helper,
-) Interface {
-	c := &backupDeletionController{
-		genericController:         newGenericController(BackupDeletion, logger),
-		deleteBackupRequestClient: deleteBackupRequestClient,
-		deleteBackupRequestLister: deleteBackupRequestInformer.Lister(),
-		backupClient:              backupClient,
-		restoreLister:             restoreLister,
-		restoreClient:             restoreClient,
-		backupTracker:             backupTracker,
-		resticMgr:                 resticMgr,
-		podvolumeBackupLister:     podvolumeBackupLister,
-		kbClient:                  kbClient,
-		snapshotLocationLister:    snapshotLocationLister,
-		csiSnapshotLister:         csiSnapshotLister,
-		csiSnapshotContentLister:  csiSnapshotContentLister,
-		csiSnapshotClient:         csiSnapshotClient,
-		metrics:                   metrics,
-		helper:                    helper,
-		// use variables to refer to these functions so they can be
-		// replaced with fakes for testing.
+	newPluginManager func(logrus.FieldLogger) clientmgmt.Manager,
+	backupStoreGetter persistence.ObjectBackupStoreGetter,
+) *backupDeletionReconciler {
+	return &backupDeletionReconciler{
+		Client:            client,
+		logger:            logger,
+		backupTracker:     backupTracker,
+		resticMgr:         resticMgr,
+		metrics:           metrics,
+		clock:             clock.RealClock{},
+		discoveryHelper:   helper,
 		newPluginManager:  newPluginManager,
 		backupStoreGetter: backupStoreGetter,
-
-		clock: &clock.RealClock{},
 	}
-
-	c.syncHandler = c.processQueueItem
-	c.processRequestFunc = c.processRequest
-
-	deleteBackupRequestInformer.Informer().AddEventHandler(
-		cache.ResourceEventHandlerFuncs{
-			AddFunc: c.enqueue,
-		},
-	)
-
-	c.resyncPeriod = time.Hour
-	c.resyncFunc = c.deleteExpiredRequests
-
-	return c
 }

-func (c *backupDeletionController) processQueueItem(key string) error {
-	log := c.logger.WithField("key", key)
-	log.Debug("Running processItem")
-
-	ns, name, err := cache.SplitMetaNamespaceKey(key)
-	if err != nil {
-		return errors.Wrap(err, "error splitting queue key")
-	}
-
-	req, err := c.deleteBackupRequestLister.DeleteBackupRequests(ns).Get(name)
-	if apierrors.IsNotFound(err) {
-		log.Debug("Unable to find DeleteBackupRequest")
-		return nil
-	}
-	if err != nil {
-		return errors.Wrap(err, "error getting DeleteBackupRequest")
-	}
-
-	switch req.Status.Phase {
-	case velerov1api.DeleteBackupRequestPhaseProcessed:
-		// Don't do anything because it's already been processed
-	default:
-		// Don't mutate the shared cache
-		reqCopy := req.DeepCopy()
-		return c.processRequestFunc(reqCopy)
-	}
-
-	return nil
+func (r *backupDeletionReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	// Make sure the expired requests can be deleted eventually
+	s := kube.NewPeriodicalEnqueueSource(r.logger, mgr.GetClient(), &velerov1api.DeleteBackupRequestList{}, time.Hour)
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&velerov1api.DeleteBackupRequest{}).
+		Watches(s, nil).
+		Complete(r)
 }

-func (c *backupDeletionController) processRequest(req *velerov1api.DeleteBackupRequest) error {
-	log := c.logger.WithFields(logrus.Fields{
-		"namespace": req.Namespace,
-		"name":      req.Name,
-		"backup":    req.Spec.BackupName,
+// +kubebuilder:rbac:groups=velero.io,resources=deletebackuprequests,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=velero.io,resources=deletebackuprequests/status,verbs=get;update;patch
+// +kubebuilder:rbac:groups=velero.io,resources=backups,verbs=delete
+
+func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	log := r.logger.WithFields(logrus.Fields{
+		"controller":          BackupDeletion,
+		"deletebackuprequest": req.String(),
 	})
+	log.Debug("Getting deletebackuprequest")
+	dbr := &velerov1api.DeleteBackupRequest{}
+	if err := r.Get(ctx, req.NamespacedName, dbr); err != nil {
+		if apierrors.IsNotFound(err) {
+			log.Debug("Unable to find the deletebackuprequest")
+			return ctrl.Result{}, nil
+		}
+		log.WithError(err).Error("Error getting deletebackuprequest")
+		return ctrl.Result{}, err
+	}

-	var err error
+	// Since we use the reconciler along with the PeriodicalEnqueueSource, there may be reconciliation triggered by
+	// stale requests.
+	if dbr.Status.Phase == velerov1api.DeleteBackupRequestPhaseProcessed {
+		age := r.clock.Now().Sub(dbr.CreationTimestamp.Time)
+		if age >= deleteBackupRequestMaxAge { // delete the expired request
+			log.Debug("The request is expired, deleting it.")
+			if err := r.Delete(ctx, dbr); err != nil {
+				log.WithError(err).Error("Error deleting DeleteBackupRequest")
+			}
+		} else {
+			log.Info("The request has been processed, skip.")
+		}
+		return ctrl.Result{}, nil
+	}

 	// Make sure we have the backup name
-	if req.Spec.BackupName == "" {
-		_, err = c.patchDeleteBackupRequest(req, func(r *velerov1api.DeleteBackupRequest) {
-			r.Status.Phase = velerov1api.DeleteBackupRequestPhaseProcessed
-			r.Status.Errors = []string{"spec.backupName is required"}
+	if dbr.Spec.BackupName == "" {
+		_, err := r.patchDeleteBackupRequest(ctx, dbr, func(res *velerov1api.DeleteBackupRequest) {
+			res.Status.Phase = velerov1api.DeleteBackupRequestPhaseProcessed
+			res.Status.Errors = []string{"spec.backupName is required"}
 		})
-		return err
+		return ctrl.Result{}, err
 	}

+	log = log.WithField("backup", dbr.Spec.BackupName)
+
 	// Remove any existing deletion requests for this backup so we only have
 	// one at a time
-	if errs := c.deleteExistingDeletionRequests(req, log); errs != nil {
-		return kubeerrs.NewAggregate(errs)
+	if errs := r.deleteExistingDeletionRequests(ctx, dbr, log); errs != nil {
+		return ctrl.Result{}, kubeerrs.NewAggregate(errs)
 	}

 	// Don't allow deleting an in-progress backup
-	if c.backupTracker.Contains(req.Namespace, req.Spec.BackupName) {
-		_, err = c.patchDeleteBackupRequest(req, func(r *velerov1api.DeleteBackupRequest) {
+	if r.backupTracker.Contains(dbr.Namespace, dbr.Spec.BackupName) {
+		_, err := r.patchDeleteBackupRequest(ctx, dbr, func(r *velerov1api.DeleteBackupRequest) {
 			r.Status.Phase = velerov1api.DeleteBackupRequestPhaseProcessed
 			r.Status.Errors = []string{"backup is still in progress"}
 		})
-
-		return err
+		return ctrl.Result{}, err
 	}

 	// Get the backup we're trying to delete
-	backup, err := c.backupClient.Backups(req.Namespace).Get(context.TODO(), req.Spec.BackupName, metav1.GetOptions{})
-	if apierrors.IsNotFound(err) {
+	backup := &velerov1api.Backup{}
+	if err := r.Get(ctx, types.NamespacedName{
+		Namespace: dbr.Namespace,
+		Name:      dbr.Spec.BackupName,
+	}, backup); apierrors.IsNotFound(err) {
 		// Couldn't find backup - update status to Processed and record the not-found error
-		req, err = c.patchDeleteBackupRequest(req, func(r *velerov1api.DeleteBackupRequest) {
+		_, err = r.patchDeleteBackupRequest(ctx, dbr, func(r *velerov1api.DeleteBackupRequest) {
 			r.Status.Phase = velerov1api.DeleteBackupRequestPhaseProcessed
 			r.Status.Errors = []string{"backup not found"}
 		})
-
-		return err
-	}
-	if err != nil {
-		return errors.Wrap(err, "error getting backup")
+		return ctrl.Result{}, err
+	} else if err != nil {
+		return ctrl.Result{}, errors.Wrap(err, "error getting backup")
 	}

 	// Don't allow deleting backups in read-only storage locations
 	location := &velerov1api.BackupStorageLocation{}
-	if err := c.kbClient.Get(context.Background(), client.ObjectKey{
+	if err := r.Get(context.Background(), client.ObjectKey{
 		Namespace: backup.Namespace,
 		Name:      backup.Spec.StorageLocation,
 	}, location); err != nil {
 		if apierrors.IsNotFound(err) {
-			_, err := c.patchDeleteBackupRequest(req, func(r *velerov1api.DeleteBackupRequest) {
+			_, err := r.patchDeleteBackupRequest(ctx, dbr, func(r *velerov1api.DeleteBackupRequest) {
 				r.Status.Phase = velerov1api.DeleteBackupRequestPhaseProcessed
 				r.Status.Errors = append(r.Status.Errors, fmt.Sprintf("backup storage location %s not found", backup.Spec.StorageLocation))
 			})
-			return err
+			return ctrl.Result{}, err
 		}
-		return errors.Wrap(err, "error getting backup storage location")
+		return ctrl.Result{}, errors.Wrap(err, "error getting backup storage location")
 	}

 	if location.Spec.AccessMode == velerov1api.BackupStorageLocationAccessModeReadOnly {
-		_, err := c.patchDeleteBackupRequest(req, func(r *velerov1api.DeleteBackupRequest) {
+		_, err := r.patchDeleteBackupRequest(ctx, dbr, func(r *velerov1api.DeleteBackupRequest) {
 			r.Status.Phase = velerov1api.DeleteBackupRequestPhaseProcessed
 			r.Status.Errors = append(r.Status.Errors, fmt.Sprintf("cannot delete backup because backup storage location %s is currently in read-only mode", location.Name))
 		})
-		return err
+		return ctrl.Result{}, err
 	}

 	// if the request object has no labels defined, initialise an empty map since
 	// we will be updating labels
-	if req.Labels == nil {
-		req.Labels = map[string]string{}
+	if dbr.Labels == nil {
+		dbr.Labels = map[string]string{}
 	}
-
-	// Update status to InProgress and set backup-name label if needed
-	req, err = c.patchDeleteBackupRequest(req, func(r *velerov1api.DeleteBackupRequest) {
+	// Update status to InProgress and set backup-name and backup-uid label if needed
+	dbr, err := r.patchDeleteBackupRequest(ctx, dbr, func(r *velerov1api.DeleteBackupRequest) {
 		r.Status.Phase = velerov1api.DeleteBackupRequestPhaseInProgress

-		if req.Labels[velerov1api.BackupNameLabel] == "" {
-			req.Labels[velerov1api.BackupNameLabel] = label.GetValidName(req.Spec.BackupName)
+		if r.Labels[velerov1api.BackupNameLabel] == "" {
+			r.Labels[velerov1api.BackupNameLabel] = label.GetValidName(dbr.Spec.BackupName)
+		}
+
+		if r.Labels[velerov1api.BackupUIDLabel] == "" {
+			r.Labels[velerov1api.BackupUIDLabel] = string(backup.UID)
 		}
 	})
 	if err != nil {
-		return err
-	}
-
-	// Set backup-uid label if needed
-	if req.Labels[velerov1api.BackupUIDLabel] == "" {
-		req, err = c.patchDeleteBackupRequest(req, func(r *velerov1api.DeleteBackupRequest) {
-			req.Labels[velerov1api.BackupUIDLabel] = string(backup.UID)
-		})
-		if err != nil {
-			return err
-		}
+		return ctrl.Result{}, err
 	}

 	// Set backup status to Deleting
-	backup, err = c.patchBackup(backup, func(b *velerov1api.Backup) {
+	backup, err = r.patchBackup(ctx, backup, func(b *velerov1api.Backup) {
 		b.Status.Phase = velerov1api.BackupPhaseDeleting
 	})
 	if err != nil {
 		log.WithError(errors.WithStack(err)).Error("Error setting backup phase to deleting")
-		return err
+		return ctrl.Result{}, err
 	}

 	backupScheduleName := backup.GetLabels()[velerov1api.ScheduleNameLabel]
-	c.metrics.RegisterBackupDeletionAttempt(backupScheduleName)
+	r.metrics.RegisterBackupDeletionAttempt(backupScheduleName)

-	var errs []string
-
-	pluginManager := c.newPluginManager(log)
+	pluginManager := r.newPluginManager(log)
 	defer pluginManager.CleanupClients()

-	backupStore, err := c.backupStoreGetter.Get(location, pluginManager, log)
+	backupStore, err := r.backupStoreGetter.Get(location, pluginManager, log)
 	if err != nil {
-		return errors.Wrap(err, "error getting the backup store")
+		return ctrl.Result{}, errors.Wrap(err, "error getting the backup store")
 	}

 	actions, err := pluginManager.GetDeleteItemActions()
 	log.Debugf("%d actions before invoking actions", len(actions))
 	if err != nil {
-		return errors.Wrap(err, "error getting delete item actions")
+		return ctrl.Result{}, errors.Wrap(err, "error getting delete item actions")
 	}
 	// don't defer CleanupClients here, since it was already called above.

@@ -308,13 +254,13 @@ func (c *backupDeletionController) processRequest(req *velerov1api.DeleteBackupR
 		if err != nil {
 			log.WithError(err).Errorf("Unable to download tarball for backup %s, skipping associated DeleteItemAction plugins", backup.Name)
 		} else {
-			defer closeAndRemoveFile(backupFile, c.logger)
+			defer closeAndRemoveFile(backupFile, r.logger)
 			ctx := &delete.Context{
 				Backup:          backup,
 				BackupReader:    backupFile,
 				Actions:         actions,
-				Log:             c.logger,
-				DiscoveryHelper: c.helper,
+				Log:             r.logger,
+				DiscoveryHelper: r.discoveryHelper,
 				Filesystem:      filesystem.NewFileSystem(),
 			}

@@ -322,11 +268,13 @@ func (c *backupDeletionController) processRequest(req *velerov1api.DeleteBackupR
 			// 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")
+				return ctrl.Result{}, errors.Wrap(err, "error invoking delete item actions")
 			}
 		}
 	}

+	var errs []string
+
 	if backupStore != nil {
 		log.Info("Removing PV snapshots")

@@ -340,7 +288,7 @@ func (c *backupDeletionController) processRequest(req *velerov1api.DeleteBackupR

 				volumeSnapshotter, ok := volumeSnapshotters[snapshot.Spec.Location]
 				if !ok {
-					if volumeSnapshotter, err = volumeSnapshotterForSnapshotLocation(backup.Namespace, snapshot.Spec.Location, c.snapshotLocationLister, pluginManager); err != nil {
+					if volumeSnapshotter, err = volumeSnapshottersForVSL(ctx, backup.Namespace, snapshot.Spec.Location, r.Client, pluginManager); err != nil {
 						errs = append(errs, err.Error())
 						continue
 					}
@@ -353,9 +301,8 @@ func (c *backupDeletionController) processRequest(req *velerov1api.DeleteBackupR
 			}
 		}
 	}
-
 	log.Info("Removing restic snapshots")
-	if deleteErrs := c.deleteResticSnapshots(backup); len(deleteErrs) > 0 {
+	if deleteErrs := r.deleteResticSnapshots(ctx, backup); len(deleteErrs) > 0 {
 		for _, err := range deleteErrs {
 			errs = append(errs, err.Error())
 		}
@@ -369,15 +316,19 @@ func (c *backupDeletionController) processRequest(req *velerov1api.DeleteBackupR
 	}

 	log.Info("Removing restores")
-	if restores, err := c.restoreLister.Restores(backup.Namespace).List(labels.Everything()); err != nil {
+	restoreList := &velerov1api.RestoreList{}
+	selector := labels.Everything()
+	if err := r.List(ctx, restoreList, &client.ListOptions{
+		Namespace:     backup.Namespace,
+		LabelSelector: selector,
+	}); err != nil {
 		log.WithError(errors.WithStack(err)).Error("Error listing restore API objects")
 	} else {
-		for _, restore := range restores {
+		for _, restore := range restoreList.Items {
 			if restore.Spec.BackupName != backup.Name {
 				continue
 			}
-
-			restoreLog := log.WithField("restore", kube.NamespaceAndName(restore))
+			restoreLog := log.WithField("restore", kube.NamespaceAndName(&restore))

 			restoreLog.Info("Deleting restore log/results from backup storage")
 			if err := backupStore.DeleteRestore(restore.Name); err != nil {
@@ -387,202 +338,160 @@ func (c *backupDeletionController) processRequest(req *velerov1api.DeleteBackupR
 			}

 			restoreLog.Info("Deleting restore referencing backup")
-			if err := c.restoreClient.Restores(restore.Namespace).Delete(context.TODO(), restore.Name, metav1.DeleteOptions{}); err != nil {
-				errs = append(errs, errors.Wrapf(err, "error deleting restore %s", kube.NamespaceAndName(restore)).Error())
+			if err := r.Delete(ctx, &restore); err != nil {
+				errs = append(errs, errors.Wrapf(err, "error deleting restore %s", kube.NamespaceAndName(&restore)).Error())
 			}
 		}
 	}

 	if len(errs) == 0 {
 		// Only try to delete the backup object from kube if everything preceding went smoothly
-		err = c.backupClient.Backups(backup.Namespace).Delete(context.TODO(), backup.Name, metav1.DeleteOptions{})
-		if err != nil {
+		if err := r.Delete(ctx, backup); err != nil {
 			errs = append(errs, errors.Wrapf(err, "error deleting backup %s", kube.NamespaceAndName(backup)).Error())
 		}
 	}

 	if len(errs) == 0 {
-		c.metrics.RegisterBackupDeletionSuccess(backupScheduleName)
+		r.metrics.RegisterBackupDeletionSuccess(backupScheduleName)
 	} else {
-		c.metrics.RegisterBackupDeletionFailed(backupScheduleName)
+		r.metrics.RegisterBackupDeletionFailed(backupScheduleName)
 	}

 	// Update status to processed and record errors
-	req, err = c.patchDeleteBackupRequest(req, func(r *velerov1api.DeleteBackupRequest) {
+	if _, err := r.patchDeleteBackupRequest(ctx, dbr, func(r *velerov1api.DeleteBackupRequest) {
 		r.Status.Phase = velerov1api.DeleteBackupRequestPhaseProcessed
 		r.Status.Errors = errs
-	})
-	if err != nil {
-		return err
+	}); err != nil {
+		return ctrl.Result{}, err
 	}
-
 	// Everything deleted correctly, so we can delete all DeleteBackupRequests for this backup
 	if len(errs) == 0 {
-		listOptions := pkgbackup.NewDeleteBackupRequestListOptions(backup.Name, string(backup.UID))
-		err = c.deleteBackupRequestClient.DeleteBackupRequests(req.Namespace).DeleteCollection(context.TODO(), metav1.DeleteOptions{}, listOptions)
+		labelSelector, err := labels.Parse(fmt.Sprintf("%s=%s,%s=%s", velerov1api.BackupNameLabel, label.GetValidName(backup.Name), velerov1api.BackupUIDLabel, backup.UID))
+		if err != nil {
+			// Should not be here
+			r.logger.WithError(err).WithField("backup", kube.NamespaceAndName(backup)).Error("error creating label selector for the backup for deleting DeleteBackupRequests")
+			return ctrl.Result{}, nil
+		}
+		alldbr := &velerov1api.DeleteBackupRequest{}
+		err = r.DeleteAllOf(ctx, alldbr, client.MatchingLabelsSelector{
+			Selector: labelSelector,
+		}, client.InNamespace(dbr.Namespace))
 		if err != nil {
 			// If this errors, all we can do is log it.
-			c.logger.WithField("backup", kube.NamespaceAndName(backup)).Error("error deleting all associated DeleteBackupRequests after successfully deleting the backup")
+			r.logger.WithError(err).WithField("backup", kube.NamespaceAndName(backup)).Error("error deleting all associated DeleteBackupRequests after successfully deleting the backup")
 		}
 	}
+	log.Infof("Reconciliation done")

-	return nil
+	return ctrl.Result{}, nil
 }

-func volumeSnapshotterForSnapshotLocation(
-	namespace, snapshotLocationName string,
-	snapshotLocationLister velerov1listers.VolumeSnapshotLocationLister,
+func volumeSnapshottersForVSL(
+	ctx context.Context,
+	namespace, vslName string,
+	client client.Client,
 	pluginManager clientmgmt.Manager,
 ) (velero.VolumeSnapshotter, error) {
-	snapshotLocation, err := snapshotLocationLister.VolumeSnapshotLocations(namespace).Get(snapshotLocationName)
+	vsl := &velerov1api.VolumeSnapshotLocation{}
+	if err := client.Get(ctx, types.NamespacedName{
+		Namespace: namespace,
+		Name:      vslName,
+	}, vsl); err != nil {
+		return nil, errors.Wrapf(err, "error getting volume snapshot location %s", vslName)
+	}
+	volumeSnapshotter, err := pluginManager.GetVolumeSnapshotter(vsl.Spec.Provider)
 	if err != nil {
-		return nil, errors.Wrapf(err, "error getting volume snapshot location %s", snapshotLocationName)
+		return nil, errors.Wrapf(err, "error getting volume snapshotter for provider %s", vsl.Spec.Provider)
 	}

-	volumeSnapshotter, err := pluginManager.GetVolumeSnapshotter(snapshotLocation.Spec.Provider)
-	if err != nil {
-		return nil, errors.Wrapf(err, "error getting volume snapshotter for provider %s", snapshotLocation.Spec.Provider)
-	}
-
-	if err = volumeSnapshotter.Init(snapshotLocation.Spec.Config); err != nil {
-		return nil, errors.Wrapf(err, "error initializing volume snapshotter for volume snapshot location %s", snapshotLocationName)
+	if err = volumeSnapshotter.Init(vsl.Spec.Config); err != nil {
+		return nil, errors.Wrapf(err, "error initializing volume snapshotter for volume snapshot location %s", vslName)
 	}

 	return volumeSnapshotter, nil
 }

-func (c *backupDeletionController) deleteExistingDeletionRequests(req *velerov1api.DeleteBackupRequest, log logrus.FieldLogger) []error {
+func (r *backupDeletionReconciler) deleteExistingDeletionRequests(ctx context.Context, req *velerov1api.DeleteBackupRequest, log logrus.FieldLogger) []error {
 	log.Info("Removing existing deletion requests for backup")
+	dbrList := &velerov1api.DeleteBackupRequestList{}
 	selector := label.NewSelectorForBackup(req.Spec.BackupName)
-	dbrs, err := c.deleteBackupRequestLister.DeleteBackupRequests(req.Namespace).List(selector)
-	if err != nil {
+	if err := r.List(ctx, dbrList, &client.ListOptions{
+		Namespace:     req.Namespace,
+		LabelSelector: selector,
+	}); err != nil {
 		return []error{errors.Wrap(err, "error listing existing DeleteBackupRequests for backup")}
 	}
-
 	var errs []error
-	for _, dbr := range dbrs {
+	for _, dbr := range dbrList.Items {
 		if dbr.Name == req.Name {
 			continue
 		}
-
-		if err := c.deleteBackupRequestClient.DeleteBackupRequests(req.Namespace).Delete(context.TODO(), dbr.Name, metav1.DeleteOptions{}); err != nil {
+		if err := r.Delete(ctx, &dbr); err != nil {
 			errs = append(errs, errors.WithStack(err))
+		} else {
+			log.Infof("deletion request '%s' removed.", dbr.Name)
 		}
 	}
-
 	return errs
 }

-func (c *backupDeletionController) deleteResticSnapshots(backup *velerov1api.Backup) []error {
-	if c.resticMgr == nil {
+func (r *backupDeletionReconciler) deleteResticSnapshots(ctx context.Context, backup *velerov1api.Backup) []error {
+	if r.resticMgr == nil {
 		return nil
 	}

-	snapshots, err := restic.GetSnapshotsInBackup(backup, c.podvolumeBackupLister)
+	snapshots, err := restic.GetSnapshotsInBackup(ctx, backup, r.Client)
 	if err != nil {
 		return []error{err}
 	}

-	ctx, cancelFunc := context.WithTimeout(context.Background(), resticTimeout)
+	ctx2, cancelFunc := context.WithTimeout(ctx, resticTimeout)
 	defer cancelFunc()

 	var errs []error
 	for _, snapshot := range snapshots {
-		if err := c.resticMgr.Forget(ctx, snapshot); err != nil {
+		if err := r.resticMgr.Forget(ctx2, snapshot); err != nil {
 			errs = append(errs, err)
 		}
 	}
-
 	return errs
 }

-const deleteBackupRequestMaxAge = 24 * time.Hour
-
-func (c *backupDeletionController) deleteExpiredRequests() {
-	c.logger.Info("Checking for expired DeleteBackupRequests")
-	defer c.logger.Info("Done checking for expired DeleteBackupRequests")
-
-	// Our shared informer factory filters on a single namespace, so asking for all is ok here.
-	requests, err := c.deleteBackupRequestLister.List(labels.Everything())
+func (r *backupDeletionReconciler) patchDeleteBackupRequest(ctx context.Context, req *velerov1api.DeleteBackupRequest, mutate func(*velerov1api.DeleteBackupRequest)) (*velerov1api.DeleteBackupRequest, error) {
+	patchHelper, err := patch.NewHelper(req, r.Client)
 	if err != nil {
-		c.logger.WithError(err).Error("unable to check for expired DeleteBackupRequests")
-		return
+		return nil, errors.Wrap(err, "unable to get the patch helper")
 	}
-
-	now := c.clock.Now()
-
-	for _, req := range requests {
-		if req.Status.Phase != velerov1api.DeleteBackupRequestPhaseProcessed {
-			continue
-		}
-
-		age := now.Sub(req.CreationTimestamp.Time)
-		if age >= deleteBackupRequestMaxAge {
-			reqLog := c.logger.WithFields(logrus.Fields{"namespace": req.Namespace, "name": req.Name})
-			reqLog.Info("Deleting expired DeleteBackupRequest")
-
-			err = c.deleteBackupRequestClient.DeleteBackupRequests(req.Namespace).Delete(context.TODO(), req.Name, metav1.DeleteOptions{})
-			if err != nil {
-				reqLog.WithError(err).Error("Error deleting DeleteBackupRequest")
-			}
-		}
-	}
-}
-
-func (c *backupDeletionController) patchDeleteBackupRequest(req *velerov1api.DeleteBackupRequest, mutate func(*velerov1api.DeleteBackupRequest)) (*velerov1api.DeleteBackupRequest, error) {
-	// Record original json
-	oldData, err := json.Marshal(req)
-	if err != nil {
-		return nil, errors.Wrap(err, "error marshalling original DeleteBackupRequest")
-	}
-
 	// Mutate
 	mutate(req)
-
-	// Record new json
-	newData, err := json.Marshal(req)
-	if err != nil {
-		return nil, errors.Wrap(err, "error marshalling updated DeleteBackupRequest")
+	if err := patchHelper.Patch(ctx, req); err != nil {
+		return nil, errors.Wrap(err, "error patching the deletebackuprquest")
 	}
-
-	patchBytes, err := jsonpatch.CreateMergePatch(oldData, newData)
-	if err != nil {
-		return nil, errors.Wrap(err, "error creating json merge patch for DeleteBackupRequest")
-	}
-
-	req, err = c.deleteBackupRequestClient.DeleteBackupRequests(req.Namespace).Patch(context.TODO(), req.Name, types.MergePatchType, patchBytes, metav1.PatchOptions{})
-	if err != nil {
-		return nil, errors.Wrap(err, "error patching DeleteBackupRequest")
-	}
-
 	return req, nil
 }

-func (c *backupDeletionController) patchBackup(backup *velerov1api.Backup, mutate func(*velerov1api.Backup)) (*velerov1api.Backup, error) {
+func (r *backupDeletionReconciler) patchBackup(ctx context.Context, backup *velerov1api.Backup, mutate func(*velerov1api.Backup)) (*velerov1api.Backup, error) {
+	//TODO: The patchHelper can't be used here because the `backup/xxx/status` does not exist, until the bakcup resource is refactored
+
 	// Record original json
 	oldData, err := json.Marshal(backup)
 	if err != nil {
 		return nil, errors.Wrap(err, "error marshalling original Backup")
 	}

-	// Mutate
-	mutate(backup)
-
-	// Record new json
-	newData, err := json.Marshal(backup)
+	newBackup := backup.DeepCopy()
+	mutate(newBackup)
+	newData, err := json.Marshal(newBackup)
 	if err != nil {
 		return nil, errors.Wrap(err, "error marshalling updated Backup")
 	}
-
 	patchBytes, err := jsonpatch.CreateMergePatch(oldData, newData)
 	if err != nil {
 		return nil, errors.Wrap(err, "error creating json merge patch for Backup")
 	}

-	backup, err = c.backupClient.Backups(backup.Namespace).Patch(context.TODO(), backup.Name, types.MergePatchType, patchBytes, metav1.PatchOptions{})
-	if err != nil {
+	if err := r.Client.Patch(ctx, backup, client.RawPatch(types.MergePatchType, patchBytes)); err != nil {
 		return nil, errors.Wrap(err, "error patching Backup")
 	}
-
 	return backup, nil
 }
@@ -20,7 +20,9 @@ import (
 	"context"
 	"time"

+	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
 	snapshotterClientSet "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned"
+	snapshotv1listers "github.com/kubernetes-csi/external-snapshotter/client/v4/listers/volumesnapshot/v1"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	kuberrs "k8s.io/apimachinery/pkg/api/errors"
@@ -29,6 +31,8 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/kubernetes"

+	"github.com/vmware-tanzu/velero/pkg/util/kube"
+
 	"github.com/vmware-tanzu/velero/internal/storage"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/features"
@@ -48,6 +52,7 @@ type backupSyncController struct {
 	kbClient                client.Client
 	podVolumeBackupClient   velerov1client.PodVolumeBackupsGetter
 	backupLister            velerov1listers.BackupLister
+	csiVSLister             snapshotv1listers.VolumeSnapshotLister
 	csiSnapshotClient       *snapshotterClientSet.Clientset
 	kubeClient              kubernetes.Interface
 	namespace               string
@@ -62,6 +67,7 @@ func NewBackupSyncController(
 	kbClient client.Client,
 	podVolumeBackupClient velerov1client.PodVolumeBackupsGetter,
 	backupLister velerov1listers.BackupLister,
+	csiVSLister snapshotv1listers.VolumeSnapshotLister,
 	syncPeriod time.Duration,
 	namespace string,
 	csiSnapshotClient *snapshotterClientSet.Clientset,
@@ -85,6 +91,7 @@ func NewBackupSyncController(
 		defaultBackupLocation:   defaultBackupLocation,
 		defaultBackupSyncPeriod: syncPeriod,
 		backupLister:            backupLister,
+		csiVSLister:             csiVSLister,
 		csiSnapshotClient:       csiSnapshotClient,
 		kubeClient:              kubeClient,

@@ -358,11 +365,34 @@ func (c *backupSyncController) deleteOrphanedBackups(locationName string, backup
 		if backup.Status.Phase != velerov1api.BackupPhaseCompleted || backupStoreBackups.Has(backup.Name) {
 			continue
 		}
-
 		if err := c.backupClient.Backups(backup.Namespace).Delete(context.TODO(), backup.Name, metav1.DeleteOptions{}); err != nil {
 			log.WithError(errors.WithStack(err)).Error("Error deleting orphaned backup from cluster")
 		} else {
 			log.Debug("Deleted orphaned backup from cluster")
+			c.deleteCSISnapshotsByBackup(backup.Name, log)
 		}
 	}
 }
+
+func (c *backupSyncController) deleteCSISnapshotsByBackup(backupName string, log logrus.FieldLogger) {
+	if !features.IsEnabled(velerov1api.CSIFeatureFlag) {
+		return
+	}
+	m := client.MatchingLabels{velerov1api.BackupNameLabel: label.GetValidName(backupName)}
+	if vsList, err := c.csiVSLister.List(label.NewSelectorForBackup(label.GetValidName(backupName))); err != nil {
+		log.WithError(err).Warnf("Failed to list volumesnapshots for backup: %s, the deletion will be skipped", backupName)
+	} else {
+		for _, vs := range vsList {
+			name := kube.NamespaceAndName(vs.GetObjectMeta())
+			log.Debugf("Deleting volumesnapshot %s", name)
+			if err := c.kbClient.Delete(context.TODO(), vs); err != nil {
+				log.WithError(err).Warnf("Failed to delete volumesnapshot %s", name)
+			}
+		}
+	}
+	vsc := &snapshotv1api.VolumeSnapshotContent{}
+	log.Debugf("Deleting volumesnapshotcontents for backup: %s", backupName)
+	if err := c.kbClient.DeleteAllOf(context.TODO(), vsc, m); err != nil {
+		log.WithError(err).Warnf("Failed to delete volumesnapshotcontents for backup: %s", backupName)
+	}
+}
@@ -344,6 +344,7 @@ func TestBackupSyncControllerRun(t *testing.T) {
 				fakeClient,
 				client.VeleroV1(),
 				sharedInformers.Velero().V1().Backups().Lister(),
+				nil, // csiVSLister
 				time.Duration(0),
 				test.namespace,
 				nil, // csiSnapshotClient
@@ -565,6 +566,7 @@ func TestDeleteOrphanedBackups(t *testing.T) {
 				fakeClient,
 				client.VeleroV1(),
 				sharedInformers.Velero().V1().Backups().Lister(),
+				nil, // csiVSLister
 				time.Duration(0),
 				test.namespace,
 				nil, // csiSnapshotClient
@@ -659,6 +661,7 @@ func TestStorageLabelsInDeleteOrphanedBackups(t *testing.T) {
 				fakeClient,
 				client.VeleroV1(),
 				sharedInformers.Velero().V1().Backups().Lister(),
+				nil, // csiVSLister
 				time.Duration(0),
 				test.namespace,
 				nil, // csiSnapshotClient
@@ -30,7 +30,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/clock"
-	corev1listers "k8s.io/client-go/listers/core/v1"
 	"sigs.k8s.io/cluster-api/util/patch"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -60,9 +59,6 @@ type PodVolumeBackupReconciler struct {
 	FileSystem     filesystem.Interface
 	ResticExec     BackupExecuter
 	Log            logrus.FieldLogger
-
-	PvLister  corev1listers.PersistentVolumeLister
-	PvcLister corev1listers.PersistentVolumeClaimLister
 }

 // +kubebuilder:rbac:groups=velero.io,resources=podvolumebackups,verbs=get;list;watch;create;update;patch;delete
@@ -302,7 +298,7 @@ type resticDetails struct {
 }

 func (r *PodVolumeBackupReconciler) buildResticCommand(ctx context.Context, log *logrus.Entry, pvb *velerov1api.PodVolumeBackup, pod *corev1.Pod, details *resticDetails) (*restic.Command, error) {
-	volDir, err := kube.GetVolumeDirectory(log, pod, pvb.Spec.Volume, r.PvcLister, r.PvLister, r.Client)
+	volDir, err := kube.GetVolumeDirectory(ctx, log, pod, pvb.Spec.Volume, r.Client)
 	if err != nil {
 		return nil, errors.Wrap(err, "getting volume directory name")
 	}
@@ -18,13 +18,11 @@ package controller

 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"os"
 	"path/filepath"

-	jsonpatch "github.com/evanphx/json-patch"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	corev1api "k8s.io/api/core/v1"
@@ -33,191 +31,176 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/clock"
-	corev1informers "k8s.io/client-go/informers/core/v1"
-	corev1listers "k8s.io/client-go/listers/core/v1"
-	"k8s.io/client-go/tools/cache"
-	k8scache "sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/cluster-api/util/patch"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"

 	"github.com/vmware-tanzu/velero/internal/credentials"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
-	velerov1client "github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned/typed/velero/v1"
-	informers "github.com/vmware-tanzu/velero/pkg/generated/informers/externalversions/velero/v1"
-	listers "github.com/vmware-tanzu/velero/pkg/generated/listers/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/restic"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )

-type podVolumeRestoreController struct {
-	*genericController
-
-	podVolumeRestoreClient velerov1client.PodVolumeRestoresGetter
-	podVolumeRestoreLister listers.PodVolumeRestoreLister
-	podLister              corev1listers.PodLister
-	pvcLister              corev1listers.PersistentVolumeClaimLister
-	pvLister               corev1listers.PersistentVolumeLister
-	backupLocationInformer k8scache.Informer
-	kbClient               client.Client
-	nodeName               string
-	credentialsFileStore   credentials.FileStore
-
-	processRestoreFunc func(*velerov1api.PodVolumeRestore) error
-	fileSystem         filesystem.Interface
-	clock              clock.Clock
+func NewPodVolumeRestoreReconciler(logger logrus.FieldLogger, client client.Client, credentialsFileStore credentials.FileStore) *PodVolumeRestoreReconciler {
+	return &PodVolumeRestoreReconciler{
+		Client:               client,
+		logger:               logger.WithField("controller", "PodVolumeRestore"),
+		credentialsFileStore: credentialsFileStore,
+		fileSystem:           filesystem.NewFileSystem(),
+		clock:                &clock.RealClock{},
+	}
 }

-// NewPodVolumeRestoreController creates a new pod volume restore controller.
-func NewPodVolumeRestoreController(
-	logger logrus.FieldLogger,
-	podVolumeRestoreInformer informers.PodVolumeRestoreInformer,
-	podVolumeRestoreClient velerov1client.PodVolumeRestoresGetter,
-	podInformer cache.SharedIndexInformer,
-	pvcInformer corev1informers.PersistentVolumeClaimInformer,
-	pvInformer corev1informers.PersistentVolumeInformer,
-	kbClient client.Client,
-	nodeName string,
-	credentialsFileStore credentials.FileStore,
-) Interface {
-	c := &podVolumeRestoreController{
-		genericController:      newGenericController(PodVolumeRestore, logger),
-		podVolumeRestoreClient: podVolumeRestoreClient,
-		podVolumeRestoreLister: podVolumeRestoreInformer.Lister(),
-		podLister:              corev1listers.NewPodLister(podInformer.GetIndexer()),
-		pvcLister:              pvcInformer.Lister(),
-		pvLister:               pvInformer.Lister(),
-		kbClient:               kbClient,
-		nodeName:               nodeName,
-		credentialsFileStore:   credentialsFileStore,
-
-		fileSystem: filesystem.NewFileSystem(),
-		clock:      &clock.RealClock{},
-	}
-
-	c.syncHandler = c.processQueueItem
-	c.cacheSyncWaiters = append(
-		c.cacheSyncWaiters,
-		podVolumeRestoreInformer.Informer().HasSynced,
-		podInformer.HasSynced,
-		pvcInformer.Informer().HasSynced,
-	)
-	c.processRestoreFunc = c.processRestore
-
-	podVolumeRestoreInformer.Informer().AddEventHandler(
-		cache.ResourceEventHandlerFuncs{
-			AddFunc: c.pvrHandler,
-			UpdateFunc: func(_, obj interface{}) {
-				c.pvrHandler(obj)
-			},
-		},
-	)
-
-	podInformer.AddEventHandler(
-		cache.ResourceEventHandlerFuncs{
-			AddFunc: c.podHandler,
-			UpdateFunc: func(_, obj interface{}) {
-				c.podHandler(obj)
-			},
-		},
-	)
-
-	return c
+type PodVolumeRestoreReconciler struct {
+	client.Client
+	logger               logrus.FieldLogger
+	credentialsFileStore credentials.FileStore
+	fileSystem           filesystem.Interface
+	clock                clock.Clock
 }

-func (c *podVolumeRestoreController) pvrHandler(obj interface{}) {
-	pvr := obj.(*velerov1api.PodVolumeRestore)
-	log := loggerForPodVolumeRestore(c.logger, pvr)
+// +kubebuilder:rbac:groups=velero.io,resources=podvolumerestores,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=velero.io,resources=podvolumerestores/status,verbs=get;update;patch
+// +kubebuilder:rbac:groups="",resources=pods,verbs=get
+// +kubebuilder:rbac:groups="",resources=persistentvolumes,verbs=get
+// +kubebuilder:rbac:groups="",resources=persistentvolumerclaims,verbs=get

-	if !isPVRNew(pvr) {
-		log.Debugf("Restore is not new, not enqueuing")
-		return
-	}
+func (c *PodVolumeRestoreReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	log := c.logger.WithField("PodVolumeRestore", req.NamespacedName.String())

-	pod, err := c.podLister.Pods(pvr.Spec.Pod.Namespace).Get(pvr.Spec.Pod.Name)
-	if apierrors.IsNotFound(err) {
-		log.WithError(err).Debugf("Restore's pod %s/%s not found, not enqueueing.", pvr.Spec.Pod.Namespace, pvr.Spec.Pod.Name)
-		return
-	}
-	if err != nil {
-		log.WithError(err).Errorf("Unable to get restore's pod %s/%s, not enqueueing.", pvr.Spec.Pod.Namespace, pvr.Spec.Pod.Name)
-		return
-	}
-
-	if !isPodOnNode(pod, c.nodeName) {
-		log.Debugf("Restore's pod is not on this node, not enqueuing")
-		return
-	}
-
-	if !isResticInitContainerRunning(pod) {
-		log.Debug("Restore's pod is not running restic-wait init container, not enqueuing")
-		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)
-}
-
-func (c *podVolumeRestoreController) podHandler(obj interface{}) {
-	pod := obj.(*corev1api.Pod)
-	log := c.logger.WithField("key", kube.NamespaceAndName(pod))
-
-	// the pod should always be for this node since the podInformer is filtered
-	// based on node, so this is just a failsafe.
-	if !isPodOnNode(pod, c.nodeName) {
-		return
-	}
-
-	if !isResticInitContainerRunning(pod) {
-		log.Debug("Pod is not running restic-wait init container, not enqueuing restores for pod")
-		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()
-
-	pvrs, err := c.podVolumeRestoreLister.List(selector)
-	if err != nil {
-		log.WithError(err).Error("Unable to list pod volume restores")
-		return
-	}
-
-	if len(pvrs) == 0 {
-		return
-	}
-
-	for _, pvr := range pvrs {
-		log := loggerForPodVolumeRestore(log, pvr)
-		if !isPVRNew(pvr) {
-			log.Debug("Restore is not new, not enqueuing")
-			continue
+	pvr := &velerov1api.PodVolumeRestore{}
+	if err := c.Get(ctx, types.NamespacedName{Namespace: req.Namespace, Name: req.Name}, pvr); err != nil {
+		if apierrors.IsNotFound(err) {
+			log.Warn("PodVolumeRestore not found, skip")
+			return ctrl.Result{}, nil
 		}
-		log.Debug("Enqueuing")
-		c.enqueue(pvr)
+		log.WithError(err).Error("Unable to get the PodVolumeRestore")
+		return ctrl.Result{}, err
 	}
+	log = log.WithField("pod", fmt.Sprintf("%s/%s", pvr.Spec.Pod.Namespace, pvr.Spec.Pod.Name))
+	if len(pvr.OwnerReferences) == 1 {
+		log = log.WithField("restore", fmt.Sprintf("%s/%s", pvr.Namespace, pvr.OwnerReferences[0].Name))
+	}
+
+	shouldProcess, pod, err := c.shouldProcess(ctx, log, pvr)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+	if !shouldProcess {
+		return ctrl.Result{}, nil
+	}
+
+	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)
+	}
+
+	patchHelper, err := patch.NewHelper(pvr, c.Client)
+	if err != nil {
+		log.WithError(err).Error("Unable to new patch helper")
+		return ctrl.Result{}, err
+	}
+
+	log.Info("Restore starting")
+	pvr.Status.Phase = velerov1api.PodVolumeRestorePhaseInProgress
+	pvr.Status.StartTimestamp = &metav1.Time{Time: c.clock.Now()}
+	if err = patchHelper.Patch(ctx, pvr); err != nil {
+		log.WithError(err).Error("Unable to update status to in progress")
+		return ctrl.Result{}, err
+	}
+	if err = c.processRestore(ctx, pvr, pod, log); err != nil {
+		pvr.Status.Phase = velerov1api.PodVolumeRestorePhaseFailed
+		pvr.Status.Message = err.Error()
+		pvr.Status.CompletionTimestamp = &metav1.Time{Time: c.clock.Now()}
+		if e := patchHelper.Patch(ctx, pvr); e != nil {
+			log.WithError(err).Error("Unable to update status to failed")
+		}
+
+		log.WithError(err).Error("Unable to process the PodVolumeRestore")
+		return ctrl.Result{}, err
+	}
+
+	pvr.Status.Phase = velerov1api.PodVolumeRestorePhaseCompleted
+	pvr.Status.CompletionTimestamp = &metav1.Time{Time: c.clock.Now()}
+	if err = patchHelper.Patch(ctx, pvr); err != nil {
+		log.WithError(err).Error("Unable to update status to completed")
+		return ctrl.Result{}, err
+	}
+	log.Info("Restore completed")
+	return ctrl.Result{}, nil
+}
+
+func (c *PodVolumeRestoreReconciler) shouldProcess(ctx context.Context, log logrus.FieldLogger, pvr *velerov1api.PodVolumeRestore) (bool, *corev1api.Pod, error) {
+	if !isPVRNew(pvr) {
+		log.Debug("PodVolumeRestore is not new, skip")
+		return false, nil, nil
+	}
+
+	// we filter the pods during the initialization of cache, if we can get a pod here, the pod must be in the same node with the controller
+	// so we don't need to compare the node anymore
+	pod := &corev1api.Pod{}
+	if err := c.Get(ctx, types.NamespacedName{Namespace: pvr.Spec.Pod.Namespace, Name: pvr.Spec.Pod.Name}, pod); err != nil {
+		if apierrors.IsNotFound(err) {
+			log.WithError(err).Debug("Pod not found on this node, skip")
+			return false, nil, nil
+		}
+		log.WithError(err).Error("Unable to get pod")
+		return false, nil, err
+	}
+
+	if !isResticInitContainerRunning(pod) {
+		log.Debug("Pod is not running restic-wait init container, skip")
+		return false, nil, nil
+	}
+
+	return true, pod, nil
+}
+
+func (c *PodVolumeRestoreReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	mgr.GetConfig()
+
+	// The pod may not being scheduled at the point when its PVRs are initially reconciled.
+	// By watching the pods, we can trigger the PVR reconciliation again once the pod is finally scheduled on the node.
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&velerov1api.PodVolumeRestore{}).
+		Watches(&source.Kind{Type: &corev1api.Pod{}}, handler.EnqueueRequestsFromMapFunc(c.findVolumeRestoresForPod)).
+		Complete(c)
+}
+
+func (c *PodVolumeRestoreReconciler) findVolumeRestoresForPod(pod client.Object) []reconcile.Request {
+	list := &velerov1api.PodVolumeRestoreList{}
+	options := &client.ListOptions{
+		LabelSelector: labels.Set(map[string]string{
+			velerov1api.PodUIDLabel: string(pod.GetUID()),
+		}).AsSelector(),
+	}
+	if err := c.List(context.TODO(), list, options); err != nil {
+		c.logger.WithField("pod", fmt.Sprintf("%s/%s", pod.GetNamespace(), pod.GetName())).WithError(err).
+			Error("unable to list PodVolumeRestores")
+		return []reconcile.Request{}
+	}
+	requests := make([]reconcile.Request, len(list.Items))
+	for i, item := range list.Items {
+		requests[i] = reconcile.Request{
+			NamespacedName: types.NamespacedName{
+				Namespace: item.GetNamespace(),
+				Name:      item.GetName(),
+			},
+		}
+	}
+	return requests
 }

 func isPVRNew(pvr *velerov1api.PodVolumeRestore) bool {
 	return pvr.Status.Phase == "" || pvr.Status.Phase == velerov1api.PodVolumeRestorePhaseNew
 }

-func isPodOnNode(pod *corev1api.Pod, node string) bool {
-	return pod.Spec.NodeName == node
-}
-
 func isResticInitContainerRunning(pod *corev1api.Pod) bool {
 	// Restic wait container can be anywhere in the list of init containers, but must be running.
 	i := getResticInitContainerIndex(pod)
@@ -237,92 +220,6 @@ func getResticInitContainerIndex(pod *corev1api.Pod) int {
 	return -1
 }

-func (c *podVolumeRestoreController) processQueueItem(key string) error {
-	log := c.logger.WithField("key", key)
-	log.Debug("Running processQueueItem")
-
-	ns, name, err := cache.SplitMetaNamespaceKey(key)
-	if err != nil {
-		log.WithError(errors.WithStack(err)).Error("error splitting queue key")
-		return nil
-	}
-
-	req, err := c.podVolumeRestoreLister.PodVolumeRestores(ns).Get(name)
-	if apierrors.IsNotFound(err) {
-		log.Debug("Unable to find PodVolumeRestore")
-		return nil
-	}
-	if err != nil {
-		return errors.Wrap(err, "error getting PodVolumeRestore")
-	}
-
-	// Don't mutate the shared cache
-	reqCopy := req.DeepCopy()
-	return c.processRestoreFunc(reqCopy)
-}
-
-func loggerForPodVolumeRestore(baseLogger logrus.FieldLogger, req *velerov1api.PodVolumeRestore) logrus.FieldLogger {
-	log := baseLogger.WithFields(logrus.Fields{
-		"namespace": req.Namespace,
-		"name":      req.Name,
-	})
-
-	if len(req.OwnerReferences) == 1 {
-		log = log.WithField("restore", fmt.Sprintf("%s/%s", req.Namespace, req.OwnerReferences[0].Name))
-	}
-
-	return log
-}
-
-func (c *podVolumeRestoreController) processRestore(req *velerov1api.PodVolumeRestore) error {
-	log := loggerForPodVolumeRestore(c.logger, req)
-
-	log.Info("Restore starting")
-
-	var err error
-
-	// update status to InProgress
-	req, err = c.patchPodVolumeRestore(req, func(r *velerov1api.PodVolumeRestore) {
-		r.Status.Phase = velerov1api.PodVolumeRestorePhaseInProgress
-		r.Status.StartTimestamp = &metav1.Time{Time: c.clock.Now()}
-	})
-	if err != nil {
-		log.WithError(err).Error("Error setting PodVolumeRestore startTimestamp and phase to InProgress")
-		return errors.WithStack(err)
-	}
-
-	pod, err := c.podLister.Pods(req.Spec.Pod.Namespace).Get(req.Spec.Pod.Name)
-	if err != nil {
-		log.WithError(err).Errorf("Error getting pod %s/%s", req.Spec.Pod.Namespace, req.Spec.Pod.Name)
-		return c.failRestore(req, errors.Wrap(err, "error getting pod").Error(), log)
-	}
-
-	volumeDir, err := kube.GetVolumeDirectory(log, pod, req.Spec.Volume, c.pvcLister, c.pvLister, c.kbClient)
-	if err != nil {
-		log.WithError(err).Error("Error getting volume directory name")
-		return c.failRestore(req, errors.Wrap(err, "error getting volume directory name").Error(), log)
-	}
-
-	// execute the restore process
-	if err := c.restorePodVolume(req, volumeDir, log); err != nil {
-		log.WithError(err).Error("Error restoring volume")
-		return c.failRestore(req, errors.Wrap(err, "error restoring volume").Error(), log)
-	}
-
-	// update status to Completed
-	if _, err = c.patchPodVolumeRestore(req, func(r *velerov1api.PodVolumeRestore) {
-		r.Status.Phase = velerov1api.PodVolumeRestorePhaseCompleted
-		r.Status.CompletionTimestamp = &metav1.Time{Time: c.clock.Now()}
-	}); err != nil {
-		log.WithError(err).Error("Error setting PodVolumeRestore completionTimestamp and phase to Completed")
-		return err
-	}
-
-	log.Info("Restore completed")
-
-	return nil
-}
-
 func singlePathMatch(path string) (string, error) {
 	matches, err := filepath.Glob(path)
 	if err != nil {
@@ -336,7 +233,12 @@ func singlePathMatch(path string) (string, error) {
 	return matches[0], nil
 }

-func (c *podVolumeRestoreController) restorePodVolume(req *velerov1api.PodVolumeRestore, volumeDir string, log logrus.FieldLogger) error {
+func (c *PodVolumeRestoreReconciler) processRestore(ctx context.Context, req *velerov1api.PodVolumeRestore, pod *corev1api.Pod, log logrus.FieldLogger) error {
+	volumeDir, err := kube.GetVolumeDirectory(ctx, log, pod, req.Spec.Volume, c.Client)
+	if err != nil {
+		return errors.Wrap(err, "error getting volume directory name")
+	}
+
 	// Get the full path of the new volume's directory as mounted in the daemonset pod, which
 	// will look like: /host_pods/<new-pod-uid>/volumes/<volume-plugin-name>/<volume-dir>
 	volumePath, err := singlePathMatch(fmt.Sprintf("/host_pods/%s/volumes/*/%s", string(req.Spec.Pod.UID), volumeDir))
@@ -346,8 +248,7 @@ func (c *podVolumeRestoreController) restorePodVolume(req *velerov1api.PodVolume

 	credsFile, err := c.credentialsFileStore.Path(restic.RepoKeySelector())
 	if err != nil {
-		log.WithError(err).Error("Error creating temp restic credentials file")
-		return c.failRestore(req, errors.Wrap(err, "error creating temp restic credentials file").Error(), log)
+		return errors.Wrap(err, "error creating temp restic credentials file")
 	}
 	// ignore error since there's nothing we can do and it's a temp file.
 	defer os.Remove(credsFile)
@@ -360,11 +261,11 @@ func (c *podVolumeRestoreController) restorePodVolume(req *velerov1api.PodVolume
 	)

 	backupLocation := &velerov1api.BackupStorageLocation{}
-	if err := c.kbClient.Get(context.Background(), client.ObjectKey{
+	if err := c.Get(ctx, client.ObjectKey{
 		Namespace: req.Namespace,
 		Name:      req.Spec.BackupStorageLocation,
 	}, backupLocation); err != nil {
-		return c.failRestore(req, errors.Wrap(err, "error getting backup storage location").Error(), log)
+		return errors.Wrap(err, "error getting backup storage location")
 	}

 	// if there's a caCert on the ObjectStorage, write it to disk so that it can be passed to restic
@@ -381,7 +282,7 @@ func (c *podVolumeRestoreController) restorePodVolume(req *velerov1api.PodVolume

 	env, err := restic.CmdEnv(backupLocation, c.credentialsFileStore)
 	if err != nil {
-		return c.failRestore(req, errors.Wrap(err, "error setting restic cmd env").Error(), log)
+		return errors.Wrap(err, "error setting restic cmd env")
 	}
 	resticCmd.Env = env

@@ -432,55 +333,18 @@ func (c *podVolumeRestoreController) restorePodVolume(req *velerov1api.PodVolume
 	return nil
 }

-func (c *podVolumeRestoreController) patchPodVolumeRestore(req *velerov1api.PodVolumeRestore, mutate func(*velerov1api.PodVolumeRestore)) (*velerov1api.PodVolumeRestore, error) {
-	// Record original json
-	oldData, err := json.Marshal(req)
-	if err != nil {
-		return nil, errors.Wrap(err, "error marshalling original PodVolumeRestore")
-	}
-
-	// Mutate
-	mutate(req)
-
-	// Record new json
-	newData, err := json.Marshal(req)
-	if err != nil {
-		return nil, errors.Wrap(err, "error marshalling updated PodVolumeRestore")
-	}
-
-	patchBytes, err := jsonpatch.CreateMergePatch(oldData, newData)
-	if err != nil {
-		return nil, errors.Wrap(err, "error creating json merge patch for PodVolumeRestore")
-	}
-
-	req, err = c.podVolumeRestoreClient.PodVolumeRestores(req.Namespace).Patch(context.TODO(), req.Name, types.MergePatchType, patchBytes, metav1.PatchOptions{})
-	if err != nil {
-		return nil, errors.Wrap(err, "error patching PodVolumeRestore")
-	}
-
-	return req, nil
-}
-
-func (c *podVolumeRestoreController) failRestore(req *velerov1api.PodVolumeRestore, msg string, log logrus.FieldLogger) error {
-	if _, err := c.patchPodVolumeRestore(req, func(pvr *velerov1api.PodVolumeRestore) {
-		pvr.Status.Phase = velerov1api.PodVolumeRestorePhaseFailed
-		pvr.Status.Message = msg
-		pvr.Status.CompletionTimestamp = &metav1.Time{Time: c.clock.Now()}
-	}); err != nil {
-		log.WithError(err).Error("Error setting PodVolumeRestore phase to Failed")
-		return err
-	}
-	return nil
-}
-
 // updateRestoreProgressFunc returns a func that takes progress info and patches
 // the PVR with the new progress
-func (c *podVolumeRestoreController) updateRestoreProgressFunc(req *velerov1api.PodVolumeRestore, log logrus.FieldLogger) func(velerov1api.PodVolumeOperationProgress) {
+func (c *PodVolumeRestoreReconciler) updateRestoreProgressFunc(req *velerov1api.PodVolumeRestore, log logrus.FieldLogger) func(velerov1api.PodVolumeOperationProgress) {
 	return func(progress velerov1api.PodVolumeOperationProgress) {
-		if _, err := c.patchPodVolumeRestore(req, func(r *velerov1api.PodVolumeRestore) {
-			r.Status.Progress = progress
-		}); err != nil {
-			log.WithError(err).Error("error updating PodVolumeRestore progress")
+		helper, err := patch.NewHelper(req, c.Client)
+		if err != nil {
+			log.WithError(err).Error("Unable to new patch helper")
+			return
+		}
+		req.Status.Progress = progress
+		if err = helper.Patch(context.Background(), req); err != nil {
+			log.WithError(err).Error("Unable to update PodVolumeRestore progress")
 		}
 	}
 }
@@ -17,64 +17,64 @@ limitations under the License.
 package controller

 import (
+	"context"
 	"testing"
 	"time"

-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"k8s.io/apimachinery/pkg/runtime"
+
+	"github.com/sirupsen/logrus"
+
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	"github.com/stretchr/testify/assert"
 	corev1api "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/sets"
-	corev1listers "k8s.io/client-go/listers/core/v1"
-	"k8s.io/client-go/tools/cache"

 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
-	velerofake "github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned/fake"
-	veleroinformers "github.com/vmware-tanzu/velero/pkg/generated/informers/externalversions"
-	velerov1listers "github.com/vmware-tanzu/velero/pkg/generated/listers/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/restic"
-	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 )

-func TestPVRHandler(t *testing.T) {
+func TestShouldProcess(t *testing.T) {
 	controllerNode := "foo"

 	tests := []struct {
-		name          string
-		obj           *velerov1api.PodVolumeRestore
-		pod           *corev1api.Pod
-		shouldEnqueue bool
+		name            string
+		obj             *velerov1api.PodVolumeRestore
+		pod             *corev1api.Pod
+		shouldProcessed bool
 	}{
 		{
-			name: "InProgress phase pvr should not be enqueued",
+			name: "InProgress phase pvr should not be processed",
 			obj: &velerov1api.PodVolumeRestore{
 				Status: velerov1api.PodVolumeRestoreStatus{
 					Phase: velerov1api.PodVolumeRestorePhaseInProgress,
 				},
 			},
-			shouldEnqueue: false,
+			shouldProcessed: false,
 		},
 		{
-			name: "Completed phase pvr should not be enqueued",
+			name: "Completed phase pvr should not be processed",
 			obj: &velerov1api.PodVolumeRestore{
 				Status: velerov1api.PodVolumeRestoreStatus{
 					Phase: velerov1api.PodVolumeRestorePhaseCompleted,
 				},
 			},
-			shouldEnqueue: false,
+			shouldProcessed: false,
 		},
 		{
-			name: "Failed phase pvr should not be enqueued",
+			name: "Failed phase pvr should not be processed",
 			obj: &velerov1api.PodVolumeRestore{
 				Status: velerov1api.PodVolumeRestoreStatus{
 					Phase: velerov1api.PodVolumeRestorePhaseFailed,
 				},
 			},
-			shouldEnqueue: false,
+			shouldProcessed: false,
 		},
 		{
-			name: "Unable to get pvr's pod should not be enqueued",
+			name: "Unable to get pvr's pod should not be processed",
 			obj: &velerov1api.PodVolumeRestore{
 				Spec: velerov1api.PodVolumeRestoreSpec{
 					Pod: corev1api.ObjectReference{
@@ -86,50 +86,10 @@ func TestPVRHandler(t *testing.T) {
 					Phase: "",
 				},
 			},
-			shouldEnqueue: false,
+			shouldProcessed: false,
 		},
 		{
-			name: "Empty phase pvr with pod not on node running init container should not be enqueued",
-			obj: &velerov1api.PodVolumeRestore{
-				Spec: velerov1api.PodVolumeRestoreSpec{
-					Pod: corev1api.ObjectReference{
-						Namespace: "ns-1",
-						Name:      "pod-1",
-					},
-				},
-				Status: velerov1api.PodVolumeRestoreStatus{
-					Phase: "",
-				},
-			},
-			pod: &corev1api.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					Namespace: "ns-1",
-					Name:      "pod-1",
-				},
-				Spec: corev1api.PodSpec{
-					NodeName: "some-other-node",
-					InitContainers: []corev1api.Container{
-						{
-							Name: restic.InitContainer,
-						},
-					},
-				},
-				Status: corev1api.PodStatus{
-					InitContainerStatuses: []corev1api.ContainerStatus{
-						{
-							State: corev1api.ContainerState{
-								Running: &corev1api.ContainerStateRunning{
-									StartedAt: metav1.Time{Time: time.Now()},
-								},
-							},
-						},
-					},
-				},
-			},
-			shouldEnqueue: false,
-		},
-		{
-			name: "Empty phase pvr with pod on node not running init container should not be enqueued",
+			name: "Empty phase pvr with pod on node not running init container should not be processed",
 			obj: &velerov1api.PodVolumeRestore{
 				Spec: velerov1api.PodVolumeRestoreSpec{
 					Pod: corev1api.ObjectReference{
@@ -162,7 +122,7 @@ func TestPVRHandler(t *testing.T) {
 					},
 				},
 			},
-			shouldEnqueue: false,
+			shouldProcessed: false,
 		},
 		{
 			name: "Empty phase pvr with pod on node running init container should be enqueued",
@@ -202,220 +162,23 @@ func TestPVRHandler(t *testing.T) {
 					},
 				},
 			},
-			shouldEnqueue: true,
+			shouldProcessed: true,
 		},
 	}

 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			var (
-				podInformer = cache.NewSharedIndexInformer(nil, new(corev1api.Pod), 0, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
-				c           = &podVolumeRestoreController{
-					genericController: newGenericController(PodVolumeRestore, velerotest.NewLogger()),
-					podLister:         corev1listers.NewPodLister(podInformer.GetIndexer()),
-					nodeName:          controllerNode,
-				}
-			)
-
+			builder := fake.NewClientBuilder()
 			if test.pod != nil {
-				require.NoError(t, podInformer.GetStore().Add(test.pod))
+				builder.WithObjects(test.pod)
+			}
+			c := &PodVolumeRestoreReconciler{
+				logger: logrus.New(),
+				Client: builder.Build(),
 			}

-			c.pvrHandler(test.obj)
-
-			if !test.shouldEnqueue {
-				assert.Equal(t, 0, c.queue.Len())
-				return
-			}
-
-			require.Equal(t, 1, c.queue.Len())
-		})
-	}
-}
-
-func TestPodHandler(t *testing.T) {
-	controllerNode := "foo"
-
-	tests := []struct {
-		name              string
-		pod               *corev1api.Pod
-		podVolumeRestores []*velerov1api.PodVolumeRestore
-		expectedEnqueues  sets.String
-	}{
-		{
-			name: "pod on controller node running restic init container with multiple PVRs has new ones enqueued",
-			pod: &corev1api.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					Namespace: "ns-1",
-					Name:      "pod-1",
-					UID:       types.UID("uid"),
-				},
-				Spec: corev1api.PodSpec{
-					NodeName: controllerNode,
-					InitContainers: []corev1api.Container{
-						{
-							Name: restic.InitContainer,
-						},
-					},
-				},
-				Status: corev1api.PodStatus{
-					InitContainerStatuses: []corev1api.ContainerStatus{
-						{
-							State: corev1api.ContainerState{
-								Running: &corev1api.ContainerStateRunning{StartedAt: metav1.Time{Time: time.Now()}},
-							},
-						},
-					},
-				},
-			},
-			podVolumeRestores: []*velerov1api.PodVolumeRestore{
-				{
-					ObjectMeta: metav1.ObjectMeta{
-						Namespace: "ns-1",
-						Name:      "pvr-1",
-						Labels: map[string]string{
-							velerov1api.PodUIDLabel: "uid",
-						},
-					},
-				},
-				{
-					ObjectMeta: metav1.ObjectMeta{
-						Namespace: "ns-1",
-						Name:      "pvr-2",
-						Labels: map[string]string{
-							velerov1api.PodUIDLabel: "uid",
-						},
-					},
-				},
-				{
-					ObjectMeta: metav1.ObjectMeta{
-						Namespace: "ns-1",
-						Name:      "pvr-3",
-						Labels: map[string]string{
-							velerov1api.PodUIDLabel: "uid",
-						},
-					},
-					Status: velerov1api.PodVolumeRestoreStatus{
-						Phase: velerov1api.PodVolumeRestorePhaseInProgress,
-					},
-				},
-				{
-					ObjectMeta: metav1.ObjectMeta{
-						Namespace: "ns-1",
-						Name:      "pvr-4",
-						Labels: map[string]string{
-							velerov1api.PodUIDLabel: "some-other-pod",
-						},
-					},
-				},
-			},
-			expectedEnqueues: sets.NewString("ns-1/pvr-1", "ns-1/pvr-2"),
-		},
-		{
-			name: "pod on controller node not running restic init container doesn't have PVRs enqueued",
-			pod: &corev1api.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					Namespace: "ns-1",
-					Name:      "pod-1",
-					UID:       types.UID("uid"),
-				},
-				Spec: corev1api.PodSpec{
-					NodeName: controllerNode,
-					InitContainers: []corev1api.Container{
-						{
-							Name: restic.InitContainer,
-						},
-					},
-				},
-				Status: corev1api.PodStatus{
-					InitContainerStatuses: []corev1api.ContainerStatus{
-						{
-							State: corev1api.ContainerState{},
-						},
-					},
-				},
-			},
-			podVolumeRestores: []*velerov1api.PodVolumeRestore{
-				{
-					ObjectMeta: metav1.ObjectMeta{
-						Namespace: "ns-1",
-						Name:      "pvr-1",
-						Labels: map[string]string{
-							velerov1api.PodUIDLabel: "uid",
-						},
-					},
-				},
-			},
-		},
-		{
-			name: "pod not running on controller node doesn't have PVRs enqueued",
-			pod: &corev1api.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					Namespace: "ns-1",
-					Name:      "pod-1",
-					UID:       types.UID("uid"),
-				},
-				Spec: corev1api.PodSpec{
-					NodeName: "some-other-node",
-					InitContainers: []corev1api.Container{
-						{
-							Name: restic.InitContainer,
-						},
-					},
-				},
-				Status: corev1api.PodStatus{
-					InitContainerStatuses: []corev1api.ContainerStatus{
-						{
-							State: corev1api.ContainerState{
-								Running: &corev1api.ContainerStateRunning{StartedAt: metav1.Time{Time: time.Now()}},
-							},
-						},
-					},
-				},
-			},
-			podVolumeRestores: []*velerov1api.PodVolumeRestore{
-				{
-					ObjectMeta: metav1.ObjectMeta{
-						Namespace: "ns-1",
-						Name:      "pvr-1",
-						Labels: map[string]string{
-							velerov1api.PodUIDLabel: "uid",
-						},
-					},
-				},
-			},
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			var (
-				client      = velerofake.NewSimpleClientset()
-				informers   = veleroinformers.NewSharedInformerFactory(client, 0)
-				pvrInformer = informers.Velero().V1().PodVolumeRestores()
-				c           = &podVolumeRestoreController{
-					genericController:      newGenericController(PodVolumeRestore, velerotest.NewLogger()),
-					podVolumeRestoreLister: velerov1listers.NewPodVolumeRestoreLister(pvrInformer.Informer().GetIndexer()),
-					nodeName:               controllerNode,
-				}
-			)
-
-			if len(test.podVolumeRestores) > 0 {
-				for _, pvr := range test.podVolumeRestores {
-					require.NoError(t, pvrInformer.Informer().GetStore().Add(pvr))
-				}
-			}
-
-			c.podHandler(test.pod)
-
-			require.Equal(t, len(test.expectedEnqueues), c.queue.Len())
-
-			itemCount := c.queue.Len()
-
-			for i := 0; i < itemCount; i++ {
-				item, _ := c.queue.Get()
-				assert.True(t, test.expectedEnqueues.Has(item.(string)))
-			}
+			shouldProcess, _, _ := c.shouldProcess(context.Background(), c.logger, test.obj)
+			require.Equal(t, test.shouldProcessed, shouldProcess)
 		})
 	}
 }
@@ -437,17 +200,6 @@ func TestIsPVRNew(t *testing.T) {
 	}
 }

-func TestIsPodOnNode(t *testing.T) {
-	pod := &corev1api.Pod{}
-	assert.False(t, isPodOnNode(pod, "bar"))
-
-	pod.Spec.NodeName = "foo"
-	assert.False(t, isPodOnNode(pod, "bar"))
-
-	pod.Spec.NodeName = "bar"
-	assert.True(t, isPodOnNode(pod, "bar"))
-}
-
 func TestIsResticContainerRunning(t *testing.T) {
 	tests := []struct {
 		name     string
@@ -720,3 +472,44 @@ func TestGetResticInitContainerIndex(t *testing.T) {
 		})
 	}
 }
+
+func TestFindVolumeRestoresForPod(t *testing.T) {
+	pod := &corev1api.Pod{}
+	pod.UID = "uid"
+
+	scheme := runtime.NewScheme()
+	scheme.AddKnownTypes(velerov1api.SchemeGroupVersion, &velerov1api.PodVolumeRestore{}, &velerov1api.PodVolumeRestoreList{})
+	clientBuilder := fake.NewClientBuilder().WithScheme(scheme)
+
+	// no matching PVR
+	reconciler := &PodVolumeRestoreReconciler{
+		Client: clientBuilder.Build(),
+		logger: logrus.New(),
+	}
+	requests := reconciler.findVolumeRestoresForPod(pod)
+	assert.Len(t, requests, 0)
+
+	// contain one matching PVR
+	reconciler.Client = clientBuilder.WithLists(&velerov1api.PodVolumeRestoreList{
+		Items: []velerov1api.PodVolumeRestore{
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "pvr1",
+					Labels: map[string]string{
+						velerov1api.PodUIDLabel: string(pod.GetUID()),
+					},
+				},
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "pvr2",
+					Labels: map[string]string{
+						velerov1api.PodUIDLabel: "non-matching-uid",
+					},
+				},
+			},
+		},
+	}).Build()
+	requests = reconciler.findVolumeRestoresForPod(pod)
+	assert.Len(t, requests, 1)
+}
@@ -144,13 +144,12 @@ func NewRestoreController(
 				restore := obj.(*api.Restore)

 				switch restore.Status.Phase {
-				case "", api.RestorePhaseNew:
-					// only process new restores
+				case "", api.RestorePhaseNew, api.RestorePhaseInProgress:
 				default:
 					c.logger.WithFields(logrus.Fields{
 						"restore": kubeutil.NamespaceAndName(restore),
 						"phase":   restore.Status.Phase,
-					}).Debug("Restore is not new, skipping")
+					}).Debug("Restore is not new or in-progress, skipping")
 					return
 				}

@@ -202,7 +201,21 @@ func (c *restoreController) processQueueItem(key string) error {
 	// is ("" | New)
 	switch restore.Status.Phase {
 	case "", api.RestorePhaseNew:
-		// only process new restores
+	case api.RestorePhaseInProgress:
+		// A restore may stay in-progress forever because of
+		// 1) the controller restarts during the processing of a restore
+		// 2) the restore with in-progress status isn't updated to completed or failed status successfully
+		// So we try to mark such restores as failed to avoid it
+		updated := restore.DeepCopy()
+		updated.Status.Phase = api.RestorePhaseFailed
+		updated.Status.FailureReason = fmt.Sprintf("got a Restore with unexpected status %q, this may be due to a restart of the controller during the restore, mark it as %q",
+			api.RestorePhaseInProgress, updated.Status.Phase)
+		_, err = patchRestore(restore, updated, c.restoreClient)
+		if err != nil {
+			return errors.Wrapf(err, "error updating Restore status to %s", updated.Status.Phase)
+		}
+		log.Warn(updated.Status.FailureReason)
+		return nil
 	default:
 		return nil
 	}
@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"fmt"
 	"io/ioutil"
 	"testing"
 	"time"
@@ -170,11 +171,6 @@ func TestProcessQueueItemSkips(t *testing.T) {
 			restoreKey:  "foo/bar",
 			expectError: true,
 		},
-		{
-			name:       "restore with phase InProgress does not get processed",
-			restoreKey: "foo/bar",
-			restore:    builder.ForRestore("foo", "bar").Phase(velerov1api.RestorePhaseInProgress).Result(),
-		},
 		{
 			name:       "restore with phase Completed does not get processed",
 			restoreKey: "foo/bar",
@@ -226,6 +222,31 @@ func TestProcessQueueItemSkips(t *testing.T) {
 	}
 }

+func TestMarkInProgressRestoreAsFailed(t *testing.T) {
+	var (
+		restore         = builder.ForRestore("velero", "bar").Phase(velerov1api.RestorePhaseInProgress).Result()
+		client          = fake.NewSimpleClientset(restore)
+		sharedInformers = informers.NewSharedInformerFactory(client, 0)
+		logger          = velerotest.NewLogger()
+	)
+
+	c := restoreController{
+		genericController: newGenericController("restore-test", logger),
+		restoreClient:     client.VeleroV1(),
+		restoreLister:     sharedInformers.Velero().V1().Restores().Lister(),
+	}
+
+	err := sharedInformers.Velero().V1().Restores().Informer().GetStore().Add(restore)
+	require.Nil(t, err)
+
+	err = c.processQueueItem(fmt.Sprintf("%s/%s", restore.Namespace, restore.Name))
+	require.Nil(t, err)
+
+	res, err := c.restoreClient.Restores(restore.Namespace).Get(context.Background(), restore.Name, metav1.GetOptions{})
+	require.Nil(t, err)
+	assert.Equal(t, velerov1api.RestorePhaseFailed, res.Status.Phase)
+}
+
 func TestProcessQueueItem(t *testing.T) {

 	defaultStorageLocation := builder.ForBackupStorageLocation("velero", "default").Provider("myCloud").Bucket("bucket").Result()