Cache dynamic workflow #254
Conversation
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
| } | ||
| return dynamicWorkflowContext{ | ||
| isDynamic: true, | ||
| subWorkflow: compiledWf, |
There was a problem hiding this comment.
Shouldn't we set subWorkflowClosure here as well?
There was a problem hiding this comment.
we don't cache that though, right? (if i'm understanding correctly)
There was a problem hiding this comment.
Do we send the dynamic workflow spec only the first time? or on every event (that will be too much)?
If it's only needed the first time, then I guess that's alright...
There was a problem hiding this comment.
Yeah we only send the spec the first time (when it's uncached) and don't need to bog down subsequent messages
| @@ -329,4 +333,88 @@ func Test_dynamicNodeHandler_buildContextualDynamicWorkflow_withLaunchPlans(t *t | |||
| assert.Error(t, err) | |||
| assert.True(t, callsAdmin) | |||
| }) | |||
| t.Run("dynamic wf cached", func(t *testing.T) { | |||
There was a problem hiding this comment.
can we write a test for the failure case? Where the cache read fails, or the cache write fails
|
I am assuming you do not have the event sending in this PR right? |
| trns.WithInfo(trns.Info().WithInfo(&handler.ExecutionInfo{TaskNodeInfo: &handler.TaskNodeInfo{TaskNodeMetadata: &event.TaskNodeMetadata{CacheStatus: status.GetCacheStatus(), CatalogKey: status.GetMetadata()}}})) | ||
| taskNodeInfoMetadata := &event.TaskNodeMetadata{CacheStatus: status.GetCacheStatus(), CatalogKey: status.GetMetadata()} | ||
| if dCtx.subWorkflowClosure != nil && dCtx.subWorkflowClosure.Primary != nil && dCtx.subWorkflowClosure.Primary.Template != nil { | ||
| taskNodeInfoMetadata.DynamicWorkflow = &event.DynamicWorkflowNodeMetadata{ |
There was a problem hiding this comment.
This is pretty static (it gets generated once and never changes for the duration of that node)... should we keep sending it to Admin on every event? isn't that too much?
There was a problem hiding this comment.
I would prefer we send it only once - does this need a state?
There was a problem hiding this comment.
yes we only send it once. for cached reads to the dynamic workflow we don't send it
There was a problem hiding this comment.
I don't think we should rely on caching to not send it... can we make it based on the phase maybe? we only send it on queued or something like that... ? if there isn't a good existing node phase to control this, we can introduce one here...
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
| @@ -116,6 +120,19 @@ func (d dynamicNodeTaskNodeHandler) buildDynamicWorkflowTemplate(ctx context.Con | |||
| }, nil | |||
| } | |||
|
|
|||
| // Updates a dynamically generated subworkflow version to be deterministically generated from the computed closure. | |||
| func setDeterministicVersion(ctx context.Context, workflowClosure *core.CompiledWorkflowClosure) error { | |||
There was a problem hiding this comment.
this does not need to be done any more right? Why is this needed?
|
PTAL @EngHabu |
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Katrina Rogan <[email protected]>
Signed-off-by: Ketan Umare <[email protected]>
TL;DR
Cache dynamic workflow. Verified the cache is read during execution.
Also sends workflow spec to admin
Type
Are all requirements met?
Complete description
How did you fix the bug, make the feature etc. Link to any design docs etc
Tracking Issue
flyteorg/flyte#899
flyteorg/flyteconsole#295
flyteorg/flyte#928
Follow-up issue
NA