Data mover micro service node agent resume

changelogs/unreleased/8085-Lyndon-Li

@@ -0,0 +1 @@
+According to design #7576, after node-agent restarts, if a DU/DD is in InProgress status, re-capture the data mover ms pod and continue the execution

pkg/builder/data_download_builder.go

@@ -19,6 +19,7 @@ package builder
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 )
 
@@ -122,3 +123,21 @@ func (d *DataDownloadBuilder) StartTimestamp(startTime *metav1.Time) *DataDownlo
 	d.object.Status.StartTimestamp = startTime
 	return d
 }
+
+// CompletionTimestamp sets the DataDownload's StartTimestamp.
+func (d *DataDownloadBuilder) CompletionTimestamp(completionTimestamp *metav1.Time) *DataDownloadBuilder {
+	d.object.Status.CompletionTimestamp = completionTimestamp
+	return d
+}
+
+// Progress sets the DataDownload's Progress.
+func (d *DataDownloadBuilder) Progress(progress shared.DataMoveOperationProgress) *DataDownloadBuilder {
+	d.object.Status.Progress = progress
+	return d
+}
+
+// Node sets the DataDownload's Node.
+func (d *DataDownloadBuilder) Node(node string) *DataDownloadBuilder {
+	d.object.Status.Node = node
+	return d
+}

pkg/builder/data_upload_builder.go

@@ -133,7 +133,14 @@ func (d *DataUploadBuilder) Labels(labels map[string]string) *DataUploadBuilder
 	return d
 }
 
+// Progress sets the DataUpload's Progress.
 func (d *DataUploadBuilder) Progress(progress shared.DataMoveOperationProgress) *DataUploadBuilder {
 	d.object.Status.Progress = progress
 	return d
 }
+
+// Node sets the DataUpload's Node.
+func (d *DataUploadBuilder) Node(node string) *DataUploadBuilder {
+	d.object.Status.Node = node
+	return d
+}

pkg/cmd/cli/nodeagent/server.go

