Add maintenance job and data mover pod's labels and annotations setting.

Add wait in file_system_test's async test cases.
Add related documents.

Signed-off-by: Xun Jiang <xun.jiang@broadcom.com>

pkg/controller/data_download_controller.go

@@ -77,6 +77,8 @@ type DataDownloadReconciler struct {
 	cancelledDataDownload map[string]time.Time
 	dataMovePriorityClass string
 	repoConfigMgr         repository.ConfigManager
+	podLabels             map[string]string
+	podAnnotations        map[string]string
 }
 
 func NewDataDownloadReconciler(
@@ -96,6 +98,8 @@ func NewDataDownloadReconciler(
 	metrics *metrics.ServerMetrics,
 	dataMovePriorityClass string,
 	repoConfigMgr repository.ConfigManager,
+	podLabels map[string]string,
+	podAnnotations map[string]string,
 ) *DataDownloadReconciler {
 	return &DataDownloadReconciler{
 		client:                client,
@@ -117,6 +121,8 @@ func NewDataDownloadReconciler(
 		cancelledDataDownload: make(map[string]time.Time),
 		dataMovePriorityClass: dataMovePriorityClass,
 		repoConfigMgr:         repoConfigMgr,
+		podLabels:             podLabels,
+		podAnnotations:        podAnnotations,
 	}
 }
 
@@ -860,25 +866,37 @@ func (r *DataDownloadReconciler) setupExposeParam(dd *velerov2alpha1api.DataDown
 	}
 
 	hostingPodLabels := map[string]string{velerov1api.DataDownloadLabel: dd.Name}
-	for _, k := range util.ThirdPartyLabels {
-		if v, err := nodeagent.GetLabelValue(context.Background(), r.kubeClient, dd.Namespace, k, nodeOS); err != nil {
-			if err != nodeagent.ErrNodeAgentLabelNotFound {
-				log.WithError(err).Warnf("Failed to check node-agent label, skip adding host pod label %s", k)
-			}
-		} else {
+	if len(r.podLabels) > 0 {
+		for k, v := range r.podLabels {
 			hostingPodLabels[k] = v
 		}
+	} else {
+		for _, k := range util.ThirdPartyLabels {
+			if v, err := nodeagent.GetLabelValue(context.Background(), r.kubeClient, dd.Namespace, k, nodeOS); err != nil {
+				if err != nodeagent.ErrNodeAgentLabelNotFound {
+					log.WithError(err).Warnf("Failed to check node-agent label, skip adding host pod label %s", k)
+				}
+			} else {
+				hostingPodLabels[k] = v
+			}
+		}
 	}
 
 	hostingPodAnnotation := map[string]string{}
-	for _, k := range util.ThirdPartyAnnotations {
-		if v, err := nodeagent.GetAnnotationValue(context.Background(), r.kubeClient, dd.Namespace, k, nodeOS); err != nil {
-			if err != nodeagent.ErrNodeAgentAnnotationNotFound {
-				log.WithError(err).Warnf("Failed to check node-agent annotation, skip adding host pod annotation %s", k)
-			}
-		} else {
+	if len(r.podAnnotations) > 0 {
+		for k, v := range r.podAnnotations {
 			hostingPodAnnotation[k] = v
 		}
+	} else {
+		for _, k := range util.ThirdPartyAnnotations {
+			if v, err := nodeagent.GetAnnotationValue(context.Background(), r.kubeClient, dd.Namespace, k, nodeOS); err != nil {
+				if err != nodeagent.ErrNodeAgentAnnotationNotFound {
+					log.WithError(err).Warnf("Failed to check node-agent annotation, skip adding host pod annotation %s", k)
+				}
+			} else {
+				hostingPodAnnotation[k] = v
+			}
+		}
 	}
 
 	hostingPodTolerations := []corev1api.Toleration{}

pkg/controller/data_download_controller_test.go

@@ -129,7 +129,26 @@ func initDataDownloadReconcilerWithError(t *testing.T, objects []any, needError
 
 	dataPathMgr := datapath.NewManager(1)
 
-	return NewDataDownloadReconciler(&fakeClient, nil, fakeKubeClient, dataPathMgr, nil, nil, velerotypes.RestorePVC{}, nil, nil, corev1api.ResourceRequirements{}, "test-node", time.Minute*5, velerotest.NewLogger(), metrics.NewServerMetrics(), "", nil), nil
+	return NewDataDownloadReconciler(
+		&fakeClient,
+		nil,
+		fakeKubeClient,
+		dataPathMgr,
+		nil,
+		nil,
+		velerotypes.RestorePVC{},
+		nil,
+		nil,
+		corev1api.ResourceRequirements{},
+		"test-node",
+		time.Minute*5,
+		velerotest.NewLogger(),
+		metrics.NewServerMetrics(),
+		"",
+		nil,
+		nil, // podLabels
+		nil, // podAnnotations
+	), nil
 }
 
 func TestDataDownloadReconcile(t *testing.T) {
@@ -1292,3 +1311,127 @@ func TestResumeCancellableRestore(t *testing.T) {
 		})
 	}
 }
+
+func TestDataDownloadSetupExposeParam(t *testing.T) {
+	// Common objects for all cases
+	node := builder.ForNode("worker-1").Labels(map[string]string{kube.NodeOSLabel: kube.NodeOSLinux}).Result()
+
+	baseDataDownload := dataDownloadBuilder().Result()
+	baseDataDownload.Namespace = velerov1api.DefaultNamespace
+	baseDataDownload.Spec.OperationTimeout = metav1.Duration{Duration: time.Minute * 10}
+	baseDataDownload.Spec.SnapshotSize = 5368709120 // 5Gi
+
+	type args struct {
+		customLabels      map[string]string
+		customAnnotations map[string]string
+	}
+	type want struct {
+		labels      map[string]string
+		annotations map[string]string
+	}
+
+	tests := []struct {
+		name string
+		args args
+		want want
+	}{
+		{
+			name: "label has customize values",
+			args: args{
+				customLabels:      map[string]string{"custom-label": "label-value"},
+				customAnnotations: nil,
+			},
+			want: want{
+				labels: map[string]string{
+					velerov1api.DataDownloadLabel: baseDataDownload.Name,
+					"custom-label":                "label-value",
+				},
+				annotations: map[string]string{},
+			},
+		},
+		{
+			name: "label has no customize values",
+			args: args{
+				customLabels:      nil,
+				customAnnotations: nil,
+			},
+			want: want{
+				labels:      map[string]string{velerov1api.DataDownloadLabel: baseDataDownload.Name},
+				annotations: map[string]string{},
+			},
+		},
+		{
+			name: "annotation has customize values",
+			args: args{
+				customLabels:      nil,
+				customAnnotations: map[string]string{"custom-annotation": "annotation-value"},
+			},
+			want: want{
+				labels:      map[string]string{velerov1api.DataDownloadLabel: baseDataDownload.Name},
+				annotations: map[string]string{"custom-annotation": "annotation-value"},
+			},
+		},
+		{
+			name: "both label and annotation have customize values",
+			args: args{
+				customLabels:      map[string]string{"custom-label": "label-value"},
+				customAnnotations: map[string]string{"custom-annotation": "annotation-value"},
+			},
+			want: want{
+				labels: map[string]string{
+					velerov1api.DataDownloadLabel: baseDataDownload.Name,
+					"custom-label":                "label-value",
+				},
+				annotations: map[string]string{"custom-annotation": "annotation-value"},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Fake clients per case
+			fakeClient := FakeClient{
+				Client: velerotest.NewFakeControllerRuntimeClient(t, node, baseDataDownload.DeepCopy()),
+			}
+			fakeKubeClient := clientgofake.NewSimpleClientset(node)
+
+			// Reconciler config per case
+			preparingTimeout := time.Minute * 3
+			podRes := corev1api.ResourceRequirements{}
+			r := NewDataDownloadReconciler(
+				&fakeClient,
+				nil,
+				fakeKubeClient,
+				datapath.NewManager(1),
+				nil,
+				nil,
+				velerotypes.RestorePVC{},
+				nil,
+				nil,
+				podRes,
+				"test-node",
+				preparingTimeout,
+				velerotest.NewLogger(),
+				metrics.NewServerMetrics(),
+				"download-priority",
+				nil, // repoConfigMgr (unused when cacheVolumeConfigs is nil)
+				tt.args.customLabels,
+				tt.args.customAnnotations,
+			)
+
+			// Act
+			got, err := r.setupExposeParam(baseDataDownload)
+
+			// Assert no error
+			require.NoError(t, err)
+
+			// Core fields
+			assert.Equal(t, baseDataDownload.Spec.TargetVolume.PVC, got.TargetPVCName)
+			assert.Equal(t, baseDataDownload.Spec.TargetVolume.Namespace, got.TargetNamespace)
+
+			// Labels and Annotations
+			assert.Equal(t, tt.want.labels, got.HostingPodLabels)
+			assert.Equal(t, tt.want.annotations, got.HostingPodAnnotations)
+		})
+	}
+}

pkg/controller/data_upload_controller.go

@@ -83,6 +83,8 @@ type DataUploadReconciler struct {
 	metrics               *metrics.ServerMetrics
 	cancelledDataUpload   map[string]time.Time
 	dataMovePriorityClass string
+	podLabels             map[string]string
+	podAnnotations        map[string]string
 }
 
 func NewDataUploadReconciler(
@@ -101,6 +103,8 @@ func NewDataUploadReconciler(
 	log logrus.FieldLogger,
 	metrics *metrics.ServerMetrics,
 	dataMovePriorityClass string,
+	podLabels map[string]string,
+	podAnnotations map[string]string,
 ) *DataUploadReconciler {
 	return &DataUploadReconciler{
 		client:            client,
@@ -126,6 +130,8 @@ func NewDataUploadReconciler(
 		metrics:               metrics,
 		cancelledDataUpload:   make(map[string]time.Time),
 		dataMovePriorityClass: dataMovePriorityClass,
+		podLabels:             podLabels,
+		podAnnotations:        podAnnotations,
 	}
 }
 
@@ -936,25 +942,37 @@ func (r *DataUploadReconciler) setupExposeParam(du *velerov2alpha1api.DataUpload
 		}
 
 		hostingPodLabels := map[string]string{velerov1api.DataUploadLabel: du.Name}
-		for _, k := range util.ThirdPartyLabels {
-			if v, err := nodeagent.GetLabelValue(context.Background(), r.kubeClient, du.Namespace, k, nodeOS); err != nil {
-				if err != nodeagent.ErrNodeAgentLabelNotFound {
-					log.WithError(err).Warnf("Failed to check node-agent label, skip adding host pod label %s", k)
-				}
-			} else {
+		if len(r.podLabels) > 0 {
+			for k, v := range r.podLabels {
 				hostingPodLabels[k] = v
 			}
+		} else {
+			for _, k := range util.ThirdPartyLabels {
+				if v, err := nodeagent.GetLabelValue(context.Background(), r.kubeClient, du.Namespace, k, nodeOS); err != nil {
+					if err != nodeagent.ErrNodeAgentLabelNotFound {
+						log.WithError(err).Warnf("Failed to check node-agent label, skip adding host pod label %s", k)
+					}
+				} else {
+					hostingPodLabels[k] = v
+				}
+			}
 		}
 
 		hostingPodAnnotation := map[string]string{}
-		for _, k := range util.ThirdPartyAnnotations {
-			if v, err := nodeagent.GetAnnotationValue(context.Background(), r.kubeClient, du.Namespace, k, nodeOS); err != nil {
-				if err != nodeagent.ErrNodeAgentAnnotationNotFound {
-					log.WithError(err).Warnf("Failed to check node-agent annotation, skip adding host pod annotation %s", k)
-				}
-			} else {
+		if len(r.podAnnotations) > 0 {
+			for k, v := range r.podAnnotations {
 				hostingPodAnnotation[k] = v
 			}
+		} else {
+			for _, k := range util.ThirdPartyAnnotations {
+				if v, err := nodeagent.GetAnnotationValue(context.Background(), r.kubeClient, du.Namespace, k, nodeOS); err != nil {
+					if err != nodeagent.ErrNodeAgentAnnotationNotFound {
+						log.WithError(err).Warnf("Failed to check node-agent annotation, skip adding host pod annotation %s", k)
+					}
+				} else {
+					hostingPodAnnotation[k] = v
+				}
+			}
 		}
 
 		hostingPodTolerations := []corev1api.Toleration{}

pkg/controller/data_upload_controller_test.go

@@ -248,7 +248,9 @@ func initDataUploaderReconcilerWithError(needError ...error) (*DataUploadReconci
 		time.Minute*5,
 		velerotest.NewLogger(),
 		metrics.NewServerMetrics(),
-		"", // dataMovePriorityClass
+		"",  // dataMovePriorityClass
+		nil, // podLabels
+		nil, // podAnnotations
 	), nil
 }
 
@@ -1384,3 +1386,149 @@ func TestResumeCancellableBackup(t *testing.T) {
 		})
 	}
 }
+
+func TestDataUploadSetupExposeParam(t *testing.T) {
+	// Common objects for all cases
+	fileMode := corev1api.PersistentVolumeFilesystem
+	node := builder.ForNode("worker-1").Labels(map[string]string{kube.NodeOSLabel: kube.NodeOSLinux}).Result()
+
+	pvc := &corev1api.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "app-ns",
+			Name:      "test-pvc",
+		},
+		Spec: corev1api.PersistentVolumeClaimSpec{
+			VolumeName: "test-pv",
+			VolumeMode: &fileMode,
+			Resources: corev1api.VolumeResourceRequirements{
+				Requests: corev1api.ResourceList{
+					corev1api.ResourceStorage: resource.MustParse("10Gi"),
+				},
+			},
+		},
+	}
+
+	pv := &corev1api.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-pv",
+		},
+	}
+
+	baseDataUpload := dataUploadBuilder().Result()
+	baseDataUpload.Spec.SourceNamespace = "app-ns"
+	baseDataUpload.Spec.SourcePVC = "test-pvc"
+	baseDataUpload.Namespace = velerov1api.DefaultNamespace
+	baseDataUpload.Spec.OperationTimeout = metav1.Duration{Duration: time.Minute * 10}
+
+	type args struct {
+		customLabels      map[string]string
+		customAnnotations map[string]string
+	}
+	type want struct {
+		labels      map[string]string
+		annotations map[string]string
+	}
+
+	tests := []struct {
+		name string
+		args args
+		want want
+	}{
+		{
+			name: "label has customize values",
+			args: args{
+				customLabels:      map[string]string{"custom-label": "label-value"},
+				customAnnotations: nil,
+			},
+			want: want{
+				labels: map[string]string{
+					velerov1api.DataUploadLabel: baseDataUpload.Name,
+					"custom-label":              "label-value",
+				},
+				annotations: map[string]string{},
+			},
+		},
+		{
+			name: "label has no customize values",
+			args: args{
+				customLabels:      nil,
+				customAnnotations: nil,
+			},
+			want: want{
+				labels:      map[string]string{velerov1api.DataUploadLabel: baseDataUpload.Name},
+				annotations: map[string]string{},
+			},
+		},
+		{
+			name: "annotation has customize values",
+			args: args{
+				customLabels:      nil,
+				customAnnotations: map[string]string{"custom-annotation": "annotation-value"},
+			},
+			want: want{
+				labels:      map[string]string{velerov1api.DataUploadLabel: baseDataUpload.Name},
+				annotations: map[string]string{"custom-annotation": "annotation-value"},
+			},
+		},
+		{
+			name: "both label and annotation have customize values",
+			args: args{
+				customLabels:      map[string]string{"custom-label": "label-value"},
+				customAnnotations: map[string]string{"custom-annotation": "annotation-value"},
+			},
+			want: want{
+				labels: map[string]string{
+					velerov1api.DataUploadLabel: baseDataUpload.Name,
+					"custom-label":              "label-value",
+				},
+				annotations: map[string]string{"custom-annotation": "annotation-value"},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Fake clients per case
+			fakeCRClient := velerotest.NewFakeControllerRuntimeClient(t, pvc, pv, node, baseDataUpload.DeepCopy())
+			fakeKubeClient := clientgofake.NewSimpleClientset(node)
+
+			// Reconciler config per case
+			preparingTimeout := time.Minute * 3
+			podRes := corev1api.ResourceRequirements{}
+			r := NewDataUploadReconciler(
+				fakeCRClient,
+				nil,
+				fakeKubeClient,
+				nil, // snapshotClient (unused in setupExposeParam)
+				datapath.NewManager(1),
+				nil, // dataPathMgr
+				nil, // exposer (unused in setupExposeParam)
+				map[string]velerotypes.BackupPVC{},
+				podRes,
+				testclocks.NewFakeClock(time.Now()),
+				"test-node",
+				preparingTimeout,
+				velerotest.NewLogger(),
+				metrics.NewServerMetrics(),
+				"upload-priority",
+				tt.args.customLabels,
+				tt.args.customAnnotations,
+			)
+
+			// Act
+			got, err := r.setupExposeParam(baseDataUpload)
+
+			// Assert no error
+			require.NoError(t, err)
+			require.NotNil(t, got)
+
+			// Type assertion to CSISnapshotExposeParam
+			csiParam, ok := got.(*exposer.CSISnapshotExposeParam)
+			require.True(t, ok, "expected CSISnapshotExposeParam type")
+
+			// Labels and Annotations
+			assert.Equal(t, tt.want.labels, csiParam.HostingPodLabels)
+			assert.Equal(t, tt.want.annotations, csiParam.HostingPodAnnotations)
+		})
+	}
+}

pkg/controller/pod_volume_backup_controller.go

@@ -58,9 +58,23 @@ const (
 )
 
 // NewPodVolumeBackupReconciler creates the PodVolumeBackupReconciler instance
-func NewPodVolumeBackupReconciler(client client.Client, mgr manager.Manager, kubeClient kubernetes.Interface, dataPathMgr *datapath.Manager,
-	counter *exposer.VgdpCounter, nodeName string, preparingTimeout time.Duration, resourceTimeout time.Duration, podResources corev1api.ResourceRequirements,
-	metrics *metrics.ServerMetrics, logger logrus.FieldLogger, dataMovePriorityClass string, privileged bool) *PodVolumeBackupReconciler {
+func NewPodVolumeBackupReconciler(
+	client client.Client,
+	mgr manager.Manager,
+	kubeClient kubernetes.Interface,
+	dataPathMgr *datapath.Manager,
+	counter *exposer.VgdpCounter,
+	nodeName string,
+	preparingTimeout time.Duration,
+	resourceTimeout time.Duration,
+	podResources corev1api.ResourceRequirements,
+	metrics *metrics.ServerMetrics,
+	logger logrus.FieldLogger,
+	dataMovePriorityClass string,
+	privileged bool,
+	podLabels map[string]string,
+	podAnnotations map[string]string,
+) *PodVolumeBackupReconciler {
 	return &PodVolumeBackupReconciler{
 		client:                client,
 		mgr:                   mgr,
@@ -78,6 +92,8 @@ func NewPodVolumeBackupReconciler(client client.Client, mgr manager.Manager, kub
 		cancelledPVB:          make(map[string]time.Time),
 		dataMovePriorityClass: dataMovePriorityClass,
 		privileged:            privileged,
+		podLabels:             podLabels,
+		podAnnotations:        podAnnotations,
 	}
 }
 
@@ -99,6 +115,8 @@ type PodVolumeBackupReconciler struct {
 	cancelledPVB          map[string]time.Time
 	dataMovePriorityClass string
 	privileged            bool
+	podLabels             map[string]string
+	podAnnotations        map[string]string
 }
 
 // +kubebuilder:rbac:groups=velero.io,resources=podvolumebackups,verbs=get;list;watch;create;update;patch;delete
@@ -796,25 +814,37 @@ func (r *PodVolumeBackupReconciler) setupExposeParam(pvb *velerov1api.PodVolumeB
 	}
 
 	hostingPodLabels := map[string]string{velerov1api.PVBLabel: pvb.Name}
-	for _, k := range util.ThirdPartyLabels {
-		if v, err := nodeagent.GetLabelValue(context.Background(), r.kubeClient, pvb.Namespace, k, nodeOS); err != nil {
-			if err != nodeagent.ErrNodeAgentLabelNotFound {
-				log.WithError(err).Warnf("Failed to check node-agent label, skip adding host pod label %s", k)
-			}
-		} else {
+	if len(r.podLabels) > 0 {
+		for k, v := range r.podLabels {
 			hostingPodLabels[k] = v
 		}
+	} else {
+		for _, k := range util.ThirdPartyLabels {
+			if v, err := nodeagent.GetLabelValue(context.Background(), r.kubeClient, pvb.Namespace, k, nodeOS); err != nil {
+				if err != nodeagent.ErrNodeAgentLabelNotFound {
+					log.WithError(err).Warnf("Failed to check node-agent label, skip adding host pod label %s", k)
+				}
+			} else {
+				hostingPodLabels[k] = v
+			}
+		}
 	}
 
 	hostingPodAnnotation := map[string]string{}
-	for _, k := range util.ThirdPartyAnnotations {
-		if v, err := nodeagent.GetAnnotationValue(context.Background(), r.kubeClient, pvb.Namespace, k, nodeOS); err != nil {
-			if err != nodeagent.ErrNodeAgentAnnotationNotFound {
-				log.WithError(err).Warnf("Failed to check node-agent annotation, skip adding host pod annotation %s", k)
-			}
-		} else {
+	if len(r.podAnnotations) > 0 {
+		for k, v := range r.podAnnotations {
 			hostingPodAnnotation[k] = v
 		}
+	} else {
+		for _, k := range util.ThirdPartyAnnotations {
+			if v, err := nodeagent.GetAnnotationValue(context.Background(), r.kubeClient, pvb.Namespace, k, nodeOS); err != nil {
+				if err != nodeagent.ErrNodeAgentAnnotationNotFound {
+					log.WithError(err).Warnf("Failed to check node-agent annotation, skip adding host pod annotation %s", k)
+				}
+			} else {
+				hostingPodAnnotation[k] = v
+			}
+		}
 	}
 
 	hostingPodTolerations := []corev1api.Toleration{}

pkg/controller/pod_volume_backup_controller_test.go

@@ -47,13 +47,12 @@ import (
 	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/datapath"
+	datapathmocks "github.com/vmware-tanzu/velero/pkg/datapath/mocks"
 	"github.com/vmware-tanzu/velero/pkg/exposer"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
-
-	datapathmocks "github.com/vmware-tanzu/velero/pkg/datapath/mocks"
 )
 
 const pvbName = "pvb-1"
@@ -153,6 +152,8 @@ func initPVBReconcilerWithError(needError ...error) (*PodVolumeBackupReconciler,
 		velerotest.NewLogger(),
 		"",    // dataMovePriorityClass
 		false, // privileged
+		nil,   // podLabels
+		nil,   // podAnnotations
 	), nil
 }
 
@@ -1187,3 +1188,123 @@ func TestResumeCancellablePodVolumeBackup(t *testing.T) {
 		})
 	}
 }
+
+func TestPodVolumeBackupSetupExposeParam(t *testing.T) {
+	// common objects for all cases
+	node := builder.ForNode("worker-1").Labels(map[string]string{kube.NodeOSLabel: kube.NodeOSLinux}).Result()
+
+	basePVB := pvbBuilder().Result()
+	basePVB.Spec.Node = "worker-1"
+	basePVB.Spec.Pod.Namespace = "app-ns"
+	basePVB.Spec.Pod.Name = "app-pod"
+	basePVB.Spec.Volume = "data-vol"
+
+	type args struct {
+		customLabels      map[string]string
+		customAnnotations map[string]string
+	}
+	type want struct {
+		labels      map[string]string
+		annotations map[string]string
+	}
+
+	tests := []struct {
+		name string
+		args args
+		want want
+	}{
+		{
+			name: "label has customize values",
+			args: args{
+				customLabels:      map[string]string{"custom-label": "label-value"},
+				customAnnotations: nil,
+			},
+			want: want{
+				labels: map[string]string{
+					velerov1api.PVBLabel: basePVB.Name,
+					"custom-label":       "label-value",
+				},
+				annotations: map[string]string{},
+			},
+		},
+		{
+			name: "label has no customize values",
+			args: args{
+				customLabels:      nil,
+				customAnnotations: nil,
+			},
+			want: want{
+				labels:      map[string]string{velerov1api.PVBLabel: basePVB.Name},
+				annotations: map[string]string{},
+			},
+		},
+		{
+			name: "annotation has customize values",
+			args: args{
+				customLabels:      nil,
+				customAnnotations: map[string]string{"custom-annotation": "annotation-value"},
+			},
+			want: want{
+				labels:      map[string]string{velerov1api.PVBLabel: basePVB.Name},
+				annotations: map[string]string{"custom-annotation": "annotation-value"},
+			},
+		},
+		{
+			name: "annotation has no customize values",
+			args: args{
+				customLabels:      map[string]string{"another-label": "lval"},
+				customAnnotations: nil,
+			},
+			want: want{
+				labels: map[string]string{
+					velerov1api.PVBLabel: basePVB.Name,
+					"another-label":      "lval",
+				},
+				annotations: map[string]string{},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Fake clients per case
+			fakeCRClient := velerotest.NewFakeControllerRuntimeClient(t, node, basePVB.DeepCopy())
+			fakeKubeClient := clientgofake.NewSimpleClientset(node)
+
+			// Reconciler config per case
+			preparingTimeout := time.Minute * 3
+			resourceTimeout := time.Minute * 10
+			podRes := corev1api.ResourceRequirements{}
+			r := NewPodVolumeBackupReconciler(
+				fakeCRClient,
+				nil,
+				fakeKubeClient,
+				datapath.NewManager(1),
+				nil,
+				"test-node",
+				preparingTimeout,
+				resourceTimeout,
+				podRes,
+				metrics.NewServerMetrics(),
+				velerotest.NewLogger(),
+				"backup-priority",
+				true,
+				tt.args.customLabels,
+				tt.args.customAnnotations,
+			)
+
+			// Act
+			got := r.setupExposeParam(basePVB)
+
+			// Core fields
+			assert.Equal(t, exposer.PodVolumeExposeTypeBackup, got.Type)
+			assert.Equal(t, basePVB.Spec.Pod.Namespace, got.ClientNamespace)
+			assert.Equal(t, basePVB.Spec.Pod.Name, got.ClientPodName)
+			assert.Equal(t, basePVB.Spec.Volume, got.ClientPodVolume)
+
+			// Labels/Annotations
+			assert.Equal(t, tt.want.labels, got.HostingPodLabels)
+			assert.Equal(t, tt.want.annotations, got.HostingPodAnnotations)
+		})
+	}
+}

pkg/controller/pod_volume_restore_controller.go

@@ -56,10 +56,25 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
-func NewPodVolumeRestoreReconciler(client client.Client, mgr manager.Manager, kubeClient kubernetes.Interface, dataPathMgr *datapath.Manager,
-	counter *exposer.VgdpCounter, nodeName string, preparingTimeout time.Duration, resourceTimeout time.Duration, backupRepoConfigs map[string]string,
-	cacheVolumeConfigs *velerotypes.CachePVC, podResources corev1api.ResourceRequirements, logger logrus.FieldLogger, dataMovePriorityClass string,
-	privileged bool, repoConfigMgr repository.ConfigManager) *PodVolumeRestoreReconciler {
+func NewPodVolumeRestoreReconciler(
+	client client.Client,
+	mgr manager.Manager,
+	kubeClient kubernetes.Interface,
+	dataPathMgr *datapath.Manager,
+	counter *exposer.VgdpCounter,
+	nodeName string,
+	preparingTimeout time.Duration,
+	resourceTimeout time.Duration,
+	backupRepoConfigs map[string]string,
+	cacheVolumeConfigs *velerotypes.CachePVC,
+	podResources corev1api.ResourceRequirements,
+	logger logrus.FieldLogger,
+	dataMovePriorityClass string,
+	privileged bool,
+	repoConfigMgr repository.ConfigManager,
+	podLabels map[string]string,
+	podAnnotations map[string]string,
+) *PodVolumeRestoreReconciler {
 	return &PodVolumeRestoreReconciler{
 		client:                client,
 		mgr:                   mgr,
@@ -79,6 +94,8 @@ func NewPodVolumeRestoreReconciler(client client.Client, mgr manager.Manager, ku
 		dataMovePriorityClass: dataMovePriorityClass,
 		privileged:            privileged,
 		repoConfigMgr:         repoConfigMgr,
+		podLabels:             podLabels,
+		podAnnotations:        podAnnotations,
 	}
 }
 
@@ -101,6 +118,8 @@ type PodVolumeRestoreReconciler struct {
 	dataMovePriorityClass string
 	privileged            bool
 	repoConfigMgr         repository.ConfigManager
+	podLabels             map[string]string
+	podAnnotations        map[string]string
 }
 
 // +kubebuilder:rbac:groups=velero.io,resources=podvolumerestores,verbs=get;list;watch;create;update;patch;delete
@@ -863,25 +882,37 @@ func (r *PodVolumeRestoreReconciler) setupExposeParam(pvr *velerov1api.PodVolume
 	}
 
 	hostingPodLabels := map[string]string{velerov1api.PVRLabel: pvr.Name}
-	for _, k := range util.ThirdPartyLabels {
-		if v, err := nodeagent.GetLabelValue(context.Background(), r.kubeClient, pvr.Namespace, k, nodeOS); err != nil {
-			if err != nodeagent.ErrNodeAgentLabelNotFound {
-				log.WithError(err).Warnf("Failed to check node-agent label, skip adding host pod label %s", k)
-			}
-		} else {
+	if len(r.podLabels) > 0 {
+		for k, v := range r.podLabels {
 			hostingPodLabels[k] = v
 		}
+	} else {
+		for _, k := range util.ThirdPartyLabels {
+			if v, err := nodeagent.GetLabelValue(context.Background(), r.kubeClient, pvr.Namespace, k, nodeOS); err != nil {
+				if err != nodeagent.ErrNodeAgentLabelNotFound {
+					log.WithError(err).Warnf("Failed to check node-agent label, skip adding host pod label %s", k)
+				}
+			} else {
+				hostingPodLabels[k] = v
+			}
+		}
 	}
 
 	hostingPodAnnotation := map[string]string{}
-	for _, k := range util.ThirdPartyAnnotations {
-		if v, err := nodeagent.GetAnnotationValue(context.Background(), r.kubeClient, pvr.Namespace, k, nodeOS); err != nil {
-			if err != nodeagent.ErrNodeAgentAnnotationNotFound {
-				log.WithError(err).Warnf("Failed to check node-agent annotation, skip adding host pod annotation %s", k)
-			}
-		} else {
+	if len(r.podAnnotations) > 0 {
+		for k, v := range r.podAnnotations {
 			hostingPodAnnotation[k] = v
 		}
+	} else {
+		for _, k := range util.ThirdPartyAnnotations {
+			if v, err := nodeagent.GetAnnotationValue(context.Background(), r.kubeClient, pvr.Namespace, k, nodeOS); err != nil {
+				if err != nodeagent.ErrNodeAgentAnnotationNotFound {
+					log.WithError(err).Warnf("Failed to check node-agent annotation, skip adding host pod annotation %s", k)
+				}
+			} else {
+				hostingPodAnnotation[k] = v
+			}
+		}
 	}
 
 	hostingPodTolerations := []corev1api.Toleration{}

pkg/controller/pod_volume_restore_controller_test.go

@@ -617,7 +617,25 @@ func initPodVolumeRestoreReconcilerWithError(objects []runtime.Object, cliObj []
 
 	dataPathMgr := datapath.NewManager(1)
 
-	return NewPodVolumeRestoreReconciler(fakeClient, nil, fakeKubeClient, dataPathMgr, nil, "test-node", time.Minute*5, time.Minute, nil, nil, corev1api.ResourceRequirements{}, velerotest.NewLogger(), "", false, nil), nil
+	return NewPodVolumeRestoreReconciler(
+		fakeClient,
+		nil,
+		fakeKubeClient,
+		dataPathMgr,
+		nil,
+		"test-node",
+		time.Minute*5,
+		time.Minute,
+		nil,
+		nil,
+		corev1api.ResourceRequirements{},
+		velerotest.NewLogger(),
+		"",
+		false,
+		nil,
+		nil, // podLabels
+		nil, // podAnnotations
+	), nil
 }
 
 func TestPodVolumeRestoreReconcile(t *testing.T) {
@@ -1082,6 +1100,128 @@ func TestPodVolumeRestoreReconcile(t *testing.T) {
 	}
 }
 
+func TestPodVolumeRestoreSetupExposeParam(t *testing.T) {
+	// common objects for all cases
+	node := builder.ForNode("worker-1").Labels(map[string]string{kube.NodeOSLabel: kube.NodeOSLinux}).Result()
+
+	basePVR := pvrBuilder().Result()
+	basePVR.Status.Node = "worker-1"
+	basePVR.Spec.Pod.Namespace = "app-ns"
+	basePVR.Spec.Pod.Name = "app-pod"
+	basePVR.Spec.Volume = "data-vol"
+
+	type args struct {
+		customLabels      map[string]string
+		customAnnotations map[string]string
+	}
+	type want struct {
+		labels      map[string]string
+		annotations map[string]string
+	}
+
+	tests := []struct {
+		name string
+		args args
+		want want
+	}{
+		{
+			name: "label has customize values",
+			args: args{
+				customLabels:      map[string]string{"custom-label": "label-value"},
+				customAnnotations: nil,
+			},
+			want: want{
+				labels: map[string]string{
+					velerov1api.PVRLabel: basePVR.Name,
+					"custom-label":       "label-value",
+				},
+				annotations: map[string]string{},
+			},
+		},
+		{
+			name: "label has no customize values",
+			args: args{
+				customLabels:      nil,
+				customAnnotations: nil,
+			},
+			want: want{
+				labels:      map[string]string{velerov1api.PVRLabel: basePVR.Name},
+				annotations: map[string]string{},
+			},
+		},
+		{
+			name: "annotation has customize values",
+			args: args{
+				customLabels:      nil,
+				customAnnotations: map[string]string{"custom-annotation": "annotation-value"},
+			},
+			want: want{
+				labels:      map[string]string{velerov1api.PVRLabel: basePVR.Name},
+				annotations: map[string]string{"custom-annotation": "annotation-value"},
+			},
+		},
+		{
+			name: "annotation has no customize values",
+			args: args{
+				customLabels:      map[string]string{"another-label": "lval"},
+				customAnnotations: nil,
+			},
+			want: want{
+				labels: map[string]string{
+					velerov1api.PVRLabel: basePVR.Name,
+					"another-label":      "lval",
+				},
+				annotations: map[string]string{},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Fake clients per case
+			fakeCRClient := velerotest.NewFakeControllerRuntimeClient(t, node, basePVR.DeepCopy())
+			fakeKubeClient := clientgofake.NewSimpleClientset(node)
+
+			// Reconciler config per case
+			preparingTimeout := time.Minute * 3
+			resourceTimeout := time.Minute * 10
+			podRes := corev1api.ResourceRequirements{}
+			r := NewPodVolumeRestoreReconciler(
+				fakeCRClient,
+				nil,
+				fakeKubeClient,
+				datapath.NewManager(1),
+				nil,
+				"test-node",
+				preparingTimeout,
+				resourceTimeout,
+				nil, // backupRepoConfigs
+				nil, // cacheVolumeConfigs -> keep nil so CacheVolume is nil
+				podRes,
+				velerotest.NewLogger(),
+				"restore-priority",
+				true,
+				nil, // repoConfigMgr (unused when cacheVolumeConfigs is nil)
+				tt.args.customLabels,
+				tt.args.customAnnotations,
+			)
+
+			// Act
+			got := r.setupExposeParam(basePVR)
+
+			// Core fields
+			assert.Equal(t, exposer.PodVolumeExposeTypeRestore, got.Type)
+			assert.Equal(t, basePVR.Spec.Pod.Namespace, got.ClientNamespace)
+			assert.Equal(t, basePVR.Spec.Pod.Name, got.ClientPodName)
+			assert.Equal(t, basePVR.Spec.Volume, got.ClientPodVolume)
+
+			// Labels/Annotations
+			assert.Equal(t, tt.want.labels, got.HostingPodLabels)
+			assert.Equal(t, tt.want.annotations, got.HostingPodAnnotations)
+		})
+	}
+}
+
 func TestOnPodVolumeRestoreFailed(t *testing.T) {
 	for _, getErr := range []bool{true, false} {
 		ctx := t.Context()