diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index c51ee6b6e..a50f5ec32 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -68,6 +68,11 @@ 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) } @@ -325,6 +330,21 @@ func (b *Backend) cleanResources(ctx context.Context, taskId string) error { return errs } +// 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 { + if cond.Status != corev1.ConditionTrue { + continue + } + if cond.Type == v1.JobComplete || cond.Type == v1.JobFailed { + 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 @@ -402,14 +422,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(timeout time.Duration, pods *corev1.PodList) bool { + if len(pods.Items) == 0 { return false } @@ -433,14 +447,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 @@ -472,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 @@ -567,351 +565,242 @@ func (b *Backend) hasJobFailedCreateEvent(ctx context.Context, jobName string) ( return totalCount, latestMsg } -// 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 | -// | QUEUED | MISSING | SYSTEM_ERROR | -// | INITIALIZING | MISSING | SYSTEM_ERROR | -// | RUNNING | MISSING | 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. "MISSING" means the worker Job no -// longer exists in Kubernetes (e.g. it was deleted out-of-band) — after a short -// grace period the task is failed rather than left stuck (see issue #88). -// -// This loop is also used to cleanup successful jobs. -func (b *Backend) reconcile(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. +// 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", + }) + 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 +} + +func (b *Backend) cleanBacklog(ctx context.Context, disableCleanup bool) { 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 - } + return + } + + k8sJobs, err := b.listAllWorkerJobs(ctx) + if err != nil { + b.log.Error("backlog cleanup: listing jobs", err) + return + } + + 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 { + 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) - } - } + // 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 { + 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) + } + + // 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) +} +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) + } } +} - ticker := time.NewTicker(rate) - failedJobEvents := make(map[string]int) +func (b *Backend) writeSystemError(ctx context.Context, jobName string, errAttributes map[string]string, additionalMessage string) { + if additionalMessage == "" { + additionalMessage = "Kubernetes job in FAILED state" + } - // 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) + b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) + b.event.WriteEvent(ctx, events.NewSystemLog( + jobName, 0, 0, "error", additionalMessage, + errAttributes, + )) +} - for { - select { - case <-ctx.Done(): +func (b *Backend) reconcileJob(ctx context.Context, j *v1.Job, disableCleanup bool) { + jobName := j.Name + status := j.Status + schedulingTimeout := b.conf.Kubernetes.Timeout.GetDuration() + + 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: + + // 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) + } + b.writeSystemError(ctx, jobName, map[string]string{"error": errDetail}, "Kubernetes worker pod has a terminal container waiting error") + b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup) 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", - }) + } + + // 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.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(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) + } + + case status.Succeeded > 0: + b.log.Debug("reconcile: reconciled successful job", "taskID", jobName) + 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 + } + b.log.Debug("reconcile: checking is job marked as failed", "taskID", jobName) + 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 + } + 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: listing jobs", err) - continue + b.log.Error("reconcile: marshaling failed job conditions", "taskID", jobName, "error", err) } - - // 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] + errDetails := map[string]string{"error": string(conds)} + if podInfo := b.getFailedPodInfo(ctx, pods); podInfo != "" { + errDetails["executor_error"] = podInfo } + b.writeSystemError(ctx, jobName, errDetails, "") + } - // 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 { - // The task is non-terminal but has no worker Job in - // Kubernetes. This is expected briefly right after submit - // (the Job object is not created yet), so we only act once - // the job has been missing for missingJobThreshold - // consecutive passes. A Job deleted out-of-band while the - // task is non-terminal would otherwise leave it stuck in - // QUEUED/INITIALIZING/RUNNING forever (issue #88). - // - // Once we have crossed the threshold and written the - // SYSTEM_ERROR event we keep the counter pinned (rather than - // deleting it) so that we do not re-emit the event on every - // subsequent pass while the task's terminal state propagates - // and it drops out of the non-terminal list. - if missingJobCounts[taskID] >= missingJobThreshold { - continue - } - - missingJobCounts[taskID]++ - if missingJobCounts[taskID] < missingJobThreshold { - b.log.Debug("reconcile: non-terminal task has no worker job, waiting before failing", - "taskID", taskID, "state", task.State, "misses", missingJobCounts[taskID]) - continue - } - - b.log.Info("reconcile: worker job missing for non-terminal task, marking SYSTEM_ERROR", - "taskID", taskID, "state", task.State, "misses", missingJobCounts[taskID]) - b.event.WriteEvent(ctx, events.NewState(taskID, tes.SystemError)) - b.event.WriteEvent(ctx, events.NewSystemLog( - taskID, 0, 0, "error", - "Kubernetes worker job no longer exists", - map[string]string{"error": "worker job was deleted or never started while the task was non-terminal"}, - )) - // Best-effort cleanup of any remaining task resources - // (PVC/PV/ServiceAccount) left behind by the deleted job. - if !disableCleanup { - if err := b.cleanResources(ctx, taskID); err != nil { - b.log.Error("reconcile: failed to clean resources for missing job", "taskID", taskID, "error", err) - } - } - continue - } - - // The job exists again — clear any prior missing-job streak. - delete(missingJobCounts, taskID) - - 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(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 >= 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)} - if podInfo := b.getFailedPodInfo(ctx, jobName); 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) - } - } + 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 + // 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.writeSystemError(ctx, jobName, map[string]string{"error": reason}, "Kubernetes worker job failed to create pod") + b.cleanResourcesIfEnabled(ctx, jobName, disableCleanup) + } + } +} - // 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) +// 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 { + 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 { + 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 { + 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 + if pageToken == "" { + break + } + time.Sleep(100 * time.Millisecond) + } + } + + // 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.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 + } + b.cleanResourcesIfEnabled(ctx, taskID, disableCleanup) + } + +} + +// reconcile is the ticker-based loop used when ExternalReconciler is false. +func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableCleanup bool) { + ticker := time.NewTicker(rate) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + b.reconcileOnce(ctx, disableCleanup) } } } @@ -969,7 +858,7 @@ func (b *Backend) CleanOrphanedResources(ctx context.Context) { // PVs (cluster-scoped; cannot be owned by a namespaced Job, so must be cleaned explicitly) pvs, err := b.client.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{LabelSelector: fmt.Sprintf("app=funnel,namespace=%s", namespace)}) if err != nil { - b.log.Error("backlog cleanup: listing PVs", err) + b.log.Error("CleanOrphanedResources: listing PVs", "error", err) } else { for _, r := range pvs.Items { if id, ok := r.Labels["taskId"]; ok { @@ -982,7 +871,7 @@ func (b *Backend) CleanOrphanedResources(ctx context.Context) { // if they were created before ownerRef support was added) sas, err := b.client.CoreV1().ServiceAccounts(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel"}) if err != nil { - b.log.Error("backlog cleanup: listing ServiceAccounts", err) + b.log.Error("CleanOrphanedResources: listing ServiceAccounts", "error", err) } else { for _, r := range sas.Items { if id, ok := r.Labels["taskId"]; ok { @@ -1005,21 +894,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) } } 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" diff --git a/compute/kubernetes/resources/job.go b/compute/kubernetes/resources/job.go index faab66bf0..715fc7abb 100644 --- a/compute/kubernetes/resources/job.go +++ b/compute/kubernetes/resources/job.go @@ -111,14 +111,6 @@ func CreateJob(ctx context.Context, task *tes.Task, conf *config.Config, client return nil, fmt.Errorf("failed to decode job spec") } - // 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 { diff --git a/compute/kubernetes/resources/pvc.go b/compute/kubernetes/resources/pvc.go index e42b68586..7dbaf2b99 100644 --- a/compute/kubernetes/resources/pvc.go +++ b/compute/kubernetes/resources/pvc.go @@ -79,6 +79,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 { @@ -100,7 +101,7 @@ func DeletePVC(ctx context.Context, taskID string, namespace string, client kube return 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 @@ -114,6 +115,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 } 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)) } }) } diff --git a/tests/kubernetes/kubernetes_test.go b/tests/kubernetes/kubernetes_test.go index eca7eee21..0e0881f98 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) }