@@ -293,17 +293,29 @@
 		loadAffinity = s.dataPathConfigs.LoadAffinity[0]
 	}
 	dataUploadReconciler := controller.NewDataUploadReconciler(s.mgr.GetClient(), s.mgr, s.kubeClient, s.csiSnapshotClient.SnapshotV1(), s.dataPathMgr, loadAffinity, repoEnsurer, clock.RealClock{}, credentialGetter, s.nodeName, s.fileSystem, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
-	s.attemptDataUploadResume(dataUploadReconciler)
 	if err = dataUploadReconciler.SetupWithManager(s.mgr); err != nil {
 		s.logger.WithError(err).Fatal("Unable to create the data upload controller")
 	}
 
 	dataDownloadReconciler := controller.NewDataDownloadReconciler(s.mgr.GetClient(), s.mgr, s.kubeClient, s.dataPathMgr, repoEnsurer, credentialGetter, s.nodeName, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
-	s.attemptDataDownloadResume(dataDownloadReconciler)
 	if err = dataDownloadReconciler.SetupWithManager(s.mgr); err != nil {
 		s.logger.WithError(err).Fatal("Unable to create the data download controller")
 	}
 
+	go func() {
+		s.mgr.GetCache().WaitForCacheSync(s.ctx)
+
+		if err := dataUploadReconciler.AttemptDataUploadResume(s.ctx, s.mgr.GetClient(), s.logger.WithField("node", s.nodeName), s.namespace); err != nil {
+			s.logger.WithError(errors.WithStack(err)).Error("failed to attempt data upload resume")
+		}
+
+		if err := dataDownloadReconciler.AttemptDataDownloadResume(s.ctx, s.mgr.GetClient(), s.logger.WithField("node", s.nodeName), s.namespace); err != nil {
+			s.logger.WithError(errors.WithStack(err)).Error("failed to attempt data download resume")
+		}
+
+		s.logger.Info("Attempt complete to resume dataUploads and dataDownloads")
+	}()
+
 	s.logger.Info("Controllers starting...")
 
 	if err := s.mgr.Start(ctrl.SetupSignalHandler()); err != nil {
@@ -373,31 +385,6 @@
 	s.markInProgressPVRsFailed(client)
 }
 
-func (s *nodeAgentServer) attemptDataUploadResume(r *controller.DataUploadReconciler) {
-	// the function is called before starting the controller manager, the embedded client isn't ready to use, so create a new one here
-	client, err := ctrlclient.New(s.mgr.GetConfig(), ctrlclient.Options{Scheme: s.mgr.GetScheme()})
-	if err != nil {
-		s.logger.WithError(errors.WithStack(err)).Error("failed to create client")
-		return
-	}
-	if err := r.AttemptDataUploadResume(s.ctx, client, s.logger.WithField("node", s.nodeName), s.namespace); err != nil {
-		s.logger.WithError(errors.WithStack(err)).Error("failed to attempt data upload resume")
-	}
-}
-
-func (s *nodeAgentServer) attemptDataDownloadResume(r *controller.DataDownloadReconciler) {
-	// the function is called before starting the controller manager, the embedded client isn't ready to use, so create a new one here
-	client, err := ctrlclient.New(s.mgr.GetConfig(), ctrlclient.Options{Scheme: s.mgr.GetScheme()})
-	if err != nil {
-		s.logger.WithError(errors.WithStack(err)).Error("failed to create client")
-		return
-	}
-
-	if err := r.AttemptDataDownloadResume(s.ctx, client, s.logger.WithField("node", s.nodeName), s.namespace); err != nil {
-		s.logger.WithError(errors.WithStack(err)).Error("failed to attempt data download resume")
-	}
-}
-
 func (s *nodeAgentServer) markInProgressPVBsFailed(client ctrlclient.Client) {
 	pvbs := &velerov1api.PodVolumeBackupList{}
 	if err := client.List(s.ctx, pvbs, &ctrlclient.ListOptions{Namespace: s.namespace}); err != nil {

pkg/cmd/server/server.go

@@ -1148,9 +1148,15 @@
 			du.Status.Phase == velerov2alpha1api.DataUploadPhaseNew ||
 			du.Status.Phase == "" {
 			err := controller.UpdateDataUploadWithRetry(ctx, client, types.NamespacedName{Namespace: du.Namespace, Name: du.Name}, log.WithField("dataupload", du.Name),
-				func(dataUpload *velerov2alpha1api.DataUpload) {
+				func(dataUpload *velerov2alpha1api.DataUpload) bool {
+					if dataUpload.Spec.Cancel {
+						return false
+					}
+
 					dataUpload.Spec.Cancel = true
-					dataUpload.Status.Message = fmt.Sprintf("found a dataupload with status %q during the velero server starting, mark it as cancel", du.Status.Phase)
+					dataUpload.Status.Message = fmt.Sprintf("Dataupload is in status %q during the velero server starting, mark it as cancel", du.Status.Phase)
+
+					return true
 				})
 
 			if err != nil {
@@ -1183,9 +1189,15 @@
 			dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseNew ||
 			dd.Status.Phase == "" {
 			err := controller.UpdateDataDownloadWithRetry(ctx, client, types.NamespacedName{Namespace: dd.Namespace, Name: dd.Name}, log.WithField("datadownload", dd.Name),
-				func(dataDownload *velerov2alpha1api.DataDownload) {
+				func(dataDownload *velerov2alpha1api.DataDownload) bool {
+					if dataDownload.Spec.Cancel {
+						return false
+					}
+
 					dataDownload.Spec.Cancel = true
-					dataDownload.Status.Message = fmt.Sprintf("found a datadownload with status %q during the velero server starting, mark it as cancel", dd.Status.Phase)
+					dataDownload.Status.Message = fmt.Sprintf("Datadownload is in status %q during the velero server starting, mark it as cancel", dd.Status.Phase)
+
+					return true
 				})
 
 			if err != nil {