feat: added when clause for cronworkflows

Signed-off-by: isubasinghe <[email protected]>

docs/fields.md

@@ -1253,6 +1253,7 @@ CronWorkflowSpec is the specification of a CronWorkflow
 |`successfulJobsHistoryLimit`|`integer`|SuccessfulJobsHistoryLimit is the number of successful jobs to be kept at a time|
 |`suspend`|`boolean`|Suspend is a flag that will stop new CronWorkflows from running if set to true|
 |`timezone`|`string`|Timezone is the timezone against which the cron schedule will be calculated, e.g. "Asia/Tokyo". Default is machine's local time.|
+|`when`|`string`|v3.6 and after: When clause can be used to determine a run should or shouldn't be scheduled.|
 |`workflowMetadata`|[`ObjectMeta`](#objectmeta)|WorkflowMetadata contains some metadata of the workflow to be run|
 |`workflowSpec`|[`WorkflowSpec`](#workflowspec)|WorkflowSpec is the spec of the workflow to be run|
 

pkg/apis/workflow/v1alpha1/cron_workflow_types.go

@@ -65,6 +65,9 @@ type CronWorkflowSpec struct {
 	StopStrategy *StopStrategy `json:"stopStrategy,omitempty" protobuf:"bytes,10,opt,name=stopStrategy"`
 	// Schedules is a list of schedules to run the Workflow in Cron format
 	Schedules []string `json:"schedules,omitempty" protobuf:"bytes,11,opt,name=schedules"`
+	// v3.6 and after: When clause can be used to determine a run should or shouldn't be scheduled.
+	// This new When clause allows for the full expressivity of expr-lang.
+	When string `json:"when,omitempty" protobuf:"bytes,12,opt,name=when"`
 }
 
 // v3.6 and after: StopStrategy defines if the CronWorkflow should stop scheduling based on a condition

util/template/replace.go

@@ -15,8 +15,11 @@ func Replace(s string, replaceMap map[string]string, allowUnresolved bool) (stri
 	if err != nil {
 		return "", err
 	}
-
-	replacedString, err := t.Replace(replaceMap, allowUnresolved)
+	interReplaceMap := make(map[string]interface{})
+	for k, v := range replaceMap {
+		interReplaceMap[k] = v
+	}
+	replacedString, err := t.Replace(interReplaceMap, allowUnresolved)
 	if err != nil {
 		return s, err
 	}

util/template/simple_template.go

@@ -11,17 +11,20 @@ import (
 	"github.com/argoproj/argo-workflows/v3/errors"
 )
 
-func simpleReplace(w io.Writer, tag string, replaceMap map[string]string, allowUnresolved bool) (int, error) {
+func simpleReplace(w io.Writer, tag string, replaceMap map[string]interface{}, allowUnresolved bool) (int, error) {
 	replacement, ok := replaceMap[strings.TrimSpace(tag)]
 	if !ok {
 		// Attempt to resolve nested tags, if possible
 		if index := strings.LastIndex(tag, "{{"); index > 0 {
 			nestedTagPrefix := tag[:index]
 			nestedTag := tag[index+2:]
 			if replacement, ok := replaceMap[nestedTag]; ok {
-				replacement = strconv.Quote(replacement)
-				replacement = replacement[1 : len(replacement)-1]
-				return w.Write([]byte("{{" + nestedTagPrefix + replacement))
+				replacement, isStr := replacement.(string)
+				if isStr {
+					replacement = strconv.Quote(replacement)
+					replacement = replacement[1 : len(replacement)-1]
+					return w.Write([]byte("{{" + nestedTagPrefix + replacement))
+				}
 			}
 		}
 		if allowUnresolved {
@@ -31,9 +34,14 @@ func simpleReplace(w io.Writer, tag string, replaceMap map[string]string, allowU
 		}
 		return 0, errors.Errorf(errors.CodeBadRequest, "failed to resolve {{%s}}", tag)
 	}
+
+	replacementStr, isStr := replacement.(string)
+	if !isStr {
+		return 0, errors.Errorf(errors.CodeBadRequest, "failed to resolve {{%s}} to string", tag)
+	}
 	// The following escapes any special characters (e.g. newlines, tabs, etc...)
 	// in preparation for substitution
-	replacement = strconv.Quote(replacement)
-	replacement = replacement[1 : len(replacement)-1]
-	return w.Write([]byte(replacement))
+	replacementStr = strconv.Quote(replacementStr)
+	replacementStr = replacementStr[1 : len(replacementStr)-1]
+	return w.Write([]byte(replacementStr))
 }

util/template/template.go

@@ -15,7 +15,7 @@ const (
 )
 
 type Template interface {
-	Replace(replaceMap map[string]string, allowUnresolved bool) (string, error)
+	Replace(replaceMap map[string]interface{}, allowUnresolved bool) (string, error)
 }
 
 func NewTemplate(s string) (Template, error) {
@@ -30,13 +30,13 @@ type impl struct {
 	*fasttemplate.Template
 }
 
-func (t *impl) Replace(replaceMap map[string]string, allowUnresolved bool) (string, error) {
+func (t *impl) Replace(replaceMap map[string]interface{}, allowUnresolved bool) (string, error) {
 	replacedTmpl := &bytes.Buffer{}
 	_, err := t.Template.ExecuteFunc(replacedTmpl, func(w io.Writer, tag string) (int, error) {
 		kind, expression := parseTag(tag)
 		switch kind {
 		case kindExpression:
-			env := exprenv.GetFuncMap(EnvMap(replaceMap))
+			env := exprenv.GetFuncMap(replaceMap)
 			return expressionReplace(w, expression, env, allowUnresolved)
 		default:
 			return simpleReplace(w, tag, replaceMap, allowUnresolved)

workflow/controller/operator.go

@@ -3526,14 +3526,18 @@ func processItem(tmpl template.Template, name string, index int, item wfv1.Item,
 		return "", errors.Errorf(errors.CodeBadRequest, "withItems[%d] expected string, number, list, or map. received: %v", index, item)
 	}
 	var newStepStr string
+	interReplaceMap := make(map[string]interface{})
+	for k, v := range replaceMap {
+		interReplaceMap[k] = v
+	}
 	// If when is not parameterised and evaluated to false, we are not executing nor resolving artifact,
 	// we allow parameter substitution to be Unresolved
 	// The parameterised when will get handle by the task-expansion
 	proceed, err := shouldExecute(whenCondition)
 	if err == nil && !proceed {
-		newStepStr, err = tmpl.Replace(replaceMap, true)
+		newStepStr, err = tmpl.Replace(interReplaceMap, true)
 	} else {
-		newStepStr, err = tmpl.Replace(replaceMap, false)
+		newStepStr, err = tmpl.Replace(interReplaceMap, false)
 	}
 	if err != nil {
 		return "", err

workflow/cron/operator.go

@@ -7,10 +7,13 @@ import (
 	"sort"
 	"time"
 
+	"github.com/Knetic/govaluate"
 	"github.com/robfig/cron/v3"
 	log "github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+
+	argoerrs "github.com/argoproj/argo-workflows/v3/errors"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/util/retry"
@@ -21,6 +24,7 @@ import (
 	wfextvv1alpha1 "github.com/argoproj/argo-workflows/v3/pkg/client/informers/externalversions/workflow/v1alpha1"
 	errorsutil "github.com/argoproj/argo-workflows/v3/util/errors"
 	"github.com/argoproj/argo-workflows/v3/util/expr/argoexpr"
+	"github.com/argoproj/argo-workflows/v3/util/template"
 	waitutil "github.com/argoproj/argo-workflows/v3/util/wait"
 	"github.com/argoproj/argo-workflows/v3/workflow/common"
 	"github.com/argoproj/argo-workflows/v3/workflow/metrics"
@@ -178,6 +182,67 @@ func (woc *cronWfOperationCtx) patch(ctx context.Context, patch map[string]inter
 	}
 }
 
+// TODO: refactor shouldExecute in steps.go
+func shouldExecute(when string) (bool, error) {
+	if when == "" {
+		return true, nil
+	}
+	expression, err := govaluate.NewEvaluableExpression(when)
+	if err != nil {
+		return false, err
+	}
+
+	result, err := expression.Evaluate(nil)
+	if err != nil {
+		return false, err
+	}
+
+	boolRes, ok := result.(bool)
+	if !ok {
+		return false, argoerrs.Errorf(argoerrs.CodeBadRequest, "Expected boolean evaluation for '%s'. Got %v", when, result)
+	}
+	return boolRes, nil
+}
+
+func evalWhen(cron *v1alpha1.CronWorkflow) (bool, error) {
+	if cron.Spec.When == "" {
+		return true, nil
+	}
+	cronBytes, err := json.Marshal(cron)
+	if err != nil {
+		return false, err
+	}
+	m := make(map[string]interface{})
+	tm := time.Date(0, 1,
+		1, 0, 0, 0, 0, time.UTC)
+
+	m["lastScheduledTimeNull"] = false
+
+	if cron.Status.LastScheduledTime != nil {
+		tm = cron.Status.LastScheduledTime.Time
+		m["lastScheduledTimeNull"] = true
+	}
+
+	m["lastScheduledTime"] = tm
+
+	t, err := template.NewTemplate(string(cronBytes))
+	if err != nil {
+		return false, err
+	}
+
+	newCronStr, err := t.Replace(m, false)
+	if err != nil {
+		return false, err
+	}
+
+	var newCron v1alpha1.CronWorkflow
+	err = json.Unmarshal([]byte(newCronStr), &newCron)
+	if err != nil {
+		return false, err
+	}
+	return shouldExecute(newCron.Spec.When)
+}
+
 func (woc *cronWfOperationCtx) enforceRuntimePolicy(ctx context.Context) (bool, error) {
 	if woc.cronWf.Spec.Suspend {
 		woc.log.Infof("%s is suspended, skipping execution", woc.name)
@@ -189,6 +254,11 @@ func (woc *cronWfOperationCtx) enforceRuntimePolicy(ctx context.Context) (bool,
 		return false, nil
 	}
 
+	canProceed, err := evalWhen(woc.cronWf)
+	if err != nil || !canProceed {
+		return canProceed, err
+	}
+
 	if woc.cronWf.Spec.ConcurrencyPolicy != "" {
 		switch woc.cronWf.Spec.ConcurrencyPolicy {
 		case v1alpha1.AllowConcurrent, "":

workflow/cron/operator_test.go

@@ -386,8 +386,8 @@ func TestMissedScheduleAfterCronScheduleWithForbid(t *testing.T) {
 	var cronWf v1alpha1.CronWorkflow
 	v1alpha1.MustUnmarshal([]byte(forbidMissedSchedule), &cronWf)
 	// StartingDeadlineSeconds is after the current second, so cron should be run
-	//startingDeadlineSeconds := int64(35)
-	//cronWf.Spec.StartingDeadlineSeconds = &startingDeadlineSeconds
+	// startingDeadlineSeconds := int64(35)
+	// cronWf.Spec.StartingDeadlineSeconds = &startingDeadlineSeconds
 	t.Run("ForbiddenWithMissedScheduleAfterCron", func(t *testing.T) {
 		cronWf.Spec.StartingDeadlineSeconds = nil
 		woc := &cronWfOperationCtx{
@@ -676,3 +676,35 @@ func TestRunOutstandingWorkflowsWithMultipleSchedules(t *testing.T) {
 	require.NoError(t, err)
 	assert.True(t, missedExecutionTime.IsZero())
 }
+
+func TestEvaluateWhen(t *testing.T) {
+	var cronWf v1alpha1.CronWorkflow
+	v1alpha1.MustUnmarshal([]byte(scheduledWf), &cronWf)
+
+	cronWf.Spec.When = "{{= lastScheduledTimeNull || ( (now() - lastScheduledTime).Seconds() > 30) }}"
+	result, err := evalWhen(&cronWf)
+	require.NoError(t, err)
+	require.Equal(t, true, result)
+
+	cronWf.Spec.When = "{{= !lastScheduledTimeNull && ( (now() - lastScheduledTime).Seconds() > 30) }}"
+	result, err = evalWhen(&cronWf)
+	require.NoError(t, err)
+	require.Equal(t, false, result)
+
+	cronWf.Status.LastScheduledTime = nil
+	cronWf.Spec.When = "{{= !lastScheduledTimeNull }}"
+	result, err = evalWhen(&cronWf)
+	require.NoError(t, err)
+	require.Equal(t, true, result)
+
+	cronWf.Status.LastScheduledTime = &v1.Time{Time: time.Now().Add(time.Minute * -30)}
+	cronWf.Spec.When = "{{= (now() - lastScheduledTime).Minutes() >= 30 }}"
+	result, err = evalWhen(&cronWf)
+	require.NoError(t, err)
+	require.Equal(t, true, result)
+
+	cronWf.Spec.When = "{{= (now() - lastScheduledTime).Minutes() <  50 }}"
+	result, err = evalWhen(&cronWf)
+	require.NoError(t, err)
+	require.Equal(t, true, result)
+}