From 21c120687420bb5298cac4668120e658d5e7a673 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Thu, 7 May 2026 14:29:48 -0500 Subject: [PATCH 01/35] Add reconcile_modular draft code --- compute/kubernetes/backend.go | 206 ++++++++++++++++++++++++++++++++++ 1 file changed, 206 insertions(+) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 43c270cd1..94788fa6c 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -384,6 +384,212 @@ func (b *Backend) isJobSchedulingTimedOut(ctx context.Context, jobName string, t return false } +/***********[WIP] Attempt to refactor reconcile loop into smaller functions for better readability and testability ***********/ +// cleanResourcesIfEnabled deletes resources for a job unless cleanup has been +// disabled (e.g. for debugging or dry-run modes). Always clears the job from +// failedJobEvents regardless, since the job has reached a terminal state. +func (b *Backend) cleanResourcesIfEnabled(ctx context.Context, jobName string, disableCleanup bool, failedJobEvents map[string]int) { + delete(failedJobEvents, jobName) + if disableCleanup { + return + } + if err := b.cleanResources(ctx, jobName); err != nil { + b.log.Error("reconcile: failed to clean resources", "taskID", jobName, "error", err) + } +} + +// listActiveK8sJobs returns a map of taskID -> Job for all funnel-worker jobs. +func (b *Backend) listActiveK8sJobs(ctx context.Context) (map[string]*v1.Job, error) { + jobs, err := b.client.BatchV1().Jobs(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ + LabelSelector: "app=funnel-worker", + }) + if err != nil { + return nil, err + } + k8sJobs := make(map[string]*v1.Job, len(jobs.Items)) + for i := range jobs.Items { + k8sJobs[jobs.Items[i].Name] = &jobs.Items[i] + } + return k8sJobs, nil +} + +// cleanBacklog clears completed and orphaned jobs left over from a previous +// server run. Called once at startup. +func (b *Backend) cleanBacklog(ctx context.Context) { + k8sJobs, err := b.listActiveK8sJobs(ctx) + if err != nil { + b.log.Error("backlog cleanup: listing jobs", err) + return + } + + // Determine stale jobs and clean up resources. Stale jobs include: + // 1. Completed jobs (Succeeded/Failed) that were not cleaned up before the server restarted. + // 2. Orphaned jobs (Active) whose task no longer exists in the Funnel DB — left over from a previous deployment or server crash. + for taskID, j := range k8sJobs { + s := j.Status + + if s.Succeeded > 0 || s.Failed > 0 { + b.log.Debug("backlog cleanup: deleting completed job", "taskID", taskID) + if err := b.cleanResources(ctx, taskID); err != nil { + b.log.Error("backlog cleanup: failed to clean resources", "taskID", taskID, "error", err) + } + continue + } + + if s.Active > 0 { + _, err := b.database.GetTask(ctx, &tes.GetTaskRequest{Id: taskID, View: tes.View_MINIMAL.String()}) + if err != nil { + b.log.Info("backlog cleanup: deleting orphaned active job with no matching task", "taskID", taskID) + if err := b.cleanResources(ctx, taskID); err != nil { + b.log.Error("backlog cleanup: failed to clean orphaned resources", "taskID", taskID, "error", err) + } + } + } + } + + // In case there are any orphaned resources that were missed by the above loop (e.g. due to a transient DB error), + // do one final sweep of all resources with no matching task. + b.CleanOrphanedResources(ctx) +} + +// reconcileOnce performs a single reconciliation pass. +func (b *Backend) reconcileOnce(ctx context.Context, disableCleanup bool, failedJobEvents map[string]int) { + if failedJobEvents == nil { + failedJobEvents = make(map[string]int) + } + + k8sJobs, err := b.listActiveK8sJobs(ctx) + if err != nil { + b.log.Error("reconcile: listing jobs", err) + return + } + + nonTerminalStates := []tes.State{tes.State_QUEUED, tes.State_INITIALIZING, tes.State_RUNNING} + for _, state := range nonTerminalStates { + if err := b.reconcileTasksForState(ctx, state, k8sJobs, disableCleanup, failedJobEvents); err != nil { + b.log.Error("reconcile: error reconciling tasks", "state", state, "error", err) + } + } + + // Any jobs still in k8sJobs were not matched to any Funnel task — orphaned. + if !disableCleanup { + for taskID := range k8sJobs { + b.log.Info("reconcile: cleaning up orphaned job with no matching Funnel task", "taskID", taskID) + if err := b.cleanResources(ctx, taskID); err != nil { + b.log.Error("reconcile: failed to clean orphaned resources", "taskID", taskID, "error", err) + } + delete(failedJobEvents, taskID) + } + } + +} + +// reconcileTasksForState pages through all Funnel tasks in a given state and +// reconciles each against its corresponding K8s Job, removing matched jobs +// from k8sJobs so that any remainder can be identified as orphaned. +func (b *Backend) reconcileTasksForState(ctx context.Context, state tes.State, k8sJobs map[string]*v1.Job, disableCleanup bool, failedJobEvents map[string]int) error { + pageToken := "" + for { + lresp, err := b.database.ListTasks(ctx, &tes.ListTasksRequest{ + State: state, + PageSize: 100, + PageToken: pageToken, + }) + if err != nil { + return fmt.Errorf("listing tasks (state=%s): %w", state, err) + } + + for _, task := range lresp.Tasks { + j, exists := k8sJobs[task.Id] + delete(k8sJobs, task.Id) // matched — remove so it isn't treated as orphaned + if exists { + b.reconcileJob(ctx, j, disableCleanup, failedJobEvents) + } + } + + pageToken = lresp.NextPageToken + if pageToken == "" { + return nil + } + time.Sleep(100 * time.Millisecond) + } +} + +// reconcileJob reconciles a single K8s Job against its Funnel task state. +func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bool, failedJobEvents map[string]int) { + jobName := j.Name + status := j.Status + const maxErrEventWrites = 2 + + switch { + case status.Active > 0: + // If any pod is stuck in active state for longer than the scheduling timeout, + // write a SystemError event and clean up resources if enabled. + if b.conf.Kubernetes.Timeout.GetDuration() == nil { + return + } + timeout := b.conf.Kubernetes.Timeout.GetDuration().AsDuration() + if !b.isJobSchedulingTimedOut(ctx, jobName, timeout) { + return + } + b.log.Debug("reconcile: worker pod scheduling timed out", "taskID", jobName) + b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) + b.event.WriteEvent(ctx, events.NewSystemLog( + jobName, 0, 0, "error", + "Kubernetes job in FAILED state", + map[string]string{"error": "worker pod scheduling timed out"}, + )) + b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup, failedJobEvents) + + case status.Succeeded > 0: + b.log.Debug("reconcile: cleaning up successful job", "taskID", jobName) + b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup, failedJobEvents) + + case status.Failed > 0: + if failedJobEvents[jobName] >= maxErrEventWrites { + return + } + b.log.Debug("reconcile: writing system error event for failed job", "taskID", jobName) + conds, err := json.Marshal(status.Conditions) + if err != nil { + b.log.Error("reconcile: marshal failed job conditions", "taskID", jobName, "error", err) + } + b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) + b.event.WriteEvent(ctx, events.NewSystemLog( + jobName, 0, 0, "error", + "Kubernetes job in FAILED state", + map[string]string{"error": string(conds)}, + )) + failedJobEvents[jobName]++ + b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup, failedJobEvents) + } +} + +// reconcile is the ticker-based loop used when ExternalReconciler is false. +func (b *Backend) reconcile_modular(ctx context.Context, rate time.Duration, disableCleanup bool) { + if !disableCleanup { + b.cleanBacklog(ctx) + } + + ticker := time.NewTicker(rate) + defer ticker.Stop() + + // failedJobEvents tracks how many times we've written a SystemError event + // for a given job, so we don't spam events on every tick. + failedJobEvents := make(map[string]int) + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + b.reconcileOnce(ctx, disableCleanup, failedJobEvents) + } + } +} + +/******************************** End of refactor attempt [WIP] ********************************/ + // Reconcile loops through tasks and checks the status from Funnel's database // against the status reported by Kubernetes. This allows the backend to report // system error's that prevented the worker process from running. From d84fa3a7c29ef7c11ed3fdb5ae715b3545efaaf1 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Fri, 8 May 2026 10:52:21 -0500 Subject: [PATCH 02/35] Improve reconciler modular code --- compute/kubernetes/backend.go | 274 +++++++++++++++------------------- 1 file changed, 120 insertions(+), 154 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 94788fa6c..7ead1a966 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -260,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) } @@ -385,21 +385,9 @@ func (b *Backend) isJobSchedulingTimedOut(ctx context.Context, jobName string, t } /***********[WIP] Attempt to refactor reconcile loop into smaller functions for better readability and testability ***********/ -// cleanResourcesIfEnabled deletes resources for a job unless cleanup has been -// disabled (e.g. for debugging or dry-run modes). Always clears the job from -// failedJobEvents regardless, since the job has reached a terminal state. -func (b *Backend) cleanResourcesIfEnabled(ctx context.Context, jobName string, disableCleanup bool, failedJobEvents map[string]int) { - delete(failedJobEvents, jobName) - if disableCleanup { - return - } - if err := b.cleanResources(ctx, jobName); err != nil { - b.log.Error("reconcile: failed to clean resources", "taskID", jobName, "error", err) - } -} -// listActiveK8sJobs returns a map of taskID -> Job for all funnel-worker jobs. -func (b *Backend) listActiveK8sJobs(ctx context.Context) (map[string]*v1.Job, error) { +// listAllWorkerJobs returns a map of taskID -> Job for all funnel-worker jobs. +func (b *Backend) listAllWorkerJobs(ctx context.Context) (map[string]*v1.Job, error) { jobs, err := b.client.BatchV1().Jobs(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ LabelSelector: "app=funnel-worker", }) @@ -413,21 +401,17 @@ func (b *Backend) listActiveK8sJobs(ctx context.Context) (map[string]*v1.Job, er return k8sJobs, nil } -// cleanBacklog clears completed and orphaned jobs left over from a previous -// server run. Called once at startup. func (b *Backend) cleanBacklog(ctx context.Context) { - k8sJobs, err := b.listActiveK8sJobs(ctx) + k8sJobs, err := b.listAllWorkerJobs(ctx) if err != nil { b.log.Error("backlog cleanup: listing jobs", err) return } - // Determine stale jobs and clean up resources. Stale jobs include: - // 1. Completed jobs (Succeeded/Failed) that were not cleaned up before the server restarted. - // 2. Orphaned jobs (Active) whose task no longer exists in the Funnel DB — left over from a previous deployment or server crash. for taskID, j := range k8sJobs { s := j.Status + // Completed jobs (Succeeded/Failed) that were not cleaned up before the server restarted. if s.Succeeded > 0 || s.Failed > 0 { b.log.Debug("backlog cleanup: deleting completed job", "taskID", taskID) if err := b.cleanResources(ctx, taskID); err != nil { @@ -436,6 +420,7 @@ func (b *Backend) cleanBacklog(ctx context.Context) { continue } + // Orphaned jobs (Active) whose task no longer exists in the Funnel DB — left over from a previous deployment or server crash. if s.Active > 0 { _, err := b.database.GetTask(ctx, &tes.GetTaskRequest{Id: taskID, View: tes.View_MINIMAL.String()}) if err != nil { @@ -452,119 +437,106 @@ func (b *Backend) cleanBacklog(ctx context.Context) { b.CleanOrphanedResources(ctx) } -// reconcileOnce performs a single reconciliation pass. -func (b *Backend) reconcileOnce(ctx context.Context, disableCleanup bool, failedJobEvents map[string]int) { - if failedJobEvents == nil { - failedJobEvents = make(map[string]int) +func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bool) { + jobName := j.Name + status := j.Status + schedulingTimeout := b.conf.Kubernetes.Timeout.GetDuration() + + writeSystemError := func(reason string) { + b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) + b.event.WriteEvent(ctx, events.NewSystemLog( + jobName, 0, 0, "error", + "Kubernetes job in FAILED state", + map[string]string{"error": reason}, + )) + } + + cleanResourcesIfEnabled := func() { + if !disableCleanup { + b.log.Debug("reconcile: cleaning up job", "taskID", jobName) + if err := b.cleanResources(ctx, jobName); err != nil { + b.log.Error("reconcile: failed to clean resources", "taskID", jobName, "error", err) + } + } + } + + switch { + case status.Active > 0: + if schedulingTimeout != nil && b.isJobSchedulingTimedOut(ctx, jobName, schedulingTimeout.AsDuration()) { + b.log.Debug("reconcile: worker pod scheduling timed out.", "taskID", jobName) + writeSystemError("worker pod scheduling timed out") + cleanResourcesIfEnabled() + } + + case status.Succeeded > 0: + b.log.Debug("reconcile: reconciled successful job", "taskID", jobName) + cleanResourcesIfEnabled() + + case status.Failed > 0: + task, err := b.database.GetTask(ctx, &tes.GetTaskRequest{Id: jobName, View: tes.View_MINIMAL.String()}) + if err != nil || task.State != tes.State_SYSTEM_ERROR { + b.log.Debug("reconcile: writing system error event for failed job", "taskID", jobName) + conds, err := json.Marshal(status.Conditions) + if err != nil { + b.log.Error("reconcile: marshaling failed job conditions", "taskID", jobName, "error", err) + } + writeSystemError(string(conds)) + } + b.log.Debug("reconcile: reconciled failed job", "taskID", jobName) + cleanResourcesIfEnabled() } +} - k8sJobs, err := b.listActiveK8sJobs(ctx) +// reconcileOnce performs a single reconciliation pass. +func (b *Backend) reconcileOnce(ctx context.Context, disableCleanup bool) { + k8sJobs, err := b.listAllWorkerJobs(ctx) if err != nil { b.log.Error("reconcile: listing jobs", err) return } + // Page through all non-terminal Funnel tasks and reconcile against K8s Jobs. + // Matched jobs are removed from k8sJobs so any remainder can be identified as orphaned. nonTerminalStates := []tes.State{tes.State_QUEUED, tes.State_INITIALIZING, tes.State_RUNNING} for _, state := range nonTerminalStates { - if err := b.reconcileTasksForState(ctx, state, k8sJobs, disableCleanup, failedJobEvents); err != nil { - b.log.Error("reconcile: error reconciling tasks", "state", state, "error", err) + pageToken := "" + for { + lresp, err := b.database.ListTasks(ctx, &tes.ListTasksRequest{ + State: state, + PageSize: 100, + PageToken: pageToken, + }) + if err != nil { + b.log.Error("reconcile: listing tasks", "state", state, "error", err) + break + } + for _, task := range lresp.Tasks { + j, exists := k8sJobs[task.Id] + delete(k8sJobs, task.Id) // matched — remove so it isn't treated as orphaned + if exists { + b.reconcileJob(ctx, j, disableCleanup) + } + } + pageToken = lresp.NextPageToken + if pageToken == "" { + break + } + time.Sleep(100 * time.Millisecond) } } // Any jobs still in k8sJobs were not matched to any Funnel task — orphaned. if !disableCleanup { for taskID := range k8sJobs { - b.log.Info("reconcile: cleaning up orphaned job with no matching Funnel task", "taskID", taskID) + b.log.Info("reconcile: cleaning up job for terminal or missing Funnel task", "taskID", taskID) if err := b.cleanResources(ctx, taskID); err != nil { b.log.Error("reconcile: failed to clean orphaned resources", "taskID", taskID, "error", err) } - delete(failedJobEvents, taskID) } } } -// reconcileTasksForState pages through all Funnel tasks in a given state and -// reconciles each against its corresponding K8s Job, removing matched jobs -// from k8sJobs so that any remainder can be identified as orphaned. -func (b *Backend) reconcileTasksForState(ctx context.Context, state tes.State, k8sJobs map[string]*v1.Job, disableCleanup bool, failedJobEvents map[string]int) error { - pageToken := "" - for { - lresp, err := b.database.ListTasks(ctx, &tes.ListTasksRequest{ - State: state, - PageSize: 100, - PageToken: pageToken, - }) - if err != nil { - return fmt.Errorf("listing tasks (state=%s): %w", state, err) - } - - for _, task := range lresp.Tasks { - j, exists := k8sJobs[task.Id] - delete(k8sJobs, task.Id) // matched — remove so it isn't treated as orphaned - if exists { - b.reconcileJob(ctx, j, disableCleanup, failedJobEvents) - } - } - - pageToken = lresp.NextPageToken - if pageToken == "" { - return nil - } - time.Sleep(100 * time.Millisecond) - } -} - -// reconcileJob reconciles a single K8s Job against its Funnel task state. -func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bool, failedJobEvents map[string]int) { - jobName := j.Name - status := j.Status - const maxErrEventWrites = 2 - - switch { - case status.Active > 0: - // If any pod is stuck in active state for longer than the scheduling timeout, - // write a SystemError event and clean up resources if enabled. - if b.conf.Kubernetes.Timeout.GetDuration() == nil { - return - } - timeout := b.conf.Kubernetes.Timeout.GetDuration().AsDuration() - if !b.isJobSchedulingTimedOut(ctx, jobName, timeout) { - return - } - b.log.Debug("reconcile: worker pod scheduling timed out", "taskID", jobName) - b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) - b.event.WriteEvent(ctx, events.NewSystemLog( - jobName, 0, 0, "error", - "Kubernetes job in FAILED state", - map[string]string{"error": "worker pod scheduling timed out"}, - )) - b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup, failedJobEvents) - - case status.Succeeded > 0: - b.log.Debug("reconcile: cleaning up successful job", "taskID", jobName) - b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup, failedJobEvents) - - case status.Failed > 0: - if failedJobEvents[jobName] >= maxErrEventWrites { - return - } - b.log.Debug("reconcile: writing system error event for failed job", "taskID", jobName) - conds, err := json.Marshal(status.Conditions) - if err != nil { - b.log.Error("reconcile: marshal failed job conditions", "taskID", jobName, "error", err) - } - b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) - b.event.WriteEvent(ctx, events.NewSystemLog( - jobName, 0, 0, "error", - "Kubernetes job in FAILED state", - map[string]string{"error": string(conds)}, - )) - failedJobEvents[jobName]++ - b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup, failedJobEvents) - } -} - // reconcile is the ticker-based loop used when ExternalReconciler is false. func (b *Backend) reconcile_modular(ctx context.Context, rate time.Duration, disableCleanup bool) { if !disableCleanup { @@ -574,16 +546,12 @@ func (b *Backend) reconcile_modular(ctx context.Context, rate time.Duration, dis ticker := time.NewTicker(rate) defer ticker.Stop() - // failedJobEvents tracks how many times we've written a SystemError event - // for a given job, so we don't spam events on every tick. - failedJobEvents := make(map[string]int) - for { select { case <-ctx.Done(): return case <-ticker.C: - b.reconcileOnce(ctx, disableCleanup, failedJobEvents) + b.reconcileOnce(ctx, disableCleanup) } } } @@ -832,23 +800,23 @@ 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) - } + pvcs, err := b.client.CoreV1().PersistentVolumeClaims(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel"}) + if err != nil { + b.log.Error("CleanOrphanedResources: listing PVCs", "error", err) + } else { for _, r := range pvcs.Items { - if id, ok := r.Labels["taskId"]; ok { + if id, ok := r.Labels["taskId"]; ok && id != "" { taskIDs[id] = struct{}{} } } } - // 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, err := b.client.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{ + LabelSelector: fmt.Sprintf("app=funnel,namespace=%s", namespace), + }) + if err != nil { + b.log.Error("CleanOrphanedResources: listing PVs", "error", err) + } else { for _, r := range pvs.Items { if id, ok := r.Labels["taskId"]; ok { taskIDs[id] = struct{}{} @@ -856,12 +824,13 @@ 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) - } + // ConfigMaps: label-first, name-based fallback for resources missing the taskId label + cms, err := b.client.CoreV1().ConfigMaps(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel"}) + if err != nil { + b.log.Error("CleanOrphanedResources: listing ConfigMaps", "error", err) + } else { const cmPrefix = "funnel-worker-config-" + //TODO: Deprecate the name-based fallback in favor of enforcing the presence of the taskId label on all resources for _, r := range cms.Items { if id, ok := r.Labels["taskId"]; ok { taskIDs[id] = struct{}{} @@ -871,11 +840,10 @@ func (b *Backend) CleanOrphanedResources(ctx context.Context) { } } - // 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) - } + sas, err := b.client.CoreV1().ServiceAccounts(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel"}) + if err != nil { + b.log.Error("CleanOrphanedResources: listing ServiceAccounts", "error", err) + } else { for _, r := range sas.Items { if id, ok := r.Labels["taskId"]; ok { taskIDs[id] = struct{}{} @@ -883,11 +851,10 @@ 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) - } + roles, err := b.client.RbacV1().Roles(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel"}) + if err != nil { + b.log.Error("CleanOrphanedResources: listing Roles", "error", err) + } else { for _, r := range roles.Items { if id, ok := r.Labels["taskId"]; ok { taskIDs[id] = struct{}{} @@ -895,11 +862,10 @@ func (b *Backend) CleanOrphanedResources(ctx context.Context) { } } - // 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) - } + rbs, err := b.client.RbacV1().RoleBindings(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel"}) + if err != nil { + b.log.Error("CleanOrphanedResources: listing RoleBindings", "error", err) + } else { for _, r := range rbs.Items { if id, ok := r.Labels["taskId"]; ok { taskIDs[id] = struct{}{} @@ -908,21 +874,21 @@ func (b *Backend) CleanOrphanedResources(ctx context.Context) { } for taskID := range taskIDs { - clean, err := b.isResourceCleanupNeeded(ctx, taskID) + needsCleanup, err := b.isResourceCleanupNeeded(ctx, taskID) if err != nil { - b.log.Error("backlog cleanup: checking task state", "taskID", taskID, "error", err) + b.log.Error("CleanOrphanedResources: checking task state", "taskID", taskID, "error", err) continue } - if !clean { + if !needsCleanup { continue } - b.log.Info("backlog cleanup: cleaning up resources for task", "taskID", taskID) + b.log.Info("CleanOrphanedResources: cleaning up resources for task", "taskID", taskID) if err := b.cleanResources(ctx, taskID); err != nil { - b.log.Error("backlog cleanup: failed to clean resources", "taskID", taskID, "error", err) + b.log.Error("CleanOrphanedResources: failed to clean resources", "taskID", taskID, "error", err) } - // Sleep briefly between deletions to avoid overwhelming the API server if there are many orphaned resources. - // Test this + see if better to sleep ~1 second for every 10 tasks... + // Brief pause between deletions to avoid hammering the K8s API server under high orphan counts. + // TODO: consider batching (e.g. 1s per 10 tasks if any performance delays are observed). time.Sleep(500 * time.Millisecond) } } From 9e4854d449396dbd0296489a8dc1ace838a823f0 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Sun, 10 May 2026 12:25:09 -0500 Subject: [PATCH 03/35] switch to reconclie_modular --- compute/kubernetes/backend.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 7ead1a966..68b0ffe27 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -538,7 +538,7 @@ func (b *Backend) reconcileOnce(ctx context.Context, disableCleanup bool) { } // reconcile is the ticker-based loop used when ExternalReconciler is false. -func (b *Backend) reconcile_modular(ctx context.Context, rate time.Duration, disableCleanup bool) { +func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableCleanup bool) { if !disableCleanup { b.cleanBacklog(ctx) } @@ -575,7 +575,7 @@ func (b *Backend) reconcile_modular(ctx context.Context, rate time.Duration, dis // one or more terminal states for the backend. // // This loop is also used to cleanup successful jobs. -func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableCleanup bool) { +func (b *Backend) reconcile_monolith(ctx context.Context, rate time.Duration, disableCleanup bool) { // Clears all resources that still exist from jobs that have run before this server started. // This handles two cases: // 1. Completed jobs (Succeeded/Failed) that were not cleaned up before the server restarted. From 661f61e459b17b4120c4dba90d8e6d6a6e7a6c73 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Sun, 10 May 2026 13:04:50 -0500 Subject: [PATCH 04/35] Add Debug logs for PVC deletiojn --- compute/kubernetes/resources/pvc.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/compute/kubernetes/resources/pvc.go b/compute/kubernetes/resources/pvc.go index b78ab813e..808ce58d4 100644 --- a/compute/kubernetes/resources/pvc.go +++ b/compute/kubernetes/resources/pvc.go @@ -78,6 +78,7 @@ func DeletePVC(ctx context.Context, taskID string, namespace string, client kube const maxRetries = 5 delay := 100 * time.Millisecond for i := range maxRetries { + log.Debug("Attempting to delete Worker PVC", "taskID", taskID, "attempt", i+1) // The PVC may not exist (no I/O task, or already deleted). pvc, err := client.CoreV1().PersistentVolumeClaims(namespace).Get(ctx, name, metav1.GetOptions{}) if err != nil { @@ -93,7 +94,7 @@ func DeletePVC(ctx context.Context, taskID string, namespace string, client kube _, err = client.CoreV1().PersistentVolumeClaims(namespace).Update(ctx, pvc, metav1.UpdateOptions{}) if err != nil { if errors.IsConflict(err) && i < maxRetries-1 { - log.Debug("conflict removing PVC finalizers, retrying", "pvc", name, "attempt", i+1) + log.Debug("conflict removing PVC finalizers, retrying", "pvc", name, "err", err, "attempt", i+1) time.Sleep(delay) delay *= 2 continue @@ -107,6 +108,8 @@ func DeletePVC(ctx context.Context, taskID string, namespace string, client kube if err != nil && !errors.IsNotFound(err) { return fmt.Errorf("deleting PVC %s: %v", name, err) } + log.Debug("deleting Worker PVC succeeded", "taskID", taskID, "attempt", i+1) + return nil } From e847336ef20693046310e21bb029bd7e35f9d7f3 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Sun, 10 May 2026 14:36:57 -0500 Subject: [PATCH 05/35] handle reconciler when job restarts happen --- compute/kubernetes/backend.go | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 68b0ffe27..2197c5b29 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -473,6 +473,22 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo cleanResourcesIfEnabled() case status.Failed > 0: + // Only act if K8s has marked the Job as permanently failed (backoffLimit exhausted). + // If Active > 0 is also set, K8s is still retrying — don't intervene. + if status.Active > 0 { + return + } + jobFailed := false + for _, cond := range status.Conditions { + if cond.Type == v1.JobFailed && cond.Status == corev1.ConditionTrue { + jobFailed = true + break + } + } + if !jobFailed { + // K8s hasn't given up yet — still within backoffLimit, retrying. + return + } task, err := b.database.GetTask(ctx, &tes.GetTaskRequest{Id: jobName, View: tes.View_MINIMAL.String()}) if err != nil || task.State != tes.State_SYSTEM_ERROR { b.log.Debug("reconcile: writing system error event for failed job", "taskID", jobName) From 6953a4f897875e55fda4715136569888768b54b6 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Sun, 10 May 2026 15:21:55 -0500 Subject: [PATCH 06/35] cause exec_error to maje a sys_error --- main.go | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/main.go b/main.go index a9484a196..a13001422 100644 --- a/main.go +++ b/main.go @@ -1,12 +1,10 @@ package main import ( - "errors" "os" "github.com/ohsu-comp-bio/funnel/cmd" "github.com/ohsu-comp-bio/funnel/logger" - "github.com/ohsu-comp-bio/funnel/worker" ) func main() { @@ -17,10 +15,10 @@ func main() { // worker catches the error and updates the task state, but the worker itself should not // fail. This avoids unwanted worker retries when the issue is a failure in the user's // script run by the executor. - var execErr *worker.K8sExecutorErr - if errors.As(err, &execErr) { - os.Exit(0) - } + // var execErr *worker.K8sExecutorErr + // if errors.As(err, &execErr) { + // os.Exit(0) + // } os.Exit(1) } From 697843aa8556e9176881c5396dd01a24a2854bc2 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Sun, 10 May 2026 19:09:36 -0500 Subject: [PATCH 07/35] Revert exec_err to sys_err, exec pod crash is no longer sys_err if job backof limit is not exhausted --- main.go | 10 ++++++---- worker/kubernetes.go | 31 +++++++++++++++++++++++++++---- 2 files changed, 33 insertions(+), 8 deletions(-) diff --git a/main.go b/main.go index a13001422..a9484a196 100644 --- a/main.go +++ b/main.go @@ -1,10 +1,12 @@ package main import ( + "errors" "os" "github.com/ohsu-comp-bio/funnel/cmd" "github.com/ohsu-comp-bio/funnel/logger" + "github.com/ohsu-comp-bio/funnel/worker" ) func main() { @@ -15,10 +17,10 @@ func main() { // worker catches the error and updates the task state, but the worker itself should not // fail. This avoids unwanted worker retries when the issue is a failure in the user's // script run by the executor. - // var execErr *worker.K8sExecutorErr - // if errors.As(err, &execErr) { - // os.Exit(0) - // } + var execErr *worker.K8sExecutorErr + if errors.As(err, &execErr) { + os.Exit(0) + } os.Exit(1) } diff --git a/worker/kubernetes.go b/worker/kubernetes.go index 0d682575e..726826a16 100644 --- a/worker/kubernetes.go +++ b/worker/kubernetes.go @@ -237,11 +237,17 @@ func (kcmd KubernetesCommand) Run(ctx context.Context) error { logger.Debug("Streaming pod logs", "podName", pod.Name) err = streamPodLogs(ctx, kcmd.JobsNamespace, pod.Name, kcmd.Stdout, kcmd.Stderr) if err != nil { - return &K8sSystemErr{ - Reason: "LogStreamingFailed", - Message: fmt.Sprintf("Failed to stream logs from pod %s", pod.Name), - Err: err, + lastAttempt, checkErr := isWorkerLastAttempt(ctx, taskId, kcmd.JobsNamespace) + if checkErr != nil || lastAttempt { + logger.Debug("Error checking if worker attempt is last attempt", "error", checkErr) + return &K8sSystemErr{ + Reason: "LogStreamingFailed", + Message: fmt.Sprintf("Failed to stream logs from pod %s", pod.Name), + Err: err, + } } + logger.Debug("Log streaming error not marked as SYSTEM_ERROR because this is not the last worker attempt", "podName", pod.Name, "error", err) + return fmt.Errorf("Transient system error, will retry. error: failed to stream logs from pod %s: %v", pod.Name, err) } if len(pod.Status.ContainerStatuses) == 0 { @@ -285,6 +291,23 @@ func (kcmd KubernetesCommand) Run(ctx context.Context) error { return nil } +func isWorkerLastAttempt(ctx context.Context, taskID, namespace string) (bool, error) { + + clientset, err := getKubernetesClientset() + job, err := clientset.BatchV1().Jobs(namespace).Get(ctx, taskID, metav1.GetOptions{}) + if err != nil { + return true, err + } + limit := int32(0) + if job.Spec.BackoffLimit != nil { + limit = *job.Spec.BackoffLimit + } + // failed count includes this attempt only after the pod exits, + // so while we're still running, status.Failed reflects *completed* failed attempts. + // If failed >= backoffLimit, the Job controller won't retry us. + return job.Status.Failed >= limit, nil +} + // streamPodLogs streams logs from a pod regardless of its state // This works for Running, Succeeded, and Failed pods (as long as they haven't been deleted) func streamPodLogs(ctx context.Context, namespace string, podName string, stdout io.Writer, stderr io.Writer) error { From 37ea9957652bae9c24112c015b3c08c9cb01de73 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Mon, 18 May 2026 20:40:25 -0500 Subject: [PATCH 08/35] Remove lastAttempt check --- worker/kubernetes.go | 14 ++++---------- 1 file changed, 4 insertions(+), 10 deletions(-) diff --git a/worker/kubernetes.go b/worker/kubernetes.go index 726826a16..848afbca9 100644 --- a/worker/kubernetes.go +++ b/worker/kubernetes.go @@ -237,17 +237,11 @@ func (kcmd KubernetesCommand) Run(ctx context.Context) error { logger.Debug("Streaming pod logs", "podName", pod.Name) err = streamPodLogs(ctx, kcmd.JobsNamespace, pod.Name, kcmd.Stdout, kcmd.Stderr) if err != nil { - lastAttempt, checkErr := isWorkerLastAttempt(ctx, taskId, kcmd.JobsNamespace) - if checkErr != nil || lastAttempt { - logger.Debug("Error checking if worker attempt is last attempt", "error", checkErr) - return &K8sSystemErr{ - Reason: "LogStreamingFailed", - Message: fmt.Sprintf("Failed to stream logs from pod %s", pod.Name), - Err: err, - } + return &K8sSystemErr{ + Reason: "LogStreamingFailed", + Message: fmt.Sprintf("Failed to stream logs from pod %s", pod.Name), + Err: err, } - logger.Debug("Log streaming error not marked as SYSTEM_ERROR because this is not the last worker attempt", "podName", pod.Name, "error", err) - return fmt.Errorf("Transient system error, will retry. error: failed to stream logs from pod %s: %v", pod.Name, err) } if len(pod.Status.ContainerStatuses) == 0 { From 7e435763eec749fff98d607bed86d2bd33f237e3 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Tue, 19 May 2026 17:12:25 -0500 Subject: [PATCH 09/35] Remove TTLSecondsAfterFinished from worker pod. --- compute/kubernetes/resources/job.go | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/compute/kubernetes/resources/job.go b/compute/kubernetes/resources/job.go index bd40069b8..c5cd20fa3 100644 --- a/compute/kubernetes/resources/job.go +++ b/compute/kubernetes/resources/job.go @@ -111,13 +111,18 @@ func CreateJob(ctx context.Context, task *tes.Task, conf *config.Config, client return nil, fmt.Errorf("failed to decode job spec") } + //FIXME: This should be taken care by the reconciler when DisableJobCleanup is false. + // Setting TTLSecondsAfterFinished here causes the job to be deleted + // regardless of the DisableJobCleanup setting, which can lead to unexpected job + // deletions if the reconciler is disabled. + // Ensure completed jobs are garbage-collected by the Kubernetes TTL Controller // so that Succeeded/Failed pods don't accumulate on nodes and block Karpenter // consolidation. Funnel's own reconciler only handles non-terminal tasks. - if job.Spec.TTLSecondsAfterFinished == nil { - var ttl int32 = 300 - job.Spec.TTLSecondsAfterFinished = &ttl - } + // if job.Spec.TTLSecondsAfterFinished == nil { + // var ttl int32 = 300 + // job.Spec.TTLSecondsAfterFinished = &ttl + // } log.Debug("Creating job", "Job", job.Name, "JobsNamespace", conf.Kubernetes.JobsNamespace) created, err := client.BatchV1().Jobs(conf.Kubernetes.JobsNamespace).Create(ctx, job, metav1.CreateOptions{}) From 5d296f9cab0f8e5d26be06d05fe1c831c812df9d Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Tue, 19 May 2026 17:16:37 -0500 Subject: [PATCH 10/35] Set quay to cdis --- .github/workflows/docker.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/docker.yaml b/.github/workflows/docker.yaml index e429acd4a..8ef92715f 100644 --- a/.github/workflows/docker.yaml +++ b/.github/workflows/docker.yaml @@ -12,7 +12,7 @@ concurrency: cancel-in-progress: true env: - QUAY_REPO: quay.io/ohsu-comp-bio/funnel + QUAY_REPO: quay.io/cdis/funnel jobs: build: From ffa4030e9488c20291895dc367c46f54a7781c0b Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Thu, 21 May 2026 23:42:13 -0500 Subject: [PATCH 11/35] Remove old code and WIP comments --- compute/kubernetes/backend.go | 222 +--------------------------------- 1 file changed, 1 insertion(+), 221 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 8a9045a49..cc5143fc3 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -384,8 +384,6 @@ func (b *Backend) isJobSchedulingTimedOut(ctx context.Context, jobName string, t return false } -/***********[WIP] Attempt to refactor reconcile loop into smaller functions for better readability and testability ***********/ - // listAllWorkerJobs returns a map of taskID -> Job for all funnel-worker jobs. func (b *Backend) listAllWorkerJobs(ctx context.Context) (map[string]*v1.Job, error) { jobs, err := b.client.BatchV1().Jobs(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ @@ -541,7 +539,7 @@ func (b *Backend) reconcileOnce(ctx context.Context, disableCleanup bool) { } } - // Any jobs still in k8sJobs were not matched to any Funnel task — orphaned. + // Any jobs still in k8sJobs were not matched to any Funnel task — orphaned or in terminal states. if !disableCleanup { for taskID := range k8sJobs { b.log.Info("reconcile: cleaning up job for terminal or missing Funnel task", "taskID", taskID) @@ -572,224 +570,6 @@ func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableClea } } -/******************************** End of refactor attempt [WIP] ********************************/ - -// Reconcile loops through tasks and checks the status from Funnel's database -// against the status reported by Kubernetes. This allows the backend to report -// system error's that prevented the worker process from running. -// -// Currently this handles a narrow set of cases: -// -// |---------------------|-----------------|--------------------| -// | Funnel State | Backend State | Reconciled State | -// |---------------------|-----------------|--------------------| -// | QUEUED | FAILED | SYSTEM_ERROR | -// | INITIALIZING | FAILED | SYSTEM_ERROR | -// | RUNNING | FAILED | SYSTEM_ERROR | -// -// In this context a "FAILED" state is being used as a generic term that captures -// one or more terminal states for the backend. -// -// This loop is also used to cleanup successful jobs. -func (b *Backend) reconcile_monolith(ctx context.Context, rate time.Duration, disableCleanup bool) { - // Clears all resources that still exist from jobs that have run before this server started. - // This handles two cases: - // 1. Completed jobs (Succeeded/Failed) that were not cleaned up before the server restarted. - // 2. Orphaned jobs (Active) whose task no longer exists in the Funnel DB — left over from - // a previous deployment or server crash. - if !disableCleanup { - jobs, err := b.client.BatchV1().Jobs(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ - LabelSelector: "app=funnel-worker", - }) - if err != nil { - b.log.Error("backlog cleanup: listing jobs", err) - } else { - for _, j := range jobs.Items { - s := j.Status - taskID := j.Name - - // Always clean up completed jobs from prior runs. - if s.Succeeded > 0 || s.Failed > 0 { - b.log.Debug("backlog cleanup: deleting completed job", "taskID", taskID) - if err := b.cleanResources(ctx, taskID); err != nil { - b.log.Error("backlog cleanup: failed to clean resources", "taskID", taskID, "error", err) - } - continue - } - - // For active jobs, check whether the task still exists in the DB. - // If it doesn't, the job is orphaned from a previous deployment. - if s.Active > 0 { - _, err := b.database.GetTask(ctx, &tes.GetTaskRequest{Id: taskID, View: tes.View_MINIMAL.String()}) - if err != nil { - b.log.Info("backlog cleanup: deleting orphaned active job with no matching task", "taskID", taskID) - if err := b.cleanResources(ctx, taskID); err != nil { - b.log.Error("backlog cleanup: failed to clean orphaned resources", "taskID", taskID, "error", err) - } - } - } - } - } - b.CleanOrphanedResources(ctx) - - } - - ticker := time.NewTicker(rate) - failedJobEvents := make(map[string]int) - const maxErrEventWrites = 2 - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - - // List worker jobs only (label selector excludes executor jobs and unrelated jobs). - // Bug: If K8s Job is not created by the time reconciler runs, then the TES Task itself will be prematurely marked as SYSTEM_ERROR - jobs, err := b.client.BatchV1().Jobs(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ - LabelSelector: "app=funnel-worker", - }) - if err != nil { - b.log.Error("reconcile: listing jobs", err) - continue - } - - // Index worker jobs by task ID. We use a label selector to avoid - // picking up executor jobs (app=funnel-executor) or unrelated jobs. - k8sJobs := make(map[string]*v1.Job) - for i := range jobs.Items { - k8sJobs[jobs.Items[i].Name] = &jobs.Items[i] - } - - // List non-terminal tasks from Funnel's database - states := []tes.State{tes.State_QUEUED, tes.State_INITIALIZING, tes.State_RUNNING} - for _, s := range states { - pageToken := "" - for { - lresp, err := b.database.ListTasks(ctx, &tes.ListTasksRequest{ - State: s, - PageSize: 100, - PageToken: pageToken, - }) - if err != nil { - b.log.Error("reconcile: listing non-terminal tasks from Funnel DB", err) - break - } - pageToken = lresp.NextPageToken - - // Compare Funnel Tasks against K8s Jobs - for _, task := range lresp.Tasks { - taskID := task.Id - - // If the job exists, check its current status (Active, Succeeded, Failed) - j := k8sJobs[taskID] - - // Remove matched jobs so that any remaining entries after this - // loop represent orphaned K8s jobs with no Funnel task. - delete(k8sJobs, taskID) - - if j == nil { - continue - } - - jobName := j.Name - status := j.Status - switch { - case status.Active > 0: - // If a scheduling timeout is configured, check whether the worker - // pod has been stuck in Pending beyond that duration. This catches - // scheduling failures (bad NodeSelector, insufficient resources, etc.) - // that the context timeout cannot detect because the Job API call - // itself succeeds immediately. - if b.conf.Kubernetes.Timeout.GetDuration() != nil { - timeout := b.conf.Kubernetes.Timeout.GetDuration().AsDuration() - if b.isJobSchedulingTimedOut(ctx, jobName, timeout) { - b.log.Debug("reconcile: worker pod scheduling timed out", "taskID", jobName) - b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) - b.event.WriteEvent(ctx, events.NewSystemLog( - jobName, 0, 0, "error", - "Kubernetes job in FAILED state", - map[string]string{"error": "worker pod scheduling timed out"}, - )) - if !disableCleanup { - if err := b.cleanResources(ctx, jobName); err != nil { - b.log.Error("failed to clean resources", "taskID", jobName, "error", err) - } - } - } - } - continue - case status.Succeeded > 0: - if disableCleanup { - continue - } - b.log.Debug("reconcile: cleaning up successful job", "taskID", jobName) - - // Delete resources - if err := b.cleanResources(ctx, jobName); err != nil { - b.log.Error("failed to clean resources", "taskID", jobName, "error", err) - continue - } - delete(failedJobEvents, jobName) - - case status.Failed > 0: - if count, exists := failedJobEvents[jobName]; exists && count >= maxErrEventWrites { - continue - } - - b.log.Debug("reconcile: writing system error event for failed job", "taskID", jobName) - conds, err := json.Marshal(status.Conditions) - if err != nil { - b.log.Error("reconcile: marshal failed job conditions", "taskID", jobName, "error", err) - } - - b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) - b.event.WriteEvent( - ctx, - events.NewSystemLog( - jobName, 0, 0, "error", - "Kubernetes job in FAILED state", - map[string]string{"error": string(conds)}, - ), - ) - - failedJobEvents[jobName]++ - if disableCleanup { - continue - } - - b.log.Debug("reconcile: cleaning up failed job", "taskID", jobName) - if err := b.cleanResources(ctx, jobName); err != nil { - b.log.Error("failed to clean resources", "taskID", jobName, "error", err) - continue - } - delete(failedJobEvents, jobName) - } - } - - // Continue to next page from ListTasks if a token exists - if pageToken == "" { - break - } - time.Sleep(time.Millisecond * 100) - } - } - - // Any jobs remaining in k8sJobs were not matched to a Funnel task — - // they are orphaned and should be cleaned up. - if !disableCleanup { - for taskID := range k8sJobs { - b.log.Info("reconcile: cleaning up orphaned job with no matching Funnel task", "taskID", taskID) - if err := b.cleanResources(ctx, taskID); err != nil { - b.log.Error("reconcile: failed to clean orphaned resources", "taskID", taskID, "error", err) - } - delete(failedJobEvents, taskID) - } - } - } - } -} - // 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) { From adb1fe87ad3800a530b519272a31fa9cf847e3d4 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Thu, 21 May 2026 23:42:54 -0500 Subject: [PATCH 12/35] TTLSecondsAfterFinished to worker job removed. Jobcleanup takes care of this. --- compute/kubernetes/resources/job.go | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/compute/kubernetes/resources/job.go b/compute/kubernetes/resources/job.go index c5cd20fa3..79f8657f3 100644 --- a/compute/kubernetes/resources/job.go +++ b/compute/kubernetes/resources/job.go @@ -111,19 +111,6 @@ func CreateJob(ctx context.Context, task *tes.Task, conf *config.Config, client return nil, fmt.Errorf("failed to decode job spec") } - //FIXME: This should be taken care by the reconciler when DisableJobCleanup is false. - // Setting TTLSecondsAfterFinished here causes the job to be deleted - // regardless of the DisableJobCleanup setting, which can lead to unexpected job - // deletions if the reconciler is disabled. - - // Ensure completed jobs are garbage-collected by the Kubernetes TTL Controller - // so that Succeeded/Failed pods don't accumulate on nodes and block Karpenter - // consolidation. Funnel's own reconciler only handles non-terminal tasks. - // if job.Spec.TTLSecondsAfterFinished == nil { - // var ttl int32 = 300 - // job.Spec.TTLSecondsAfterFinished = &ttl - // } - log.Debug("Creating job", "Job", job.Name, "JobsNamespace", conf.Kubernetes.JobsNamespace) created, err := client.BatchV1().Jobs(conf.Kubernetes.JobsNamespace).Create(ctx, job, metav1.CreateOptions{}) if err != nil { From 777acfd83118859b383f8096bb07acada207b7e2 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Fri, 22 May 2026 01:37:52 -0500 Subject: [PATCH 13/35] Print SQL logs to identify bug --- compute/kubernetes/backend.go | 1 + database/postgres/tes.go | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index cc5143fc3..307f0c904 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -525,6 +525,7 @@ func (b *Backend) reconcileOnce(ctx context.Context, disableCleanup bool) { break } for _, task := range lresp.Tasks { + b.log.Debug("reconcile: processing task", "taskID", task.Id, "state", task.State) j, exists := k8sJobs[task.Id] delete(k8sJobs, task.Id) // matched — remove so it isn't treated as orphaned if exists { diff --git a/database/postgres/tes.go b/database/postgres/tes.go index 1c17a18bf..b47121451 100644 --- a/database/postgres/tes.go +++ b/database/postgres/tes.go @@ -128,7 +128,7 @@ func (db *Postgres) ListTasks(ctx context.Context, req *tes.ListTasksRequest) (* args = append(args, pageSize) selectSQL := fmt.Sprintf("SELECT data FROM tasks %s %s %s", whereClause, orderByClause, limitClause) - + fmt.Printf("ListTasks SQL: %s\n", selectSQL) rows, err := db.client.Query(ctx, selectSQL, args...) if err != nil { return nil, err From 7f60650ad8bab7d8b6a5b5785a0593aa9c17c3df Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Fri, 22 May 2026 01:54:32 -0500 Subject: [PATCH 14/35] Add state filter in ListTasks --- compute/kubernetes/backend.go | 2 +- database/postgres/tes.go | 8 ++++++++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 307f0c904..59198f761 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -525,7 +525,7 @@ func (b *Backend) reconcileOnce(ctx context.Context, disableCleanup bool) { break } for _, task := range lresp.Tasks { - b.log.Debug("reconcile: processing task", "taskID", task.Id, "state", task.State) + fmt.Println("DEBUG: Reconciling task", task.Id, "with state", task.State) j, exists := k8sJobs[task.Id] delete(k8sJobs, task.Id) // matched — remove so it isn't treated as orphaned if exists { diff --git a/database/postgres/tes.go b/database/postgres/tes.go index b47121451..57859097a 100644 --- a/database/postgres/tes.go +++ b/database/postgres/tes.go @@ -89,6 +89,14 @@ func (db *Postgres) ListTasks(ctx context.Context, req *tes.ListTasksRequest) (* paramCount++ } + // State filter. FIXME: The logic fails to fetch the records for tasks in the UNKNOWN state, but we currently have no tasks in that state, so it isn't a problem. + // We should review this logic when we add support for UNKNOWN state tasks. + if req.State != tes.State_UNKNOWN { + whereClauses = append(whereClauses, fmt.Sprintf("state = $%d", paramCount)) + args = append(args, req.State) + paramCount++ + } + // Authorization filter if userInfo := server.GetUser(ctx); !userInfo.CanSeeAllTasks() { whereClauses = append(whereClauses, fmt.Sprintf("owner = $%d", paramCount)) From ede3f33ff403b5a9b640e6dc3d0345cc8a526e76 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Fri, 22 May 2026 02:00:32 -0500 Subject: [PATCH 15/35] Add state in the ListTasks tests --- database/postgres/postgres_test.go | 46 +++++++++++++++++++----------- 1 file changed, 30 insertions(+), 16 deletions(-) diff --git a/database/postgres/postgres_test.go b/database/postgres/postgres_test.go index 7946f3df8..7e498797a 100644 --- a/database/postgres/postgres_test.go +++ b/database/postgres/postgres_test.go @@ -150,9 +150,31 @@ func TestPostgresOperations(t *testing.T) { } }) + // Cancel Task + t.Run("CancelTask", func(t *testing.T) { + event := events.NewState(testTaskID, tes.State_CANCELED) + if err := db.WriteEvent(ctx, event); err != nil { + t.Fatalf("Failed to cancel task: %v", err) + } + }) + + // Check Canceled Task + t.Run("CheckCancel", func(t *testing.T) { + req := &tes.GetTaskRequest{Id: testTaskID, View: tes.View_MINIMAL.String()} + fetchedTask, err := db.GetTask(ctx, req) + + if err != nil { + t.Fatalf("Failed to get canceled task: %v", err) + } + if fetchedTask.State != tes.State_CANCELED { + t.Errorf("Task state not CANCELED. Expected %s, Got %s", tes.State_CANCELED, fetchedTask.State) + } + }) + // List Tasks t.Run("ListTasks", func(t *testing.T) { req := &tes.ListTasksRequest{ + State: tes.State_CANCELED, PageSize: 10, View: tes.View_MINIMAL.String(), } @@ -166,26 +188,18 @@ func TestPostgresOperations(t *testing.T) { if resp.Tasks[0].Id != testTaskID { t.Errorf("ListTasks returned wrong ID: %s", resp.Tasks[0].Id) } - }) - // Cancel Task - t.Run("CancelTask", func(t *testing.T) { - event := events.NewState(testTaskID, tes.State_CANCELED) - if err := db.WriteEvent(ctx, event); err != nil { - t.Fatalf("Failed to cancel task: %v", err) + req = &tes.ListTasksRequest{ + State: tes.State_COMPLETE, + PageSize: 10, + View: tes.View_MINIMAL.String(), } - }) - - // Check Canceled Task - t.Run("CheckCancel", func(t *testing.T) { - req := &tes.GetTaskRequest{Id: testTaskID, View: tes.View_MINIMAL.String()} - fetchedTask, err := db.GetTask(ctx, req) - + resp, err = db.ListTasks(ctx, req) if err != nil { - t.Fatalf("Failed to get canceled task: %v", err) + t.Fatalf("Failed to list tasks: %v", err) } - if fetchedTask.State != tes.State_CANCELED { - t.Errorf("Task state not CANCELED. Expected %s, Got %s", tes.State_CANCELED, fetchedTask.State) + if len(resp.Tasks) != 0 { // We only created one task and it's canceled, so there should be 0 complete tasks. + t.Fatalf("Expected 0 tasks in list, got %d", len(resp.Tasks)) } }) } From 88b33900dc0c60afb5bd1fe552a57ef7e85ee035 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Fri, 22 May 2026 02:51:00 -0500 Subject: [PATCH 16/35] Remove sql debug logs --- database/postgres/tes.go | 1 - 1 file changed, 1 deletion(-) diff --git a/database/postgres/tes.go b/database/postgres/tes.go index 57859097a..bc7d6eb4a 100644 --- a/database/postgres/tes.go +++ b/database/postgres/tes.go @@ -136,7 +136,6 @@ func (db *Postgres) ListTasks(ctx context.Context, req *tes.ListTasksRequest) (* args = append(args, pageSize) selectSQL := fmt.Sprintf("SELECT data FROM tasks %s %s %s", whereClause, orderByClause, limitClause) - fmt.Printf("ListTasks SQL: %s\n", selectSQL) rows, err := db.client.Query(ctx, selectSQL, args...) if err != nil { return nil, err From 97009c7f93580e914338b822fa5bc595d351a775 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Tue, 16 Jun 2026 20:23:03 -0500 Subject: [PATCH 17/35] Add updated reconcile_monolith and modular reconciler --- compute/kubernetes/backend.go | 90 +++++++++++++++++++++++++---------- 1 file changed, 66 insertions(+), 24 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index e45ef3ff1..5045ca451 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -401,14 +401,8 @@ func FetchPodWarningEvents(ctx context.Context, clientset kubernetes.Interface, // isJobSchedulingTimedOut returns true if all pods for the given job have been // stuck in Pending (with a scheduling condition) for longer than timeout. // It returns false if any pod has been scheduled, or if pod status cannot be determined. -func (b *Backend) isJobSchedulingTimedOut(ctx context.Context, jobName string, timeout time.Duration) bool { - pods, err := b.client.CoreV1().Pods(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ - LabelSelector: fmt.Sprintf("job-name=%s", jobName), - }) - if err != nil { - b.log.Error("reconcile: listing pods for job", "taskID", jobName, "error", err) - return false - } +func (b *Backend) isJobSchedulingTimedOut(ctx context.Context, timeout time.Duration, pods *corev1.PodList) bool { + if len(pods.Items) == 0 { return false } @@ -432,14 +426,7 @@ func (b *Backend) isJobSchedulingTimedOut(ctx context.Context, jobName string, t // getFailedPodInfo returns a human-readable summary of why the most recently // terminated pod for jobName failed: "exit code N (Reason): Message". It is // best-effort; an empty string is returned when no useful information is found. -func (b *Backend) getFailedPodInfo(ctx context.Context, jobName string) string { - pods, err := b.client.CoreV1().Pods(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ - LabelSelector: fmt.Sprintf("job-name=%s", jobName), - }) - if err != nil { - b.log.Error("reconcile: listing pods for failed job", "taskID", jobName, "error", err) - return "" - } +func (b *Backend) getFailedPodInfo(ctx context.Context, pods *corev1.PodList) string { var latestFinish metav1.Time var result string @@ -613,12 +600,15 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo status := j.Status schedulingTimeout := b.conf.Kubernetes.Timeout.GetDuration() - writeSystemError := func(reason string) { + writeSystemError := func(errAttributes map[string]string, additionalMessage string) { + if additionalMessage == "" { + additionalMessage = "Kubernetes job in FAILED state" + } + b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) b.event.WriteEvent(ctx, events.NewSystemLog( - jobName, 0, 0, "error", - "Kubernetes job in FAILED state", - map[string]string{"error": reason}, + jobName, 0, 0, "error", additionalMessage, + errAttributes, )) } @@ -631,11 +621,46 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo } } + pods, err := b.client.CoreV1().Pods(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ + LabelSelector: fmt.Sprintf("job-name=%s", jobName), + }) + if err != nil { + b.log.Error("reconcile: failed to list pods for job", "taskID", jobName, "error", err) + return + } + switch { case status.Active > 0: - if schedulingTimeout != nil && b.isJobSchedulingTimedOut(ctx, jobName, schedulingTimeout.AsDuration()) { + + // Check for container waiting errors that will never self-resolve + // (e.g. CreateContainerConfigError). These keep the Job Active + // indefinitely, so we must detect and fail them explicitly. + if terminal, reason := hasTerminalContainerWaitingError(pods); terminal { + b.log.Debug("reconcile: worker pod has terminal container waiting error", "taskID", jobName, "reason", reason) + errDetail := reason + if podEvents := FetchPodWarningEvents(ctx, b.client, b.conf.Kubernetes.JobsNamespace, pods); podEvents != "" { + errDetail = fmt.Sprintf("%s\n%s", reason, podEvents) + } + writeSystemError(map[string]string{"error": errDetail}, "Kubernetes worker pod has a terminal container waiting error") + cleanResourcesIfEnabled() + return + } + + // Check for FailedCreate events on the Job itself. This catches + // cases where pod creation is rejected before a pod object is + // ever persisted (e.g. Pod Security Admission enforcement blocks + // the pod), so there are no pod container statuses to inspect. + b.log.Debug("checking for FailedCreate events on job", "taskID", jobName) + if count, reason := b.hasJobFailedCreateEvent(ctx, jobName); count > 0 { + b.log.Debug("reconcile: worker job has FailedCreate event", "taskID", jobName, "count", count, "reason", reason) + writeSystemError(map[string]string{"error": reason}, "worker job failed to create pod") + cleanResourcesIfEnabled() + return + } + + if schedulingTimeout != nil && b.isJobSchedulingTimedOut(ctx, schedulingTimeout.AsDuration(), pods) { b.log.Debug("reconcile: worker pod scheduling timed out.", "taskID", jobName) - writeSystemError("worker pod scheduling timed out") + writeSystemError(map[string]string{"error": "worker pod scheduling timed out"}, "") cleanResourcesIfEnabled() } @@ -667,10 +692,27 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo if err != nil { b.log.Error("reconcile: marshaling failed job conditions", "taskID", jobName, "error", err) } - writeSystemError(string(conds)) + errDetails := map[string]string{"error": string(conds)} + if podInfo := b.getFailedPodInfo(ctx, pods); podInfo != "" { + errDetails["executor_error"] = podInfo + } + writeSystemError(errDetails, "") } b.log.Debug("reconcile: reconciled failed job", "taskID", jobName) cleanResourcesIfEnabled() + + default: + // All status counters are zero: the Job controller has not yet + // recorded any Active/Succeeded/Failed pods. This happens when + // every pod creation attempt is rejected before Kubernetes + // persists a pod object (e.g. Pod Security Admission blocks the + // pod). Check for FailedCreate events which are the only signal + // available in this state. + if count, reason := b.hasJobFailedCreateEvent(ctx, jobName); count > 0 { + b.log.Debug("reconcile: worker job has FailedCreate event (zero-status)", "taskID", jobName, "count", count, "reason", reason) + writeSystemError(map[string]string{"error": reason}, "Kubernetes worker job failed to create pod") + cleanResourcesIfEnabled() + } } } @@ -924,7 +966,7 @@ func (b *Backend) reconcile_monolith(ctx context.Context, rate time.Duration, di // itself succeeds immediately. if b.conf.Kubernetes.Timeout.GetDuration() != nil { timeout := b.conf.Kubernetes.Timeout.GetDuration().AsDuration() - if b.isJobSchedulingTimedOut(ctx, jobName, timeout) { + if b.isJobSchedulingTimedOut(ctx, timeout, pods) { b.log.Debug("reconcile: worker pod scheduling timed out", "taskID", jobName) b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) b.event.WriteEvent(ctx, events.NewSystemLog( From 082a1a15d6dbcded3bc84357606ada4a655f4f3a Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Wed, 17 Jun 2026 12:52:34 -0500 Subject: [PATCH 18/35] Retry pods when job backoff limit is not reached yet --- compute/kubernetes/backend.go | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 5045ca451..c6242d707 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -598,6 +598,7 @@ func (b *Backend) cleanBacklog(ctx context.Context) { func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bool) { jobName := j.Name status := j.Status + jobBackoffLimit := j.Spec.BackoffLimit schedulingTimeout := b.conf.Kubernetes.Timeout.GetDuration() writeSystemError := func(errAttributes map[string]string, additionalMessage string) { @@ -698,9 +699,13 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo } writeSystemError(errDetails, "") } - b.log.Debug("reconcile: reconciled failed job", "taskID", jobName) - cleanResourcesIfEnabled() + b.log.Debug("reconcile: reconciled failed job", "taskID", jobName) + if jobBackoffLimit != nil && status.Failed > *jobBackoffLimit { + cleanResourcesIfEnabled() + } else { + b.log.Debug("reconcile: job backoff limit not exceeded. Kubernetes will retry the job.", "taskID", jobName, "failed", status.Failed, "backoffLimit", *jobBackoffLimit) + } default: // All status counters are zero: the Job controller has not yet // recorded any Active/Succeeded/Failed pods. This happens when From 0c3f225aee5129385aa096e50a1e8f31923ca07d Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Wed, 17 Jun 2026 13:01:34 -0500 Subject: [PATCH 19/35] Fix compiler error in reconciler_monolith --- compute/kubernetes/backend.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index c6242d707..c609d75bd 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -1012,7 +1012,14 @@ func (b *Backend) reconcile_monolith(ctx context.Context, rate time.Duration, di } errDetails := map[string]string{"error": string(conds)} - if podInfo := b.getFailedPodInfo(ctx, jobName); podInfo != "" { + pods, err := b.client.CoreV1().Pods(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ + LabelSelector: fmt.Sprintf("job-name=%s", jobName), + }) + if err != nil { + b.log.Error("reconcile: failed to list pods for job", "taskID", jobName, "error", err) + return + } + if podInfo := b.getFailedPodInfo(ctx, pods); podInfo != "" { errDetails["executor_error"] = podInfo } From 60ba5725655d08d72563d2f98e7f5d301cd1640a Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Wed, 17 Jun 2026 13:39:40 -0500 Subject: [PATCH 20/35] Replace inner function to class level function and mark completed task in Running as System error --- compute/kubernetes/backend.go | 76 +++++++++++++++++------------------ 1 file changed, 37 insertions(+), 39 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index c609d75bd..126b0e873 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -595,33 +595,33 @@ func (b *Backend) cleanBacklog(ctx context.Context) { b.CleanOrphanedResources(ctx) } +func (b *Backend) cleanResourcesIfEnabled(ctx context.Context, jobName string, disableCleanup bool) { + if !disableCleanup { + b.log.Debug("reconcile: cleaning up job", "taskID", jobName) + if err := b.cleanResources(ctx, jobName); err != nil { + b.log.Error("reconcile: failed to clean resources", "taskID", jobName, "error", err) + } + } +} + +func (b *Backend) writeSystemError(ctx context.Context, jobName string, errAttributes map[string]string, additionalMessage string) { + if additionalMessage == "" { + additionalMessage = "Kubernetes job in FAILED state" + } + + b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) + b.event.WriteEvent(ctx, events.NewSystemLog( + jobName, 0, 0, "error", additionalMessage, + errAttributes, + )) +} + func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bool) { jobName := j.Name status := j.Status jobBackoffLimit := j.Spec.BackoffLimit schedulingTimeout := b.conf.Kubernetes.Timeout.GetDuration() - writeSystemError := func(errAttributes map[string]string, additionalMessage string) { - if additionalMessage == "" { - additionalMessage = "Kubernetes job in FAILED state" - } - - b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) - b.event.WriteEvent(ctx, events.NewSystemLog( - jobName, 0, 0, "error", additionalMessage, - errAttributes, - )) - } - - cleanResourcesIfEnabled := func() { - if !disableCleanup { - b.log.Debug("reconcile: cleaning up job", "taskID", jobName) - if err := b.cleanResources(ctx, jobName); err != nil { - b.log.Error("reconcile: failed to clean resources", "taskID", jobName, "error", err) - } - } - } - pods, err := b.client.CoreV1().Pods(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ LabelSelector: fmt.Sprintf("job-name=%s", jobName), }) @@ -642,8 +642,8 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo if podEvents := FetchPodWarningEvents(ctx, b.client, b.conf.Kubernetes.JobsNamespace, pods); podEvents != "" { errDetail = fmt.Sprintf("%s\n%s", reason, podEvents) } - writeSystemError(map[string]string{"error": errDetail}, "Kubernetes worker pod has a terminal container waiting error") - cleanResourcesIfEnabled() + b.writeSystemError(ctx, jobName, map[string]string{"error": errDetail}, "Kubernetes worker pod has a terminal container waiting error") + b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup) return } @@ -654,20 +654,20 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo b.log.Debug("checking for FailedCreate events on job", "taskID", jobName) if count, reason := b.hasJobFailedCreateEvent(ctx, jobName); count > 0 { b.log.Debug("reconcile: worker job has FailedCreate event", "taskID", jobName, "count", count, "reason", reason) - writeSystemError(map[string]string{"error": reason}, "worker job failed to create pod") - cleanResourcesIfEnabled() + b.writeSystemError(ctx, jobName, map[string]string{"error": reason}, "worker job failed to create pod") + b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup) return } if schedulingTimeout != nil && b.isJobSchedulingTimedOut(ctx, schedulingTimeout.AsDuration(), pods) { b.log.Debug("reconcile: worker pod scheduling timed out.", "taskID", jobName) - writeSystemError(map[string]string{"error": "worker pod scheduling timed out"}, "") - cleanResourcesIfEnabled() + b.writeSystemError(ctx, jobName, map[string]string{"error": "worker pod scheduling timed out"}, "") + b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup) } case status.Succeeded > 0: b.log.Debug("reconcile: reconciled successful job", "taskID", jobName) - cleanResourcesIfEnabled() + b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup) case status.Failed > 0: // Only act if K8s has marked the Job as permanently failed (backoffLimit exhausted). @@ -697,12 +697,12 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo if podInfo := b.getFailedPodInfo(ctx, pods); podInfo != "" { errDetails["executor_error"] = podInfo } - writeSystemError(errDetails, "") + b.writeSystemError(ctx, jobName, errDetails, "") } b.log.Debug("reconcile: reconciled failed job", "taskID", jobName) if jobBackoffLimit != nil && status.Failed > *jobBackoffLimit { - cleanResourcesIfEnabled() + b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup) } else { b.log.Debug("reconcile: job backoff limit not exceeded. Kubernetes will retry the job.", "taskID", jobName, "failed", status.Failed, "backoffLimit", *jobBackoffLimit) } @@ -715,8 +715,8 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo // available in this state. if count, reason := b.hasJobFailedCreateEvent(ctx, jobName); count > 0 { b.log.Debug("reconcile: worker job has FailedCreate event (zero-status)", "taskID", jobName, "count", count, "reason", reason) - writeSystemError(map[string]string{"error": reason}, "Kubernetes worker job failed to create pod") - cleanResourcesIfEnabled() + b.writeSystemError(ctx, jobName, map[string]string{"error": reason}, "Kubernetes worker job failed to create pod") + b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup) } } } @@ -750,6 +750,9 @@ func (b *Backend) reconcileOnce(ctx context.Context, disableCleanup bool) { delete(k8sJobs, task.Id) // matched — remove so it isn't treated as orphaned if exists { b.reconcileJob(ctx, j, disableCleanup) + } else { + b.log.Debug("reconcile: job not found for task", "taskID", task.Id) + b.writeSystemError(ctx, task.Id, map[string]string{"error": "job not found"}, "Kubernetes job not found for task") } } pageToken = lresp.NextPageToken @@ -761,13 +764,8 @@ func (b *Backend) reconcileOnce(ctx context.Context, disableCleanup bool) { } // Any jobs still in k8sJobs were not matched to any Funnel task — orphaned or in terminal states. - if !disableCleanup { - for taskID := range k8sJobs { - b.log.Info("reconcile: cleaning up job for terminal or missing Funnel task", "taskID", taskID) - if err := b.cleanResources(ctx, taskID); err != nil { - b.log.Error("reconcile: failed to clean orphaned resources", "taskID", taskID, "error", err) - } - } + for taskID := range k8sJobs { + b.cleanResourcesIfEnabled(ctx, taskID, disableCleanup) } } From ee31f49d217b6c419c996c8dde7eaf30c0f56fce Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Wed, 17 Jun 2026 14:29:13 -0500 Subject: [PATCH 21/35] Remove tests linked to TTLSecondsAfterFinished --- tests/kubernetes/kubernetes_test.go | 84 +++-------------------------- 1 file changed, 7 insertions(+), 77 deletions(-) diff --git a/tests/kubernetes/kubernetes_test.go b/tests/kubernetes/kubernetes_test.go index 91e13350b..f48c0fd23 100644 --- a/tests/kubernetes/kubernetes_test.go +++ b/tests/kubernetes/kubernetes_test.go @@ -259,9 +259,8 @@ func TestHelloWorld(t *testing.T) { } } -// jobTemplateNoTTL is a minimal WorkerTemplate without ttlSecondsAfterFinished; -// used to verify that a default TTL of 300 seconds is injected automatically. -const jobTemplateNoTTL = `apiVersion: batch/v1 +// jobTemplate is a minimal WorkerTemplate +const jobTemplate = `apiVersion: batch/v1 kind: Job metadata: name: funnel-{{.TaskId}} @@ -280,24 +279,6 @@ spec: command: ["echo", "{{.TaskName}}"] ` -// jobTemplateWithTTL has an explicit ttlSecondsAfterFinished of 600; -// used to verify that a template-specified TTL is not overwritten by the default. -const jobTemplateWithTTL = `apiVersion: batch/v1 -kind: Job -metadata: - name: funnel-{{.TaskId}} - namespace: {{.JobsNamespace}} -spec: - ttlSecondsAfterFinished: 600 - backoffLimit: {{.BackoffLimit}} - template: - spec: - restartPolicy: Never - containers: - - name: worker - image: alpine -` - // Tests for SanitizeLabelValue: converts task names into valid Kubernetes label values. func TestSanitizeLabelValue_ValidInputPassthrough(t *testing.T) { cases := []struct{ in, want string }{ @@ -349,57 +330,6 @@ func TestSanitizeLabelValue_StripsLeadingTrailingNonAlphanumeric(t *testing.T) { } } -// Tests for CreateJob: ttlSecondsAfterFinished defaults to 300 when absent from the template. -func TestCreateJob_DefaultTTLIsSet(t *testing.T) { - client := fakeClientWithFunnelPod(unitTestNS) - task := &tes.Task{ - Id: "ttl-default", - Name: "TTL Default Test", - Resources: &tes.Resources{CpuCores: 1, RamGb: 1.0}, - } - - if _, err := resources.CreateJob(context.Background(), task, baseJobConfig(jobTemplateNoTTL), client, unitTestLog); err != nil { - t.Fatalf("CreateJob: %v", err) - } - - job, err := client.BatchV1().Jobs(unitTestJobsNS).Get(context.Background(), "funnel-"+task.Id, metav1.GetOptions{}) - if err != nil { - t.Fatalf("get job: %v", err) - } - - if job.Spec.TTLSecondsAfterFinished == nil { - t.Fatal("TTLSecondsAfterFinished is nil; want 300") - } - if got := *job.Spec.TTLSecondsAfterFinished; got != 300 { - t.Errorf("TTLSecondsAfterFinished = %d; want 300", got) - } -} - -// Tests for CreateJob: a ttlSecondsAfterFinished already in the template is not overwritten. -func TestCreateJob_ExistingTTLIsPreserved(t *testing.T) { - client := fakeClientWithFunnelPod(unitTestNS) - task := &tes.Task{ - Id: "ttl-preserve", - Resources: &tes.Resources{CpuCores: 1, RamGb: 1.0}, - } - - if _, err := resources.CreateJob(context.Background(), task, baseJobConfig(jobTemplateWithTTL), client, unitTestLog); err != nil { - t.Fatalf("CreateJob: %v", err) - } - - job, err := client.BatchV1().Jobs(unitTestJobsNS).Get(context.Background(), "funnel-"+task.Id, metav1.GetOptions{}) - if err != nil { - t.Fatalf("get job: %v", err) - } - - if job.Spec.TTLSecondsAfterFinished == nil { - t.Fatal("TTLSecondsAfterFinished is nil; want 600") - } - if got := *job.Spec.TTLSecondsAfterFinished; got != 600 { - t.Errorf("TTLSecondsAfterFinished = %d; want 600 (template value must not be overwritten)", got) - } -} - // Tests for CreateJob: BackoffLimit falls back to 10 when not set via backend_parameters. func TestCreateJob_DefaultBackoffLimit(t *testing.T) { client := fakeClientWithFunnelPod(unitTestNS) @@ -408,7 +338,7 @@ func TestCreateJob_DefaultBackoffLimit(t *testing.T) { Resources: &tes.Resources{CpuCores: 1}, } - if _, err := resources.CreateJob(context.Background(), task, baseJobConfig(jobTemplateNoTTL), client, unitTestLog); err != nil { + if _, err := resources.CreateJob(context.Background(), task, baseJobConfig(jobTemplate), client, unitTestLog); err != nil { t.Fatalf("CreateJob: %v", err) } @@ -436,7 +366,7 @@ func TestCreateJob_BackoffLimitFromBackendParameters(t *testing.T) { }, } - if _, err := resources.CreateJob(context.Background(), task, baseJobConfig(jobTemplateNoTTL), client, unitTestLog); err != nil { + if _, err := resources.CreateJob(context.Background(), task, baseJobConfig(jobTemplate), client, unitTestLog); err != nil { t.Fatalf("CreateJob: %v", err) } @@ -463,7 +393,7 @@ func TestCreateJob_BackoffLimitInvalidValueFallsBackToDefault(t *testing.T) { }, } - if _, err := resources.CreateJob(context.Background(), task, baseJobConfig(jobTemplateNoTTL), client, unitTestLog); err != nil { + if _, err := resources.CreateJob(context.Background(), task, baseJobConfig(jobTemplate), client, unitTestLog); err != nil { t.Fatalf("CreateJob: %v", err) } @@ -490,7 +420,7 @@ func TestCreateJob_NegativeBackoffLimitFallsBackToDefault(t *testing.T) { }, } - if _, err := resources.CreateJob(context.Background(), task, baseJobConfig(jobTemplateNoTTL), client, unitTestLog); err != nil { + if _, err := resources.CreateJob(context.Background(), task, baseJobConfig(jobTemplate), client, unitTestLog); err != nil { t.Fatalf("CreateJob: %v", err) } @@ -516,7 +446,7 @@ func TestCreateJob_TaskNameLabelIsSanitized(t *testing.T) { Resources: &tes.Resources{CpuCores: 1}, } - if _, err := resources.CreateJob(context.Background(), task, baseJobConfig(jobTemplateNoTTL), client, unitTestLog); err != nil { + if _, err := resources.CreateJob(context.Background(), task, baseJobConfig(jobTemplate), client, unitTestLog); err != nil { t.Fatalf("CreateJob: %v", err) } From e9da4c6090187a67e817118c64888f01050eb47b Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Wed, 17 Jun 2026 15:18:20 -0500 Subject: [PATCH 22/35] Update backofflimit check when task is in terminal state. --- compute/kubernetes/backend.go | 36 +++++++++++++++++++---------------- 1 file changed, 20 insertions(+), 16 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 126b0e873..2d1f5c37f 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -324,6 +324,15 @@ func (b *Backend) cleanResources(ctx context.Context, taskId string) error { return errs } +func (b *Backend) isJobMarkedAsFailed(jobStatus v1.JobStatus) bool { + for _, cond := range jobStatus.Conditions { + if cond.Type == v1.JobFailed && cond.Status == corev1.ConditionTrue { + return true + } + } + return false +} + // hasTerminalContainerWaitingError returns true if any pod in pods has a // container stuck in a waiting state whose reason is known to be permanent // (e.g. CreateContainerConfigError). These pods will never transition to a @@ -619,7 +628,6 @@ func (b *Backend) writeSystemError(ctx context.Context, jobName string, errAttri func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bool) { jobName := j.Name status := j.Status - jobBackoffLimit := j.Spec.BackoffLimit schedulingTimeout := b.conf.Kubernetes.Timeout.GetDuration() pods, err := b.client.CoreV1().Pods(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ @@ -670,20 +678,16 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup) case status.Failed > 0: + b.log.Debug("reconcile: Job has non zero failed status", "taskID", jobName, "failed", status.Failed) // Only act if K8s has marked the Job as permanently failed (backoffLimit exhausted). // If Active > 0 is also set, K8s is still retrying — don't intervene. if status.Active > 0 { return } - jobFailed := false - for _, cond := range status.Conditions { - if cond.Type == v1.JobFailed && cond.Status == corev1.ConditionTrue { - jobFailed = true - break - } - } - if !jobFailed { + + if !b.isJobMarkedAsFailed(status) { // K8s hasn't given up yet — still within backoffLimit, retrying. + b.log.Debug("reconcile: K8s hasn't given up yet — still within backoffLimit, retrying.", "taskID", jobName) return } task, err := b.database.GetTask(ctx, &tes.GetTaskRequest{Id: jobName, View: tes.View_MINIMAL.String()}) @@ -700,12 +704,7 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo b.writeSystemError(ctx, jobName, errDetails, "") } - b.log.Debug("reconcile: reconciled failed job", "taskID", jobName) - if jobBackoffLimit != nil && status.Failed > *jobBackoffLimit { - b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup) - } else { - b.log.Debug("reconcile: job backoff limit not exceeded. Kubernetes will retry the job.", "taskID", jobName, "failed", status.Failed, "backoffLimit", *jobBackoffLimit) - } + b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup) default: // All status counters are zero: the Job controller has not yet // recorded any Active/Succeeded/Failed pods. This happens when @@ -764,7 +763,12 @@ func (b *Backend) reconcileOnce(ctx context.Context, disableCleanup bool) { } // Any jobs still in k8sJobs were not matched to any Funnel task — orphaned or in terminal states. - for taskID := range k8sJobs { + for taskID, job := range k8sJobs { + b.log.Debug("reconcile: Task is either orphaned or in a terminal state", "taskID", taskID) + if !b.isJobMarkedAsFailed(job.Status) { + b.log.Debug("reconcile: K8s hasn't given up yet — still within backoffLimit, retrying.", "taskID", taskID, "failed", job.Status.Failed, "backoffLimit", *job.Spec.BackoffLimit) + continue + } b.cleanResourcesIfEnabled(ctx, taskID, disableCleanup) } From a6d75e4d8fa11dcecc993fb6f56b3c062268d139 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Tue, 16 Jun 2026 15:52:28 -0700 Subject: [PATCH 23/35] task: Revert cancel response coe back to 200 From e12f2de1d75dbebeb1aa55efda2f50af34ae79df Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Wed, 17 Jun 2026 15:21:52 -0700 Subject: [PATCH 24/35] Update integration_tests_on_kind.yaml --- .github/workflows/integration_tests_on_kind.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/integration_tests_on_kind.yaml b/.github/workflows/integration_tests_on_kind.yaml index 173157df1..7419bbcdc 100644 --- a/.github/workflows/integration_tests_on_kind.yaml +++ b/.github/workflows/integration_tests_on_kind.yaml @@ -11,7 +11,7 @@ permissions: jobs: integration_tests: name: Integration tests - uses: uc-cdis/.github/.github/workflows/integration_tests.yaml@master + uses: uc-cdis/.github/.github/workflows/integration_tests.yaml@28f8020ea479e0b79e9a86bdfb9da6566806cf3b with: QUAY_ORG: ohsu-comp-bio EXTERNAL_TO_CTDS: "true" From a4f4fbcc8e0dce00d74126324717423b189ad2b0 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Sat, 20 Jun 2026 22:56:37 -0500 Subject: [PATCH 25/35] Fix isJobMarkedAsFailed for tasks that are completed. --- compute/kubernetes/backend.go | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 2d1f5c37f..b2bfa7027 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -325,12 +325,19 @@ func (b *Backend) cleanResources(ctx context.Context, taskId string) error { } func (b *Backend) isJobMarkedAsFailed(jobStatus v1.JobStatus) bool { + var isComplete, isFailed bool for _, cond := range jobStatus.Conditions { - if cond.Type == v1.JobFailed && cond.Status == corev1.ConditionTrue { - return true + if cond.Status != corev1.ConditionTrue { + continue + } + switch cond.Type { + case v1.JobComplete: + isComplete = true + case v1.JobFailed: + isFailed = true } } - return false + return isFailed && !isComplete } // hasTerminalContainerWaitingError returns true if any pod in pods has a From 72b368f99f8953c28f697dd4d7af4a9d48f1a9b8 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Sat, 20 Jun 2026 23:11:51 -0500 Subject: [PATCH 26/35] Add debug logs to understand the jobConditions for successful tasks --- compute/kubernetes/backend.go | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index b2bfa7027..4a8abd1f0 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -70,6 +70,7 @@ func NewBackend(ctx context.Context, conf *config.Config, reader tes.ReadOnlySer if !conf.Kubernetes.DisableReconciler { rate := conf.Kubernetes.ReconcileRate.AsDuration() go b.reconcile(ctx, rate, conf.Kubernetes.DisableJobCleanup) + } return b, nil @@ -327,6 +328,7 @@ func (b *Backend) cleanResources(ctx context.Context, taskId string) error { func (b *Backend) isJobMarkedAsFailed(jobStatus v1.JobStatus) bool { var isComplete, isFailed bool for _, cond := range jobStatus.Conditions { + b.log.Debug("checking job condition", "type", cond.Type, "status", cond.Status, "reason", cond.Reason, "message", cond.Message) if cond.Status != corev1.ConditionTrue { continue } @@ -417,7 +419,7 @@ func FetchPodWarningEvents(ctx context.Context, clientset kubernetes.Interface, // isJobSchedulingTimedOut returns true if all pods for the given job have been // stuck in Pending (with a scheduling condition) for longer than timeout. // It returns false if any pod has been scheduled, or if pod status cannot be determined. -func (b *Backend) isJobSchedulingTimedOut(ctx context.Context, timeout time.Duration, pods *corev1.PodList) bool { +func (b *Backend) isJobSchedulingTimedOut(timeout time.Duration, pods *corev1.PodList) bool { if len(pods.Items) == 0 { return false @@ -674,7 +676,7 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo return } - if schedulingTimeout != nil && b.isJobSchedulingTimedOut(ctx, schedulingTimeout.AsDuration(), pods) { + if schedulingTimeout != nil && b.isJobSchedulingTimedOut(schedulingTimeout.AsDuration(), pods) { b.log.Debug("reconcile: worker pod scheduling timed out.", "taskID", jobName) b.writeSystemError(ctx, jobName, map[string]string{"error": "worker pod scheduling timed out"}, "") b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup) @@ -691,7 +693,7 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo if status.Active > 0 { return } - + b.log.Debug("reconcile: checking is job marked as failed", "taskID", jobName) if !b.isJobMarkedAsFailed(status) { // K8s hasn't given up yet — still within backoffLimit, retrying. b.log.Debug("reconcile: K8s hasn't given up yet — still within backoffLimit, retrying.", "taskID", jobName) @@ -980,7 +982,7 @@ func (b *Backend) reconcile_monolith(ctx context.Context, rate time.Duration, di // itself succeeds immediately. if b.conf.Kubernetes.Timeout.GetDuration() != nil { timeout := b.conf.Kubernetes.Timeout.GetDuration().AsDuration() - if b.isJobSchedulingTimedOut(ctx, timeout, pods) { + if b.isJobSchedulingTimedOut(timeout, pods) { b.log.Debug("reconcile: worker pod scheduling timed out", "taskID", jobName) b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) b.event.WriteEvent(ctx, events.NewSystemLog( From 6dda0823ef4c0c956ccbe9deddb18d8e6b6086ce Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Sat, 20 Jun 2026 23:33:05 -0500 Subject: [PATCH 27/35] isJobDone --- compute/kubernetes/backend.go | 20 +++++++++----------- 1 file changed, 9 insertions(+), 11 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 4a8abd1f0..1e66f192f 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -325,21 +325,19 @@ func (b *Backend) cleanResources(ctx context.Context, taskId string) error { return errs } -func (b *Backend) isJobMarkedAsFailed(jobStatus v1.JobStatus) bool { - var isComplete, isFailed bool +// isJobDone reports whether the job has finished — either succeeded +// or permanently failed (backoffLimit exhausted) — as opposed to still +// being retried or in progress. +func (b *Backend) isJobDone(jobStatus v1.JobStatus) bool { for _, cond := range jobStatus.Conditions { - b.log.Debug("checking job condition", "type", cond.Type, "status", cond.Status, "reason", cond.Reason, "message", cond.Message) if cond.Status != corev1.ConditionTrue { continue } - switch cond.Type { - case v1.JobComplete: - isComplete = true - case v1.JobFailed: - isFailed = true + if cond.Type == v1.JobComplete || cond.Type == v1.JobFailed { + return true } } - return isFailed && !isComplete + return false } // hasTerminalContainerWaitingError returns true if any pod in pods has a @@ -694,7 +692,7 @@ func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bo return } b.log.Debug("reconcile: checking is job marked as failed", "taskID", jobName) - if !b.isJobMarkedAsFailed(status) { + if !b.isJobDone(status) { // K8s hasn't given up yet — still within backoffLimit, retrying. b.log.Debug("reconcile: K8s hasn't given up yet — still within backoffLimit, retrying.", "taskID", jobName) return @@ -774,7 +772,7 @@ func (b *Backend) reconcileOnce(ctx context.Context, disableCleanup bool) { // Any jobs still in k8sJobs were not matched to any Funnel task — orphaned or in terminal states. for taskID, job := range k8sJobs { b.log.Debug("reconcile: Task is either orphaned or in a terminal state", "taskID", taskID) - if !b.isJobMarkedAsFailed(job.Status) { + if !b.isJobDone(job.Status) { b.log.Debug("reconcile: K8s hasn't given up yet — still within backoffLimit, retrying.", "taskID", taskID, "failed", job.Status.Failed, "backoffLimit", *job.Spec.BackoffLimit) continue } From fd848b38d3d1e73b77a5af52593dfc748f93e4d5 Mon Sep 17 00:00:00 2001 From: Sai Shanmukha Date: Sun, 21 Jun 2026 00:19:27 -0500 Subject: [PATCH 28/35] Final commits and comment improvements before making PR. --- .github/workflows/docker.yaml | 2 +- compute/kubernetes/backend.go | 317 ++-------------------------------- database/postgres/tes.go | 1 + worker/kubernetes.go | 17 -- 4 files changed, 12 insertions(+), 325 deletions(-) diff --git a/.github/workflows/docker.yaml b/.github/workflows/docker.yaml index f732a7412..aedbaa184 100644 --- a/.github/workflows/docker.yaml +++ b/.github/workflows/docker.yaml @@ -12,7 +12,7 @@ concurrency: cancel-in-progress: true env: - QUAY_REPO: quay.io/cdis/funnel + QUAY_REPO: quay.io/ohsu-comp-bio/funnel jobs: build: diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 1e66f192f..a6d334e96 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -68,9 +68,13 @@ func NewBackend(ctx context.Context, conf *config.Config, reader tes.ReadOnlySer } if !conf.Kubernetes.DisableReconciler { + b.cleanBacklog(ctx, conf.Kubernetes.DisableJobCleanup) + + // TODO: Add a condition based on whether ExternalReconciler is enabled or not, + // so that we don't start the internal reconciler when an external one is configured. + // This is to avoid the possibility of multiple concurrent reconcilers running for each server instance. rate := conf.Kubernetes.ReconcileRate.AsDuration() go b.reconcile(ctx, rate, conf.Kubernetes.DisableJobCleanup) - } return b, nil @@ -575,7 +579,11 @@ func (b *Backend) listAllWorkerJobs(ctx context.Context) (map[string]*v1.Job, er return k8sJobs, nil } -func (b *Backend) cleanBacklog(ctx context.Context) { +func (b *Backend) cleanBacklog(ctx context.Context, disableCleanup bool) { + if !disableCleanup { + return + } + k8sJobs, err := b.listAllWorkerJobs(ctx) if err != nil { b.log.Error("backlog cleanup: listing jobs", err) @@ -783,10 +791,6 @@ func (b *Backend) reconcileOnce(ctx context.Context, disableCleanup bool) { // reconcile is the ticker-based loop used when ExternalReconciler is false. func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableCleanup bool) { - if !disableCleanup { - b.cleanBacklog(ctx) - } - ticker := time.NewTicker(rate) defer ticker.Stop() @@ -800,307 +804,6 @@ func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableClea } } -// Reconcile loops through tasks and checks the status from Funnel's database -// against the status reported by Kubernetes. This allows the backend to report -// system error's that prevented the worker process from running. -// -// Currently this handles a narrow set of cases: -// -// |---------------------|-----------------|--------------------| -// | Funnel State | Backend State | Reconciled State | -// |---------------------|-----------------|--------------------| -// | QUEUED | FAILED | SYSTEM_ERROR | -// | INITIALIZING | FAILED | SYSTEM_ERROR | -// | RUNNING | FAILED | SYSTEM_ERROR | -// -// In this context a "FAILED" state is being used as a generic term that captures -// one or more terminal states for the backend. -// -// This loop is also used to cleanup successful jobs. -func (b *Backend) reconcile_monolith(ctx context.Context, rate time.Duration, disableCleanup bool) { - // Clears all resources that still exist from jobs that have run before this server started. - // This handles two cases: - // 1. Completed jobs (Succeeded/Failed) that were not cleaned up before the server restarted. - // 2. Orphaned jobs (Active) whose task no longer exists in the Funnel DB — left over from - // a previous deployment or server crash. - if !disableCleanup { - jobs, err := b.client.BatchV1().Jobs(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ - LabelSelector: "app=funnel-worker", - }) - if err != nil { - b.log.Error("backlog cleanup: listing jobs", err) - } else { - for _, j := range jobs.Items { - s := j.Status - taskID := j.Name - - // Always clean up completed jobs from prior runs. - if s.Succeeded > 0 || s.Failed > 0 { - b.log.Debug("backlog cleanup: deleting completed job", "taskID", taskID) - if err := b.cleanResources(ctx, taskID); err != nil { - b.log.Error("backlog cleanup: failed to clean resources", "taskID", taskID, "error", err) - } - continue - } - - // For active jobs, check whether the task still exists in the DB. - // If it doesn't, the job is orphaned from a previous deployment. - if s.Active > 0 { - _, err := b.database.GetTask(ctx, &tes.GetTaskRequest{Id: taskID, View: tes.View_MINIMAL.String()}) - if err != nil { - b.log.Info("backlog cleanup: deleting orphaned active job with no matching task", "taskID", taskID) - if err := b.cleanResources(ctx, taskID); err != nil { - b.log.Error("backlog cleanup: failed to clean orphaned resources", "taskID", taskID, "error", err) - } - } - } - } - } - b.CleanOrphanedResources(ctx) - - } - - ticker := time.NewTicker(rate) - failedJobEvents := make(map[string]int) - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - // List worker jobs only (label selector excludes executor jobs and unrelated jobs). - // Bug: If K8s Job is not created by the time reconciler runs, then the TES Task itself will be prematurely marked as SYSTEM_ERROR - jobs, err := b.client.BatchV1().Jobs(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ - LabelSelector: "app=funnel-worker", - }) - if err != nil { - b.log.Error("reconcile: listing jobs", err) - continue - } - - // Index worker jobs by task ID. We use a label selector to avoid - // picking up executor jobs (app=funnel-executor) or unrelated jobs. - k8sJobs := make(map[string]*v1.Job) - for i := range jobs.Items { - k8sJobs[jobs.Items[i].Name] = &jobs.Items[i] - } - - // List non-terminal tasks from Funnel's database - states := []tes.State{tes.State_QUEUED, tes.State_INITIALIZING, tes.State_RUNNING} - for _, s := range states { - pageToken := "" - for { - lresp, err := b.database.ListTasks(ctx, &tes.ListTasksRequest{ - State: s, - PageSize: 100, - PageToken: pageToken, - }) - if err != nil { - b.log.Error("reconcile: listing non-terminal tasks from Funnel DB", err) - break - } - pageToken = lresp.NextPageToken - - // Compare Funnel Tasks against K8s Jobs - for _, task := range lresp.Tasks { - taskID := task.Id - - // If the job exists, check its current status (Active, Succeeded, Failed) - j := k8sJobs[taskID] - - // Remove matched jobs so that any remaining entries after this - // loop represent orphaned K8s jobs with no Funnel task. - delete(k8sJobs, taskID) - - if j == nil { - continue - } - - jobName := j.Name - status := j.Status - switch { - case status.Active > 0: - // Fetch pods once and pass to both checks to avoid redundant K8s API calls. - pods, err := b.client.CoreV1().Pods(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ - LabelSelector: fmt.Sprintf("job-name=%s", jobName), - }) - if err != nil { - b.log.Error("reconcile: listing pods for job", "taskID", jobName, "error", err) - continue - } - - // Check for container waiting errors that will never self-resolve - // (e.g. CreateContainerConfigError). These keep the Job Active - // indefinitely, so we must detect and fail them explicitly. - if terminal, reason := hasTerminalContainerWaitingError(pods); terminal { - b.log.Debug("reconcile: worker pod has terminal container waiting error", "taskID", jobName, "reason", reason) - b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) - errDetail := reason - if podEvents := FetchPodWarningEvents(ctx, b.client, b.conf.Kubernetes.JobsNamespace, pods); podEvents != "" { - errDetail = fmt.Sprintf("%s\n%s", reason, podEvents) - } - b.event.WriteEvent(ctx, events.NewSystemLog( - jobName, 0, 0, "error", - "Kubernetes worker pod has a terminal container waiting error", - map[string]string{"error": errDetail}, - )) - if !disableCleanup { - if err := b.cleanResources(ctx, jobName); err != nil { - b.log.Error("failed to clean resources", "taskID", jobName, "error", err) - } - } - continue - } - - // Check for FailedCreate events on the Job itself. This catches - // cases where pod creation is rejected before a pod object is - // ever persisted (e.g. Pod Security Admission enforcement blocks - // the pod), so there are no pod container statuses to inspect. - b.log.Debug("checking for FailedCreate events on job", "taskID", jobName) - if count, reason := b.hasJobFailedCreateEvent(ctx, jobName); count > 0 { - b.log.Debug("reconcile: worker job has FailedCreate event", "taskID", jobName, "count", count, "reason", reason) - b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) - b.event.WriteEvent(ctx, events.NewSystemLog( - jobName, 0, 0, "error", - "Kubernetes worker job failed to create pod", - map[string]string{"error": reason}, - )) - if !disableCleanup { - if err := b.cleanResources(ctx, jobName); err != nil { - b.log.Error("failed to clean resources", "taskID", jobName, "error", err) - } - } - continue - } - - // If a scheduling timeout is configured, check whether the worker - // pod has been stuck in Pending beyond that duration. This catches - // scheduling failures (bad NodeSelector, insufficient resources, etc.) - // that the context timeout cannot detect because the Job API call - // itself succeeds immediately. - if b.conf.Kubernetes.Timeout.GetDuration() != nil { - timeout := b.conf.Kubernetes.Timeout.GetDuration().AsDuration() - if b.isJobSchedulingTimedOut(timeout, pods) { - b.log.Debug("reconcile: worker pod scheduling timed out", "taskID", jobName) - b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) - b.event.WriteEvent(ctx, events.NewSystemLog( - jobName, 0, 0, "error", - "Kubernetes job in FAILED state", - map[string]string{"error": "worker pod scheduling timed out"}, - )) - if !disableCleanup { - if err := b.cleanResources(ctx, jobName); err != nil { - b.log.Error("failed to clean resources", "taskID", jobName, "error", err) - } - } - } - } - continue - case status.Succeeded > 0: - if disableCleanup { - continue - } - b.log.Debug("reconcile: cleaning up successful job", "taskID", jobName) - - // Delete resources - if err := b.cleanResources(ctx, jobName); err != nil { - b.log.Error("failed to clean resources", "taskID", jobName, "error", err) - continue - } - delete(failedJobEvents, jobName) - - case status.Failed > 0: - if count, exists := failedJobEvents[jobName]; exists && count >= failedCreateThreshold { - continue - } - - b.log.Debug("reconcile: writing system error event for failed job", "taskID", jobName) - conds, err := json.Marshal(status.Conditions) - if err != nil { - b.log.Error("reconcile: marshal failed job conditions", "taskID", jobName, "error", err) - } - - errDetails := map[string]string{"error": string(conds)} - pods, err := b.client.CoreV1().Pods(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{ - LabelSelector: fmt.Sprintf("job-name=%s", jobName), - }) - if err != nil { - b.log.Error("reconcile: failed to list pods for job", "taskID", jobName, "error", err) - return - } - if podInfo := b.getFailedPodInfo(ctx, pods); podInfo != "" { - errDetails["executor_error"] = podInfo - } - - b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) - b.event.WriteEvent( - ctx, - events.NewSystemLog( - jobName, 0, 0, "error", - "Kubernetes job in FAILED state", - errDetails, - ), - ) - - failedJobEvents[jobName]++ - if disableCleanup { - continue - } - - b.log.Debug("reconcile: cleaning up failed job", "taskID", jobName) - if err := b.cleanResources(ctx, jobName); err != nil { - b.log.Error("failed to clean resources", "taskID", jobName, "error", err) - continue - } - delete(failedJobEvents, jobName) - - default: - // All status counters are zero: the Job controller has not yet - // recorded any Active/Succeeded/Failed pods. This happens when - // every pod creation attempt is rejected before Kubernetes - // persists a pod object (e.g. Pod Security Admission blocks the - // pod). Check for FailedCreate events which are the only signal - // available in this state. - if count, reason := b.hasJobFailedCreateEvent(ctx, jobName); count > 0 { - b.log.Debug("reconcile: worker job has FailedCreate event (zero-status)", "taskID", jobName, "count", count, "reason", reason) - b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) - b.event.WriteEvent(ctx, events.NewSystemLog( - jobName, 0, 0, "error", - "Kubernetes worker job failed to create pod", - map[string]string{"error": reason}, - )) - if !disableCleanup { - if err := b.cleanResources(ctx, jobName); err != nil { - b.log.Error("failed to clean resources", "taskID", jobName, "error", err) - } - } - } - } - } - - // Continue to next page from ListTasks if a token exists - if pageToken == "" { - break - } - time.Sleep(time.Millisecond * 100) - } - } - - // Any jobs remaining in k8sJobs were not matched to a Funnel task — - // they are orphaned and should be cleaned up. - if !disableCleanup { - for taskID := range k8sJobs { - b.log.Info("reconcile: cleaning up orphaned job with no matching Funnel task", "taskID", taskID) - if err := b.cleanResources(ctx, taskID); err != nil { - b.log.Error("reconcile: failed to clean orphaned resources", "taskID", taskID, "error", err) - } - delete(failedJobEvents, taskID) - } - } - } - } -} - // 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. diff --git a/database/postgres/tes.go b/database/postgres/tes.go index 8d4f87645..d1a87bbed 100644 --- a/database/postgres/tes.go +++ b/database/postgres/tes.go @@ -156,6 +156,7 @@ func (db *Postgres) ListTasks(ctx context.Context, req *tes.ListTasksRequest) (* args = append(args, pageSize) selectSQL := fmt.Sprintf("SELECT data FROM tasks %s %s %s", whereClause, orderByClause, limitClause) + rows, err := db.client.Query(ctx, selectSQL, args...) if err != nil { return nil, err diff --git a/worker/kubernetes.go b/worker/kubernetes.go index 1e27e650f..577d06f35 100644 --- a/worker/kubernetes.go +++ b/worker/kubernetes.go @@ -330,23 +330,6 @@ func (kcmd KubernetesCommand) Run(ctx context.Context) error { return nil } -func isWorkerLastAttempt(ctx context.Context, taskID, namespace string) (bool, error) { - - clientset, err := getKubernetesClientset() - job, err := clientset.BatchV1().Jobs(namespace).Get(ctx, taskID, metav1.GetOptions{}) - if err != nil { - return true, err - } - limit := int32(0) - if job.Spec.BackoffLimit != nil { - limit = *job.Spec.BackoffLimit - } - // failed count includes this attempt only after the pod exits, - // so while we're still running, status.Failed reflects *completed* failed attempts. - // If failed >= backoffLimit, the Job controller won't retry us. - return job.Status.Failed >= limit, nil -} - // streamPodLogs streams logs from a pod regardless of its state // This works for Running, Succeeded, and Failed pods (as long as they haven't been deleted) func streamPodLogs(ctx context.Context, namespace string, podName string, stdout io.Writer, stderr io.Writer) error { From 9cb2bd983297410888fe69f9f36874207b0134e4 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Mon, 22 Jun 2026 12:55:03 -0700 Subject: [PATCH 29/35] debug: Integration tests --- .github/workflows/integration_tests_on_kind.yaml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/.github/workflows/integration_tests_on_kind.yaml b/.github/workflows/integration_tests_on_kind.yaml index 7419bbcdc..24e41bb3c 100644 --- a/.github/workflows/integration_tests_on_kind.yaml +++ b/.github/workflows/integration_tests_on_kind.yaml @@ -16,6 +16,11 @@ jobs: QUAY_ORG: ohsu-comp-bio EXTERNAL_TO_CTDS: "true" SERVICE_TO_TEST: gen3_workflow + # Pin the integration test code (gen3-code-vigil) to the last known-good + # commit. On master, conftest.py:180 now calls get_navigation_url(), which + # crashes with KeyError: 'navigation' against the kind portal config. + # 891ea9e is the commit from the last green run (calypr/funnel run 27659443374). + TEST_REPO_BRANCH: "891ea9eaf18fa3ce114483af5cbcdd9356ce31c9" SETUP_SCRIPT_1: https://raw.githubusercontent.com/uc-cdis/gen3-workflow/refs/heads/master/.github/workflows/integration_tests_on_kind/ci_start_kind_cluster.sh SETUP_SCRIPT_2: https://raw.githubusercontent.com/uc-cdis/gen3-workflow/refs/heads/master/.github/workflows/integration_tests_on_kind/ci_override_config_and_start_minio.sh SETUP_SCRIPT_3: https://raw.githubusercontent.com/uc-cdis/gen3-workflow/refs/heads/master/.github/workflows/integration_tests_on_kind/ci_expose_kind_cluster.sh From 544446a73e0b27561e9765a02b2935968d116f48 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Mon, 22 Jun 2026 12:59:13 -0700 Subject: [PATCH 30/35] ci: re-trigger integration tests From ccd2042d58adc371d96d66c103bb582662e704dd Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Mon, 22 Jun 2026 13:02:00 -0700 Subject: [PATCH 31/35] ci: pin gen3-code-vigil to correct green commit 69c620a 891ea9e was the funnel merge commit, not gen3-code-vigil, so the checkout failed with 'not our ref'. 69c620a (Gen3SDK Tests #558) is the gen3-code-vigil commit from the last green run, and its conftest.py predates the get_navigation_url() call that crashes on master. Co-Authored-By: Claude Opus 4.8 Assisted-by: Claude Code:claude [Claude Code] --- .github/workflows/integration_tests_on_kind.yaml | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/.github/workflows/integration_tests_on_kind.yaml b/.github/workflows/integration_tests_on_kind.yaml index 24e41bb3c..fee2501d5 100644 --- a/.github/workflows/integration_tests_on_kind.yaml +++ b/.github/workflows/integration_tests_on_kind.yaml @@ -19,8 +19,10 @@ jobs: # Pin the integration test code (gen3-code-vigil) to the last known-good # commit. On master, conftest.py:180 now calls get_navigation_url(), which # crashes with KeyError: 'navigation' against the kind portal config. - # 891ea9e is the commit from the last green run (calypr/funnel run 27659443374). - TEST_REPO_BRANCH: "891ea9eaf18fa3ce114483af5cbcdd9356ce31c9" + # 69c620a ("Gen3SDK Tests (#558)") is the gen3-code-vigil commit from the + # last green run (calypr/funnel run 27659443374); its conftest.py predates + # the get_navigation_url call. + TEST_REPO_BRANCH: "69c620a62f3deb4be8ca19a858701519d18a462b" SETUP_SCRIPT_1: https://raw.githubusercontent.com/uc-cdis/gen3-workflow/refs/heads/master/.github/workflows/integration_tests_on_kind/ci_start_kind_cluster.sh SETUP_SCRIPT_2: https://raw.githubusercontent.com/uc-cdis/gen3-workflow/refs/heads/master/.github/workflows/integration_tests_on_kind/ci_override_config_and_start_minio.sh SETUP_SCRIPT_3: https://raw.githubusercontent.com/uc-cdis/gen3-workflow/refs/heads/master/.github/workflows/integration_tests_on_kind/ci_expose_kind_cluster.sh From 3968358327af83de6f16d45e9beecfc29b09ff80 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Mon, 22 Jun 2026 13:48:12 -0700 Subject: [PATCH 32/35] debug: CI --- .github/workflows/integration_tests_on_kind.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/integration_tests_on_kind.yaml b/.github/workflows/integration_tests_on_kind.yaml index fee2501d5..461b8b07b 100644 --- a/.github/workflows/integration_tests_on_kind.yaml +++ b/.github/workflows/integration_tests_on_kind.yaml @@ -11,7 +11,7 @@ permissions: jobs: integration_tests: name: Integration tests - uses: uc-cdis/.github/.github/workflows/integration_tests.yaml@28f8020ea479e0b79e9a86bdfb9da6566806cf3b + uses: uc-cdis/.github/.github/workflows/integration_tests.yaml@master with: QUAY_ORG: ohsu-comp-bio EXTERNAL_TO_CTDS: "true" From f2895a09d7497daa2cfa88dab6d06d026baa5afa Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Mon, 22 Jun 2026 14:10:23 -0700 Subject: [PATCH 33/35] fix: Cleanup code Signed-off-by: Liam Beckman --- compute/kubernetes/backend.go | 8 -------- 1 file changed, 8 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index c559d58f7..285081e63 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -804,14 +804,6 @@ func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableClea ticker := time.NewTicker(rate) defer ticker.Stop() - // missingJobCounts tracks, per task, the number of consecutive reconcile - // passes in which a non-terminal task has had no matching worker job in - // Kubernetes. A job can be legitimately absent for a short window right after - // submit (the Job API object has not been created yet), so we only treat the - // task as failed after the job has been missing for missingJobThreshold - // consecutive passes. The counter is reset as soon as the job reappears. - missingJobCounts := make(map[string]int) - for { select { case <-ctx.Done(): From 098fb86d6d7410e22dd52e94b8c25fc87e7db714 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Mon, 22 Jun 2026 15:57:05 -0700 Subject: [PATCH 34/35] task: Revert changes to better match PR #1410 Signed-off-by: Liam Beckman --- compute/kubernetes/backend.go | 9 -- compute/kubernetes/backend_test.go | 140 ----------------------------- 2 files changed, 149 deletions(-) diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 285081e63..a50f5ec32 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -479,15 +479,6 @@ const failedCreateThreshold = 5 // rapid-fire bursts in the first few seconds of a job's life. const minFailureSpan = 20 * time.Second -// missingJobThreshold is the number of consecutive reconcile passes a -// non-terminal task may have no matching worker Job in Kubernetes before the -// reconciler marks it SYSTEM_ERROR. This grace window avoids a false positive -// for the brief period between a task being submitted and its Job object being -// created. A worker Job that is deleted out-of-band (e.g. manually, or while its -// pod is still ContainerCreating) leaves the task with no Job indefinitely, so -// after this many misses the task is failed rather than left stuck. See issue #88. -const missingJobThreshold = 3 - // hasJobFailedCreateEvent returns (totalCount, message) when the job has // accumulated enough FailedCreate events spread over enough real time and no // SuccessfulCreate has occurred after the last failure. Returns (0, "") when diff --git a/compute/kubernetes/backend_test.go b/compute/kubernetes/backend_test.go index 54d29a369..82c43450a 100644 --- a/compute/kubernetes/backend_test.go +++ b/compute/kubernetes/backend_test.go @@ -670,18 +670,11 @@ func TestHasJobFailedCreateEvent(t *testing.T) { type mockReadOnlyServer struct { mu sync.Mutex tasks []*tes.Task - // onQueuedList, if set, is invoked each time ListTasks is called for the - // QUEUED state. Tests use it to count reconcile passes and cancel the - // context deterministically after a known number of passes. - onQueuedList func() } func (m *mockReadOnlyServer) ListTasks(_ context.Context, req *tes.ListTasksRequest) (*tes.ListTasksResponse, error) { m.mu.Lock() defer m.mu.Unlock() - if req.State == tes.State_QUEUED && m.onQueuedList != nil { - m.onQueuedList() - } var out []*tes.Task for _, t := range m.tasks { if t.State == req.State { @@ -732,34 +725,6 @@ func (c *capturingEventWriter) hasSystemError(taskID string) bool { return false } -// systemErrorCount returns the number of SYSTEM_ERROR state events written for taskID. -func (c *capturingEventWriter) systemErrorCount(taskID string) int { - c.mu.Lock() - defer c.mu.Unlock() - n := 0 - for _, ev := range c.events { - if ev.Id == taskID && ev.Type == events.Type_TASK_STATE && ev.GetState() == tes.State_SYSTEM_ERROR { - n++ - } - } - return n -} - -// hasSystemLog reports whether a SYSTEM_LOG event whose message contains substr -// was written for taskID. -func (c *capturingEventWriter) hasSystemLog(taskID, substr string) bool { - c.mu.Lock() - defer c.mu.Unlock() - for _, ev := range c.events { - if ev.Id == taskID && ev.Type == events.Type_SYSTEM_LOG { - if sl := ev.GetSystemLog(); sl != nil && strings.Contains(sl.Msg, substr) { - return true - } - } - } - return false -} - // TestReconcile_ZeroStatusFailedCreate verifies that a Job with all-zero // status counters (Active=0, Succeeded=0, Failed=0) but with a FailedCreate // event — as produced by Pod Security Admission enforcement — is detected by @@ -862,111 +827,6 @@ func TestReconcile_ZeroStatusFailedCreate(t *testing.T) { } } -// TestReconcile_MissingJobFailsTask verifies that a non-terminal task whose -// worker Job has been deleted out-of-band (e.g. manually, or while the pod was -// still ContainerCreating) is eventually transitioned to SYSTEM_ERROR with a -// system log, rather than being left stuck in QUEUED forever. See issue #88. -func TestReconcile_MissingJobFailsTask(t *testing.T) { - const ns = "test-ns" - const taskID = "test-task-missing-job" - - // No worker Job exists in Kubernetes, but the task is QUEUED in the DB. - fakeClient := fake.NewSimpleClientset() - - db := &mockReadOnlyServer{ - tasks: []*tes.Task{ - {Id: taskID, State: tes.State_QUEUED}, - }, - } - evWriter := &capturingEventWriter{} - - // Cancel a few passes after the threshold is crossed so we can also assert - // the SYSTEM_ERROR is emitted exactly once (not re-fired every pass). - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - var passes int - db.onQueuedList = func() { - passes++ - if passes >= missingJobThreshold+2 { - cancel() - } - } - - conf := config.DefaultConfig() - conf.Kubernetes.JobsNamespace = ns - - b := &Backend{ - client: fakeClient, - event: evWriter, - database: db, - log: logger.NewLogger("test", logger.DefaultConfig()), - conf: conf, - } - - b.reconcile(ctx, 20*time.Millisecond, true /* disableCleanup */) - - if !evWriter.hasSystemError(taskID) { - t.Errorf("expected SYSTEM_ERROR for task with missing job, got events: %+v", evWriter.events) - } - if !evWriter.hasSystemLog(taskID, "no longer exists") { - t.Errorf("expected SYSTEM_LOG describing missing job, got events: %+v", evWriter.events) - } - if n := evWriter.systemErrorCount(taskID); n != 1 { - t.Errorf("expected exactly one SYSTEM_ERROR event, got %d", n) - } -} - -// TestReconcile_MissingJobGracePeriod verifies that a non-terminal task with no -// worker Job is NOT failed within the grace window — this is the expected, -// transient state immediately after submit (before the Job object is created). -// The task must not be marked SYSTEM_ERROR until missingJobThreshold consecutive -// passes have observed the missing job. -func TestReconcile_MissingJobGracePeriod(t *testing.T) { - const ns = "test-ns" - const taskID = "test-task-grace" - - fakeClient := fake.NewSimpleClientset() - - evWriter := &capturingEventWriter{} - db := &mockReadOnlyServer{ - tasks: []*tes.Task{ - {Id: taskID, State: tes.State_QUEUED}, - }, - } - - // Cancel the context after the second reconcile pass (2 < missingJobThreshold), - // so the task should still be within its grace window and not yet failed. - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - var passes int - db.onQueuedList = func() { - passes++ - if passes >= 2 { - cancel() - } - } - - conf := config.DefaultConfig() - conf.Kubernetes.JobsNamespace = ns - - b := &Backend{ - client: fakeClient, - event: evWriter, - database: db, - log: logger.NewLogger("test", logger.DefaultConfig()), - conf: conf, - } - - b.reconcile(ctx, 20*time.Millisecond, true /* disableCleanup */) - - if passes >= missingJobThreshold { - t.Skipf("reconcile ran %d passes before cancel; cannot assert grace behavior for threshold %d", passes, missingJobThreshold) - } - if evWriter.hasSystemError(taskID) { - t.Errorf("task was failed within grace window (%d passes, threshold %d): %+v", passes, missingJobThreshold, evWriter.events) - } -} - func TestFetchPodWarningEvents(t *testing.T) { const ns = "test-ns" const jobName = "test-job" From 60c1bf484881bf5a10a2dd47882d2b9e876bb762 Mon Sep 17 00:00:00 2001 From: Liam Beckman Date: Thu, 25 Jun 2026 21:02:48 -0700 Subject: [PATCH 35/35] chore: Unpin integration tests action --- .github/workflows/integration_tests_on_kind.yaml | 7 ------- 1 file changed, 7 deletions(-) diff --git a/.github/workflows/integration_tests_on_kind.yaml b/.github/workflows/integration_tests_on_kind.yaml index 461b8b07b..173157df1 100644 --- a/.github/workflows/integration_tests_on_kind.yaml +++ b/.github/workflows/integration_tests_on_kind.yaml @@ -16,13 +16,6 @@ jobs: QUAY_ORG: ohsu-comp-bio EXTERNAL_TO_CTDS: "true" SERVICE_TO_TEST: gen3_workflow - # Pin the integration test code (gen3-code-vigil) to the last known-good - # commit. On master, conftest.py:180 now calls get_navigation_url(), which - # crashes with KeyError: 'navigation' against the kind portal config. - # 69c620a ("Gen3SDK Tests (#558)") is the gen3-code-vigil commit from the - # last green run (calypr/funnel run 27659443374); its conftest.py predates - # the get_navigation_url call. - TEST_REPO_BRANCH: "69c620a62f3deb4be8ca19a858701519d18a462b" SETUP_SCRIPT_1: https://raw.githubusercontent.com/uc-cdis/gen3-workflow/refs/heads/master/.github/workflows/integration_tests_on_kind/ci_start_kind_cluster.sh SETUP_SCRIPT_2: https://raw.githubusercontent.com/uc-cdis/gen3-workflow/refs/heads/master/.github/workflows/integration_tests_on_kind/ci_override_config_and_start_minio.sh SETUP_SCRIPT_3: https://raw.githubusercontent.com/uc-cdis/gen3-workflow/refs/heads/master/.github/workflows/integration_tests_on_kind/ci_expose_kind_cluster.sh