diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 7f5ebff4f..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" @@ -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. @@ -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 { @@ -626,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{}{} @@ -650,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{}{} @@ -677,29 +624,7 @@ 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 { clean, err := b.isResourceCleanupNeeded(ctx, taskID) diff --git a/compute/kubernetes/backend_test.go b/compute/kubernetes/backend_test.go index 6436c6966..a973eaec3 100644 --- a/compute/kubernetes/backend_test.go +++ b/compute/kubernetes/backend_test.go @@ -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 @@ -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) {