From d2e910b1b0e0db9390a764869ccb53ce4141b975 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Wed, 20 May 2026 10:38:36 -0700 Subject: [PATCH 1/9] 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/9] 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/9] 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/9] 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/9] 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 d413aa45e17ef64af0d3a87dc04ef540097626a9 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Thu, 28 May 2026 19:55:19 -0700 Subject: [PATCH 6/9] chore: Restrict fix/service-account-deletion to SA race condition only Move backend.go and backend_test.go resource cleanup changes to review/service-account-deletion (PR #1423) per reviewer request. PR #1421 should only contain the ServiceAccount deletion race condition fix. Co-Authored-By: Claude Sonnet 4.6 Assisted-by: Claude Code:claude [Claude Code] --- compute/kubernetes/backend.go | 32 ++++++++++++++++++++++++++++-- compute/kubernetes/backend_test.go | 26 +++++++----------------- 2 files changed, 37 insertions(+), 21 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 96855d944..7f5ebff4f 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -296,6 +296,29 @@ 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. @@ -314,6 +337,13 @@ 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 { @@ -671,8 +701,6 @@ 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/backend_test.go b/compute/kubernetes/backend_test.go index 59bf20d5d..6436c6966 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,18 +155,12 @@ 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. + // Verify that the ConfigMap was created configMapName := "funnel-worker-config-" + task.Id - cm, err := fakeClient.CoreV1().ConfigMaps(conf.Kubernetes.JobsNamespace).Get(context.Background(), configMapName, metav1.GetOptions{}) + _, 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) @@ -188,9 +174,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 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") + } } From 1d6801b9fa04c50756a533aa16b841a63d46f457 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Thu, 4 Jun 2026 18:09:41 -0700 Subject: [PATCH 7/9] fix: Add support for cleaning orphaned Executor jobs Signed-off-by: Liam Beckman --- compute/kubernetes/backend.go | 33 ++++++++ compute/kubernetes/backend_test.go | 130 +++++++++++++++++++++++++++++ 2 files changed, 163 insertions(+) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 7f5ebff4f..db6fa3c04 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -600,6 +600,26 @@ func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableClea } } +// extractTaskIDFromExecutorJobName parses the taskID from an executor job name. +// Executor jobs are named "{taskID}-{index}" where index is a non-negative integer. +// Returns an empty string if the name does not match the expected pattern. +func extractTaskIDFromExecutorJobName(name string) string { + idx := strings.LastIndex(name, "-") + if idx <= 0 { + return "" + } + suffix := name[idx+1:] + for _, c := range suffix { + if c < '0' || c > '9' { + return "" + } + } + if len(suffix) == 0 { + return "" + } + return name[:idx] +} + // isResourceCleanupNeeded returns true when the task is confirmed gone (NotFound) // or in a terminal state. func (b *Backend) isResourceCleanupNeeded(ctx context.Context, taskID string) (bool, error) { @@ -701,6 +721,19 @@ func (b *Backend) CleanOrphanedResources(ctx context.Context) { } } + // Executor Jobs (label app=funnel-executor; named {taskID}-{index}). + // These are not owned by the worker Job, so they must be discovered and cleaned explicitly. + executorJobs, err := b.client.BatchV1().Jobs(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel-executor"}) + if err != nil { + b.log.Error("backlog cleanup: listing executor jobs", err) + } else { + for _, j := range executorJobs.Items { + if taskID := extractTaskIDFromExecutorJobName(j.Name); taskID != "" { + taskIDs[taskID] = struct{}{} + } + } + } + for taskID := range taskIDs { clean, err := b.isResourceCleanupNeeded(ctx, taskID) if err != nil { diff --git a/compute/kubernetes/backend_test.go b/compute/kubernetes/backend_test.go index 6436c6966..132aabb5f 100644 --- a/compute/kubernetes/backend_test.go +++ b/compute/kubernetes/backend_test.go @@ -19,6 +19,25 @@ import ( k8stesting "k8s.io/client-go/testing" ) +// mockDatabase implements tes.ReadOnlyServer for testing CleanOrphanedResources. +type mockDatabase struct { + tasks map[string]*tes.Task +} + +func (m *mockDatabase) GetTask(_ context.Context, req *tes.GetTaskRequest) (*tes.Task, error) { + t, ok := m.tasks[req.Id] + if !ok { + return nil, fmt.Errorf("not found") + } + return t, nil +} + +func (m *mockDatabase) ListTasks(_ context.Context, _ *tes.ListTasksRequest) (*tes.ListTasksResponse, error) { + return &tes.ListTasksResponse{}, nil +} + +func (m *mockDatabase) Close() {} + // noopEventWriter implements events.Writer for testing. type noopEventWriter struct{} @@ -389,3 +408,114 @@ func TestCancel_SAInUse(t *testing.T) { t.Errorf("expected SA to remain while pod is still running, got: %v", err) } } + +func TestExtractTaskIDFromExecutorJobName(t *testing.T) { + tests := []struct { + name string + want string + }{ + {"abc123-0", "abc123"}, + {"abc123-42", "abc123"}, + {"task-id-with-dashes-0", "task-id-with-dashes"}, + {"abc123", ""}, // no index suffix + {"abc123-", ""}, // empty suffix + {"abc123-foo", ""}, // non-numeric suffix + {"-0", ""}, // empty task ID portion + {"", ""}, + } + for _, tt := range tests { + got := extractTaskIDFromExecutorJobName(tt.name) + if got != tt.want { + t.Errorf("extractTaskIDFromExecutorJobName(%q) = %q, want %q", tt.name, got, tt.want) + } + } +} + +// TestCleanOrphanedResources_ExecutorJobs verifies that CleanOrphanedResources +// discovers and deletes executor jobs whose parent tasks are in a terminal state. +func TestCleanOrphanedResources_ExecutorJobs(t *testing.T) { + const ns = "test-namespace" + const taskID = "orphaned-task" + ctx := context.Background() + + // Executor job left behind after the worker job was already removed. + executorJob := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: taskID + "-0", + Namespace: ns, + Labels: map[string]string{"app": "funnel-executor"}, + }, + } + + fakeClient := fake.NewSimpleClientset(executorJob) + + db := &mockDatabase{ + tasks: map[string]*tes.Task{ + taskID: {Id: taskID, State: tes.State_COMPLETE}, + }, + } + + conf := config.DefaultConfig() + conf.Kubernetes.Namespace = ns + conf.Kubernetes.JobsNamespace = ns + + backend := &Backend{ + client: fakeClient, + log: logger.NewLogger("test", logger.DefaultConfig()), + conf: conf, + event: &noopEventWriter{}, + database: db, + } + + backend.CleanOrphanedResources(ctx) + + // Executor job must be gone. + _, err := fakeClient.BatchV1().Jobs(ns).Get(ctx, taskID+"-0", metav1.GetOptions{}) + if err == nil { + t.Error("expected executor job to be deleted, but it still exists") + } +} + +// TestCleanOrphanedResources_ExecutorJobs_ActiveTask verifies that executor jobs +// for tasks that are still active are NOT deleted. +func TestCleanOrphanedResources_ExecutorJobs_ActiveTask(t *testing.T) { + const ns = "test-namespace" + const taskID = "active-task" + ctx := context.Background() + + executorJob := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: taskID + "-0", + Namespace: ns, + Labels: map[string]string{"app": "funnel-executor"}, + }, + } + + fakeClient := fake.NewSimpleClientset(executorJob) + + db := &mockDatabase{ + tasks: map[string]*tes.Task{ + taskID: {Id: taskID, State: tes.State_RUNNING}, + }, + } + + conf := config.DefaultConfig() + conf.Kubernetes.Namespace = ns + conf.Kubernetes.JobsNamespace = ns + + backend := &Backend{ + client: fakeClient, + log: logger.NewLogger("test", logger.DefaultConfig()), + conf: conf, + event: &noopEventWriter{}, + database: db, + } + + backend.CleanOrphanedResources(ctx) + + // Executor job must still be present — task is running. + _, err := fakeClient.BatchV1().Jobs(ns).Get(ctx, taskID+"-0", metav1.GetOptions{}) + if err != nil { + t.Errorf("expected executor job to remain for running task, got: %v", err) + } +} From a9d3cda24be99fcc7940c0e4c6a1f89f3e4b241a Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Fri, 5 Jun 2026 15:26:13 -0700 Subject: [PATCH 8/9] fix: Test + lint errors --- compute/kubernetes/backend.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 569dd6efb..e42167c39 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "fmt" + "strings" "time" "dario.cat/mergo" @@ -259,7 +260,7 @@ func (b *Backend) createResources(ctx context.Context, task *tes.Task, config *c b.log.Debug("creating Worker PV", "taskID", task.Id) // Check to make sure required configs are present - if config.GenericS3 == nil || len(config.GenericS3) == 0 || + if len(config.GenericS3) == 0 || config.GenericS3[0].Bucket == "" || config.GenericS3[0].Region == "" { return fmt.Errorf("Bucket or Region not found in GenericS3 config when attempting to create resources for task: %#v", task) } From 039b0f4be1d9d44680f769b6e200e81ce7727cae Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Fri, 5 Jun 2026 15:33:22 -0700 Subject: [PATCH 9/9] fix: Linting errors Signed-off-by: Liam Beckman --- compute/kubernetes/backend_test.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/compute/kubernetes/backend_test.go b/compute/kubernetes/backend_test.go index 7f2543eaf..083e1aa2b 100644 --- a/compute/kubernetes/backend_test.go +++ b/compute/kubernetes/backend_test.go @@ -425,10 +425,10 @@ func TestExtractTaskIDFromExecutorJobName(t *testing.T) { {"abc123-0", "abc123"}, {"abc123-42", "abc123"}, {"task-id-with-dashes-0", "task-id-with-dashes"}, - {"abc123", ""}, // no index suffix - {"abc123-", ""}, // empty suffix - {"abc123-foo", ""}, // non-numeric suffix - {"-0", ""}, // empty task ID portion + {"abc123", ""}, // no index suffix + {"abc123-", ""}, // empty suffix + {"abc123-foo", ""}, // non-numeric suffix + {"-0", ""}, // empty task ID portion {"", ""}, } for _, tt := range tests {