resume PVB/PVR on node-agent restart

Signed-off-by: Lyndon-Li <lyonghui@vmware.com>

pkg/cmd/cli/nodeagent/server.go

@@ -392,7 +392,17 @@ func (s *nodeAgentServer) waitCacheForResume() error {
 		return errors.Wrap(err, "error getting dd informer")
 	}
 
-	if !cacheutil.WaitForCacheSync(s.ctx.Done(), podInformer.HasSynced, duInformer.HasSynced, ddInformer.HasSynced) {
+	pvbInformer, err := s.mgr.GetCache().GetInformer(s.ctx, &velerov1api.PodVolumeBackup{})
+	if err != nil {
+		return errors.Wrap(err, "error getting PVB informer")
+	}
+
+	pvrInformer, err := s.mgr.GetCache().GetInformer(s.ctx, &velerov1api.PodVolumeRestore{})
+	if err != nil {
+		return errors.Wrap(err, "error getting PVR informer")
+	}
+
+	if !cacheutil.WaitForCacheSync(s.ctx.Done(), podInformer.HasSynced, duInformer.HasSynced, ddInformer.HasSynced, pvbInformer.HasSynced, pvrInformer.HasSynced) {
 		return errors.New("error waiting informer synced")
 	}
 

pkg/controller/data_download_controller.go

@@ -937,7 +937,7 @@ func (r *DataDownloadReconciler) AttemptDataDownloadResume(ctx context.Context,
 			if err != nil {
 				logger.WithError(errors.WithStack(err)).WithError(errors.WithStack(err)).Error("Failed to trigger datadownload cancel")
 			}
-		} else {
+		} else if !isDataDownloadInFinalState(dd) {
 			// the Prepared CR could be still handled by datadownload controller after node-agent restart
 			// the accepted CR may also suvived from node-agent restart as long as the intermediate objects are all done
 			logger.WithField("datadownload", dd.GetName()).Infof("find a datadownload with status %s", dd.Status.Phase)

pkg/controller/data_upload_controller.go

@@ -1042,7 +1042,7 @@ func (r *DataUploadReconciler) AttemptDataUploadResume(ctx context.Context, logg
 			if err != nil {
 				logger.WithField("dataupload", du.GetName()).WithError(errors.WithStack(err)).Error("Failed to trigger dataupload cancel")
 			}
-		} else {
+		} else if !isDataUploadInFinalState(du) {
 			// the Prepared CR could be still handled by dataupload controller after node-agent restart
 			// the accepted CR may also suvived from node-agent restart as long as the intermediate objects are all done
 			logger.WithField("dataupload", du.GetName()).Infof("find a dataupload with status %s", du.Status.Phase)

pkg/controller/pod_volume_backup_controller.go

@@ -885,7 +885,7 @@ func (r *PodVolumeBackupReconciler) AttemptPVBResume(ctx context.Context, logger
 			if err != nil {
 				logger.WithField("PVB", pvb.GetName()).WithError(errors.WithStack(err)).Error("Failed to trigger PVB cancel")
 			}
-		} else {
+		} else if !isPVBInFinalState(pvb) {
 			logger.WithField("PVB", pvb.GetName()).Infof("find a PVB with status %s", pvb.Status.Phase)
 		}
 	}
@@ -902,7 +902,7 @@ func (r *PodVolumeBackupReconciler) resumeCancellableDataPath(ctx context.Contex
 	}
 
 	if res == nil {
-		return errors.Errorf("expose info missed for PVB %s", pvb.Name)
+		return errors.Errorf("no expose result is available for the current node for PVB %s", pvb.Name)
 	}
 
 	callbacks := datapath.Callbacks{

pkg/controller/pod_volume_backup_controller_test.go

@@ -1086,7 +1086,7 @@ func TestResumeCancellablePodVolumeBackup(t *testing.T) {
 		{
 			name:          "no expose",
 			pvb:           pvbBuilder().Phase(velerov1api.PodVolumeBackupPhaseAccepted).Node("node-1").Result(),
-			expectedError: fmt.Sprintf("expose info missed for PVB %s", pvbName),
+			expectedError: fmt.Sprintf("no expose result is available for the current node for PVB %s", pvbName),
 		},
 		{
 			name: "watcher init error",

pkg/controller/pod_volume_restore_controller.go

@@ -953,7 +953,7 @@ func (r *PodVolumeRestoreReconciler) AttemptPVRResume(ctx context.Context, logge
 			if err != nil {
 				logger.WithField("PVR", pvr.GetName()).WithError(errors.WithStack(err)).Error("Failed to trigger PVR cancel")
 			}
-		} else {
+		} else if !isPVRInFinalState(pvr) {
 			logger.WithField("PVR", pvr.GetName()).Infof("find a PVR with status %s", pvr.Status.Phase)
 		}
 	}
@@ -970,7 +970,7 @@ func (r *PodVolumeRestoreReconciler) resumeCancellableDataPath(ctx context.Conte
 	}
 
 	if res == nil {
-		return errors.Errorf("expose info missed for PVR %s", pvr.Name)
+		return errors.Errorf("no expose result is available for the current node for PVR %s", pvr.Name)
 	}
 
 	callbacks := datapath.Callbacks{

pkg/controller/pod_volume_restore_controller_test.go

@@ -1566,7 +1566,7 @@ func TestResumeCancellablePodVolumeRestore(t *testing.T) {
 		{
 			name:          "no expose",
 			pvr:           pvrBuilder().Phase(velerov1api.PodVolumeRestorePhaseAccepted).Node("node-1").Result(),
-			expectedError: fmt.Sprintf("expose info missed for PVR %s", pvrName),
+			expectedError: fmt.Sprintf("no expose result is available for the current node for PVR %s", pvrName),
 		},
 		{
 			name: "watcher init error",