feat: Enhanced Different TTLSecondsAfterFinished depending on if job is in Succeeded, Failed or Error, Fixes

pkg/apis/workflow/v1alpha1/workflow_types.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"hash/fnv"
+
 	apiv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -106,6 +107,13 @@ type WorkflowList struct {
 var _ TemplateGetter = &Workflow{}
 var _ TemplateStorage = &Workflow{}
 
+// TTLStrategy is the strategy for the time to live depending on if the workflow succeded or failed
+type TTLStrategy struct {
+	SecondsAfterCompleted *int32 `json:"secondsAfterCompleted:,omitempty" protobuf:"bytes,1,opt,name=secondsAfterCompleted:"`
+	SecondsAfterSuccess   *int32 `json:"secondsAfterSuccess:,omitempty" protobuf:"bytes,2,opt,name=secondsAfterSuccess:"`
+	SecondsAfterFailed    *int32 `json:"secondsAfterFailed:,omitempty" protobuf:"bytes,3,opt,name=secondsAfterFailed"`
+}
+
 // WorkflowSpec is the specification of a Workflow.
 type WorkflowSpec struct {
 	// Templates is a list of workflow templates used in a workflow
@@ -199,8 +207,16 @@ type WorkflowSpec struct {
 	// deleted after ttlSecondsAfterFinished expires. If this field is unset,
 	// ttlSecondsAfterFinished will not expire. If this field is set to zero,
 	// ttlSecondsAfterFinished expires immediately after the Workflow finishes.
+	// DEPRECATED: Use TTLStrategy.SecondsAfterCompleted instead.
 	TTLSecondsAfterFinished *int32 `json:"ttlSecondsAfterFinished,omitempty" protobuf:"bytes,18,opt,name=ttlSecondsAfterFinished"`
 
+	// TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it
+	// Succeeded or Failed. If this struct is set, once the Workflow finishes, it will be
+	// deleted after the time to live expires. If this field is unset,
+	// the controller config map will hold the default values
+	// Update
+	TTLStrategy *TTLStrategy `json:"ttlStrategy,omitempty" protobuf:"bytes,28,opt,name=ttlStrategy"`
+
 	// Optional duration in seconds relative to the workflow start time which the workflow is
 	// allowed to run before the controller terminates the workflow. A value of zero is used to
 	// terminate a Running workflow
@@ -850,7 +866,7 @@ func isCompletedPhase(phase NodePhase) bool {
 		phase == NodeSkipped
 }
 
-// Remove returns whether or not the workflow has completed execution
+// Completed returns whether or not the workflow has completed execution
 func (ws *WorkflowStatus) Completed() bool {
 	return isCompletedPhase(ws.Phase)
 }
@@ -860,6 +876,11 @@ func (ws *WorkflowStatus) Successful() bool {
 	return ws.Phase == NodeSucceeded
 }
 
+// Failed return whether or not the workflow has failed
+func (ws *WorkflowStatus) Failed() bool {
+	return ws.Phase == NodeFailed
+}
+
 // Remove returns whether or not the node has completed execution
 func (n NodeStatus) Completed() bool {
 	return isCompletedPhase(n.Phase) || n.IsDaemoned() && n.Phase != NodePending

workflow/ttlcontroller/ttlcontroller.go

@@ -199,24 +199,44 @@ func (c *Controller) deleteWorkflow(key string) error {
 
 func (c *Controller) ttlExpired(wf *wfv1.Workflow) bool {
 	// We don't care about the Workflows that are going to be deleted, or the ones that don't need clean up.
-	if wf.DeletionTimestamp != nil || wf.Spec.TTLSecondsAfterFinished == nil || wf.Status.FinishedAt.IsZero() {
+	if wf.DeletionTimestamp != nil || wf.Spec.TTLStrategy == nil || wf.Status.FinishedAt.IsZero() {
 		return false
 	}
 	now := c.clock.Now()
-	expiry := wf.Status.FinishedAt.Add(time.Second * time.Duration(*wf.Spec.TTLSecondsAfterFinished))
-	return now.After(expiry)
+	if wf.Status.Failed() && wf.Spec.TTLStrategy != nil && wf.Spec.TTLStrategy.SecondsAfterFailed != nil {
+		expiry := wf.Status.FinishedAt.Add(time.Second * time.Duration(*wf.Spec.TTLStrategy.SecondsAfterFailed))
+		return now.After(expiry)
+	} else if wf.Status.Successful() && wf.Spec.TTLStrategy != nil && wf.Spec.TTLStrategy.SecondsAfterSuccess != nil {
+		expiry := wf.Status.FinishedAt.Add(time.Second * time.Duration(*wf.Spec.TTLStrategy.SecondsAfterSuccess))
+		return now.After(expiry)
+	} else {
+		expiry := wf.Status.FinishedAt.Add(time.Second * time.Duration(*wf.Spec.TTLStrategy.SecondsAfterCompleted))
+		return now.After(expiry)
+	}
 }
 
 func timeLeft(wf *wfv1.Workflow, since *time.Time) (*time.Duration, *time.Time) {
-	if wf.DeletionTimestamp != nil || wf.Spec.TTLSecondsAfterFinished == nil || wf.Status.FinishedAt.IsZero() {
+	if wf.DeletionTimestamp != nil || wf.Spec.TTLStrategy == nil || wf.Status.FinishedAt.IsZero() {
 		return nil, nil
 	}
 	sinceUTC := since.UTC()
 	finishAtUTC := wf.Status.FinishedAt.UTC()
 	if finishAtUTC.After(sinceUTC) {
 		log.Infof("Warning: Found Workflow %s/%s finished in the future. This is likely due to time skew in the cluster. Workflow cleanup will be deferred.", wf.Namespace, wf.Name)
 	}
-	expireAtUTC := finishAtUTC.Add(time.Duration(*wf.Spec.TTLSecondsAfterFinished) * time.Second)
-	remaining := expireAtUTC.Sub(sinceUTC)
-	return &remaining, &expireAtUTC
+	if wf.Status.Failed() && wf.Spec.TTLStrategy.SecondsAfterFailed != nil {
+		expireAtUTC := finishAtUTC.Add(time.Duration(*wf.Spec.TTLStrategy.SecondsAfterFailed) * time.Second)
+		remaining := expireAtUTC.Sub(sinceUTC)
+		return &remaining, &expireAtUTC
+	} else if wf.Status.Successful() && wf.Spec.TTLStrategy.SecondsAfterSuccess != nil {
+		expireAtUTC := finishAtUTC.Add(time.Duration(*wf.Spec.TTLStrategy.SecondsAfterSuccess) * time.Second)
+		remaining := expireAtUTC.Sub(sinceUTC)
+		return &remaining, &expireAtUTC
+	} else if wf.Spec.TTLStrategy.SecondsAfterCompleted != nil {
+		expireAtUTC := finishAtUTC.Add(time.Duration(*wf.Spec.TTLStrategy.SecondsAfterCompleted) * time.Second)
+		remaining := expireAtUTC.Sub(sinceUTC)
+		return &remaining, &expireAtUTC
+	} else {
+		return nil, nil
+	}
 }