Merge pull request #10273 from Lyndon-Li/fix-pvr-regression

Fix PVR regression

pkg/controller/pod_volume_restore_controller.go

@@ -236,7 +236,15 @@ func (r *PodVolumeRestoreReconciler) Reconcile(ctx context.Context, req ctrl.Req
 			return ctrl.Result{}, nil
 		}
 
-		shouldProcess, pod, err := shouldProcess(ctx, r.client, log, pvr, r.resourceTimeout)
+		pod, err := getTargetPod(ctx, r.client, log, pvr)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
+		if pod == nil {
+			return ctrl.Result{}, nil
+		}
+
+		shouldProcess, err := shouldProcess(pod, log)
 		if err != nil {
 			return r.errorOut(ctx, pvr, err, "Pod for this PVR is not ready", log)
 		}
@@ -255,12 +263,6 @@ func (r *PodVolumeRestoreReconciler) Reconcile(ctx context.Context, req ctrl.Req
 			return ctrl.Result{}, errors.Wrapf(err, "error accepting PVR %s", pvr.Name)
 		}
 
-		initContainerIndex := getInitContainerIndex(pod)
-		if initContainerIndex > 0 {
-			log.Warnf(`Init containers before the %s container may cause issues
-					  if they interfere with volumes being restored: %s index %d`, restorehelper.WaitInitContainer, restorehelper.WaitInitContainer, initContainerIndex)
-		}
-
 		log.Info("Exposing PVR")
 
 		exposeParam := r.setupExposeParam(pvr)
@@ -565,71 +567,61 @@ func UpdatePVRStatusToFailed(ctx context.Context, c client.Client, pvr *velerov1
 	return err
 }
 
