diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index 7fd0a3337..a6d334e96 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) } @@ -324,6 +329,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 @@ -401,14 +421,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 } @@ -432,14 +446,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 @@ -557,296 +564,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 | -// -// 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(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" + } - for { - select { - case <-ctx.Done(): + 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 + 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 { - 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(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) } } } @@ -904,7 +857,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 { @@ -917,7 +870,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 { @@ -940,21 +893,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/resources/job.go b/compute/kubernetes/resources/job.go index bd40069b8..79f8657f3 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 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 } 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 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) }