diff --git a/controllers/backupcronjob/backupcronjob_controller.go b/controllers/backupcronjob/backupcronjob_controller.go index 6c56f1fb2..6a4b5722a 100644 --- a/controllers/backupcronjob/backupcronjob_controller.go +++ b/controllers/backupcronjob/backupcronjob_controller.go @@ -31,6 +31,7 @@ import ( "github.com/devfile/devworkspace-operator/pkg/constants" "github.com/devfile/devworkspace-operator/pkg/infrastructure" "github.com/devfile/devworkspace-operator/pkg/library/storage" + provstorage "github.com/devfile/devworkspace-operator/pkg/provision/storage" "github.com/devfile/devworkspace-operator/pkg/secrets" "github.com/go-logr/logr" "github.com/robfig/cron/v3" @@ -456,6 +457,18 @@ func (r *BackupCronJobReconciler) createBackupJob( }, }, } + // Pin backup Job to the node where the PVC is currently mounted to avoid + // Multi-Attach errors with ReadWriteOnce PVCs on multi-node clusters. + targetNode, err := provstorage.FindNodeForPVC(ctx, r.Client, workspace.Namespace, pvc.Name) + if err != nil { + log.Error(err, "Failed to find node with PVC, backup Job will not have node affinity", "pvc", pvc.Name) + } else if targetNode == "" { + log.Info("No target node for backup job, NodeAffinity will not be defined", "pvc", pvc.Name) + } + if targetNode != "" { + job.Spec.Template.Spec.Affinity = provstorage.NodeAffinityForHostname(targetNode) + } + if registryAuthSecret != nil { job.Spec.Template.Spec.Volumes = append(job.Spec.Template.Spec.Volumes, corev1.Volume{ Name: constants.RegistryAuthVolumeName, diff --git a/controllers/backupcronjob/backupcronjob_controller_test.go b/controllers/backupcronjob/backupcronjob_controller_test.go index ecc95b34b..efaa20ff5 100644 --- a/controllers/backupcronjob/backupcronjob_controller_test.go +++ b/controllers/backupcronjob/backupcronjob_controller_test.go @@ -426,6 +426,116 @@ var _ = Describe("BackupCronJobReconciler", func() { Expect(*jobList.Items[0].Spec.BackoffLimit).To(Equal(int32(2))) }) + It("creates a Job with node affinity when a running pod mounts the PVC", func() { + enabled := true + schedule := "* * * * *" + dwoc := &controllerv1alpha1.DevWorkspaceOperatorConfig{ + ObjectMeta: metav1.ObjectMeta{Name: nameNamespace.Name, Namespace: nameNamespace.Namespace}, + Config: &controllerv1alpha1.OperatorConfiguration{ + Workspace: &controllerv1alpha1.WorkspaceConfig{ + BackupCronJob: &controllerv1alpha1.BackupCronJobConfig{ + Enable: &enabled, + Schedule: schedule, + Registry: &controllerv1alpha1.RegistryConfig{ + Path: "fake-registry", + AuthSecret: "backup-auth", + }, + }, + }, + }, + } + Expect(fakeClient.Create(ctx, dwoc)).To(Succeed()) + dw := createDevWorkspace("dw-affinity", "ns-affinity", false, metav1.NewTime(time.Now().Add(-10*time.Minute))) + dw.Status.Phase = dwv2.DevWorkspaceStatusStopped + dw.Status.DevWorkspaceId = "id-affinity" + Expect(fakeClient.Create(ctx, dw)).To(Succeed()) + + pvc := &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: "claim-devworkspace", Namespace: dw.Namespace}} + Expect(fakeClient.Create(ctx, pvc)).To(Succeed()) + + authSecret := createAuthSecret("backup-auth", nameNamespace.Namespace, map[string][]byte{}) + Expect(fakeClient.Create(ctx, authSecret)).To(Succeed()) + + // Create a running pod that mounts the PVC on a specific node + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "workspace-pod", + Namespace: dw.Namespace, + Labels: map[string]string{constants.DevWorkspaceIDLabel: "other-workspace-id"}, + }, + Spec: corev1.PodSpec{ + NodeName: "worker-node-1", + Volumes: []corev1.Volume{ + { + Name: "workspace-data", + VolumeSource: corev1.VolumeSource{ + PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ + ClaimName: "claim-devworkspace", + }, + }, + }, + }, + }, + Status: corev1.PodStatus{Phase: corev1.PodRunning}, + } + Expect(fakeClient.Create(ctx, pod)).To(Succeed()) + + Expect(reconciler.executeBackupSync(ctx, dwoc, log)).To(Succeed()) + + jobList := &batchv1.JobList{} + Expect(fakeClient.List(ctx, jobList, &client.ListOptions{Namespace: dw.Namespace})).To(Succeed()) + Expect(jobList.Items).To(HaveLen(1)) + job := jobList.Items[0] + Expect(job.Spec.Template.Spec.Affinity).ToNot(BeNil()) + Expect(job.Spec.Template.Spec.Affinity.NodeAffinity).ToNot(BeNil()) + nodeSelector := job.Spec.Template.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution + Expect(nodeSelector).ToNot(BeNil()) + Expect(nodeSelector.NodeSelectorTerms).To(HaveLen(1)) + Expect(nodeSelector.NodeSelectorTerms[0].MatchExpressions).To(HaveLen(1)) + expr := nodeSelector.NodeSelectorTerms[0].MatchExpressions[0] + Expect(expr.Key).To(Equal("kubernetes.io/hostname")) + Expect(expr.Operator).To(Equal(corev1.NodeSelectorOpIn)) + Expect(expr.Values).To(Equal([]string{"worker-node-1"})) + }) + + It("creates a Job without node affinity when no running pod mounts the PVC", func() { + enabled := true + schedule := "* * * * *" + dwoc := &controllerv1alpha1.DevWorkspaceOperatorConfig{ + ObjectMeta: metav1.ObjectMeta{Name: nameNamespace.Name, Namespace: nameNamespace.Namespace}, + Config: &controllerv1alpha1.OperatorConfiguration{ + Workspace: &controllerv1alpha1.WorkspaceConfig{ + BackupCronJob: &controllerv1alpha1.BackupCronJobConfig{ + Enable: &enabled, + Schedule: schedule, + Registry: &controllerv1alpha1.RegistryConfig{ + Path: "fake-registry", + AuthSecret: "backup-auth", + }, + }, + }, + }, + } + Expect(fakeClient.Create(ctx, dwoc)).To(Succeed()) + dw := createDevWorkspace("dw-no-affinity", "ns-no-affinity", false, metav1.NewTime(time.Now().Add(-10*time.Minute))) + dw.Status.Phase = dwv2.DevWorkspaceStatusStopped + dw.Status.DevWorkspaceId = "id-no-affinity" + Expect(fakeClient.Create(ctx, dw)).To(Succeed()) + + pvc := &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: "claim-devworkspace", Namespace: dw.Namespace}} + Expect(fakeClient.Create(ctx, pvc)).To(Succeed()) + + authSecret := createAuthSecret("backup-auth", nameNamespace.Namespace, map[string][]byte{}) + Expect(fakeClient.Create(ctx, authSecret)).To(Succeed()) + + Expect(reconciler.executeBackupSync(ctx, dwoc, log)).To(Succeed()) + + jobList := &batchv1.JobList{} + Expect(fakeClient.List(ctx, jobList, &client.ListOptions{Namespace: dw.Namespace})).To(Succeed()) + Expect(jobList.Items).To(HaveLen(1)) + Expect(jobList.Items[0].Spec.Template.Spec.Affinity).To(BeNil()) + }) + It("creates a Job with configured podSecurityContext", func() { enabled := true schedule := "* * * * *" diff --git a/pkg/provision/storage/cleanup.go b/pkg/provision/storage/cleanup.go index 036322290..111b48d7d 100644 --- a/pkg/provision/storage/cleanup.go +++ b/pkg/provision/storage/cleanup.go @@ -29,9 +29,7 @@ import ( k8sErrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/types" - "sigs.k8s.io/controller-runtime/pkg/client" k8sclient "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" @@ -121,7 +119,7 @@ func getSpecCommonPVCCleanupJob(workspace *common.DevWorkspaceWithConfig, cluste pvcName = workspace.Config.Workspace.PVCName } - targetNode, err := getTargetNodeName(workspace, clusterAPI) + targetNode, err := FindNodeForPVC(clusterAPI.Ctx, clusterAPI.Client, workspace.Namespace, pvcName) if err != nil { clusterAPI.Logger.Error(err, "Error getting target node for cleanup job") } else if targetNode == "" { @@ -197,21 +195,7 @@ func getSpecCommonPVCCleanupJob(workspace *common.DevWorkspaceWithConfig, cluste } if targetNode != "" { - job.Spec.Template.Spec.Affinity.NodeAffinity = &corev1.NodeAffinity{ - RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{ - NodeSelectorTerms: []corev1.NodeSelectorTerm{ - { - MatchExpressions: []corev1.NodeSelectorRequirement{ - { - Key: corev1.LabelHostname, - Operator: corev1.NodeSelectorOpIn, - Values: []string{targetNode}, - }, - }, - }, - }, - }, - } + job.Spec.Template.Spec.Affinity = NodeAffinityForHostname(targetNode) } podTolerations, nodeSelector, err := nsconfig.GetNamespacePodTolerationsAndNodeSelector(workspace.Namespace, clusterAPI) @@ -246,40 +230,3 @@ func commonPVCExists(workspace *common.DevWorkspaceWithConfig, clusterAPI sync.C } return true, nil } - -// getTargetNodeName returns the node name of the node a running devworkspace pod that already mounts the -// common PVC is running in. -// Returns an empty string if no such pod exists. -func getTargetNodeName(workspace *common.DevWorkspaceWithConfig, clusterAPI sync.ClusterAPI) (string, error) { - - labelSelector, err := labels.Parse(constants.DevWorkspaceIDLabel) - if err != nil { - return "", err - } - - listOptions := &client.ListOptions{ - Namespace: workspace.Namespace, - LabelSelector: labelSelector, - } - - found := &corev1.PodList{} - err = clusterAPI.Client.List(clusterAPI.Ctx, found, listOptions) - if err != nil { - return "", err - } - - return getNodeNameWithPVC(found, workspace.Config.Workspace.PVCName), nil -} - -func getNodeNameWithPVC(list *corev1.PodList, pvcName string) string { - for _, pod := range list.Items { - if pod.Status.Phase == corev1.PodRunning { - for _, volume := range pod.Spec.Volumes { - if volume.PersistentVolumeClaim != nil && volume.PersistentVolumeClaim.ClaimName == pvcName { - return pod.Spec.NodeName - } - } - } - } - return "" -} diff --git a/pkg/provision/storage/cleanup_test.go b/pkg/provision/storage/cleanup_test.go index 3d1d7258e..9852cde15 100644 --- a/pkg/provision/storage/cleanup_test.go +++ b/pkg/provision/storage/cleanup_test.go @@ -81,6 +81,68 @@ func TestGetSpecCommonPVCCleanupJobUsesConfigPodSecurityContext(t *testing.T) { assert.Equal(t, customPodSecurityContext, job.Spec.Template.Spec.SecurityContext) } +func TestGetSpecCommonPVCCleanupJobHasNodeAffinityWhenPodMountsPVC(t *testing.T) { + infrastructure.InitializeForTesting(infrastructure.Kubernetes) + + namespace := "test-ns" + pvcName := "claim-devworkspace" + + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "workspace-pod", + Namespace: namespace, + Labels: map[string]string{constants.DevWorkspaceIDLabel: "other-workspace-id"}, + }, + Spec: corev1.PodSpec{NodeName: "worker-node-1", Volumes: []corev1.Volume{{Name: "data", VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: pvcName}}}}}, + Status: corev1.PodStatus{Phase: corev1.PodRunning}, + } + + fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects( + &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}, + pod, + ).Build() + + workspace := &common.DevWorkspaceWithConfig{ + DevWorkspace: &dw.DevWorkspace{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-workspace", + Namespace: namespace, + Labels: map[string]string{ + constants.DevWorkspaceCreatorLabel: "test-creator", + }, + }, + Status: dw.DevWorkspaceStatus{ + DevWorkspaceId: "test-workspace-id", + }, + }, + Config: &v1alpha1.OperatorConfiguration{ + Workspace: &v1alpha1.WorkspaceConfig{ + PVCName: pvcName, + }, + }, + } + + clusterAPI := sync.ClusterAPI{ + Client: fakeClient, + Scheme: scheme, + Logger: zap.New(zap.UseDevMode(true)), + Ctx: context.Background(), + } + + job, err := getSpecCommonPVCCleanupJob(workspace, clusterAPI) + assert.NoError(t, err) + assert.NotNil(t, job.Spec.Template.Spec.Affinity) + assert.NotNil(t, job.Spec.Template.Spec.Affinity.NodeAffinity) + nodeSelector := job.Spec.Template.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution + assert.NotNil(t, nodeSelector) + assert.Len(t, nodeSelector.NodeSelectorTerms, 1) + assert.Len(t, nodeSelector.NodeSelectorTerms[0].MatchExpressions, 1) + expr := nodeSelector.NodeSelectorTerms[0].MatchExpressions[0] + assert.Equal(t, "kubernetes.io/hostname", expr.Key) + assert.Equal(t, corev1.NodeSelectorOpIn, expr.Operator) + assert.Equal(t, []string{"worker-node-1"}, expr.Values) +} + func TestGetSpecCommonPVCCleanupJobWithNilPodSecurityContext(t *testing.T) { infrastructure.InitializeForTesting(infrastructure.Kubernetes) diff --git a/pkg/provision/storage/shared.go b/pkg/provision/storage/shared.go index eff544d8d..f6000c684 100644 --- a/pkg/provision/storage/shared.go +++ b/pkg/provision/storage/shared.go @@ -16,6 +16,7 @@ package storage import ( + "context" "errors" "fmt" @@ -27,6 +28,7 @@ import ( k8sErrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" @@ -266,6 +268,60 @@ func getSharedPVCWorkspaceCount(namespace string, api sync.ClusterAPI) (total in return total, nil } +// FindNodeForPVC lists DevWorkspace pods in the given namespace and returns the +// node name where a running pod mounts the specified PVC. Returns an empty +// string (and nil error) when no such pod is found. +func FindNodeForPVC(ctx context.Context, k8sClient client.Client, namespace, pvcName string) (string, error) { + labelSelector, err := labels.Parse(constants.DevWorkspaceIDLabel) + if err != nil { + return "", err + } + podList := &corev1.PodList{} + if err := k8sClient.List(ctx, podList, &client.ListOptions{ + Namespace: namespace, + LabelSelector: labelSelector, + }); err != nil { + return "", err + } + return getNodeNameWithPVC(podList, pvcName), nil +} + +// NodeAffinityForHostname returns an Affinity that pins a pod to the given node +// using the kubernetes.io/hostname label. Used by both cleanup and backup Jobs +// to avoid Multi-Attach errors with ReadWriteOnce PVCs on multi-node clusters. +func NodeAffinityForHostname(nodeName string) *corev1.Affinity { + return &corev1.Affinity{ + NodeAffinity: &corev1.NodeAffinity{ + RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{ + NodeSelectorTerms: []corev1.NodeSelectorTerm{ + { + MatchExpressions: []corev1.NodeSelectorRequirement{ + { + Key: corev1.LabelHostname, + Operator: corev1.NodeSelectorOpIn, + Values: []string{nodeName}, + }, + }, + }, + }, + }, + }, + } +} + +func getNodeNameWithPVC(list *corev1.PodList, pvcName string) string { + for _, pod := range list.Items { + if pod.Status.Phase == corev1.PodRunning { + for _, volume := range pod.Spec.Volumes { + if volume.PersistentVolumeClaim != nil && volume.PersistentVolumeClaim.ClaimName == pvcName { + return pod.Spec.NodeName + } + } + } + } + return "" +} + func checkPVCTerminating(name, namespace string, api sync.ClusterAPI) (bool, error) { if name == "" { // Should not happen diff --git a/pkg/provision/storage/shared_test.go b/pkg/provision/storage/shared_test.go index e426bc5ab..00a9b20cc 100644 --- a/pkg/provision/storage/shared_test.go +++ b/pkg/provision/storage/shared_test.go @@ -16,11 +16,15 @@ package storage import ( + "context" "testing" + "github.com/devfile/devworkspace-operator/pkg/constants" "github.com/stretchr/testify/assert" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client/fake" ) func TestGetPVCSpecWithDefaultStorageAccessMode(t *testing.T) { @@ -59,3 +63,123 @@ func TestGetPVCSpecWithCustomStorageAccessMode(t *testing.T) { assert.Equal(t, []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOncePod}, pvc.Spec.AccessModes, "Access modes should match") assert.Equal(t, "5Gi", pvc.Spec.Resources.Requests.Storage().String(), "Storage size should match") } + +func TestNodeAffinityForHostname(t *testing.T) { + affinity := NodeAffinityForHostname("worker-node-1") + + assert.NotNil(t, affinity) + assert.NotNil(t, affinity.NodeAffinity) + nodeSelector := affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution + assert.NotNil(t, nodeSelector) + assert.Len(t, nodeSelector.NodeSelectorTerms, 1) + assert.Len(t, nodeSelector.NodeSelectorTerms[0].MatchExpressions, 1) + + expr := nodeSelector.NodeSelectorTerms[0].MatchExpressions[0] + assert.Equal(t, "kubernetes.io/hostname", expr.Key) + assert.Equal(t, corev1.NodeSelectorOpIn, expr.Operator) + assert.Equal(t, []string{"worker-node-1"}, expr.Values) +} + +func TestFindNodeForPVC(t *testing.T) { + pvcVolume := func(claimName string) corev1.Volume { + return corev1.Volume{ + Name: "workspace-data", + VolumeSource: corev1.VolumeSource{ + PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ + ClaimName: claimName, + }, + }, + } + } + makePod := func(name, nodeName, namespace string, phase corev1.PodPhase, volumes ...corev1.Volume) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + Labels: map[string]string{constants.DevWorkspaceIDLabel: "some-id"}, + }, + Spec: corev1.PodSpec{NodeName: nodeName, Volumes: volumes}, + Status: corev1.PodStatus{Phase: phase}, + } + } + + const ns = "test-ns" + + tests := []struct { + name string + pods []*corev1.Pod + pvcName string + expected string + }{ + { + name: "returns node for running pod with matching PVC", + pods: []*corev1.Pod{makePod("pod-1", "node-1", ns, corev1.PodRunning, pvcVolume("claim-devworkspace"))}, + pvcName: "claim-devworkspace", + expected: "node-1", + }, + { + name: "ignores non-running pods", + pods: []*corev1.Pod{makePod("pod-1", "node-1", ns, corev1.PodPending, pvcVolume("claim-devworkspace"))}, + pvcName: "claim-devworkspace", + expected: "", + }, + { + name: "returns empty for non-matching PVC name", + pods: []*corev1.Pod{makePod("pod-1", "node-1", ns, corev1.PodRunning, pvcVolume("other-pvc"))}, + pvcName: "claim-devworkspace", + expected: "", + }, + { + name: "returns empty when no pods exist", + pods: nil, + pvcName: "claim-devworkspace", + expected: "", + }, + { + name: "returns first matching node when multiple pods mount same PVC", + pods: []*corev1.Pod{ + makePod("pod-1", "node-1", ns, corev1.PodRunning, pvcVolume("claim-devworkspace")), + makePod("pod-2", "node-2", ns, corev1.PodRunning, pvcVolume("claim-devworkspace")), + }, + pvcName: "claim-devworkspace", + expected: "node-1", + }, + { + name: "ignores pods without PVC volumes", + pods: []*corev1.Pod{ + makePod("pod-1", "node-1", ns, corev1.PodRunning, corev1.Volume{ + Name: "config", + VolumeSource: corev1.VolumeSource{ConfigMap: &corev1.ConfigMapVolumeSource{}}, + }), + }, + pvcName: "claim-devworkspace", + expected: "", + }, + { + name: "only considers pods with DevWorkspaceIDLabel", + pods: []*corev1.Pod{ + { + ObjectMeta: metav1.ObjectMeta{Name: "unlabeled-pod", Namespace: ns}, + Spec: corev1.PodSpec{NodeName: "node-1", Volumes: []corev1.Volume{pvcVolume("claim-devworkspace")}}, + Status: corev1.PodStatus{Phase: corev1.PodRunning}, + }, + }, + pvcName: "claim-devworkspace", + expected: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + builder := fake.NewClientBuilder().WithScheme(scheme) + for _, p := range tt.pods { + builder = builder.WithObjects(p) + } + fakeClient := builder.Build() + + result, err := FindNodeForPVC(context.Background(), fakeClient, ns, tt.pvcName) + assert.NoError(t, err) + assert.Equal(t, tt.expected, result) + }) + } +}