Skip to content

fix: Update Funnel resource cleanup all K8s resources (including Worker + Executor jobs) #1423

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lbeckman314 merged 11 commits into from
Jun 5, 2026
+31 −98
Merged

fix: Update Funnel resource cleanup all K8s resources (including Worker + Executor jobs) #1423

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
d2e910b
fix: Resolve ServiceAccount deletion race condition (thanks @nss10!)
lbeckman314 May 20, 2026
0a8117a
fix: K8s Unit Tests
lbeckman314 May 20, 2026
7a5616f
chore: Revert function name in serviceaccount.go
lbeckman314 May 21, 2026
fdc9b71
chore: Update comment
lbeckman314 May 21, 2026
7c134e7
chore: Remove unused parameter
lbeckman314 May 21, 2026
142dfdd
fix: remove GC-owned/namespace-scoped resources from CleanOrphanedRes…
lbeckman314 May 27, 2026
d804efb
Merge branch 'develop' into review/service-account-deletion
lbeckman314 Jun 3, 2026
d0c6b39
Merge branch 'develop' into review/service-account-deletion
lbeckman314 Jun 3, 2026
03ef2ad
Merge branch 'develop' into review/service-account-deletion
lbeckman314 Jun 4, 2026
e42f2e8
Merge branch 'develop' into review/service-account-deletion
lbeckman314 Jun 4, 2026
9d27519
Merge branch 'develop' into review/service-account-deletion
lbeckman314 Jun 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
105 changes: 15 additions & 90 deletions compute/kubernetes/backend.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ import (
"context"
"encoding/json"
"fmt"
"strings"
"time"

"dario.cat/mergo"
Expand Down Expand Up @@ -296,29 +295,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.
Expand All @@ -337,13 +313,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 {
Expand Down Expand Up @@ -626,80 +595,36 @@ 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{}{}
}
}
}

// 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{}{}
}
}
}

// 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

@lbeckman314 lbeckman314 Jun 3, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  • Never added support here to clean orphaned Executor Jobs (orphaned worker jobs are already handled in cleanOrphanedResources)

Recommended Steps (Credit: @nss10):

    1. Need to make sure that job completed (terminal)
    1. Ensure DeleteJob is triggered (to clean up the Executor job)

Important

  • Take initial pass to add this support along with new tests
  • Then update team to review/pair-program as needed!

for taskID := range taskIDs {
clean, err := b.isResourceCleanupNeeded(ctx, taskID)
Expand Down
24 changes: 16 additions & 8 deletions compute/kubernetes/backend_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,11 +155,20 @@ spec:
t.Errorf("expected Job name '%s', got '%s'", task.Id, job.Name)
}

// Verify that the ConfigMap was created
configMapName := "funnel-worker-config-" + task.Id
_, 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)
t.Fatalf("failed to create test PV: %v", err)
}

// Clean up resources
Expand All @@ -174,12 +183,11 @@ 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{})
// Verify that the PV was deleted
_, err = fakeClient.CoreV1().PersistentVolumes().Get(context.Background(), pvName, metav1.GetOptions{})
if err == nil {
t.Error("expected ConfigMap to be deleted, but it still exists")
t.Error("expected PV to be deleted, but it still exists")
}

}

func TestSubmit_AppliesNodeSelectorAndTolerationsToWorkerJob(t *testing.T) {
Expand Down
Loading