Persist flyte deck during workflow recovery (flyteorg#495)

Signed-off-by: Kevin Su <[email protected]> Signed-off-by: Kevin Su <[email protected]> Co-authored-by: Dan Rammer <[email protected]>
eapolinario · Nov 1, 2022 · fbe6f23 · fbe6f23
1 parent 4db2319
commit fbe6f23
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 1 deletion.
diff --git a/flytepropeller/pkg/controller/nodes/executor.go b/flytepropeller/pkg/controller/nodes/executor.go
@@ -263,7 +263,7 @@ func (c *nodeExecutor) attemptRecovery(ctx context.Context, nCtx handler.NodeExe
 		OutputURI: outputFile,
 	}
 
-	deckFile := v1alpha1.GetDeckFile(nCtx.NodeStatus().GetOutputDir())
+	deckFile := storage.DataReference(recovered.Closure.GetDeckUri())
 	metadata, err := nCtx.DataStore().Head(ctx, deckFile)
 	if err != nil {
 		logger.Errorf(ctx, "Failed to check the existence of deck file. Error: %v", err)

diff --git a/flytepropeller/pkg/controller/nodes/executor_test.go b/flytepropeller/pkg/controller/nodes/executor_test.go
@@ -2099,6 +2099,7 @@ func TestRecover(t *testing.T) {
 					OutputResult: &admin.NodeExecutionClosure_OutputUri{
 						OutputUri: "outputuri.pb",
 					},
+					DeckUri: deckPath,
 				},
 			}, nil)
 
@@ -2154,6 +2155,7 @@ func TestRecover(t *testing.T) {
 							CacheStatus: core.CatalogCacheStatus_CACHE_HIT,
 						},
 					},
+					DeckUri: deckPath,
 				},
 			}, nil)
 
@@ -2237,6 +2239,7 @@ func TestRecover(t *testing.T) {
 							},
 						},
 					},
+					DeckUri: deckPath,
 				},
 			}, nil)
 
@@ -2319,6 +2322,7 @@ func TestRecover(t *testing.T) {
 					OutputResult: &admin.NodeExecutionClosure_OutputUri{
 						OutputUri: "outputuri.pb",
 					},
+					DeckUri: deckPath,
 				},
 			}, nil)
 
@@ -2364,6 +2368,7 @@ func TestRecover(t *testing.T) {
 					OutputResult: &admin.NodeExecutionClosure_OutputUri{
 						OutputUri: "outputuri.pb",
 					},
+					DeckUri: deckPath,
 				},
 			}, nil)