Add node-agent pod check to DataMover backup path

When SnapshotMoveData is enabled but no node-agent pods are running,
DataUpload CRs sit unprocessed until timeout. This adds a pre-flight
check in the PVC backup item action that verifies running node-agent
pods exist before creating DataUpload CRs, mirroring the check FSB
already performs. Cleans up the VolumeSnapshot on failure to prevent
orphaned resources.

Signed-off-by: Joseph <jvaikath@redhat.com>

Author: Joeavaikath

pkg/backup/actions/csi/pvc_action.go

@@ -49,6 +49,7 @@ import (
 	veleroclient "github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	"github.com/vmware-tanzu/velero/pkg/label"
+	"github.com/vmware-tanzu/velero/pkg/nodeagent"
 	plugincommon "github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
 	"github.com/vmware-tanzu/velero/pkg/plugin/utils/volumehelper"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
@@ -384,6 +385,12 @@ func (p *pvcBackupItemAction) Execute(
 
 		dataUploadLog.Info("Starting data upload of backup")
 
+		if err := nodeagent.HasRunningPods(context.Background(), backup.Namespace, p.crClient); err != nil {
+			dataUploadLog.WithError(err).Error("failed to check for running node-agent pods")
+			csi.CleanupVolumeSnapshot(vs, p.crClient, p.log)
+			return nil, nil, "", nil, errors.Wrap(err, "snapshot data movement requires a running node-agent daemonset; ensure node-agent is deployed and running")
+		}
+
 		dataUpload, err := createDataUpload(
 			context.Background(),
 			backup,

pkg/backup/actions/csi/pvc_action_test.go

@@ -94,6 +94,7 @@ func TestExecute(t *testing.T) {
 		expectedDataUpload *velerov2alpha1.DataUpload
 		expectedPVC        *corev1api.PersistentVolumeClaim
 		resourcePolicy     *corev1api.ConfigMap
+		extraObjects       []runtime.Object
 		failVSCreate       bool
 		skipVSReadyUpdate  bool // New flag to control VS readiness
 	}{
@@ -122,12 +123,24 @@ func TestExecute(t *testing.T) {
 			expectErr:         true, // Expect an error, but the exact message can vary
 		},
 		{
-			name:        "Test SnapshotMoveData",
+			name:        "Fail SnapshotMoveData when no node-agent pods",
 			backup:      builder.ForBackup("velero", "test").SnapshotMoveData(true).CSISnapshotTimeout(1 * time.Minute).Result(),
 			pvc:         builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").StorageClass("testSC").Phase(corev1api.ClaimBound).Result(),
 			pv:          builder.ForPersistentVolume("testPV").CSI("hostpath", "testVolume").Result(),
 			sc:          builder.ForStorageClass("testSC").Provisioner("hostpath").Result(),
 			vsClass:     builder.ForVolumeSnapshotClass("testVSClass").Driver("hostpath").ObjectMeta(builder.WithLabels(velerov1api.VolumeSnapshotClassSelectorLabel, "")).Result(),
+			expectedErr: errors.New("snapshot data movement requires a running node-agent daemonset; ensure node-agent is deployed and running: no running node-agent pods found"),
+		},
+		{
+			name:    "Test SnapshotMoveData",
+			backup:  builder.ForBackup("velero", "test").SnapshotMoveData(true).CSISnapshotTimeout(1 * time.Minute).Result(),
+			pvc:     builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").StorageClass("testSC").Phase(corev1api.ClaimBound).Result(),
+			pv:      builder.ForPersistentVolume("testPV").CSI("hostpath", "testVolume").Result(),
+			sc:      builder.ForStorageClass("testSC").Provisioner("hostpath").Result(),
+			vsClass: builder.ForVolumeSnapshotClass("testVSClass").Driver("hostpath").ObjectMeta(builder.WithLabels(velerov1api.VolumeSnapshotClassSelectorLabel, "")).Result(),
+			extraObjects: []runtime.Object{
+				builder.ForPod("velero", "node-agent-pod").Labels(map[string]string{"role": "node-agent"}).Phase(corev1api.PodRunning).NodeName("test-node").Result(),
+			},
 			operationID: ".",
 			expectedDataUpload: &velerov2alpha1.DataUpload{
 				TypeMeta: metav1.TypeMeta{
@@ -167,12 +180,15 @@ func TestExecute(t *testing.T) {
 			},
 		},
 		{
-			name:        "Verify PVC is modified as expected",
-			backup:      builder.ForBackup("velero", "test").SnapshotMoveData(true).CSISnapshotTimeout(1 * time.Minute).Result(),
-			pvc:         builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").StorageClass("testSC").Phase(corev1api.ClaimBound).Result(),
-			pv:          builder.ForPersistentVolume("testPV").CSI("hostpath", "testVolume").Result(),
-			sc:          builder.ForStorageClass("testSC").Provisioner("hostpath").Result(),
-			vsClass:     builder.ForVolumeSnapshotClass("tescVSClass").Driver("hostpath").ObjectMeta(builder.WithLabels(velerov1api.VolumeSnapshotClassSelectorLabel, "")).Result(),
+			name:    "Verify PVC is modified as expected",
+			backup:  builder.ForBackup("velero", "test").SnapshotMoveData(true).CSISnapshotTimeout(1 * time.Minute).Result(),
+			pvc:     builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").StorageClass("testSC").Phase(corev1api.ClaimBound).Result(),
+			pv:      builder.ForPersistentVolume("testPV").CSI("hostpath", "testVolume").Result(),
+			sc:      builder.ForStorageClass("testSC").Provisioner("hostpath").Result(),
+			vsClass: builder.ForVolumeSnapshotClass("tescVSClass").Driver("hostpath").ObjectMeta(builder.WithLabels(velerov1api.VolumeSnapshotClassSelectorLabel, "")).Result(),
+			extraObjects: []runtime.Object{
+				builder.ForPod("velero", "node-agent-pod").Labels(map[string]string{"role": "node-agent"}).Phase(corev1api.PodRunning).NodeName("test-node").Result(),
+			},
 			operationID: ".",
 			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").
 				ObjectMeta(builder.WithAnnotations(velerov1api.MustIncludeAdditionalItemAnnotation, "true", velerov1api.DataUploadNameAnnotation, "velero/"),
@@ -210,6 +226,7 @@ func TestExecute(t *testing.T) {
 			if tc.resourcePolicy != nil {
 				objects = append(objects, tc.resourcePolicy)
 			}
+			objects = append(objects, tc.extraObjects...)
 
 			var crClient crclient.Client
 			if tc.failVSCreate {

pkg/nodeagent/node_agent.go

@@ -80,6 +80,31 @@ func KbClientIsRunningInNode(ctx context.Context, namespace string, nodeName str
 	return isRunningInNode(ctx, namespace, nodeName, nil, kubeClient)
 }
 
+// HasRunningPods checks if any node agent pod is running in the namespace through controller client. If not, return the error found.
+func HasRunningPods(ctx context.Context, namespace string, crClient ctrlclient.Client) error {
+	pods := new(corev1api.PodList)
+	parsedSelector, err := labels.Parse(fmt.Sprintf("role=%s", nodeAgentRole))
+	if err != nil {
+		return errors.Wrap(err, "fail to parse selector")
+	}
+
+	err = crClient.List(ctx, pods, &ctrlclient.ListOptions{
+		LabelSelector: parsedSelector,
+		Namespace:     namespace,
+	})
+	if err != nil {
+		return errors.Wrap(err, "failed to list node-agent pods")
+	}
+
+	for i := range pods.Items {
+		if kube.IsPodRunning(&pods.Items[i]) == nil {
+			return nil
+		}
+	}
+
+	return errors.New("no running node-agent pods found")
+}
+
 // IsRunningInNode checks if the node agent pod is running properly in a specified node through controller client. If not, return the error found
 func IsRunningInNode(ctx context.Context, namespace string, nodeName string, crClient ctrlclient.Client) error {
 	return isRunningInNode(ctx, namespace, nodeName, crClient, nil)

pkg/nodeagent/node_agent_test.go

@@ -213,6 +213,72 @@ func TestIsRunningInNode(t *testing.T) {
 	}
 }
 
+func TestHasRunningPods(t *testing.T) {
+	scheme := runtime.NewScheme()
+	corev1api.AddToScheme(scheme)
+
+	nonNodeAgentPod := builder.ForPod("fake-ns", "fake-pod").Result()
+	nodeAgentPodNotRunning := builder.ForPod("fake-ns", "fake-pod-pending").Labels(map[string]string{"role": "node-agent"}).Result()
+	nodeAgentPodRunning := builder.ForPod("fake-ns", "fake-pod-running").
+		Labels(map[string]string{"role": "node-agent"}).
+		Phase(corev1api.PodRunning).
+		NodeName("fake-node").
+		Result()
+
+	tests := []struct {
+		name          string
+		kubeClientObj []runtime.Object
+		namespace     string
+		expectErr     string
+	}{
+		{
+			name:      "no pods at all",
+			namespace: "fake-ns",
+			expectErr: "no running node-agent pods found",
+		},
+		{
+			name:      "only non-node-agent pods",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				nonNodeAgentPod,
+			},
+			expectErr: "no running node-agent pods found",
+		},
+		{
+			name:      "node-agent pods exist but none running",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				nodeAgentPodNotRunning,
+			},
+			expectErr: "no running node-agent pods found",
+		},
+		{
+			name:      "at least one running node-agent pod",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				nodeAgentPodNotRunning,
+				nodeAgentPodRunning,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fakeClientBuilder := clientFake.NewClientBuilder()
+			fakeClientBuilder = fakeClientBuilder.WithScheme(scheme)
+
+			fakeClient := fakeClientBuilder.WithRuntimeObjects(test.kubeClientObj...).Build()
+
+			err := HasRunningPods(t.Context(), test.namespace, fakeClient)
+			if test.expectErr == "" {
+				assert.NoError(t, err)
+			} else {
+				assert.EqualError(t, err, test.expectErr)
+			}
+		})
+	}
+}
+
 func TestGetPodSpec(t *testing.T) {
 	podSpec := corev1api.PodSpec{
 		NodeName: "fake-node",