diff --git a/controllers/backupcronjob/backupcronjob_controller.go b/controllers/backupcronjob/backupcronjob_controller.go index 7a64ff346..8bfb4908d 100644 --- a/controllers/backupcronjob/backupcronjob_controller.go +++ b/controllers/backupcronjob/backupcronjob_controller.go @@ -394,6 +394,11 @@ func (r *BackupCronJobReconciler) createBackupJob( BackoffLimit: backUpConfig.BackoffLimit, Template: corev1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{ + constants.DevWorkspaceIDLabel: dwID, + constants.DevWorkspaceNameLabel: workspace.Name, + constants.DevWorkspaceBackupJobLabel: "true", + }, Annotations: map[string]string{ "io.kubernetes.cri-o.Devices": "/dev/fuse", }, @@ -459,7 +464,9 @@ 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) + // The backup job name is generated at creation time, so it cannot be excluded here; + // all running pods mounting the PVC are considered when computing the affinity. + 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 == "" { diff --git a/controllers/backupcronjob/backupcronjob_controller_test.go b/controllers/backupcronjob/backupcronjob_controller_test.go index 2807aca0f..e55464b99 100644 --- a/controllers/backupcronjob/backupcronjob_controller_test.go +++ b/controllers/backupcronjob/backupcronjob_controller_test.go @@ -365,6 +365,10 @@ var _ = Describe("BackupCronJobReconciler", func() { Expect(jobList.Items).To(HaveLen(1)) job := jobList.Items[0] Expect(job.Labels[constants.DevWorkspaceIDLabel]).To(Equal("id-recent")) + Expect(job.Spec.Template.Labels).To(HaveKeyWithValue(constants.DevWorkspaceIDLabel, "id-recent"), + "backup job pod template must carry the workspace ID label so deleteCommonPVC can see pods still using the PVC") + Expect(job.Spec.Template.Labels).To(HaveKeyWithValue(constants.DevWorkspaceNameLabel, "dw-recent")) + Expect(job.Spec.Template.Labels).To(HaveKeyWithValue(constants.DevWorkspaceBackupJobLabel, "true")) Expect(job.Spec.Template.Spec.ServiceAccountName).To(Equal("devworkspace-job-runner-id-recent")) container := job.Spec.Template.Spec.Containers[0] expectedEnvs := []corev1.EnvVar{ @@ -456,12 +460,23 @@ var _ = Describe("BackupCronJobReconciler", func() { 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 + // The ReplicaSet owner reference is required + // because getNodeNameWithPVC only considers deployment-owned pods -- job + // pods (e.g. cleanup jobs) and orphaned pods that mount the PVC are ignored + // so they cannot skew the node affinity. Without this owner + // reference the pod would be skipped and no node affinity would be set. pod := &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: "workspace-pod", Namespace: dw.Namespace, Labels: map[string]string{constants.DevWorkspaceIDLabel: "other-workspace-id"}, + OwnerReferences: []metav1.OwnerReference{{ + APIVersion: "apps/v1", + Kind: "ReplicaSet", + Name: "other-workspace", + UID: "other-workspace-rs-uid", + Controller: func() *bool { b := true; return &b }(), + }}, }, Spec: corev1.PodSpec{ NodeName: "worker-node-1", diff --git a/pkg/provision/storage/cleanup.go b/pkg/provision/storage/cleanup.go index aaed82243..81bee8986 100644 --- a/pkg/provision/storage/cleanup.go +++ b/pkg/provision/storage/cleanup.go @@ -18,6 +18,7 @@ package storage import ( "fmt" "path" + "slices" "time" "github.com/devfile/devworkspace-operator/pkg/dwerrors" @@ -29,6 +30,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" k8sclient "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" @@ -119,7 +121,10 @@ func getSpecCommonPVCCleanupJob(workspace *common.DevWorkspaceWithConfig, cluste pvcName = workspace.Config.Workspace.PVCName } - targetNode, err := FindNodeForPVC(clusterAPI.Ctx, clusterAPI.Client, workspace.Namespace, pvcName) + // Exclude the cleanup job's own pods so that recomputing the job spec does not + // pin it to the node of its own (possibly leftover) pods; other Job pods using + // the PVC (e.g. backup or concurrent cleanup jobs) are still followed. + targetNode, err := FindNodeForPVC(clusterAPI.Ctx, clusterAPI.Client, workspace.Namespace, pvcName, common.PVCCleanupJobName(workspaceId)) if err != nil { clusterAPI.Logger.Error(err, "Error getting target node for cleanup job") } else if targetNode == "" { @@ -230,3 +235,139 @@ func commonPVCExists(workspace *common.DevWorkspaceWithConfig, clusterAPI sync.C } return true, nil } + +// deleteCommonPVC deletes the shared PVC used for workspace storage in the workspace's +// namespace, coordinating with cleanup jobs and leftover pods that still reference it. +// Deleting the PVC while such references exist leaves it stuck in a Terminating state +// (the pvc-protection finalizer holds it as long as any pod uses it), which blocks new +// workspaces in the namespace from starting. +func deleteCommonPVC(workspace *common.DevWorkspaceWithConfig, clusterAPI sync.ClusterAPI) error { + namespace := workspace.Namespace + + usingAlternatePVC, pvcName, err := checkForAlternatePVC(namespace, clusterAPI) + if err != nil { + return err + } + if pvcName == "" { + pvcName = workspace.Config.Workspace.PVCName + } + + labelSelector, err := labels.Parse(constants.DevWorkspaceIDLabel) + if err != nil { + return err + } + + // Resolve leftover pods that reference the PVC. + podList := &corev1.PodList{} + if err := clusterAPI.Client.List(clusterAPI.Ctx, podList, &k8sclient.ListOptions{ + Namespace: namespace, + LabelSelector: labelSelector, + }); err != nil { + return err + } + for _, pod := range podList.Items { + if !isPVCMounted(pod.Spec, pvcName) { + continue + } + // Job-owned pods are not removed when the job finishes, so a finished + // cleanup job can leave its pod holding the PVC in a Terminating state; + // delete the pod so the pvc-protection finalizer can release the PVC. + jobName := getOwningJobName(&pod) + if jobName != "" { + job := &batchv1.Job{} + if err := clusterAPI.Client.Get(clusterAPI.Ctx, types.NamespacedName{Name: jobName, Namespace: pod.Namespace}, job); err != nil { + if !k8sErrors.IsNotFound(err) { + return err + } + // The owning job is gone, so the pod will be garbage-collected via its owner reference. + return &dwerrors.RetryError{ + Message: fmt.Sprintf( + "Waiting for pod %s/%s of deleted job %s to be garbage-collected; it still uses PVC %s", + pod.Namespace, pod.Name, jobName, pvcName), + RequeueAfter: 3 * time.Second, + } + } else if isJobFinished(job) { + if err := clusterAPI.Client.Delete(clusterAPI.Ctx, &pod); err != nil && !k8sErrors.IsNotFound(err) { + return err + } + return &dwerrors.RetryError{ + Message: fmt.Sprintf("Deleted pod %s of finished job %s using PVC %s", pod.Name, jobName, pvcName), + RequeueAfter: 3 * time.Second, + } + } + } + if len(pod.OwnerReferences) == 0 { + if err := clusterAPI.Client.Delete(clusterAPI.Ctx, &pod); err != nil && !k8sErrors.IsNotFound(err) { + return err + } + return &dwerrors.RetryError{ + Message: fmt.Sprintf("Deleted orphaned pod %s using PVC %s", pod.Name, pvcName), + RequeueAfter: 3 * time.Second, + } + } + + jobInfo := "" + if jobName != "" { + jobInfo = fmt.Sprintf(" of job %s", jobName) + } + return &dwerrors.RetryError{ + Message: fmt.Sprintf("Waiting for pod %s%s using PVC %s to be deleted", pod.Name, jobInfo, pvcName), + RequeueAfter: 10 * time.Second, + } + } + + if !usingAlternatePVC { + pvc := &corev1.PersistentVolumeClaim{} + namespacedName := types.NamespacedName{Name: pvcName, Namespace: namespace} + err = clusterAPI.Client.Get(clusterAPI.Ctx, namespacedName, pvc) + if err != nil { + if k8sErrors.IsNotFound(err) { + return nil + } + return err + } + + err = clusterAPI.Client.Delete(clusterAPI.Ctx, pvc) + if err != nil && !k8sErrors.IsNotFound(err) { + return err + } + } else { + clusterAPI.Logger.Info( + fmt.Sprintf("Skipping deletion of PVC %s as it is not provisioned by the DevWorkspace Operator", pvcName), + "namespace", namespace, + ) + } + + return nil +} + +func isPVCMounted(spec corev1.PodSpec, pvcName string) bool { + return slices.ContainsFunc( + spec.Volumes, + func(volume corev1.Volume) bool { + return volume.PersistentVolumeClaim != nil && volume.PersistentVolumeClaim.ClaimName == pvcName + }, + ) +} + +// getOwningJobName returns the name of the Job that controls the pod, or an +// empty string if the pod is not controlled by a Job. +func getOwningJobName(pod *corev1.Pod) string { + if ref := metav1.GetControllerOf(pod); ref != nil && ref.Kind == "Job" { + return ref.Name + } + return "" +} + +func isJobFinished(job *batchv1.Job) bool { + for _, condition := range job.Status.Conditions { + if condition.Status != corev1.ConditionTrue { + continue + } + switch condition.Type { + case batchv1.JobComplete, batchv1.JobFailed: + return true + } + } + return false +} diff --git a/pkg/provision/storage/cleanup_test.go b/pkg/provision/storage/cleanup_test.go index 9852cde15..ea92dc8e3 100644 --- a/pkg/provision/storage/cleanup_test.go +++ b/pkg/provision/storage/cleanup_test.go @@ -21,14 +21,20 @@ import ( dw "github.com/devfile/api/v2/pkg/apis/workspaces/v1alpha2" "github.com/stretchr/testify/assert" + batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" + 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/types" + "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" "sigs.k8s.io/controller-runtime/pkg/log/zap" "github.com/devfile/devworkspace-operator/apis/controller/v1alpha1" "github.com/devfile/devworkspace-operator/pkg/common" "github.com/devfile/devworkspace-operator/pkg/constants" + "github.com/devfile/devworkspace-operator/pkg/dwerrors" "github.com/devfile/devworkspace-operator/pkg/infrastructure" "github.com/devfile/devworkspace-operator/pkg/provision/sync" ) @@ -92,6 +98,13 @@ func TestGetSpecCommonPVCCleanupJobHasNodeAffinityWhenPodMountsPVC(t *testing.T) Name: "workspace-pod", Namespace: namespace, Labels: map[string]string{constants.DevWorkspaceIDLabel: "other-workspace-id"}, + OwnerReferences: []metav1.OwnerReference{{ + APIVersion: "apps/v1", + Kind: "ReplicaSet", + Name: "other-workspace", + UID: "rs-uid", + Controller: func() *bool { b := true; return &b }(), + }}, }, 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}, @@ -184,3 +197,235 @@ func TestGetSpecCommonPVCCleanupJobWithNilPodSecurityContext(t *testing.T) { assert.NoError(t, err) assert.Nil(t, job.Spec.Template.Spec.SecurityContext) } + +// makeTerminatingWorkspace returns a DevWorkspace that is being deleted, for use as the +// workspace whose storage cleanup is running. +func makeTerminatingWorkspace(name, namespace string) *dw.DevWorkspace { + now := metav1.Now() + return &dw.DevWorkspace{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + UID: types.UID(name + "-uid"), + DeletionTimestamp: &now, + Finalizers: []string{constants.StorageCleanupFinalizer}, + }, + Status: dw.DevWorkspaceStatus{ + DevWorkspaceId: name + "-id", + }, + } +} + +// makeCleanupJob returns a cleanup Job fixture labelled with the workspace ID label and +// mounting the given PVC. If ownerUID is set, the job is owned by the DevWorkspace with +// that UID; if complete is set, the job has a JobComplete condition. +func makeCleanupJob(name, namespace, workspaceId, pvcName, ownerUID string, complete bool) *batchv1.Job { + job := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + Labels: map[string]string{constants.DevWorkspaceIDLabel: workspaceId}, + }, + Spec: batchv1.JobSpec{ + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Volumes: []corev1.Volume{{ + Name: pvcName, + VolumeSource: corev1.VolumeSource{ + PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: pvcName}, + }, + }}, + }, + }, + }, + } + if ownerUID != "" { + job.OwnerReferences = []metav1.OwnerReference{{ + APIVersion: "workspace.devfile.io/v1alpha2", + Kind: "DevWorkspace", + Name: "owner-workspace", + UID: types.UID(ownerUID), + Controller: func() *bool { return new(true) }(), + }} + } + if complete { + job.Status.Conditions = []batchv1.JobCondition{{ + Type: batchv1.JobComplete, + Status: corev1.ConditionTrue, + }} + } + return job +} + +// makeCleanupPod returns a pod fixture labelled with the workspace ID label and mounting +// the given PVC. If ownerJobName is set, the pod is controlled by the Job with that name. +func makeCleanupPod(name, namespace, workspaceId, pvcName, ownerJobName string) *corev1.Pod { + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + Labels: map[string]string{constants.DevWorkspaceIDLabel: workspaceId}, + }, + Spec: corev1.PodSpec{ + Volumes: []corev1.Volume{{ + Name: pvcName, + VolumeSource: corev1.VolumeSource{ + PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: pvcName}, + }, + }}, + }, + Status: corev1.PodStatus{Phase: corev1.PodSucceeded}, + } + if ownerJobName != "" { + pod.OwnerReferences = []metav1.OwnerReference{{ + APIVersion: "batch/v1", + Kind: "Job", + Name: ownerJobName, + UID: types.UID(ownerJobName + "-uid"), + Controller: func() *bool { return new(true) }(), + }} + } + return pod +} + +func TestCleanupWorkspaceStorageCoordinatesSharedPVCDeletion(t *testing.T) { + const namespace = "test-ns" + const pvcName = "claim-devworkspace" + + makeClusterAPI := func(objects ...client.Object) (sync.ClusterAPI, client.Client) { + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(objects...). + Build() + return sync.ClusterAPI{ + Client: fakeClient, + Scheme: scheme, + Logger: zap.New(zap.UseDevMode(true)), + Ctx: context.Background(), + }, fakeClient + } + + makeWorkspaceWithConfig := func(workspace *dw.DevWorkspace) *common.DevWorkspaceWithConfig { + return &common.DevWorkspaceWithConfig{ + DevWorkspace: workspace, + Config: &v1alpha1.OperatorConfiguration{ + Workspace: &v1alpha1.WorkspaceConfig{PVCName: pvcName}, + }, + } + } + + getPVC := func(cli client.Client) error { + return cli.Get(context.Background(), types.NamespacedName{Name: pvcName, Namespace: namespace}, &corev1.PersistentVolumeClaim{}) + } + + makeSharedPVC := func() *corev1.PersistentVolumeClaim { + pvc, err := getPVCSpec(pvcName, namespace, nil, resource.MustParse("10Gi"), nil) + if err != nil { + t.Fatal(err) + } + return pvc + } + + t.Run("deletes PVC when no jobs or pods reference it", func(t *testing.T) { + currentWorkspace := makeTerminatingWorkspace("ws-current", namespace) + clusterAPI, fakeClient := makeClusterAPI(makeSharedPVC(), currentWorkspace) + + err := (&CommonStorageProvisioner{}).CleanupWorkspaceStorage(makeWorkspaceWithConfig(currentWorkspace), clusterAPI) + assert.NoError(t, err) + assert.True(t, k8sErrors.IsNotFound(getPVC(fakeClient)), "PVC should be deleted") + }) + + t.Run("waits for in-flight cleanup job owned by another workspace", func(t *testing.T) { + currentWorkspace := makeTerminatingWorkspace("ws-current", namespace) + otherWorkspace := makeTerminatingWorkspace("ws-other", namespace) + inFlightJob := makeCleanupJob("cleanup-other", namespace, otherWorkspace.Status.DevWorkspaceId, pvcName, string(otherWorkspace.UID), false) + jobPod := makeCleanupPod("cleanup-other-pod", namespace, otherWorkspace.Status.DevWorkspaceId, pvcName, "cleanup-other") + clusterAPI, fakeClient := makeClusterAPI(makeSharedPVC(), currentWorkspace, otherWorkspace, inFlightJob, jobPod) + + err := (&CommonStorageProvisioner{}).CleanupWorkspaceStorage(makeWorkspaceWithConfig(currentWorkspace), clusterAPI) + + var retryErr *dwerrors.RetryError + assert.ErrorAs(t, err, &retryErr, "Should return RetryError while cleanup job is in flight") + assert.False(t, k8sErrors.IsNotFound(getPVC(fakeClient)), "PVC should still exist") + assert.NoError(t, fakeClient.Get(context.Background(), types.NamespacedName{Name: "cleanup-other", Namespace: namespace}, &batchv1.Job{}), + "In-flight job should not be deleted") + assert.NoError(t, fakeClient.Get(context.Background(), types.NamespacedName{Name: "cleanup-other-pod", Namespace: namespace}, &corev1.Pod{}), + "Pod of in-flight job should not be deleted") + }) + + t.Run("waits for running cleanup job before deleting PVC", func(t *testing.T) { + currentWorkspace := makeTerminatingWorkspace("ws-current", namespace) + goneWorkspace := makeTerminatingWorkspace("ws-gone", namespace) + runningJob := makeCleanupJob("cleanup-running", namespace, goneWorkspace.Status.DevWorkspaceId, pvcName, string(goneWorkspace.UID), false) + jobPod := makeCleanupPod("cleanup-running-pod", namespace, goneWorkspace.Status.DevWorkspaceId, pvcName, "cleanup-running") + clusterAPI, fakeClient := makeClusterAPI(makeSharedPVC(), currentWorkspace, runningJob, jobPod) + + err := (&CommonStorageProvisioner{}).CleanupWorkspaceStorage(makeWorkspaceWithConfig(currentWorkspace), clusterAPI) + + var retryErr *dwerrors.RetryError + assert.ErrorAs(t, err, &retryErr, "Should return RetryError while a cleanup job is still running") + assert.Contains(t, retryErr.Message, "cleanup-running", "Wait message should mention the owning job for debuggability") + assert.False(t, k8sErrors.IsNotFound(getPVC(fakeClient)), "PVC should still exist") + assert.NoError(t, fakeClient.Get(context.Background(), types.NamespacedName{Name: "cleanup-running", Namespace: namespace}, &batchv1.Job{}), + "Running job should not be deleted") + assert.NoError(t, fakeClient.Get(context.Background(), types.NamespacedName{Name: "cleanup-running-pod", Namespace: namespace}, &corev1.Pod{}), + "Pod of running job should not be deleted") + }) + + t.Run("deletes orphaned cleanup pods before deleting PVC", func(t *testing.T) { + currentWorkspace := makeTerminatingWorkspace("ws-current", namespace) + orphanedPod := makeCleanupPod("orphaned-cleanup-pod", namespace, "ws-gone-id", pvcName, "") + clusterAPI, fakeClient := makeClusterAPI(makeSharedPVC(), currentWorkspace, orphanedPod) + + err := (&CommonStorageProvisioner{}).CleanupWorkspaceStorage(makeWorkspaceWithConfig(currentWorkspace), clusterAPI) + + // The orphaned pod is deleted, but the PVC deletion is requeued to confirm on + // the next reconcile that no references remain. + var retryErr *dwerrors.RetryError + assert.ErrorAs(t, err, &retryErr, "Should return RetryError after deleting orphaned pod") + assert.True(t, k8sErrors.IsNotFound(fakeClient.Get(context.Background(), + types.NamespacedName{Name: "orphaned-cleanup-pod", Namespace: namespace}, &corev1.Pod{})), "Orphaned pod should be deleted") + assert.False(t, k8sErrors.IsNotFound(getPVC(fakeClient)), "PVC should still exist") + + err = (&CommonStorageProvisioner{}).CleanupWorkspaceStorage(makeWorkspaceWithConfig(currentWorkspace), clusterAPI) + assert.NoError(t, err) + assert.True(t, k8sErrors.IsNotFound(getPVC(fakeClient)), "PVC should be deleted once no pods reference it") + }) + + t.Run("deletes pods of finished cleanup jobs before deleting PVC", func(t *testing.T) { + currentWorkspace := makeTerminatingWorkspace("ws-current", namespace) + finishedJob := makeCleanupJob("cleanup-done", namespace, "ws-gone-id", pvcName, "", true) + jobPod := makeCleanupPod("cleanup-done-pod", namespace, "ws-gone-id", pvcName, "cleanup-done") + clusterAPI, fakeClient := makeClusterAPI(makeSharedPVC(), currentWorkspace, finishedJob, jobPod) + + err := (&CommonStorageProvisioner{}).CleanupWorkspaceStorage(makeWorkspaceWithConfig(currentWorkspace), clusterAPI) + + // The finished job's pod is deleted, but the PVC deletion is requeued to confirm on + // the next reconcile that no references remain. + var retryErr *dwerrors.RetryError + assert.ErrorAs(t, err, &retryErr, "Should return RetryError after deleting finished job's pod") + assert.True(t, k8sErrors.IsNotFound(fakeClient.Get(context.Background(), + types.NamespacedName{Name: "cleanup-done-pod", Namespace: namespace}, &corev1.Pod{})), "Finished job's pod should be deleted") + assert.False(t, k8sErrors.IsNotFound(getPVC(fakeClient)), "PVC should still exist") + + err = (&CommonStorageProvisioner{}).CleanupWorkspaceStorage(makeWorkspaceWithConfig(currentWorkspace), clusterAPI) + assert.NoError(t, err) + assert.True(t, k8sErrors.IsNotFound(getPVC(fakeClient)), "PVC should be deleted once no pods reference it") + }) + + t.Run("ignores jobs and pods that do not mount the shared PVC", func(t *testing.T) { + currentWorkspace := makeTerminatingWorkspace("ws-current", namespace) + otherJob := makeCleanupJob("async-job", namespace, "ws-other-id", "claim-async", "ws-other-uid", false) + otherPod := makeCleanupPod("async-pod", namespace, "ws-other-id", "claim-async", "") + clusterAPI, fakeClient := makeClusterAPI(makeSharedPVC(), currentWorkspace, otherJob, otherPod) + + err := (&CommonStorageProvisioner{}).CleanupWorkspaceStorage(makeWorkspaceWithConfig(currentWorkspace), clusterAPI) + + assert.NoError(t, err) + assert.True(t, k8sErrors.IsNotFound(getPVC(fakeClient)), "PVC should be deleted") + assert.NoError(t, fakeClient.Get(context.Background(), types.NamespacedName{Name: "async-job", Namespace: namespace}, &batchv1.Job{}), + "Job mounting another PVC should not be deleted") + assert.NoError(t, fakeClient.Get(context.Background(), types.NamespacedName{Name: "async-pod", Namespace: namespace}, &corev1.Pod{}), + "Pod mounting another PVC should not be deleted") + }) +} diff --git a/pkg/provision/storage/commonStorage.go b/pkg/provision/storage/commonStorage.go index c95c93bf3..8223ae990 100644 --- a/pkg/provision/storage/commonStorage.go +++ b/pkg/provision/storage/commonStorage.go @@ -25,10 +25,6 @@ import ( "github.com/devfile/devworkspace-operator/pkg/library/overrides/restrictions" "github.com/devfile/devworkspace-operator/pkg/provision/sync" - corev1 "k8s.io/api/core/v1" - k8sErrors "k8s.io/apimachinery/pkg/api/errors" - "k8s.io/apimachinery/pkg/types" - "github.com/devfile/devworkspace-operator/apis/controller/v1alpha1" ) @@ -105,25 +101,8 @@ func (p *CommonStorageProvisioner) CleanupWorkspaceStorage(workspace *common.Dev // delete common PVC instead of running cleanup job if totalWorkspaces > 0 { return runCommonPVCCleanupJob(workspace, clusterAPI) - } else { - sharedPVC := &corev1.PersistentVolumeClaim{} - namespacedName := types.NamespacedName{Name: workspace.Config.Workspace.PVCName, Namespace: workspace.Namespace} - err := clusterAPI.Client.Get(clusterAPI.Ctx, namespacedName, sharedPVC) - - if err != nil { - if k8sErrors.IsNotFound(err) { - return nil - } - return err - } - - err = clusterAPI.Client.Delete(clusterAPI.Ctx, sharedPVC) - if err != nil && !k8sErrors.IsNotFound(err) { - return err - } } - - return nil + return deleteCommonPVC(workspace, clusterAPI) } // rewriteContainerVolumeMounts rewrites the VolumeMounts in a set of PodAdditions according to the 'common' PVC strategy diff --git a/pkg/provision/storage/shared.go b/pkg/provision/storage/shared.go index 90903dfa8..59b880a10 100644 --- a/pkg/provision/storage/shared.go +++ b/pkg/provision/storage/shared.go @@ -271,8 +271,9 @@ func getSharedPVCWorkspaceCount(namespace string, api sync.ClusterAPI) (total in // 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) { +// string (and nil error) when no such pod is found. Pods controlled by the Job +// named excludeJob are skipped; pass an empty string to consider all pods. +func FindNodeForPVC(ctx context.Context, k8sClient client.Client, namespace, pvcName, excludeJob string) (string, error) { labelSelector, err := labels.Parse(constants.DevWorkspaceIDLabel) if err != nil { return "", err @@ -284,7 +285,7 @@ func FindNodeForPVC(ctx context.Context, k8sClient client.Client, namespace, pvc }); err != nil { return "", err } - return getNodeNameWithPVC(podList, pvcName), nil + return getNodeNameWithPVC(podList, pvcName, excludeJob), nil } // NodeAffinityForHostname returns an Affinity that pins a pod to the given node @@ -310,13 +311,23 @@ func NodeAffinityForHostname(nodeName string) *corev1.Affinity { } } -func getNodeNameWithPVC(list *corev1.PodList, pvcName string) string { +// getNodeNameWithPVC returns the node name where the shared PVC is currently in use, +// determined from the first running pod in the list that mounts it. All running pods +// are considered (workspace pods, async-storage pods, and Job pods such as backup or +// concurrent cleanup Jobs), since any pod mounting a ReadWriteOnce PVC forces the same +// node. The only exception is pods controlled by the Job named excludeJob: a Job +// computing its own node affinity must not follow its own leftover pods. +func getNodeNameWithPVC(list *corev1.PodList, pvcName, excludeJob 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 - } + if pod.Status.Phase != corev1.PodRunning { + continue + } + if excludeJob != "" && getOwningJobName(&pod) == excludeJob { + continue + } + for _, volume := range pod.Spec.Volumes { + if volume.PersistentVolumeClaim != nil && volume.PersistentVolumeClaim.ClaimName == pvcName { + return pod.Spec.NodeName } } } diff --git a/pkg/provision/storage/shared_test.go b/pkg/provision/storage/shared_test.go index 2d2b7ef50..90058d280 100644 --- a/pkg/provision/storage/shared_test.go +++ b/pkg/provision/storage/shared_test.go @@ -97,6 +97,14 @@ func TestFindNodeForPVC(t *testing.T) { Name: name, Namespace: namespace, Labels: map[string]string{constants.DevWorkspaceIDLabel: "some-id"}, + // Workspace pods are created by the workspace Deployment via a ReplicaSet + OwnerReferences: []metav1.OwnerReference{{ + APIVersion: "apps/v1", + Kind: "ReplicaSet", + Name: "some-workspace", + UID: "rs-uid", + Controller: func() *bool { b := true; return &b }(), + }}, }, Spec: corev1.PodSpec{NodeName: nodeName, Volumes: volumes}, Status: corev1.PodStatus{Phase: phase}, @@ -106,10 +114,11 @@ func TestFindNodeForPVC(t *testing.T) { const ns = "test-ns" tests := []struct { - name string - pods []*corev1.Pod - pvcName string - expected string + name string + pods []*corev1.Pod + pvcName string + excludeJob string + expected string }{ { name: "returns node for running pod with matching PVC", @@ -167,6 +176,86 @@ func TestFindNodeForPVC(t *testing.T) { pvcName: "claim-devworkspace", expected: "", }, + { + name: "ignores running pods owned by the excluded Job", + pods: func() []*corev1.Pod { + // A cleanup job computing its own affinity must not follow its own + // (possibly leftover) pods. + pod := makePod("cleanup-pod", "node-1", ns, corev1.PodRunning, pvcVolume("claim-devworkspace")) + pod.OwnerReferences = []metav1.OwnerReference{{ + APIVersion: "batch/v1", + Kind: "Job", + Name: "cleanup-workspace123", + UID: "job-uid", + Controller: func() *bool { b := true; return &b }(), + }} + return []*corev1.Pod{pod} + }(), + pvcName: "claim-devworkspace", + excludeJob: "cleanup-workspace123", + expected: "", + }, + { + name: "considers running pods owned by other Jobs", + pods: func() []*corev1.Pod { + // E.g. a backup job or a concurrent cleanup job still holds the PVC; + // following its node avoids Multi-Attach errors. + pod := makePod("backup-pod", "node-2", ns, corev1.PodRunning, pvcVolume("claim-devworkspace")) + pod.OwnerReferences = []metav1.OwnerReference{{ + APIVersion: "batch/v1", + Kind: "Job", + Name: "backup-workspace123", + UID: "job-uid", + Controller: func() *bool { b := true; return &b }(), + }} + return []*corev1.Pod{pod} + }(), + pvcName: "claim-devworkspace", + excludeJob: "cleanup-workspace123", + expected: "node-2", + }, + { + name: "considers running pods owned by ReplicaSets", + pods: func() []*corev1.Pod { + pod := makePod("workspace-pod", "node-2", ns, corev1.PodRunning, pvcVolume("claim-devworkspace")) + pod.OwnerReferences = []metav1.OwnerReference{{ + APIVersion: "apps/v1", + Kind: "ReplicaSet", + Name: "workspace123", + UID: "rs-uid", + Controller: func() *bool { b := true; return &b }(), + }} + return []*corev1.Pod{pod} + }(), + pvcName: "claim-devworkspace", + expected: "node-2", + }, + { + name: "considers pods without owner references", + pods: func() []*corev1.Pod { + pod := makePod("standalone-pod", "node-1", ns, corev1.PodRunning, pvcVolume("claim-devworkspace")) + pod.OwnerReferences = nil + return []*corev1.Pod{pod} + }(), + pvcName: "claim-devworkspace", + expected: "node-1", + }, + { + name: "considers pods owned by StatefulSets", + pods: func() []*corev1.Pod { + pod := makePod("statefulset-pod", "node-1", ns, corev1.PodRunning, pvcVolume("claim-devworkspace")) + pod.OwnerReferences = []metav1.OwnerReference{{ + APIVersion: "apps/v1", + Kind: "StatefulSet", + Name: "not-a-workspace", + UID: "sts-uid", + Controller: func() *bool { b := true; return &b }(), + }} + return []*corev1.Pod{pod} + }(), + pvcName: "claim-devworkspace", + expected: "node-1", + }, } for _, tt := range tests { @@ -177,7 +266,7 @@ func TestFindNodeForPVC(t *testing.T) { } fakeClient := builder.Build() - result, err := FindNodeForPVC(context.Background(), fakeClient, ns, tt.pvcName) + result, err := FindNodeForPVC(context.Background(), fakeClient, ns, tt.pvcName, tt.excludeJob) assert.NoError(t, err) assert.Equal(t, tt.expected, result) }) diff --git a/pkg/provision/sync/diff.go b/pkg/provision/sync/diff.go index fbf9f01de..7b6f68287 100644 --- a/pkg/provision/sync/diff.go +++ b/pkg/provision/sync/diff.go @@ -250,7 +250,6 @@ func routingDiffFunc(spec, cluster crclient.Object) (delete, update bool) { func jobDiffFunc(spec, cluster crclient.Object) (delete, update bool) { specJob := spec.(*batchv1.Job) clusterJob := cluster.(*batchv1.Job) - // TODO: previously, this delete was specified with a background deletion policy, which is currently unsupported. return !equality.Semantic.DeepDerivative(specJob.Spec, clusterJob.Spec), false } diff --git a/pkg/provision/sync/sync.go b/pkg/provision/sync/sync.go index 25cfd9a36..f8e78b448 100644 --- a/pkg/provision/sync/sync.go +++ b/pkg/provision/sync/sync.go @@ -14,6 +14,7 @@ package sync import ( + "context" "fmt" "reflect" @@ -27,6 +28,7 @@ import ( networkingv1 "k8s.io/api/networking/v1" rbacv1 "k8s.io/api/rbac/v1" k8sErrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" crclient "sigs.k8s.io/controller-runtime/pkg/client" ) @@ -67,7 +69,7 @@ func SyncObjectWithCluster(specObj crclient.Object, api ClusterAPI) (crclient.Ob shouldDelete, shouldUpdate := diffFunc(specObj, clusterObj) if shouldDelete { printDiff(specObj, clusterObj, api.Logger) - err := api.Client.Delete(api.Ctx, specObj) + err := deleteObject(api.Ctx, api.Client, specObj) if err != nil { return nil, err } @@ -119,7 +121,7 @@ func SyncUnrecognizedObjectWithCluster(specObj crclient.Object, api ClusterAPI) } } if delete { - if err := api.Client.Delete(api.Ctx, clusterObj); err != nil { + if err := deleteObject(api.Ctx, api.Client, clusterObj); err != nil { if k8sErrors.IsForbidden(err) { return nil, &UnrecoverableSyncError{fmt.Errorf("failed to delete object %s: %w", clusterObj.GetName(), err)} } @@ -155,7 +157,7 @@ func updateObjectGeneric(specObj, clusterObj crclient.Object, api ClusterAPI) er updateFunc := getUpdateFunc(specObj) updatedObj, err := updateFunc(specObj, clusterObj) if err != nil { - if err := api.Client.Delete(api.Ctx, specObj); err != nil { + if err := deleteObject(api.Ctx, api.Client, specObj); err != nil { return err } api.Logger.Info("Deleted object", "kind", reflect.TypeOf(specObj).Elem().String(), "name", specObj.GetName()) @@ -186,6 +188,12 @@ func isMutableObject(obj crclient.Object) bool { } } +// deleteObject deletes an object from the cluster with background propagation. On +// Kubernetes, deleting a Job without an explicit propagation policy orphans its pods. +func deleteObject(ctx context.Context, c crclient.Client, obj crclient.Object) error { + return c.Delete(ctx, obj, crclient.PropagationPolicy(metav1.DeletePropagationBackground)) +} + func printDiff(specObj, clusterObj crclient.Object, log logr.Logger) { if config.IsSetUp() && config.ExperimentalFeaturesEnabled() { var diffOpts cmp.Options