Merge pull request #9697 from Joeavaikath/add-nodeagent-check-datamover

Fast-fail backup when built-in data mover has no running node-agent

changelogs/unreleased/9697-Joeavaikath

@@ -0,0 +1 @@
+Fail backup validation when built-in data mover is requested but no node-agent pods are running

pkg/backup/actions/csi/pvc_action.go

@@ -48,6 +48,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"
@@ -55,6 +56,7 @@ import (
 	uploaderUtil "github.com/vmware-tanzu/velero/pkg/uploader/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/csi"
+	datamover "github.com/vmware-tanzu/velero/pkg/util/datamover"
 	kubeutil "github.com/vmware-tanzu/velero/pkg/util/kube"
 	podvolumeutil "github.com/vmware-tanzu/velero/pkg/util/podvolume"
 	vhutil "github.com/vmware-tanzu/velero/pkg/util/volumehelper"
@@ -340,6 +342,17 @@ func (p *pvcBackupItemAction) Execute(
 		return nil, nil, "", nil, err
 	}
 
+	// validate that the node-agent daemonset is ready when snapshot data movement with
+	// the built-in data mover is requested. Without this, the DataUpload CR will be
+	// created but never processed (the DataUpload controller runs inside node-agent),
+	// causing the backup to hang until itemOperationTimeout expires.
+	if boolptr.IsSetToTrue(backup.Spec.SnapshotMoveData) && datamover.IsBuiltInDataMover(backup.Spec.DataMover) {
+		if err := nodeagent.IsReady(context.TODO(), backup.Namespace, p.crClient, p.log); err != nil {
+			p.log.WithError(err).Error("cannot perform snapshot data movement without running node-agent pods")
+			return nil, nil, "", nil, errors.Wrap(err, "CSI PVC BIA cannot proceed: node-agent is not ready for snapshot data movement")
+		}
+	}
+
 	policySnapshotClass, scErr := vh.GetSnapshotClass(item, kuberesource.PersistentVolumeClaims)
 	if scErr != nil {
 		p.log.WithError(scErr).Warn("failed to get snapshotClass from volume policy, proceeding without it")

pkg/backup/actions/csi/pvc_action_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	appsv1api "k8s.io/api/apps/v1"
 	corev1api "k8s.io/api/core/v1"
 	storagev1api "k8s.io/api/storage/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -92,6 +93,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
 		expectedVSClassName string
@@ -121,12 +123,21 @@ func TestExecute(t *testing.T) {
 			expectErr:         true, // Expect an error, but the exact message can vary
 		},
 		{
-			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(),
+			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{
+				&corev1api.Node{
+					ObjectMeta: metav1.ObjectMeta{Name: "linux-node", Labels: map[string]string{"kubernetes.io/os": "linux"}},
+				},
+				&appsv1api.DaemonSet{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "velero", Name: "node-agent"},
+					Status:     appsv1api.DaemonSetStatus{NumberReady: 3},
+				},
+			},
 			operationID: ".",
 			expectedDataUpload: &velerov2alpha1.DataUpload{
 				TypeMeta: metav1.TypeMeta{
@@ -167,18 +178,37 @@ 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{
+				&corev1api.Node{
+					ObjectMeta: metav1.ObjectMeta{Name: "linux-node", Labels: map[string]string{"kubernetes.io/os": "linux"}},
+				},
+				&appsv1api.DaemonSet{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "velero", Name: "node-agent"},
+					Status:     appsv1api.DaemonSetStatus{NumberReady: 3},
+				},
+			},
 			operationID: ".",
 			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").
 				ObjectMeta(builder.WithAnnotations(velerov1api.MustIncludeAdditionalItemAnnotation, "true", velerov1api.DataUploadNameAnnotation, "velero/"),
 					builder.WithLabels(velerov1api.BackupNameLabel, "test")).
 				VolumeName("testPV").StorageClass("testSC").Phase(corev1api.ClaimBound).Result(),
 		},
+		{
+			name:              "Test SnapshotMoveData without node-agent",
+			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(),
+			expectErr:         true,
+			skipVSReadyUpdate: true,
+		},
 		{
 			name:           "Test ResourcePolicy",
 			backup:         builder.ForBackup("velero", "test").ResourcePolicies("resourcePolicy").SnapshotVolumes(false).CSISnapshotTimeout(time.Duration(3600) * time.Second).Result(),
@@ -220,6 +250,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

@@ -22,6 +22,8 @@ import (
 	"fmt"
 
 	"github.com/cockroachdb/errors"
+	"github.com/sirupsen/logrus"
+	appsv1api "k8s.io/api/apps/v1"
 	corev1api "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -80,6 +82,34 @@ func KbClientIsRunningInNode(ctx context.Context, namespace string, nodeName str
 	return isRunningInNode(ctx, namespace, nodeName, nil, kubeClient)
 }
 
+// IsReady checks whether the node-agent daemonset has at least one ready pod
+// by inspecting the DaemonSet status. It only checks the daemonset for node
+// OS types that are present in the cluster, following the same pattern as
+// server.checkNodeAgent.
+func IsReady(ctx context.Context, namespace string, crClient ctrlclient.Client, log logrus.FieldLogger) error {
+	if kube.WithLinuxNode(ctx, crClient, log) {
+		ds := new(appsv1api.DaemonSet)
+		if err := crClient.Get(ctx, ctrlclient.ObjectKey{Namespace: namespace, Name: daemonSet}, ds); err != nil {
+			return errors.Wrap(err, "failed to get linux node-agent daemonset")
+		}
+		if ds.Status.NumberReady > 0 {
+			return nil
+		}
+	}
+
+	if kube.WithWindowsNode(ctx, crClient, log) {
+		ds := new(appsv1api.DaemonSet)
+		if err := crClient.Get(ctx, ctrlclient.ObjectKey{Namespace: namespace, Name: daemonsetWindows}, ds); err != nil {
+			return errors.Wrap(err, "failed to get windows node-agent daemonset")
+		}
+		if ds.Status.NumberReady > 0 {
+			return nil
+		}
+	}
+
+	return errors.New("node-agent is not ready: no ready 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

@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/errors"
+	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	appsv1api "k8s.io/api/apps/v1"
@@ -213,6 +214,146 @@ func TestIsRunningInNode(t *testing.T) {
 	}
 }
 
+func TestIsReady(t *testing.T) {
+	scheme := runtime.NewScheme()
+	appsv1api.AddToScheme(scheme)
+	corev1api.AddToScheme(scheme)
+
+	log := logrus.New()
+
+	linuxNode := &corev1api.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   "linux-node",
+			Labels: map[string]string{kube.NodeOSLabel: kube.NodeOSLinux},
+		},
+	}
+	windowsNode := &corev1api.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   "windows-node",
+			Labels: map[string]string{kube.NodeOSLabel: kube.NodeOSWindows},
+		},
+	}
+
+	dsLinuxNotReady := &appsv1api.DaemonSet{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "fake-ns", Name: "node-agent"},
+		Status:     appsv1api.DaemonSetStatus{NumberReady: 0},
+	}
+	dsLinuxReady := &appsv1api.DaemonSet{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "fake-ns", Name: "node-agent"},
+		Status:     appsv1api.DaemonSetStatus{NumberReady: 3},
+	}
+	dsWindowsNotReady := &appsv1api.DaemonSet{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "fake-ns", Name: "node-agent-windows"},
+		Status:     appsv1api.DaemonSetStatus{NumberReady: 0},
+	}
+	dsWindowsReady := &appsv1api.DaemonSet{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "fake-ns", Name: "node-agent-windows"},
+		Status:     appsv1api.DaemonSetStatus{NumberReady: 2},
+	}
+
+	tests := []struct {
+		name          string
+		kubeClientObj []runtime.Object
+		namespace     string
+		expectErr     string
+	}{
+		{
+			name:      "no nodes in cluster",
+			namespace: "fake-ns",
+			expectErr: "node-agent is not ready: no ready pods found",
+		},
+		{
+			name:      "linux node exists but daemonset not found",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				linuxNode,
+			},
+			expectErr: "failed to get linux node-agent daemonset",
+		},
+		{
+			name:      "linux node and daemonset exist but no ready pods",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				linuxNode,
+				dsLinuxNotReady,
+			},
+			expectErr: "node-agent is not ready: no ready pods found",
+		},
+		{
+			name:      "linux node and daemonset with ready pods",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				linuxNode,
+				dsLinuxReady,
+			},
+		},
+		{
+			name:      "windows node and daemonset with ready pods",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				windowsNode,
+				dsWindowsReady,
+			},
+		},
+		{
+			name:      "windows node and daemonset with no ready pods",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				windowsNode,
+				dsWindowsNotReady,
+			},
+			expectErr: "node-agent is not ready: no ready pods found",
+		},
+		{
+			name:      "both node types with both daemonsets ready",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				linuxNode,
+				windowsNode,
+				dsLinuxReady,
+				dsWindowsReady,
+			},
+		},
+		{
+			name:      "both node types but neither daemonset has ready pods",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				linuxNode,
+				windowsNode,
+				dsLinuxNotReady,
+				dsWindowsNotReady,
+			},
+			expectErr: "node-agent is not ready: no ready pods found",
+		},
+		{
+			name:      "linux not ready but windows ready",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				linuxNode,
+				windowsNode,
+				dsLinuxNotReady,
+				dsWindowsReady,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fakeClient := clientFake.NewClientBuilder().
+				WithScheme(scheme).
+				WithRuntimeObjects(test.kubeClientObj...).
+				Build()
+
+			err := IsReady(t.Context(), test.namespace, fakeClient, log)
+			if test.expectErr == "" {
+				assert.NoError(t, err)
+			} else {
+				assert.ErrorContains(t, err, test.expectErr)
+			}
+		})
+	}
+}
+
 func TestGetPodSpec(t *testing.T) {
 	podSpec := corev1api.PodSpec{
 		NodeName: "fake-node",