Implement workflow execution recovery (flyteorg#290)

eapolinario · Jul 20, 2021 · 59069d2 · 59069d2
1 parent dc5b435
commit 59069d2
Show file tree

Hide file tree

Showing 61 changed files with 1,269 additions and 760 deletions.
diff --git a/go.mod b/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/DiSiqueira/GoTree v1.0.1-0.20180907134536-53a8e837f295
 	github.com/benlaurie/objecthash v0.0.0-20180202135721-d1e3d6079fc1
 	github.com/fatih/color v1.10.0
-	github.com/flyteorg/flyteidl v0.19.5
+	github.com/flyteorg/flyteidl v0.19.14
 	github.com/flyteorg/flyteplugins v0.5.59
 	github.com/flyteorg/flytestdlib v0.3.27
 	github.com/ghodss/yaml v1.0.0

diff --git a/go.sum b/go.sum
@@ -230,8 +230,8 @@ github.com/fatih/color v1.10.0 h1:s36xzo75JdqLaaWoiEHk767eHiwo0598uUxyfiPkDsg=
 github.com/fatih/color v1.10.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM=
 github.com/fatih/structtag v1.2.0/go.mod h1:mBJUNpUnHmRKrKlQQlmCrh5PuhftFbNv8Ys4/aAZl94=
 github.com/flyteorg/flyteidl v0.19.2/go.mod h1:576W2ViEyjTpT+kEVHAGbrTP3HARNUZ/eCwrNPmdx9U=
-github.com/flyteorg/flyteidl v0.19.5 h1:qNhNK6mhCTuOms7zJmBtog6bLQJhBj+iScf1IlHdqeg=
-github.com/flyteorg/flyteidl v0.19.5/go.mod h1:576W2ViEyjTpT+kEVHAGbrTP3HARNUZ/eCwrNPmdx9U=
+github.com/flyteorg/flyteidl v0.19.14 h1:OLg2eT9uYllcfMMjEZJoXQ+2WXcrNbUxD+yaCrz2AlI=
+github.com/flyteorg/flyteidl v0.19.14/go.mod h1:576W2ViEyjTpT+kEVHAGbrTP3HARNUZ/eCwrNPmdx9U=
 github.com/flyteorg/flyteplugins v0.5.59 h1:Uw1xlrlx5rSTpdTMwJTo7mbqHI7X7p7CFVm3473iRjo=
 github.com/flyteorg/flyteplugins v0.5.59/go.mod h1:nesnW7pJhXEysFQg9TnSp36ao33ie0oA/TI4sYPaeyw=
 github.com/flyteorg/flytestdlib v0.3.13/go.mod h1:Tz8JCECAbX6VWGwFT6cmEQ+RJpZ/6L9pswu3fzWs220=

diff --git a/pkg/apis/flyteworkflow/v1alpha1/execution_config.go b/pkg/apis/flyteworkflow/v1alpha1/execution_config.go
@@ -1,6 +1,8 @@
 package v1alpha1
 
-import "github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
+import (
+	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
+)
 
 // This contains an OutputLocationPrefix. When running against AWS, this should be something of the form
 // s3://my-bucket, or s3://my-bucket/  A sharding string will automatically be appended to this prefix before
@@ -20,6 +22,8 @@ type ExecutionConfig struct {
 	TaskPluginImpls map[string]TaskPluginOverride
 	// Can be used to control the number of parallel nodes to run within the workflow. This is useful to achieve fairness.
 	MaxParallelism uint32
+	// Defines execution behavior for processing nodes.
+	RecoveryExecution WorkflowExecutionIdentifier
 }
 
 type TaskPluginOverride struct {

diff --git a/pkg/apis/flyteworkflow/v1alpha1/iface.go b/pkg/apis/flyteworkflow/v1alpha1/iface.go
@@ -64,6 +64,7 @@ const (
 	NodePhaseTimingOut
 	NodePhaseTimedOut
 	NodePhaseDynamicRunning
+	NodePhaseRecovered
 )
 
 func (p NodePhase) String() string {
@@ -92,6 +93,8 @@ func (p NodePhase) String() string {
 		return "RetryableFailure"
 	case NodePhaseDynamicRunning:
 		return "DynamicRunning"
+	case NodePhaseRecovered:
+		return "NodePhaseRecovered"
 	}
 
 	return "Unknown"

diff --git a/pkg/apis/flyteworkflow/v1alpha1/node_status.go b/pkg/apis/flyteworkflow/v1alpha1/node_status.go
@@ -397,7 +397,7 @@ func (in *NodeStatus) GetMessage() string {
 }
 
 func IsPhaseTerminal(phase NodePhase) bool {
-	return phase == NodePhaseSucceeded || phase == NodePhaseFailed || phase == NodePhaseSkipped || phase == NodePhaseTimedOut
+	return phase == NodePhaseSucceeded || phase == NodePhaseFailed || phase == NodePhaseSkipped || phase == NodePhaseTimedOut || phase == NodePhaseRecovered
 }
 
 func (in *NodeStatus) GetOrCreateTaskStatus() MutableTaskNodeStatus {
@@ -576,7 +576,7 @@ func (in *NodeStatus) GetNodeExecutionStatus(ctx context.Context, id NodeID) Exe
 }
 
 func (in *NodeStatus) IsTerminated() bool {
-	return in.GetPhase() == NodePhaseFailed || in.GetPhase() == NodePhaseSkipped || in.GetPhase() == NodePhaseSucceeded
+	return in.GetPhase() == NodePhaseFailed || in.GetPhase() == NodePhaseSkipped || in.GetPhase() == NodePhaseSucceeded || in.GetPhase() == NodePhaseRecovered
 }
 
 func (in *NodeStatus) GetDataDir() DataReference {

diff --git a/pkg/compiler/common/mocks/workflow_builder.go b/pkg/compiler/common/mocks/workflow_builder.go
diff --git a/pkg/compiler/test/testdata/app-workflows-work-one-python-task-w-f.pb b/pkg/compiler/test/testdata/app-workflows-work-one-python-task-w-f.pb
diff --git a/pkg/compiler/test/testdata/app-workflows-work-one-python-task-w-f.yaml b/pkg/compiler/test/testdata/app-workflows-work-one-python-task-w-f.yaml
@@ -24,6 +24,12 @@ tasks:
       value: testValue2
     - key: testKey3
       value: testValue3
+    - key: testKey1
+      value: testValue1
+    - key: testKey2
+      value: testValue2
+    - key: testKey3
+      value: testValue3
     image: myflytecontainer:abc123
     resources: {}
   id:

diff --git a/pkg/compiler/test/testdata/branch/k8s/5_myapp.workflows.cereal.mycereal_2.json b/pkg/compiler/test/testdata/branch/k8s/5_myapp.workflows.cereal.mycereal_2.json
@@ -646,6 +646,7 @@
   "rawOutputDataConfig": {},
   "executionConfig": {
     "TaskPluginImpls": null,
-    "MaxParallelism": 0
+    "MaxParallelism": 0,
+    "RecoveryExecution": {}
   }
 }
diff --git a/pkg/compiler/test/testdata/branch/k8s/mycereal_condition_has_no_deps.json b/pkg/compiler/test/testdata/branch/k8s/mycereal_condition_has_no_deps.json
@@ -650,6 +650,7 @@
   "rawOutputDataConfig": {},
   "executionConfig": {
     "TaskPluginImpls": null,
-    "MaxParallelism": 0
+    "MaxParallelism": 0,
+    "RecoveryExecution": {}
   }
 }
diff --git a/pkg/compiler/test/testdata/branch/k8s/success_1.json b/pkg/compiler/test/testdata/branch/k8s/success_1.json
@@ -387,6 +387,7 @@
   "rawOutputDataConfig": {},
   "executionConfig": {
     "TaskPluginImpls": null,
-    "MaxParallelism": 0
+    "MaxParallelism": 0,
+    "RecoveryExecution": {}
   }
 }
diff --git a/pkg/compiler/test/testdata/branch/k8s/success_10_simple.json b/pkg/compiler/test/testdata/branch/k8s/success_10_simple.json
@@ -593,6 +593,7 @@
   "rawOutputDataConfig": {},
   "executionConfig": {
     "TaskPluginImpls": null,
-    "MaxParallelism": 0
+    "MaxParallelism": 0,
+    "RecoveryExecution": {}
   }
 }