-func shouldProcess(ctx context.Context, client client.Client, log logrus.FieldLogger, pvr *velerov1api.PodVolumeRestore, timeout time.Duration) (bool, *corev1api.Pod, error) {
-	if !isPVRNew(pvr) {
-		log.Debug("PVR is not new, skip")
-		return false, nil, nil
-	}
-
+func getTargetPod(ctx context.Context, client client.Client, log logrus.FieldLogger, pvr *velerov1api.PodVolumeRestore) (*corev1api.Pod, error) {
 	// 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
-	var targetPod *corev1api.Pod
-	err := wait.PollUntilContextTimeout(ctx, time.Millisecond*100, timeout, true, func(ctx context.Context) (bool, error) {
-		updated := &corev1api.Pod{}
-		if err := client.Get(ctx, types.NamespacedName{Namespace: pvr.Spec.Pod.Namespace, Name: pvr.Spec.Pod.Name}, updated); err != nil {
-			if apierrors.IsNotFound(err) {
-				return false, nil
-			}
-
-			return false, err
-		}
-
-		targetPod = updated
-
-		return true, nil
-	})
-
-	if err != nil {
-		if errors.Is(err, context.DeadlineExceeded) {
-			return false, nil, errors.Errorf("timeout to wait for pod %s/%s", pvr.Spec.Pod.Namespace, pvr.Spec.Pod.Name)
-		} else {
-			return false, nil, errors.Wrapf(err, "error waiting for pod %s/%s", pvr.Spec.Pod.Namespace, pvr.Spec.Pod.Name)
+	pod := &corev1api.Pod{}
+	if err := client.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 nil, nil
 		}
+		log.WithError(err).Error("Unable to get pod")
+		return nil, err
 	}
 
+	return pod, nil
+}
+
+func shouldProcess(targetPod *corev1api.Pod, log logrus.FieldLogger) (bool, error) {
 	if targetPod.Status.Phase == corev1api.PodFailed || targetPod.Status.Phase == corev1api.PodUnknown {
-		return false, nil, errors.Errorf("unexpected state for pod %s/%s", targetPod.Namespace, targetPod.Name)
+		return false, errors.Errorf("unexpected state for pod %s/%s", targetPod.Namespace, targetPod.Name)
 	}
 
 	idx := getInitContainerIndex(targetPod)
 	if idx < 0 {
-		return false, nil, errors.Errorf("no restore-wait init container in pod %s/%s", targetPod.Namespace, targetPod.Name)
+		return false, errors.Errorf("no restore-wait init container in pod %s/%s", targetPod.Namespace, targetPod.Name)
 	}
 
 	if len(targetPod.Status.InitContainerStatuses) <= idx {
 		log.Debug("Pod init container statuses are not fully populated yet, skip")
-		return false, nil, nil
+		return false, nil
+	}
+
+	if idx > 0 {
+		log.Warnf(`Init containers before the %s container may cause issues
+					  if they interfere with volumes being restored: %s index %d`, restorehelper.WaitInitContainer, restorehelper.WaitInitContainer, idx)
 	}
 
 	containerStatus := targetPod.Status.InitContainerStatuses[idx]
 
 	if containerStatus.State.Terminated != nil {
-		return false, nil, errors.Errorf("restore-wait init container has already completed in pod %s/%s", targetPod.Namespace, targetPod.Name)
+		return false, errors.Errorf("restore-wait init container has already completed in pod %s/%s", targetPod.Namespace, targetPod.Name)
 	}
 
 	if containerStatus.State.Waiting != nil {
 		reason := containerStatus.State.Waiting.Reason
 		if reason == "ImagePullBackOff" || reason == "ErrImageNeverPull" || reason == "CreateContainerConfigError" || reason == "CreateContainerError" || reason == "InvalidImageName" || reason == "ErrImagePull" {
-			return false, nil, errors.Errorf("restore-wait init container in pod %s/%s is in unrecoverable waiting state with reason %s", targetPod.Namespace, targetPod.Name, reason)
+			return false, errors.Errorf("restore-wait init container in pod %s/%s is in unrecoverable waiting state with reason %s", targetPod.Namespace, targetPod.Name, reason)
 		}
 	}
 
 	if containerStatus.State.Running == nil {
 		log.Debug("Pod is not running restore-wait init container, skip")
-		return false, nil, nil
+		return false, nil
 	}
 
-	return true, targetPod, nil
+	return true, nil
 }
 
 func (r *PodVolumeRestoreReconciler) closeDataPath(ctx context.Context, pvrName string) {

pkg/controller/pod_volume_restore_controller_test.go

@@ -117,8 +117,6 @@ func TestShouldProcess(t *testing.T) {
 				},
 			},
 			shouldProcessed: false,
-			expectError:     true,
-			errString:       "timeout to wait for pod ns-1/pod-1",
 		},
 		{
 			name: "Empty phase pvr with pod on node not running init container should not be processed",
@@ -470,8 +468,6 @@ func TestShouldProcess(t *testing.T) {
 
 	for _, ts := range tests {
 		t.Run(ts.name, func(t *testing.T) {
-			ctx := t.Context()
-
 			var objs []runtime.Object
 			if ts.obj != nil {
 				objs = append(objs, ts.obj)
@@ -487,7 +483,21 @@ func TestShouldProcess(t *testing.T) {
 				clock:  &clocks.RealClock{},
 			}
 
-			shouldProcess, _, err := shouldProcess(ctx, c.client, c.logger, ts.obj, time.Second)
+			if !isPVRNew(ts.obj) {
+				require.False(t, ts.shouldProcessed)
+				return
+			}
+
+			if ts.pod == nil {
+				_, err := getTargetPod(context.Background(), c.client, c.logger, ts.obj)
+				if ts.expectError {
+					require.Error(t, err)
+				}
+				require.False(t, ts.shouldProcessed)
+				return
+			}
+
+			shouldProcess, err := shouldProcess(ts.pod, c.logger)
 			require.Equal(t, ts.shouldProcessed, shouldProcess)
 			if ts.expectError {
 				require.Error(t, err)