Issue 8344: constrain data path expose

changelogs/unreleased/9064-Lyndon-Li

@@ -0,0 +1 @@
+Fix #8344, add a mechanism to soothe creation of data mover pods for DataUpload, DataDownload, PodVolumeBackup and PodVolumeRestore

pkg/builder/pod_volume_backup_builder.go

@@ -149,3 +149,9 @@ func (b *PodVolumeBackupBuilder) OwnerReference(ref metav1.OwnerReference) *PodV
 	b.object.OwnerReferences = append(b.object.OwnerReferences, ref)
 	return b
 }
+
+// Labels sets the PodVolumeBackup's Labels.
+func (b *PodVolumeBackupBuilder) Labels(label map[string]string) *PodVolumeBackupBuilder {
+	b.object.Labels = label
+	return b
+}

pkg/builder/pod_volume_restore_builder.go

@@ -133,3 +133,9 @@ func (b *PodVolumeRestoreBuilder) Node(node string) *PodVolumeRestoreBuilder {
 	b.object.Status.Node = node
 	return b
 }
+
+// Labels sets the PodVolumeRestoreBuilder's Labels.
+func (b *PodVolumeRestoreBuilder) Labels(label map[string]string) *PodVolumeRestoreBuilder {
+	b.object.Labels = label
+	return b
+}

pkg/cmd/cli/nodeagent/server.go

@@ -57,6 +57,7 @@
 	"github.com/vmware-tanzu/velero/pkg/constant"
 	"github.com/vmware-tanzu/velero/pkg/controller"
 	"github.com/vmware-tanzu/velero/pkg/datapath"
+	"github.com/vmware-tanzu/velero/pkg/exposer"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
 	"github.com/vmware-tanzu/velero/pkg/nodeagent"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
@@ -140,6 +141,7 @@
 	csiSnapshotClient *snapshotv1client.Clientset
 	dataPathMgr       *datapath.Manager
 	dataPathConfigs   *nodeagent.Configs
+	vgdpCounter       *exposer.VgdpCounter
 }
 
 func newNodeAgentServer(logger logrus.FieldLogger, factory client.Factory, config nodeAgentServerConfig) (*nodeAgentServer, error) {
@@ -300,12 +302,21 @@
 		}
 	}
 
-	pvbReconciler := controller.NewPodVolumeBackupReconciler(s.mgr.GetClient(), s.mgr, s.kubeClient, s.dataPathMgr, s.nodeName, s.config.dataMoverPrepareTimeout, s.config.resourceTimeout, podResources, s.metrics, s.logger)
+	if s.dataPathConfigs != nil && s.dataPathConfigs.LoadConcurrency != nil && s.dataPathConfigs.LoadConcurrency.PrepareQueueLength > 0 {
+		if counter, err := exposer.StartVgdpCounter(s.ctx, s.mgr, s.dataPathConfigs.LoadConcurrency.PrepareQueueLength); err != nil {
+			s.logger.WithError(err).Warnf("Failed to start VGDP counter, VDGP loads are not constrained")
+		} else {
+			s.vgdpCounter = counter
+			s.logger.Infof("VGDP loads are constrained with %d", s.dataPathConfigs.LoadConcurrency.PrepareQueueLength)
+		}
+	}
+
+	pvbReconciler := controller.NewPodVolumeBackupReconciler(s.mgr.GetClient(), s.mgr, s.kubeClient, s.dataPathMgr, s.vgdpCounter, s.nodeName, s.config.dataMoverPrepareTimeout, s.config.resourceTimeout, podResources, s.metrics, s.logger)
 	if err := pvbReconciler.SetupWithManager(s.mgr); err != nil {
 		s.logger.Fatal(err, "unable to create controller", "controller", constant.ControllerPodVolumeBackup)
 	}
 
