Add interruptible override to workflow execution config #minor (flyte…

…org#410) * Fixed minor log message formatting Signed-off-by: Nick Müller <[email protected]> * Added override for interruptible flag to execution config Signed-off-by: Nick Müller <[email protected]> * Interruptible override now uses BoolValue wrapper Allows for distinguishment between no value being provided and an override to false Signed-off-by: Nick Müller <[email protected]> * Test comments cleanup Signed-off-by: Nick Müller <[email protected]> * Use released flyteidl/flytepropeller versions Signed-off-by: Nick Müller <[email protected]>
eapolinario · May 3, 2022 · ed6056c · ed6056c
1 parent acfb63b
commit ed6056c
Show file tree

Hide file tree

Showing 12 changed files with 335 additions and 17 deletions.
diff --git a/flyteadmin/go.mod b/flyteadmin/go.mod
@@ -13,9 +13,9 @@ require (
 	github.com/cloudevents/sdk-go/v2 v2.8.0
 	github.com/coreos/go-oidc v2.2.1+incompatible
 	github.com/evanphx/json-patch v4.9.0+incompatible
-	github.com/flyteorg/flyteidl v1.0.0
+	github.com/flyteorg/flyteidl v1.1.0
 	github.com/flyteorg/flyteplugins v1.0.0
-	github.com/flyteorg/flytepropeller v1.0.0
+	github.com/flyteorg/flytepropeller v1.1.0
 	github.com/flyteorg/flytestdlib v1.0.0
 	github.com/flyteorg/stow v0.3.5
 	github.com/ghodss/yaml v1.0.0

diff --git a/flyteadmin/go.sum b/flyteadmin/go.sum
@@ -380,12 +380,13 @@ github.com/fatih/structs v1.1.0/go.mod h1:9NiDSp5zOcgEDl+j00MP/WkGVPOlPRLejGD8Ga
 github.com/fatih/structtag v1.2.0/go.mod h1:mBJUNpUnHmRKrKlQQlmCrh5PuhftFbNv8Ys4/aAZl94=
 github.com/felixge/httpsnoop v1.0.1 h1:lvB5Jl89CsZtGIWuTcDM1E/vkVs49/Ml7JJe07l8SPQ=
 github.com/felixge/httpsnoop v1.0.1/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
-github.com/flyteorg/flyteidl v1.0.0 h1:02V/h8cN3TzI6H9kzB2XNKR4XsJDmsGGfDWxbfmRZGs=
 github.com/flyteorg/flyteidl v1.0.0/go.mod h1:JW0z1ZaHS9zWvDAwSMIyGhsf+V4zrzBBgh5IuqzMFCM=
+github.com/flyteorg/flyteidl v1.1.0 h1:f8tdMXOuorS/d+4Ut2QarfDbdCOriK0S+EnlQzrwz9E=
+github.com/flyteorg/flyteidl v1.1.0/go.mod h1:JW0z1ZaHS9zWvDAwSMIyGhsf+V4zrzBBgh5IuqzMFCM=
 github.com/flyteorg/flyteplugins v1.0.0 h1:77hUJjiIxBmQ9rd3+cXjSGnzOVAFrSzCd59aIaYFB/8=
 github.com/flyteorg/flyteplugins v1.0.0/go.mod h1:4Cpn+9RfanIieTTh2XsuL6zPYXtsR5UDe8YaEmXONT4=
-github.com/flyteorg/flytepropeller v1.0.0 h1:INabpERWpJnNQWX0nw1aKeipiLayHv82MtP9Dtg1S1A=
-github.com/flyteorg/flytepropeller v1.0.0/go.mod h1:OuegUOqqmhtoMVVhsmGBpT1Am0eutzjcqJk7z4bL29E=
+github.com/flyteorg/flytepropeller v1.1.0 h1:R4iVoznTtpa10VC+CqCi4Wal/Yc+cng3dOi7hCLpi+Y=
+github.com/flyteorg/flytepropeller v1.1.0/go.mod h1:QkWYTYZ3lrCY5nXKrYkVwbaGB2IhfzDmIGAWeWzuoak=
 github.com/flyteorg/flytestdlib v1.0.0 h1:gb99ignMsVcNTUmWzArtcIDdkRjyzQQVBkWNOQakiFg=
 github.com/flyteorg/flytestdlib v1.0.0/go.mod h1:QSVN5wIM1lM9d60eAEbX7NwweQXW96t5x4jbyftn89c=
 github.com/flyteorg/stow v0.3.3/go.mod h1:HBld7ud0i4khMHwJjkO8v+NSP7ddKa/ruhf4I8fliaA=

diff --git a/flyteadmin/pkg/manager/impl/execution_manager.go b/flyteadmin/pkg/manager/impl/execution_manager.go
@@ -51,6 +51,7 @@ import (
 	"github.com/benbjohnson/clock"
 	"github.com/flyteorg/flyteadmin/pkg/manager/impl/shared"
 	"github.com/golang/protobuf/proto"
+	"github.com/golang/protobuf/ptypes/wrappers"
 )
 
 const childContainerQueueKey = "child_queue"
@@ -447,6 +448,8 @@ type WorkflowExecutionConfigInterface interface {
 	GetAnnotations() *admin.Annotations
 	// GetLabels Custom labels to be applied to a triggered execution resource.
 	GetLabels() *admin.Labels
+	// GetInterruptible indicates a workflow should be flagged as interruptible for a single execution. If omitted, the workflow's default is used.
+	GetInterruptible() *wrappers.BoolValue
 }
 
 // Merge into workflowExecConfig from spec and return true if any value has been changed
@@ -483,6 +486,17 @@ func mergeIntoExecConfig(workflowExecConfig *admin.WorkflowExecutionConfig, spec
 		workflowExecConfig.Annotations = spec.GetAnnotations()
 		isChanged = true
 	}
+
+	// Override interruptible flag if workflow execution config does not have a value set or the spec sets a different
+	// value that defined as the workflow default. This allows for workflows to have their interruptible setting
+	// explicitly turned on and off for a single execution.
+	if (workflowExecConfig.GetInterruptible() == nil && spec.GetInterruptible() != nil) ||
+		(workflowExecConfig.GetInterruptible() != nil && spec.GetInterruptible() != nil &&
+			workflowExecConfig.GetInterruptible().GetValue() != spec.GetInterruptible().GetValue()) {
+		workflowExecConfig.Interruptible = spec.GetInterruptible()
+		isChanged = true
+	}
+
 	return isChanged
 }