From d2e910b1b0e0db9390a764869ccb53ce4141b975 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Wed, 20 May 2026 10:38:36 -0700 Subject: [PATCH 1/6] fix: Resolve ServiceAccount deletion race condition (thanks @nss10!) Signed-off-by: Liam Beckman --- compute/kubernetes/backend.go | 32 ++----------------- .../kubernetes/resources/serviceaccount.go | 17 +++++++--- 2 files changed, 14 insertions(+), 35 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 7f5ebff4f..96855d944 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -296,29 +296,6 @@ func (b *Backend) cleanResources(ctx context.Context, taskId string) error { b.log.Error("deleting Job", "error", err) } - // Delete PVC - err = resources.DeletePVC(ctx, taskId, b.conf.Kubernetes.JobsNamespace, b.client, b.log) - if err != nil { - errs = multierror.Append(errs, err) - b.log.Error("deleting Worker PVC", "error", err) - } - - // Delete per-task ConfigMap only if ConfigMapTemplate was configured - if b.conf.Kubernetes.ConfigMapTemplate != "" { - err = resources.DeleteConfigMap(ctx, taskId, b.conf.Kubernetes.JobsNamespace, b.client, b.log) - if err != nil { - errs = multierror.Append(errs, err) - b.log.Error("deleting Worker ConfigMap", "error", err) - } - } - - // Delete RoleBinding - err = resources.DeleteRoleBinding(ctx, taskId, b.conf.Kubernetes.JobsNamespace, b.client, b.log) - if err != nil { - errs = multierror.Append(errs, err) - b.log.Error("deleting Job", "error", err) - } - // Determine the ServiceAccount for this task. // Default to the conventional task-scoped name; override if the task // specifies an externally-managed SA via the _WORKER_SA tag. @@ -337,13 +314,6 @@ func (b *Backend) cleanResources(ctx context.Context, taskId string) error { b.log.Error("deleting Worker ServiceAccount", "taskID", taskId, "error", err) } - // Delete Role - err = resources.DeleteRole(ctx, taskId, b.conf.Kubernetes.JobsNamespace, b.client, b.log) - if err != nil { - errs = multierror.Append(errs, err) - b.log.Error("deleting Worker Role", "error", err) - } - // Delete PV err = resources.DeletePV(ctx, taskId, b.conf.Kubernetes.JobsNamespace, b.client, b.log) if err != nil { @@ -701,6 +671,8 @@ func (b *Backend) CleanOrphanedResources(ctx context.Context) { } } + // TODO: Add Executor Jobs here beacause orphaned tasks can result in orphaned jobs + for taskID := range taskIDs { clean, err := b.isResourceCleanupNeeded(ctx, taskID) if err != nil { diff --git a/compute/kubernetes/resources/serviceaccount.go b/compute/kubernetes/resources/serviceaccount.go index 18065af19..ac7037e57 100644 --- a/compute/kubernetes/resources/serviceaccount.go +++ b/compute/kubernetes/resources/serviceaccount.go @@ -5,6 +5,7 @@ import ( "context" "fmt" "text/template" + "time" "github.com/ohsu-comp-bio/funnel/config" "github.com/ohsu-comp-bio/funnel/logger" @@ -73,9 +74,14 @@ func CreateServiceAccount(ctx context.Context, task *tes.Task, conf *config.Conf return nil } -// isServiceAccountAttachedToPods returns true as soon as it finds one active -// (non-terminating) pod using the given ServiceAccount. -func isServiceAccountAttachedToPods(ctx context.Context, saName, namespace string, client kubernetes.Interface) (bool, error) { +// isServiceAccountAttachedToOtherPods returns true if any active pod for a +// *different* task is still using the given ServiceAccount. Pods are skipped +// if they are terminating (DeletionTimestamp set) or if their owning Job has +// already been deleted (indicating the pod is in the process of being cleaned +// up even if DeletionTimestamp has not propagated yet). +func isServiceAccountAttachedToOtherPods(ctx context.Context, saName, namespace string, client kubernetes.Interface, taskID string) (bool, error) { + fmt.Println("DEBUG: Sleeping for 5s before ServiceAccount deletion to avoid Race Conditions...") + time.Sleep(5 * time.Second) pods, err := client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{ FieldSelector: fmt.Sprintf("spec.serviceAccountName=%s", saName), }) @@ -84,9 +90,10 @@ func isServiceAccountAttachedToPods(ctx context.Context, saName, namespace strin } for _, pod := range pods.Items { if pod.DeletionTimestamp == nil { - return true, nil // early return on first active pod + return true, nil } } + return false, nil } @@ -120,7 +127,7 @@ func DeleteServiceAccount(ctx context.Context, taskID, namespace string, client } if sharedSA { - inUse, err := isServiceAccountAttachedToPods(ctx, saName, namespace, client) + inUse, err := isServiceAccountAttachedToOtherPods(ctx, saName, namespace, client, taskID) if err != nil { return fmt.Errorf("checking pod attachment for ServiceAccount %s: %v", saName, err) } From 0a8117ad5c5584b994bfce617dc1d85412013f6d Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Wed, 20 May 2026 11:08:31 -0700 Subject: [PATCH 2/6] fix: K8s Unit Tests - K8s fake client does not support cascading delete from Owner field Signed-off-by: Liam Beckman --- compute/kubernetes/backend_test.go | 26 +++++++++++++++++++------- 1 file changed, 19 insertions(+), 7 deletions(-) diff --git a/compute/kubernetes/backend_test.go b/compute/kubernetes/backend_test.go index 6436c6966..59bf20d5d 100644 --- a/compute/kubernetes/backend_test.go +++ b/compute/kubernetes/backend_test.go @@ -86,6 +86,14 @@ func TestTaskSubmission(t *testing.T) { // Create a fake Kubernetes client fakeClient := fake.NewSimpleClientset() + // Inject a deterministic UID on every Job create so ownerRef propagation can be verified. + const testJobUID = "test-job-uid-1234" + fakeClient.PrependReactor("create", "jobs", func(action k8stesting.Action) (bool, runtime.Object, error) { + obj := action.(k8stesting.CreateAction).GetObject().(*batchv1.Job) + obj.UID = testJobUID + return false, obj, nil + }) + // Create a mock configuration conf := config.DefaultConfig() conf.Kubernetes.Namespace = "test-namespace" @@ -155,12 +163,18 @@ spec: t.Errorf("expected Job name '%s', got '%s'", task.Id, job.Name) } - // Verify that the ConfigMap was created + // Verify that the ConfigMap was created with the Job's UID in its ownerRef. configMapName := "funnel-worker-config-" + task.Id - _, err = fakeClient.CoreV1().ConfigMaps(conf.Kubernetes.JobsNamespace).Get(context.Background(), configMapName, metav1.GetOptions{}) + cm, err := fakeClient.CoreV1().ConfigMaps(conf.Kubernetes.JobsNamespace).Get(context.Background(), configMapName, metav1.GetOptions{}) if err != nil { t.Fatalf("failed to get ConfigMap: %v", err) } + if len(cm.OwnerReferences) == 0 { + t.Fatal("expected ConfigMap to have an ownerReference, but got none") + } + if got := cm.OwnerReferences[0].UID; got != testJobUID { + t.Errorf("expected ConfigMap ownerRef UID %q, got %q", testJobUID, got) + } // Clean up resources err = backend.cleanResources(context.Background(), task.Id) @@ -174,11 +188,9 @@ spec: t.Error("expected Job to be deleted, but it still exists") } - // Verify that the ConfigMap was deleted - _, err = fakeClient.CoreV1().ConfigMaps(conf.Kubernetes.JobsNamespace).Get(context.Background(), configMapName, metav1.GetOptions{}) - if err == nil { - t.Error("expected ConfigMap to be deleted, but it still exists") - } + // ConfigMap deletion is handled by Kubernetes garbage collection via ownerReferences, + // not explicitly by cleanResources. The fake clientset does not simulate cascading GC, + // so we only verify the ownerRef is set correctly (asserted above). } From 7a5616fcae4d9cf0b9dcfeed01209ff7363a0013 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Thu, 21 May 2026 15:16:16 -0700 Subject: [PATCH 3/6] chore: Revert function name in serviceaccount.go Signed-off-by: Liam Beckman --- compute/kubernetes/resources/serviceaccount.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/compute/kubernetes/resources/serviceaccount.go b/compute/kubernetes/resources/serviceaccount.go index ac7037e57..44e540061 100644 --- a/compute/kubernetes/resources/serviceaccount.go +++ b/compute/kubernetes/resources/serviceaccount.go @@ -74,12 +74,12 @@ func CreateServiceAccount(ctx context.Context, task *tes.Task, conf *config.Conf return nil } -// isServiceAccountAttachedToOtherPods returns true if any active pod for a +// isServiceAccountAttachedToPods returns true if any active pod for a // *different* task is still using the given ServiceAccount. Pods are skipped // if they are terminating (DeletionTimestamp set) or if their owning Job has // already been deleted (indicating the pod is in the process of being cleaned // up even if DeletionTimestamp has not propagated yet). -func isServiceAccountAttachedToOtherPods(ctx context.Context, saName, namespace string, client kubernetes.Interface, taskID string) (bool, error) { +func isServiceAccountAttachedToPods(ctx context.Context, saName, namespace string, client kubernetes.Interface, taskID string) (bool, error) { fmt.Println("DEBUG: Sleeping for 5s before ServiceAccount deletion to avoid Race Conditions...") time.Sleep(5 * time.Second) pods, err := client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{ @@ -127,7 +127,7 @@ func DeleteServiceAccount(ctx context.Context, taskID, namespace string, client } if sharedSA { - inUse, err := isServiceAccountAttachedToOtherPods(ctx, saName, namespace, client, taskID) + inUse, err := isServiceAccountAttachedToPods(ctx, saName, namespace, client, taskID) if err != nil { return fmt.Errorf("checking pod attachment for ServiceAccount %s: %v", saName, err) } From fdc9b71776be55fb38b876baa8e589e05d38aee4 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Thu, 21 May 2026 15:18:06 -0700 Subject: [PATCH 4/6] chore: Update comment [skip ci] Signed-off-by: Liam Beckman --- compute/kubernetes/resources/serviceaccount.go | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/compute/kubernetes/resources/serviceaccount.go b/compute/kubernetes/resources/serviceaccount.go index 44e540061..eea3dda2c 100644 --- a/compute/kubernetes/resources/serviceaccount.go +++ b/compute/kubernetes/resources/serviceaccount.go @@ -74,11 +74,8 @@ func CreateServiceAccount(ctx context.Context, task *tes.Task, conf *config.Conf return nil } -// isServiceAccountAttachedToPods returns true if any active pod for a -// *different* task is still using the given ServiceAccount. Pods are skipped -// if they are terminating (DeletionTimestamp set) or if their owning Job has -// already been deleted (indicating the pod is in the process of being cleaned -// up even if DeletionTimestamp has not propagated yet). +// isServiceAccountAttachedToPods returns true if any active pod for any active pod +// is still using the given ServiceAccount. Pods are skipped if they are terminating (DeletionTimestamp set) func isServiceAccountAttachedToPods(ctx context.Context, saName, namespace string, client kubernetes.Interface, taskID string) (bool, error) { fmt.Println("DEBUG: Sleeping for 5s before ServiceAccount deletion to avoid Race Conditions...") time.Sleep(5 * time.Second) From 7c134e70df57e879cfedcb5adf1d3604dda0eb60 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Thu, 21 May 2026 15:23:05 -0700 Subject: [PATCH 5/6] chore: Remove unused parameter Signed-off-by: Liam Beckman --- compute/kubernetes/resources/serviceaccount.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/compute/kubernetes/resources/serviceaccount.go b/compute/kubernetes/resources/serviceaccount.go index eea3dda2c..4e50a8293 100644 --- a/compute/kubernetes/resources/serviceaccount.go +++ b/compute/kubernetes/resources/serviceaccount.go @@ -76,8 +76,8 @@ func CreateServiceAccount(ctx context.Context, task *tes.Task, conf *config.Conf // isServiceAccountAttachedToPods returns true if any active pod for any active pod // is still using the given ServiceAccount. Pods are skipped if they are terminating (DeletionTimestamp set) -func isServiceAccountAttachedToPods(ctx context.Context, saName, namespace string, client kubernetes.Interface, taskID string) (bool, error) { - fmt.Println("DEBUG: Sleeping for 5s before ServiceAccount deletion to avoid Race Conditions...") +func isServiceAccountAttachedToPods(ctx context.Context, saName, namespace string, client kubernetes.Interface, log *logger.Logger) (bool, error) { + log.Debug("ServiceAccount", "Sleeping for 5s before ServiceAccount deletion to avoid Race Conditions...") time.Sleep(5 * time.Second) pods, err := client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{ FieldSelector: fmt.Sprintf("spec.serviceAccountName=%s", saName), @@ -124,7 +124,7 @@ func DeleteServiceAccount(ctx context.Context, taskID, namespace string, client } if sharedSA { - inUse, err := isServiceAccountAttachedToPods(ctx, saName, namespace, client, taskID) + inUse, err := isServiceAccountAttachedToPods(ctx, saName, namespace, client, log) if err != nil { return fmt.Errorf("checking pod attachment for ServiceAccount %s: %v", saName, err) } From 142dfdd4e14afa7bea4afcce3238e284968c4940 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Wed, 27 May 2026 12:30:43 -0700 Subject: [PATCH 6/6] fix: remove GC-owned/namespace-scoped resources from CleanOrphanedResources Signed-off-by: Liam Beckman Assisted-by: Claude:claude-sonnet-4-5 --- compute/kubernetes/backend.go | 75 ++++++------------------------ compute/kubernetes/backend_test.go | 40 +++++++--------- 2 files changed, 32 insertions(+), 83 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 96855d944..70efef99c 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -5,7 +5,6 @@ import ( "context" "encoding/json" "fmt" - "strings" "time" "dario.cat/mergo" @@ -596,23 +595,15 @@ func (b *Backend) CleanOrphanedResources(ctx context.Context) { namespace := b.conf.Kubernetes.JobsNamespace taskIDs := make(map[string]struct{}) - // Collect task IDs from each resource type - if pvcs, err := b.client.CoreV1().PersistentVolumeClaims(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel"}); err == nil { - if err != nil { - b.log.Error("backlog cleanup: listing PVCs", err) - } - for _, r := range pvcs.Items { - if id, ok := r.Labels["taskId"]; ok { - taskIDs[id] = struct{}{} - } - } - } + // Collect task IDs from resources that cleanResources manages directly. + // ConfigMaps, PVCs, Roles, and RoleBindings are now owned by the Job via ownerReferences + // and are garbage-collected by Kubernetes automatically — they are intentionally excluded here. - // PVs - if pvs, err := b.client.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{LabelSelector: fmt.Sprintf("app=funnel,namespace=%s", namespace)}); err == nil { - if err != nil { - b.log.Error("backlog cleanup: listing PVs", err) - } + // PVs (cluster-scoped; cannot be owned by a namespaced Job, so must be cleaned explicitly) + pvs, err := b.client.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{LabelSelector: fmt.Sprintf("app=funnel,namespace=%s", namespace)}) + if err != nil { + b.log.Error("backlog cleanup: listing PVs", err) + } else { for _, r := range pvs.Items { if id, ok := r.Labels["taskId"]; ok { taskIDs[id] = struct{}{} @@ -620,26 +611,12 @@ func (b *Backend) CleanOrphanedResources(ctx context.Context) { } } - // ConfigMaps - if cms, err := b.client.CoreV1().ConfigMaps(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel"}); err == nil { - if err != nil { - b.log.Error("backlog cleanup: listing ConfigMaps", err) - } - const cmPrefix = "funnel-worker-config-" - for _, r := range cms.Items { - if id, ok := r.Labels["taskId"]; ok { - taskIDs[id] = struct{}{} - } else if strings.HasPrefix(r.Name, cmPrefix) { - taskIDs[strings.TrimPrefix(r.Name, cmPrefix)] = struct{}{} - } - } - } - - // ServiceAccounts - if sas, err := b.client.CoreV1().ServiceAccounts(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel"}); err == nil { - if err != nil { - b.log.Error("backlog cleanup: listing ServiceAccounts", err) - } + // ServiceAccounts (shared SAs are not owned by a Job; task-scoped SAs may also be orphaned + // if they were created before ownerRef support was added) + sas, err := b.client.CoreV1().ServiceAccounts(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel"}) + if err != nil { + b.log.Error("backlog cleanup: listing ServiceAccounts", err) + } else { for _, r := range sas.Items { if id, ok := r.Labels["taskId"]; ok { taskIDs[id] = struct{}{} @@ -647,30 +624,6 @@ func (b *Backend) CleanOrphanedResources(ctx context.Context) { } } - // Roles - if roles, err := b.client.RbacV1().Roles(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel"}); err == nil { - if err != nil { - b.log.Error("backlog cleanup: listing Roles", err) - } - for _, r := range roles.Items { - if id, ok := r.Labels["taskId"]; ok { - taskIDs[id] = struct{}{} - } - } - } - - // RoleBindings - if rbs, err := b.client.RbacV1().RoleBindings(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel"}); err == nil { - if err != nil { - b.log.Error("backlog cleanup: listing RoleBindings", err) - } - for _, r := range rbs.Items { - if id, ok := r.Labels["taskId"]; ok { - taskIDs[id] = struct{}{} - } - } - } - // TODO: Add Executor Jobs here beacause orphaned tasks can result in orphaned jobs for taskID := range taskIDs { diff --git a/compute/kubernetes/backend_test.go b/compute/kubernetes/backend_test.go index 59bf20d5d..a973eaec3 100644 --- a/compute/kubernetes/backend_test.go +++ b/compute/kubernetes/backend_test.go @@ -86,14 +86,6 @@ func TestTaskSubmission(t *testing.T) { // Create a fake Kubernetes client fakeClient := fake.NewSimpleClientset() - // Inject a deterministic UID on every Job create so ownerRef propagation can be verified. - const testJobUID = "test-job-uid-1234" - fakeClient.PrependReactor("create", "jobs", func(action k8stesting.Action) (bool, runtime.Object, error) { - obj := action.(k8stesting.CreateAction).GetObject().(*batchv1.Job) - obj.UID = testJobUID - return false, obj, nil - }) - // Create a mock configuration conf := config.DefaultConfig() conf.Kubernetes.Namespace = "test-namespace" @@ -163,17 +155,20 @@ spec: t.Errorf("expected Job name '%s', got '%s'", task.Id, job.Name) } - // Verify that the ConfigMap was created with the Job's UID in its ownerRef. - configMapName := "funnel-worker-config-" + task.Id - cm, err := fakeClient.CoreV1().ConfigMaps(conf.Kubernetes.JobsNamespace).Get(context.Background(), configMapName, metav1.GetOptions{}) + // Seed a PV so we can verify cleanResources deletes it. + pvName := "funnel-worker-pv-" + task.Id + _, err = fakeClient.CoreV1().PersistentVolumes().Create(context.Background(), &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{ + Name: pvName, + Labels: map[string]string{ + "app": "funnel", + "taskId": task.Id, + "namespace": conf.Kubernetes.JobsNamespace, + }, + }, + }, metav1.CreateOptions{}) if err != nil { - t.Fatalf("failed to get ConfigMap: %v", err) - } - if len(cm.OwnerReferences) == 0 { - t.Fatal("expected ConfigMap to have an ownerReference, but got none") - } - if got := cm.OwnerReferences[0].UID; got != testJobUID { - t.Errorf("expected ConfigMap ownerRef UID %q, got %q", testJobUID, got) + t.Fatalf("failed to create test PV: %v", err) } // Clean up resources @@ -188,10 +183,11 @@ spec: t.Error("expected Job to be deleted, but it still exists") } - // ConfigMap deletion is handled by Kubernetes garbage collection via ownerReferences, - // not explicitly by cleanResources. The fake clientset does not simulate cascading GC, - // so we only verify the ownerRef is set correctly (asserted above). - + // Verify that the PV was deleted + _, err = fakeClient.CoreV1().PersistentVolumes().Get(context.Background(), pvName, metav1.GetOptions{}) + if err == nil { + t.Error("expected PV to be deleted, but it still exists") + } } func TestSubmit_AppliesNodeSelectorAndTolerationsToWorkerJob(t *testing.T) {