fix: properly clean up finished jobs and return failure when max-age exceeded (#1128)

* Skip jobs from hibernating due to culling config

* Update timeouts on job reconcile

* Extract failure message from job, skip reconciling finished jobs

* Not setting ttlSecondsAfterFinished for jobs

the job is deleted with the amalthea session; it is part of the
`needsDeletion` function determining when to remove a session.

The job sets its idleSince time when completed.

Author: eikek

api/v1alpha1/amaltheasession_children.go

@@ -180,13 +180,6 @@ func (cr *AmaltheaSession) Job(cfg config.AmaltheaSessionConfiguration) (batchv1
 		activeTTL = ptr.To(int64(cr.Spec.Culling.MaxAge.Seconds()))
 	}
 
-	// use MaxHibernatedDuration (amount of time until a hibernated session is deleted) to set the time
-	// after which the job is removed after it has completed
-	var ttlFinish *int32
-	if cr.Spec.Culling.MaxHibernatedDuration.Seconds() > 0 {
-		ttlFinish = ptr.To(int32(cr.Spec.Culling.MaxHibernatedDuration.Seconds()))
-	}
-
 	job := batchv1.Job{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:        cr.JobName(),
@@ -195,12 +188,11 @@ func (cr *AmaltheaSession) Job(cfg config.AmaltheaSessionConfiguration) (batchv1
 			Annotations: cr.Spec.Template.Metadata.Annotations,
 		},
 		Spec: batchv1.JobSpec{
-			Parallelism:             ptr.To(int32(1)),
-			Completions:             ptr.To(int32(1)),
-			BackoffLimit:            ptr.To(int32(0)),
-			ActiveDeadlineSeconds:   activeTTL,
-			TTLSecondsAfterFinished: ttlFinish,
-			Suspend:                 &cr.Spec.Hibernated,
+			Parallelism:           ptr.To(int32(1)),
+			Completions:           ptr.To(int32(1)),
+			BackoffLimit:          ptr.To(int32(0)),
+			ActiveDeadlineSeconds: activeTTL,
+			Suspend:               &cr.Spec.Hibernated,
 			Template: v1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels:      cr.childLabels(),

internal/controller/children.go

@@ -301,6 +301,11 @@ func (c ChildResource[T]) Reconcile(ctx context.Context, clnt client.Client, cr
 			if !ok {
 				return fmt.Errorf("could not cast when reconciling")
 			}
+			// finished jobs are not updated
+			if jobIsFinished(current.Status) {
+				log.Info("Job is terminated, not reconciling", "job", current.Name)
+				return nil
+			}
 			if current.CreationTimestamp.IsZero() {
 				log.Info("Creating a Job", "job", desired.Spec)
 				current.Spec = desired.Spec
@@ -334,6 +339,8 @@ func (c ChildResource[T]) Reconcile(ctx context.Context, clnt client.Client, cr
 			current.Spec.Template.Spec.NodeSelector = desired.Spec.Template.Spec.NodeSelector
 			current.Spec.Template.Spec.PriorityClassName = desired.Spec.Template.Spec.PriorityClassName
 			current.Spec.Suspend = desired.Spec.Suspend
+			current.Spec.ActiveDeadlineSeconds = desired.Spec.ActiveDeadlineSeconds
+			current.Spec.TTLSecondsAfterFinished = desired.Spec.TTLSecondsAfterFinished
 			switch strategy := cr.Spec.ReconcileStrategy; strategy {
 			case amaltheadevv1alpha1.Never:
 				return nil
@@ -538,6 +545,10 @@ func (c ChildResourceUpdates) failureMessage(pod *v1.Pod) string {
 	if msg != "" {
 		return msg
 	}
+	msg = jobFailureReason(c.Job.Manifest)
+	if msg != "" {
+		return msg
+	}
 
 	return ""
 }

internal/controller/children_conditions.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"slices"
 
+	batchv1 "k8s.io/api/batch/v1"
 	v1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 )
@@ -26,6 +27,22 @@ func handleBadWaitingState(status v1.ContainerStatus) string {
 	return fmt.Sprintf("the container %s is failing because %s", status.Name, waitingState.Reason)
 }
 
+func jobFailureReason(job *batchv1.Job) string {
+	if job == nil {
+		return ""
+	}
+
+	if job.Status.Failed > 0 {
+		for _, cond := range job.Status.Conditions {
+			if (cond.Type == batchv1.JobFailed || cond.Type == batchv1.JobFailureTarget) && cond.Status == v1.ConditionTrue {
+				return cond.Reason + ": " + cond.Message
+			}
+		}
+		return "The job failed."
+	}
+	return ""
+}
+
 func podFailureReason(pod *v1.Pod) string {
 	if pod == nil {
 		return ""

internal/controller/cull.go

@@ -30,6 +30,12 @@ import (
 )
 
 func updateHibernationState(ctx context.Context, r *AmaltheaSessionReconciler, amaltheasession *amaltheadevv1alpha1.AmaltheaSession) error {
+	if amaltheasession.Spec.SessionType == amaltheadevv1alpha1.SessionTypeNonInteractive {
+		// non-interactive sessions (jobs) never get auto-scaled down by amalthea;
+		// this is managed by k8s via the activeDeadlineSeconds setting for jobs
+		return nil
+	}
+
 	status := amaltheasession.Status
 	log := log.FromContext(ctx)
 	if !amaltheasession.Spec.Hibernated {
@@ -40,7 +46,7 @@ func updateHibernationState(ctx context.Context, r *AmaltheaSessionReconciler, a
 			if err != nil {
 				return err
 			}
-			log.Info("statefulSet scaled down or job suspended")
+			log.Info("statefulSet scaled down")
 		}
 	}
 	return nil

internal/controller/status.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	amaltheadevv1alpha1 "github.com/SwissDataScienceCenter/amalthea/api/v1alpha1"
+	batchv1 "k8s.io/api/batch/v1"
 	v1 "k8s.io/api/core/v1"
 	resource "k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -99,6 +100,15 @@ func podIsCompleted(pod *v1.Pod) bool {
 	return phaseCompleted && totalCnt == completedCnt
 }
 
+func jobIsFinished(job batchv1.JobStatus) bool {
+	for _, c := range job.Conditions {
+		if (c.Type == batchv1.JobComplete || c.Type == batchv1.JobFailed) && c.Status == v1.ConditionTrue {
+			return true
+		}
+	}
+	return false
+}
+
 func metrics(ctx context.Context, clnt metricsv1beta1.PodMetricsesGetter, cr *amaltheadevv1alpha1.AmaltheaSession) (v1.ResourceList, error) {
 	podName := fmt.Sprintf("%s-0", cr.Name)
 	podMetricses, err := clnt.PodMetricses(cr.Namespace).List(