refine what gets enqueued in PVB/PVR controllers, and log better

Signed-off-by: Steve Kriss <steve@heptio.com>

pkg/controller/pod_volume_backup_controller.go

@@ -86,17 +86,36 @@ func NewPodVolumeBackupController(
 
 	podVolumeBackupInformer.Informer().AddEventHandler(
 		cache.ResourceEventHandlerFuncs{
-			AddFunc:    c.enqueue,
-			UpdateFunc: func(_, obj interface{}) { c.enqueue(obj) },
+			AddFunc:    c.pvbHandler,
+			UpdateFunc: func(_, obj interface{}) { c.pvbHandler(obj) },
 		},
 	)
 
 	return c
 }
 
+func (c *podVolumeBackupController) pvbHandler(obj interface{}) {
+	req := obj.(*arkv1api.PodVolumeBackup)
+
+	// only enqueue items for this node
+	if req.Spec.Node != c.nodeName {
+		return
+	}
+
+	log := loggerForPodVolumeBackup(c.logger, req)
+
+	if req.Status.Phase != "" && req.Status.Phase != arkv1api.PodVolumeBackupPhaseNew {
+		log.Debug("Backup is not new, not enqueuing")
+		return
+	}
+
+	log.Debug("Enqueueing")
+	c.enqueue(obj)
+}
+
 func (c *podVolumeBackupController) processQueueItem(key string) error {
 	log := c.logger.WithField("key", key)
-	log.Debug("Running processItem")
+	log.Debug("Running processQueueItem")
 
 	ns, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
@@ -120,22 +139,29 @@ func (c *podVolumeBackupController) processQueueItem(key string) error {
 		return nil
 	}
 
-	// only process items for this node
-	if req.Spec.Node != c.nodeName {
-		return nil
-	}
-
 	// Don't mutate the shared cache
 	reqCopy := req.DeepCopy()
 	return c.processBackupFunc(reqCopy)
 }
 