-	pvrReconciler := controller.NewPodVolumeRestoreReconciler(s.mgr.GetClient(), s.mgr, s.kubeClient, s.dataPathMgr, s.nodeName, s.config.dataMoverPrepareTimeout, s.config.resourceTimeout, podResources, s.logger)
+	pvrReconciler := controller.NewPodVolumeRestoreReconciler(s.mgr.GetClient(), s.mgr, s.kubeClient, s.dataPathMgr, s.vgdpCounter, s.nodeName, s.config.dataMoverPrepareTimeout, s.config.resourceTimeout, podResources, s.logger)
 	if err := pvrReconciler.SetupWithManager(s.mgr); err != nil {
 		s.logger.WithError(err).Fatal("Unable to create the pod volume restore controller")
 	}
@@ -320,6 +331,7 @@
 		s.kubeClient,
 		s.csiSnapshotClient.SnapshotV1(),
 		s.dataPathMgr,
+		s.vgdpCounter,
 		loadAffinity,
 		backupPVCConfig,
 		podResources,
@@ -344,6 +356,7 @@
 		s.mgr,
 		s.kubeClient,
 		s.dataPathMgr,
+		s.vgdpCounter,
 		loadAffinity,
 		restorePVCConfig,
 		podResources,

pkg/controller/data_download_controller.go