diff --git a/pkg/compiler/test/testdata/branch/k8s/success_2.json b/pkg/compiler/test/testdata/branch/k8s/success_2.json
@@ -428,6 +428,7 @@
   "rawOutputDataConfig": {},
   "executionConfig": {
     "TaskPluginImpls": null,
-    "MaxParallelism": 0
+    "MaxParallelism": 0,
+    "RecoveryExecution": {}
   }
 }
diff --git a/pkg/compiler/test/testdata/branch/k8s/success_3.json b/pkg/compiler/test/testdata/branch/k8s/success_3.json
@@ -411,6 +411,7 @@
   "rawOutputDataConfig": {},
   "executionConfig": {
     "TaskPluginImpls": null,
-    "MaxParallelism": 0
+    "MaxParallelism": 0,
+    "RecoveryExecution": {}
   }
 }
diff --git a/pkg/compiler/test/testdata/branch/k8s/success_4.json b/pkg/compiler/test/testdata/branch/k8s/success_4.json
@@ -497,6 +497,7 @@
   "rawOutputDataConfig": {},
   "executionConfig": {
     "TaskPluginImpls": null,
-    "MaxParallelism": 0
+    "MaxParallelism": 0,
+    "RecoveryExecution": {}
   }
 }
diff --git a/pkg/compiler/test/testdata/branch/k8s/success_5.json b/pkg/compiler/test/testdata/branch/k8s/success_5.json
@@ -524,6 +524,7 @@
   "rawOutputDataConfig": {},
   "executionConfig": {
     "TaskPluginImpls": null,
-    "MaxParallelism": 0
+    "MaxParallelism": 0,
+    "RecoveryExecution": {}
   }
 }
