diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 70efef99c..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) } @@ -569,6 +570,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) { @@ -626,6 +647,19 @@ func (b *Backend) CleanOrphanedResources(ctx context.Context) { // TODO: Add Executor Jobs here beacause orphaned tasks can result in orphaned jobs + // 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 a973eaec3..083e1aa2b 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{} @@ -397,3 +416,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) + } +}