@@ -64,6 +64,7 @@ type DataDownloadReconciler struct {
 	restoreExposer        exposer.GenericRestoreExposer
 	nodeName              string
 	dataPathMgr           *datapath.Manager
+	vgdpCounter           *exposer.VgdpCounter
 	loadAffinity          []*kube.LoadAffinity
 	restorePVCConfig      nodeagent.RestorePVC
 	podResources          corev1api.ResourceRequirements
@@ -77,6 +78,7 @@ func NewDataDownloadReconciler(
 	mgr manager.Manager,
 	kubeClient kubernetes.Interface,
 	dataPathMgr *datapath.Manager,
+	counter *exposer.VgdpCounter,
 	loadAffinity []*kube.LoadAffinity,
 	restorePVCConfig nodeagent.RestorePVC,
 	podResources corev1api.ResourceRequirements,
@@ -95,6 +97,7 @@ func NewDataDownloadReconciler(
 		restoreExposer:        exposer.NewGenericRestoreExposer(kubeClient, logger),
 		restorePVCConfig:      restorePVCConfig,
 		dataPathMgr:           dataPathMgr,
+		vgdpCounter:           counter,
 		loadAffinity:          loadAffinity,
 		podResources:          podResources,
 		preparingTimeout:      preparingTimeout,
@@ -220,13 +223,26 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 	}
 
 	if dd.Status.Phase == "" || dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseNew {
-		log.Info("Data download starting")
+		if dd.Spec.Cancel {
+			log.Debugf("Data download is canceled in Phase %s", dd.Status.Phase)
+
+			r.tryCancelDataDownload(ctx, dd, "")
+
+			return ctrl.Result{}, nil
+		}
+
+		if r.vgdpCounter != nil && r.vgdpCounter.IsConstrained(ctx, r.logger) {
+			log.Debug("Data path initiation is constrained, requeue later")
+			return ctrl.Result{Requeue: true, RequeueAfter: time.Second * 5}, nil
+		}
 
 		if _, err := r.getTargetPVC(ctx, dd); err != nil {
 			log.WithField("error", err).Debugf("Cannot find target PVC for DataDownload yet. Retry later.")
 			return ctrl.Result{Requeue: true}, nil
 		}
 
+		log.Info("Data download starting")
+
 		accepted, err := r.acceptDataDownload(ctx, dd)
 		if err != nil {
 			return ctrl.Result{}, errors.Wrapf(err, "error accepting the data download %s", dd.Name)
@@ -239,12 +255,6 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 
 		log.Info("Data download is accepted")
 
-		if dd.Spec.Cancel {
-			log.Debugf("Data download is been canceled %s in Phase %s", dd.GetName(), dd.Status.Phase)
-			r.OnDataDownloadCancelled(ctx, dd.GetNamespace(), dd.GetName())
-			return ctrl.Result{}, nil
-		}
-
 		exposeParam, err := r.setupExposeParam(dd)
 		if err != nil {
 			return r.errorOut(ctx, dd, err, "failed to set exposer parameters", log)
@@ -312,7 +322,7 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 			dd.Name, dd.Namespace, result.ByPod.HostingPod.Name, result.ByPod.HostingContainer, dd.Name, callbacks, false, log)
 		if err != nil {
 			if err == datapath.ConcurrentLimitExceed {
-				log.Info("Data path instance is concurrent limited requeue later")
+				log.Debug("Data path instance is concurrent limited requeue later")
 				return ctrl.Result{Requeue: true, RequeueAfter: time.Second * 5}, nil
 			} else {
 				return r.errorOut(ctx, dd, err, "error to create data path", log)
@@ -337,6 +347,8 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 			dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseInProgress
 			dd.Status.StartTimestamp = &metav1.Time{Time: r.Clock.Now()}
 
+			delete(dd.Labels, exposer.ExposeOnGoingLabel)
+
 			return true
 		}); err != nil {
 			log.WithError(err).Warnf("Failed to update datadownload %s to InProgress, will data path close and retry", dd.Name)
@@ -454,6 +466,8 @@ func (r *DataDownloadReconciler) OnDataDownloadCompleted(ctx context.Context, na
 		dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseCompleted
 		dd.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
 
+		delete(dd.Labels, exposer.ExposeOnGoingLabel)
+
 		return true
 	}); err != nil {
 		log.WithError(err).Error("error updating data download status")
@@ -504,6 +518,8 @@ func (r *DataDownloadReconciler) OnDataDownloadCancelled(ctx context.Context, na
 		}
 		dd.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
 
+		delete(dd.Labels, exposer.ExposeOnGoingLabel)
+
 		return true
 	}); err != nil {
 		log.WithError(err).Error("error updating data download status")
@@ -525,6 +541,8 @@ func (r *DataDownloadReconciler) tryCancelDataDownload(ctx context.Context, dd *
 		if message != "" {
 			dataDownload.Status.Message = message
 		}
+
+		delete(dataDownload.Labels, exposer.ExposeOnGoingLabel)
 	})
 
 	if err != nil {
@@ -702,6 +720,8 @@ func (r *DataDownloadReconciler) updateStatusToFailed(ctx context.Context, dd *v
 		dd.Status.Message = errors.WithMessage(err, msg).Error()
 		dd.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
 
+		delete(dd.Labels, exposer.ExposeOnGoingLabel)
+
 		return true
 	}); patchErr != nil {
 		log.WithError(patchErr).Error("error updating DataDownload status")
@@ -724,6 +744,11 @@ func (r *DataDownloadReconciler) acceptDataDownload(ctx context.Context, dd *vel
 		datadownload.Status.Phase = velerov2alpha1api.DataDownloadPhaseAccepted
 		datadownload.Status.AcceptedByNode = r.nodeName
 		datadownload.Status.AcceptedTimestamp = &metav1.Time{Time: r.Clock.Now()}
+
+		if datadownload.Labels == nil {
+			datadownload.Labels = make(map[string]string)
+		}
+		datadownload.Labels[exposer.ExposeOnGoingLabel] = "true"
 	}
 
 	succeeded, err := funcExclusiveUpdateDataDownload(ctx, r.client, updated, updateFunc)
@@ -749,6 +774,8 @@ func (r *DataDownloadReconciler) onPrepareTimeout(ctx context.Context, dd *veler
 	succeeded, err := funcExclusiveUpdateDataDownload(ctx, r.client, dd, func(dd *velerov2alpha1api.DataDownload) {
 		dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseFailed
 		dd.Status.Message = "timeout on preparing data download"
+
+		delete(dd.Labels, exposer.ExposeOnGoingLabel)
 	})
 
 	if err != nil {

pkg/controller/data_download_controller_test.go

@@ -136,6 +136,7 @@ func initDataDownloadReconcilerWithError(t *testing.T, objects []any, needError
 		fakeKubeClient,
 		dataPathMgr,
 		nil,
+		nil,
 		nodeagent.RestorePVC{},
 		corev1api.ResourceRequirements{},
 		"test-node",
@@ -195,6 +196,7 @@ func TestDataDownloadReconcile(t *testing.T) {
 		mockCancel               bool
 		mockClose                bool
 		needExclusiveUpdateError error
+		constrained              bool
 		expected                 *velerov2alpha1api.DataDownload
 		expectDeleted            bool
 		expectCancelRecord       bool
@@ -295,6 +297,20 @@ func TestDataDownloadReconcile(t *testing.T) {
 			name: "Unknown data download status",
 			dd:   dataDownloadBuilder().Phase("Unknown").Finalizers([]string{DataUploadDownloadFinalizer}).Result(),
 		},
+		{
+			name:               "dd is cancel on new",
+			dd:                 dataDownloadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).Cancel(true).Result(),
+			targetPVC:          builder.ForPersistentVolumeClaim("test-ns", "test-pvc").Result(),
+			expectCancelRecord: true,
+			expected:           dataDownloadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).Cancel(true).Phase(velerov2alpha1api.DataDownloadPhaseCanceled).Result(),
+		},
+		{
+			name:           "new dd but constrained",
+			dd:             dataDownloadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).Result(),
+			constrained:    true,
+			expected:       dataDownloadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).Result(),
+			expectedResult: &ctrl.Result{Requeue: true, RequeueAfter: time.Second * 5},
+		},
 		{
 			name:           "new dd but no target PVC",
 			dd:             dataDownloadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).Result(),
@@ -308,12 +324,6 @@ func TestDataDownloadReconcile(t *testing.T) {
 			expected:                 dataDownloadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).Result(),
 			expectedErr:              "error accepting the data download datadownload-1: exclusive-update-error",
 		},
-		{
-			name:      "dd is cancel on accepted",
-			dd:        dataDownloadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).Cancel(true).Result(),
-			targetPVC: builder.ForPersistentVolumeClaim("test-ns", "test-pvc").Result(),
-			expected:  dataDownloadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).Cancel(true).Phase(velerov2alpha1api.DataDownloadPhaseCanceled).Result(),
-		},
 		{
 			name:        "dd is accepted but setup expose param failed",
 			dd:          dataDownloadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).NodeOS("xxx").Result(),
@@ -488,6 +498,10 @@ func TestDataDownloadReconcile(t *testing.T) {
 				r.cancelledDataDownload[test.dd.Name] = test.sportTime.Time
 			}
 
+			if test.constrained {
+				r.vgdpCounter = &exposer.VgdpCounter{}
+			}
+
 			funcExclusiveUpdateDataDownload = exclusiveUpdateDataDownload
 			if test.needExclusiveUpdateError != nil {
 				funcExclusiveUpdateDataDownload = func(context.Context, kbclient.Client, *velerov2alpha1api.DataDownload, func(*velerov2alpha1api.DataDownload)) (bool, error) {
@@ -571,13 +585,13 @@ func TestDataDownloadReconcile(t *testing.T) {
 				assert.Equal(t, test.expectedResult.RequeueAfter, actualResult.RequeueAfter)
 			}
 
-			if test.expected != nil || test.expectDeleted {
-				dd := velerov2alpha1api.DataDownload{}
-				err = r.client.Get(ctx, kbclient.ObjectKey{
-					Name:      test.dd.Name,
-					Namespace: test.dd.Namespace,
-				}, &dd)
+			dd := velerov2alpha1api.DataDownload{}
+			err = r.client.Get(ctx, kbclient.ObjectKey{
+				Name:      test.dd.Name,
+				Namespace: test.dd.Namespace,
+			}, &dd)
 
+			if test.expected != nil || test.expectDeleted {
 				if test.expectDeleted {
 					assert.True(t, apierrors.IsNotFound(err))
 				} else {
@@ -601,6 +615,12 @@ func TestDataDownloadReconcile(t *testing.T) {
 			} else {
 				assert.Empty(t, r.cancelledDataDownload)
 			}
+
+			if isDataDownloadInFinalState(&dd) || dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseInProgress {
+				assert.NotContains(t, dd.Labels, exposer.ExposeOnGoingLabel)
+			} else if dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseAccepted {
+				assert.Contains(t, dd.Labels, exposer.ExposeOnGoingLabel)
+			}
 		})
 	}
 }