diff --git a/pkg/compiler/test/testdata/branch/k8s/success_6.json b/pkg/compiler/test/testdata/branch/k8s/success_6.json
@@ -356,6 +356,7 @@
   "rawOutputDataConfig": {},
   "executionConfig": {
     "TaskPluginImpls": null,
-    "MaxParallelism": 0
+    "MaxParallelism": 0,
+    "RecoveryExecution": {}
   }
 }
diff --git a/pkg/compiler/test/testdata/branch/k8s/success_7_nested.json b/pkg/compiler/test/testdata/branch/k8s/success_7_nested.json
@@ -440,6 +440,7 @@
   "rawOutputDataConfig": {},
   "executionConfig": {
     "TaskPluginImpls": null,
-    "MaxParallelism": 0
+    "MaxParallelism": 0,
+    "RecoveryExecution": {}
   }
 }
diff --git a/pkg/compiler/test/testdata/branch/k8s/success_8_nested.json b/pkg/compiler/test/testdata/branch/k8s/success_8_nested.json
@@ -522,6 +522,7 @@
   "rawOutputDataConfig": {},
   "executionConfig": {
     "TaskPluginImpls": null,
-    "MaxParallelism": 0
+    "MaxParallelism": 0,
+    "RecoveryExecution": {}
   }
 }
diff --git a/pkg/compiler/test/testdata/branch/k8s/success_9_nested.json b/pkg/compiler/test/testdata/branch/k8s/success_9_nested.json
@@ -545,6 +545,7 @@
   "rawOutputDataConfig": {},
   "executionConfig": {
     "TaskPluginImpls": null,
-    "MaxParallelism": 0
+    "MaxParallelism": 0,
+    "RecoveryExecution": {}
   }
 }
diff --git a/pkg/controller/config/config_flags.go b/pkg/controller/config/config_flags.go