Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 5 additions & 13 deletions api/v1alpha1/amaltheasession_children.go
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand All @@ -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(),
Expand Down
11 changes: 11 additions & 0 deletions internal/controller/children.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 ""
}
Expand Down
17 changes: 17 additions & 0 deletions internal/controller/children_conditions.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand All @@ -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 ""
Expand Down
8 changes: 7 additions & 1 deletion internal/controller/cull.go
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,12 @@ import (
)

func updateHibernationState(ctx context.Context, r *AmaltheaSessionReconciler, amaltheasession *amaltheadevv1alpha1.AmaltheaSession) error {
Comment thread
eikek marked this conversation as resolved.
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 {
Expand All @@ -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
Expand Down
10 changes: 10 additions & 0 deletions internal/controller/status.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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(
Expand Down
Loading