-func (c *podVolumeBackupController) processBackup(req *arkv1api.PodVolumeBackup) error {
-	log := c.logger.WithFields(logrus.Fields{
+func loggerForPodVolumeBackup(baseLogger logrus.FieldLogger, req *arkv1api.PodVolumeBackup) logrus.FieldLogger {
+	log := baseLogger.WithFields(logrus.Fields{
 		"namespace": req.Namespace,
 		"name":      req.Name,
 	})
 
+	if len(req.OwnerReferences) == 1 {
+		log = log.WithField("backup", fmt.Sprintf("%s/%s", req.Namespace, req.OwnerReferences[0].Name))
+	}
+
+	return log
+}
+
+func (c *podVolumeBackupController) processBackup(req *arkv1api.PodVolumeBackup) error {
+	log := loggerForPodVolumeBackup(c.logger, req)
+
+	log.Info("Backup starting")
+
 	var err error
 
 	// update status to InProgress
@@ -157,11 +183,15 @@ func (c *podVolumeBackupController) processBackup(req *arkv1api.PodVolumeBackup)
 		return c.fail(req, errors.Wrap(err, "error getting volume directory name").Error(), log)
 	}
 
-	path, err := singlePathMatch(fmt.Sprintf("/host_pods/%s/volumes/*/%s", string(req.Spec.Pod.UID), volumeDir))
+	pathGlob := fmt.Sprintf("/host_pods/%s/volumes/*/%s", string(req.Spec.Pod.UID), volumeDir)
+	log.WithField("pathGlob", pathGlob).Debug("Looking for path matching glob")
+
+	path, err := singlePathMatch(pathGlob)
 	if err != nil {
 		log.WithError(err).Error("Error uniquely identifying volume path")
 		return c.fail(req, errors.Wrap(err, "error getting volume path on host").Error(), log)
 	}
+	log.WithField("path", path).Debugf("Found path matching glob")
 
 	// temp creds
 	file, err := restic.TempCredentialsFile(c.secretLister, req.Spec.Pod.Namespace)
@@ -204,6 +234,8 @@ func (c *podVolumeBackupController) processBackup(req *arkv1api.PodVolumeBackup)
 		return err
 	}
 
+	log.Info("Backup completed")
+
 	return nil
 }
 

pkg/controller/pod_volume_backup_controller_test.go

@@ -0,0 +1,171 @@
+/*
+Copyright 2018 the Heptio Ark contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	arkv1api "github.com/heptio/ark/pkg/apis/ark/v1"
+	arktest "github.com/heptio/ark/pkg/util/test"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPVBHandler(t *testing.T) {
+	controllerNode := "foo"
+
+	tests := []struct {
+		name          string
+		obj           *arkv1api.PodVolumeBackup
+		shouldEnqueue bool
+	}{
+		{
+			name: "Empty phase pvb on same node should be enqueued",
+			obj: &arkv1api.PodVolumeBackup{
+				Spec: arkv1api.PodVolumeBackupSpec{
+					Node: controllerNode,
+				},
+			},
+			shouldEnqueue: true,
+		},
+		{
+			name: "New phase pvb on same node should be enqueued",
+			obj: &arkv1api.PodVolumeBackup{
+				Spec: arkv1api.PodVolumeBackupSpec{
+					Node: controllerNode,
+				},
+				Status: arkv1api.PodVolumeBackupStatus{
+					Phase: arkv1api.PodVolumeBackupPhaseNew,
+				},
+			},
+			shouldEnqueue: true,
+		},
+		{
+			name: "InProgress phase pvb on same node should not be enqueued",
+			obj: &arkv1api.PodVolumeBackup{
+				Spec: arkv1api.PodVolumeBackupSpec{
+					Node: controllerNode,
+				},
+				Status: arkv1api.PodVolumeBackupStatus{
+					Phase: arkv1api.PodVolumeBackupPhaseInProgress,
+				},
+			},
+			shouldEnqueue: false,
+		},
+		{
+			name: "Completed phase pvb on same node should not be enqueued",
+			obj: &arkv1api.PodVolumeBackup{
+				Spec: arkv1api.PodVolumeBackupSpec{
+					Node: controllerNode,
+				},
+				Status: arkv1api.PodVolumeBackupStatus{
+					Phase: arkv1api.PodVolumeBackupPhaseCompleted,
+				},
+			},
+			shouldEnqueue: false,
+		},
+		{
+			name: "Failed phase pvb on same node should not be enqueued",
+			obj: &arkv1api.PodVolumeBackup{
+				Spec: arkv1api.PodVolumeBackupSpec{
+					Node: controllerNode,
+				},
+				Status: arkv1api.PodVolumeBackupStatus{
+					Phase: arkv1api.PodVolumeBackupPhaseFailed,
+				},
+			},
+			shouldEnqueue: false,
+		},
+
+		{
+			name: "Empty phase pvb on different node should not be enqueued",
+			obj: &arkv1api.PodVolumeBackup{
+				Spec: arkv1api.PodVolumeBackupSpec{
+					Node: "some-other-node",
+				},
+			},
+			shouldEnqueue: false,
+		},
+		{
+			name: "New phase pvb on different node should not be enqueued",
+			obj: &arkv1api.PodVolumeBackup{
+				Spec: arkv1api.PodVolumeBackupSpec{
+					Node: "some-other-node",
+				},
+				Status: arkv1api.PodVolumeBackupStatus{
+					Phase: arkv1api.PodVolumeBackupPhaseNew,
+				},
+			},
+			shouldEnqueue: false,
+		},
+		{
+			name: "InProgress phase pvb on different node should not be enqueued",
+			obj: &arkv1api.PodVolumeBackup{
+				Spec: arkv1api.PodVolumeBackupSpec{
+					Node: "some-other-node",
+				},
+				Status: arkv1api.PodVolumeBackupStatus{
+					Phase: arkv1api.PodVolumeBackupPhaseInProgress,
+				},
+			},
+			shouldEnqueue: false,
+		},
+		{
+			name: "Completed phase pvb on different node should not be enqueued",
+			obj: &arkv1api.PodVolumeBackup{
+				Spec: arkv1api.PodVolumeBackupSpec{
+					Node: "some-other-node",
+				},
+				Status: arkv1api.PodVolumeBackupStatus{
+					Phase: arkv1api.PodVolumeBackupPhaseCompleted,
+				},
+			},
+			shouldEnqueue: false,
+		},
+		{
+			name: "Failed phase pvb on different node should not be enqueued",
+			obj: &arkv1api.PodVolumeBackup{
+				Spec: arkv1api.PodVolumeBackupSpec{
+					Node: "some-other-node",
+				},
+				Status: arkv1api.PodVolumeBackupStatus{
+					Phase: arkv1api.PodVolumeBackupPhaseFailed,
+				},
+			},
+			shouldEnqueue: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			c := &podVolumeBackupController{
+				genericController: newGenericController("pod-volume-backup", arktest.NewLogger()),
+				nodeName:          controllerNode,
+			}
+
+			c.pvbHandler(test.obj)
+
+			if !test.shouldEnqueue {
+				assert.Equal(t, 0, c.queue.Len())
+				return
+			}
+
+			require.Equal(t, 1, c.queue.Len())
+		})
+	}
+}

pkg/controller/pod_volume_restore_controller.go

@@ -111,13 +111,34 @@ func NewPodVolumeRestoreController(
 
 func (c *podVolumeRestoreController) pvrHandler(obj interface{}) {
 	pvr := obj.(*arkv1api.PodVolumeRestore)
-	log := c.logger.WithField("key", kube.NamespaceAndName(pvr))
+	log := loggerForPodVolumeRestore(c.logger, pvr)
 
-	if !shouldEnqueuePVR(pvr, c.podLister, c.nodeName, log) {
+	if !isPVRNew(pvr) {
+		log.Debugf("Restore is not new, not enqueuing")
 		return
 	}
 
-	log.Debug("enqueueing")
+	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
+	}
+
+	log.Debug("Enqueueing")
 	c.enqueue(obj)
 }
 
@@ -125,84 +146,53 @@ func (c *podVolumeRestoreController) podHandler(obj interface{}) {
 	pod := obj.(*corev1api.Pod)
 	log := c.logger.WithField("key", kube.NamespaceAndName(pod))
 
-	for _, pvr := range pvrsToEnqueueForPod(pod, c.podVolumeRestoreLister, c.nodeName, log) {
-		c.enqueue(pvr)
-	}
-}
-
-func shouldProcessPod(pod *corev1api.Pod, nodeName string, log logrus.FieldLogger) bool {
-	// if the pod lister being used is filtered to pods on this node, this is superfluous,
-	// but retaining for safety.
-	if pod.Spec.NodeName != nodeName {
-		log.Debugf("Pod is scheduled on node %s, not enqueueing.", pod.Spec.NodeName)
-		return false
+	// 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
 	}
 
-	// only process items for pods that have the restic initContainer running
-	if !resticInitContainerRunning(pod) {
-		log.Debugf("Pod is not running restic initContainer, not enqueueing.")
-		return false
-	}
-
-	return true
-}
-
-func shouldProcessPVR(pvr *arkv1api.PodVolumeRestore, log logrus.FieldLogger) bool {
-	// only process new items
-	if pvr.Status.Phase != "" && pvr.Status.Phase != arkv1api.PodVolumeRestorePhaseNew {
-		log.Debugf("Item has phase %s, not enqueueing.", pvr.Status.Phase)
-		return false
-	}
-
-	return true
-}
-
-func pvrsToEnqueueForPod(pod *corev1api.Pod, pvrLister listers.PodVolumeRestoreLister, nodeName string, log logrus.FieldLogger) []*arkv1api.PodVolumeRestore {
-	if !shouldProcessPod(pod, nodeName, log) {
-		return nil
+	if !isResticInitContainerRunning(pod) {
+		log.Debug("Pod is not running restic-wait init container, not enqueuing restores for pod")
+		return
 	}
 
 	selector, err := labels.Parse(fmt.Sprintf("%s=%s", arkv1api.PodUIDLabel, pod.UID))
 	if err != nil {
 		log.WithError(err).Error("Unable to parse label selector %s", fmt.Sprintf("%s=%s", arkv1api.PodUIDLabel, pod.UID))
-		return nil
+		return
 	}
 
-	pvrs, err := pvrLister.List(selector)
+	pvrs, err := c.podVolumeRestoreLister.List(selector)
 	if err != nil {
 		log.WithError(err).Error("Unable to list pod volume restores")
-		return nil
+		return
 	}
 
-	var res []*arkv1api.PodVolumeRestore
-	for i, pvr := range pvrs {
-		if shouldProcessPVR(pvr, log) {
-			res = append(res, pvrs[i])
+	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
 		}
+		log.Debug("Enqueuing")
+		c.enqueue(pvr)
 	}
-
-	return res
 }
 
-func shouldEnqueuePVR(pvr *arkv1api.PodVolumeRestore, podLister corev1listers.PodLister, nodeName string, log logrus.FieldLogger) bool {
-	if !shouldProcessPVR(pvr, log) {
-		return false
-	}
-
-	pod, err := podLister.Pods(pvr.Spec.Pod.Namespace).Get(pvr.Spec.Pod.Name)
-	if err != nil {
-		log.WithError(err).Errorf("Unable to get item's pod %s/%s, not enqueueing.", pvr.Spec.Pod.Namespace, pvr.Spec.Pod.Name)
-		return false
-	}
-
-	if !shouldProcessPod(pod, nodeName, log) {
-		return false
-	}
-
-	return true
+func isPVRNew(pvr *arkv1api.PodVolumeRestore) bool {
+	return pvr.Status.Phase == "" || pvr.Status.Phase == arkv1api.PodVolumeRestorePhaseNew
 }
 
-func resticInitContainerRunning(pod *corev1api.Pod) bool {
+func isPodOnNode(pod *corev1api.Pod, node string) bool {
+	return pod.Spec.NodeName == node
+}
+
+func isResticInitContainerRunning(pod *corev1api.Pod) bool {
 	// no init containers, or the first one is not the ark restic one: return false
 	if len(pod.Spec.InitContainers) == 0 || pod.Spec.InitContainers[0].Name != restic.InitContainer {
 		return false
@@ -219,7 +209,7 @@ func resticInitContainerRunning(pod *corev1api.Pod) bool {
 
 func (c *podVolumeRestoreController) processQueueItem(key string) error {
 	log := c.logger.WithField("key", key)
-	log.Debug("Running processItem")
+	log.Debug("Running processQueueItem")
 
 	ns, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
@@ -241,12 +231,24 @@ func (c *podVolumeRestoreController) processQueueItem(key string) error {
 	return c.processRestoreFunc(reqCopy)
 }
 
-func (c *podVolumeRestoreController) processRestore(req *arkv1api.PodVolumeRestore) error {
-	log := c.logger.WithFields(logrus.Fields{
+func loggerForPodVolumeRestore(baseLogger logrus.FieldLogger, req *arkv1api.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 *arkv1api.PodVolumeRestore) error {
+	log := loggerForPodVolumeRestore(c.logger, req)
+
+	log.Info("Restore starting")
+
 	var err error
 
 	// update status to InProgress
@@ -288,6 +290,8 @@ func (c *podVolumeRestoreController) processRestore(req *arkv1api.PodVolumeResto
 		return err
 	}
 
+	log.Info("Restore completed")
+
 	return nil
 }
 

pkg/controller/pod_volume_restore_controller_test.go

@@ -0,0 +1,601 @@
+/*
+Copyright 2018 the Heptio Ark contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	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"
+
+	arkv1api "github.com/heptio/ark/pkg/apis/ark/v1"
+	arkfake "github.com/heptio/ark/pkg/generated/clientset/versioned/fake"
+	arkinformers "github.com/heptio/ark/pkg/generated/informers/externalversions"
+	arkv1listers "github.com/heptio/ark/pkg/generated/listers/ark/v1"
+	"github.com/heptio/ark/pkg/restic"
+	arktest "github.com/heptio/ark/pkg/util/test"
+)
+
+func TestPVRHandler(t *testing.T) {
+	controllerNode := "foo"
+
+	tests := []struct {
+		name          string
+		obj           *arkv1api.PodVolumeRestore
+		pod           *corev1api.Pod
+		shouldEnqueue bool
+	}{
+		{
+			name: "InProgress phase pvr should not be enqueued",
+			obj: &arkv1api.PodVolumeRestore{
+				Status: arkv1api.PodVolumeRestoreStatus{
+					Phase: arkv1api.PodVolumeRestorePhaseInProgress,
+				},
+			},
+			shouldEnqueue: false,
+		},
+		{
+			name: "Completed phase pvr should not be enqueued",
+			obj: &arkv1api.PodVolumeRestore{
+				Status: arkv1api.PodVolumeRestoreStatus{
+					Phase: arkv1api.PodVolumeRestorePhaseCompleted,
+				},
+			},
+			shouldEnqueue: false,
+		},
+		{
+			name: "Failed phase pvr should not be enqueued",
+			obj: &arkv1api.PodVolumeRestore{
+				Status: arkv1api.PodVolumeRestoreStatus{
+					Phase: arkv1api.PodVolumeRestorePhaseFailed,
+				},
+			},
+			shouldEnqueue: false,
+		},
+		{
+			name: "Unable to get pvr's pod should not be enqueued",
+			obj: &arkv1api.PodVolumeRestore{
+				Spec: arkv1api.PodVolumeRestoreSpec{
+					Pod: corev1api.ObjectReference{
+						Namespace: "ns-1",
+						Name:      "pod-1",
+					},
+				},
+				Status: arkv1api.PodVolumeRestoreStatus{
+					Phase: "",
+				},
+			},
+			shouldEnqueue: false,
+		},
+		{
+			name: "Empty phase pvr with pod not on node running init container should not be enqueued",
+			obj: &arkv1api.PodVolumeRestore{
+				Spec: arkv1api.PodVolumeRestoreSpec{
+					Pod: corev1api.ObjectReference{
+						Namespace: "ns-1",
+						Name:      "pod-1",
+					},
+				},
+				Status: arkv1api.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",
+			obj: &arkv1api.PodVolumeRestore{
+				Spec: arkv1api.PodVolumeRestoreSpec{
+					Pod: corev1api.ObjectReference{
+						Namespace: "ns-1",
+						Name:      "pod-1",
+					},
+				},
+				Status: arkv1api.PodVolumeRestoreStatus{
+					Phase: "",
+				},
+			},
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns-1",
+					Name:      "pod-1",
+				},
+				Spec: corev1api.PodSpec{
+					NodeName: controllerNode,
+					InitContainers: []corev1api.Container{
+						{
+							Name: restic.InitContainer,
+						},
+					},
+				},
+				Status: corev1api.PodStatus{
+					InitContainerStatuses: []corev1api.ContainerStatus{
+						{
+							State: corev1api.ContainerState{},
+						},
+					},
+				},
+			},
+			shouldEnqueue: false,
+		},
+		{
+			name: "Empty phase pvr with pod on node running init container should be enqueued",
+			obj: &arkv1api.PodVolumeRestore{
+				Spec: arkv1api.PodVolumeRestoreSpec{
+					Pod: corev1api.ObjectReference{
+						Namespace: "ns-1",
+						Name:      "pod-1",
+					},
+				},
+				Status: arkv1api.PodVolumeRestoreStatus{
+					Phase: "",
+				},
+			},
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns-1",
+					Name:      "pod-1",
+				},
+				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()},
+								},
+							},
+						},
+					},
+				},
+			},
+			shouldEnqueue: 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("pod-volume-restore", arktest.NewLogger()),
+					podLister:         corev1listers.NewPodLister(podInformer.GetIndexer()),
+					nodeName:          controllerNode,
+				}
+			)
+
+			if test.pod != nil {
+				require.NoError(t, podInformer.GetStore().Add(test.pod))
+			}
+
+			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 []*arkv1api.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: []*arkv1api.PodVolumeRestore{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns-1",
+						Name:      "pvr-1",
+						Labels: map[string]string{
+							arkv1api.PodUIDLabel: "uid",
+						},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns-1",
+						Name:      "pvr-2",
+						Labels: map[string]string{
+							arkv1api.PodUIDLabel: "uid",
+						},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns-1",
+						Name:      "pvr-3",
+						Labels: map[string]string{
+							arkv1api.PodUIDLabel: "uid",
+						},
+					},
+					Status: arkv1api.PodVolumeRestoreStatus{
+						Phase: arkv1api.PodVolumeRestorePhaseInProgress,
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns-1",
+						Name:      "pvr-4",
+						Labels: map[string]string{
+							arkv1api.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: []*arkv1api.PodVolumeRestore{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns-1",
+						Name:      "pvr-1",
+						Labels: map[string]string{
+							arkv1api.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: []*arkv1api.PodVolumeRestore{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns-1",
+						Name:      "pvr-1",
+						Labels: map[string]string{
+							arkv1api.PodUIDLabel: "uid",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var (
+				client      = arkfake.NewSimpleClientset()
+				informers   = arkinformers.NewSharedInformerFactory(client, 0)
+				pvrInformer = informers.Ark().V1().PodVolumeRestores()
+				c           = &podVolumeRestoreController{
+					genericController:      newGenericController("pod-volume-restore", arktest.NewLogger()),
+					podVolumeRestoreLister: arkv1listers.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)))
+			}
+		})
+	}
+}
+
+func TestIsPVRNew(t *testing.T) {
+	pvr := &arkv1api.PodVolumeRestore{}
+
+	expectationByStatus := map[arkv1api.PodVolumeRestorePhase]bool{
+		"": true,
+		arkv1api.PodVolumeRestorePhaseNew:        true,
+		arkv1api.PodVolumeRestorePhaseInProgress: false,
+		arkv1api.PodVolumeRestorePhaseCompleted:  false,
+		arkv1api.PodVolumeRestorePhaseFailed:     false,
+	}
+
+	for phase, expected := range expectationByStatus {
+		pvr.Status.Phase = phase
+		assert.Equal(t, expected, isPVRNew(pvr))
+	}
+}
+
+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
+		pod      *corev1api.Pod
+		expected bool
+	}{
+		{
+			name: "pod with no init containers should return false",
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns-1",
+					Name:      "pod-1",
+				},
+			},
+			expected: false,
+		},
+		{
+			name: "pod with running init container that's not restic should return false",
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns-1",
+					Name:      "pod-1",
+				},
+				Spec: corev1api.PodSpec{
+					InitContainers: []corev1api.Container{
+						{
+							Name: "non-restic-init",
+						},
+					},
+				},
+				Status: corev1api.PodStatus{
+					InitContainerStatuses: []corev1api.ContainerStatus{
+						{
+							State: corev1api.ContainerState{
+								Running: &corev1api.ContainerStateRunning{StartedAt: metav1.Time{Time: time.Now()}},
+							},
+						},
+					},
+				},
+			},
+			expected: false,
+		},
+		{
+			name: "pod with running restic init container that's not first should return false",
+			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,
+						},
+					},
+				},
+				Status: corev1api.PodStatus{
+					InitContainerStatuses: []corev1api.ContainerStatus{
+						{
+							State: corev1api.ContainerState{
+								Running: &corev1api.ContainerStateRunning{StartedAt: metav1.Time{Time: time.Now()}},
+							},
+						},
+						{
+							State: corev1api.ContainerState{
+								Running: &corev1api.ContainerStateRunning{StartedAt: metav1.Time{Time: time.Now()}},
+							},
+						},
+					},
+				},
+			},
+			expected: false,
+		},
+		{
+			name: "pod with restic init container as first initContainer that's not running should return false",
+			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",
+						},
+					},
+				},
+				Status: corev1api.PodStatus{
+					InitContainerStatuses: []corev1api.ContainerStatus{
+						{
+							State: corev1api.ContainerState{},
+						},
+						{
+							State: corev1api.ContainerState{
+								Running: &corev1api.ContainerStateRunning{StartedAt: metav1.Time{Time: time.Now()}},
+							},
+						},
+					},
+				},
+			},
+			expected: false,
+		},
+		{
+			name: "pod with running restic init container as first initContainer should return true",
+			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",
+						},
+					},
+				},
+				Status: corev1api.PodStatus{
+					InitContainerStatuses: []corev1api.ContainerStatus{
+						{
+							State: corev1api.ContainerState{
+								Running: &corev1api.ContainerStateRunning{StartedAt: metav1.Time{Time: time.Now()}},
+							},
+						},
+						{
+							State: corev1api.ContainerState{
+								Running: &corev1api.ContainerStateRunning{StartedAt: metav1.Time{Time: time.Now()}},
+							},
+						},
+					},
+				},
+			},
+			expected: true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			assert.Equal(t, test.expected, isResticInitContainerRunning(test.pod))
+		})
+	}
+}