wip: feat(engine): dags-as-durable-tasks#4266
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
|
|
|
|
|
|
|
This reverts commit abc0dc7.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| ), | ||
| cell: ({ row }) => { | ||
| if (row.getCanExpand()) { | ||
| if (row.original.type === 'DAG') { |
There was a problem hiding this comment.
this lets us open the workflow run detail page for an operator DAG before any of the tasks have actually been spawned
| [taskRuns], | ||
| () => | ||
| taskRuns | ||
| .filter((t) => t.isDurable && t.taskExternalId !== id) |
There was a problem hiding this comment.
there's a bunch of stuff like this - basically removing the orchestrator from the ids
| if e.isDagOrchestrator { | ||
| continue | ||
| } |
There was a problem hiding this comment.
don't want to render spans for the orchestrator task since it messes up nesting and is "internal" / not user facing
| d.l.Warn().Ctx(ctx).Err(err).Msg("failed to unmarshal input") | ||
| continue | ||
| if len(currInput.DagParentTaskRunIds) > 0 { | ||
| dagParentOutputs, err := d.repov1.Tasks().GetDagParentOutputs(ctx, tenantId, currInput.DagParentTaskRunIds) |
There was a problem hiding this comment.
not sure what to do about this - seems very suboptimal to need a db lookup here, will circle back
| continue | ||
| } | ||
| } else { | ||
| childEvents, err := s.repov1.Tasks().ListDurableOrchestratorChildOutputEvents(ctx, tenantId, event.TaskExternalId) |
There was a problem hiding this comment.
same here - really not ideal to need the db lookup, will circle back
| return nil | ||
| } | ||
|
|
||
| func (d *DispatcherServiceImpl) TriggerDAGStep(ctx context.Context, tenantId uuid.UUID, req *operator.DAGStepTriggerRequest) (*operator.DAGStepTriggerResult, error) { |
There was a problem hiding this comment.
this is one of the important methods, since it's what gets called by the operator to trigger tasks
| TriggerTaskData: &v1.TriggerTaskData{ | ||
| WorkflowName: req.WorkflowName, | ||
| TargetActionId: &req.ActionId, | ||
| UserMessage: &stepLabel, | ||
| Data: []byte(req.Input), | ||
| AdditionalMetadata: task.AdditionalMetadata, | ||
| ParentExternalId: &task.ExternalID, | ||
| ParentTaskId: &task.ID, | ||
| ParentTaskInsertedAt: &task.InsertedAt.Time, | ||
| ChildIndex: &childIndex, | ||
| DagParentTaskRunIds: req.DagParentTaskRunIds, | ||
| IsSkipped: req.IsSkipped, | ||
| IsCancelled: req.IsCancelled, | ||
| DesiredWorkerLabels: req.DesiredWorkerLabels, | ||
| WorkflowRunId: &orchestratorWorkflowRunId, |
There was a problem hiding this comment.
important stuff here:
- we basically pass most things along (e.g. the metadata, the input, etc.)
- we tie the workflow run id manually to the orchestrator's workflow run id
| if t.output != nil { | ||
| b, err := json.Marshal(t.output) | ||
| if err == nil { | ||
| output[t.readableId] = json.RawMessage(b) |
There was a problem hiding this comment.
this results in a DAG output that maps step ids to outputs, similar to what we see in the dash and over the API
1 participant
Description
WIP
Initial implementation of the dag operator, which basically creates an internal durable task that processes the dag via child spawning
to do:
workflow_run_idfor the dag children correspond's to the durable task external idwait_for(sleep), we get a bunch of "could not send task to worker'Fixes # (issue)
Type of change
What's Changed
Checklist
Changes have been:
🤖 AI Disclosure