Workflow graph structure

[antho1404]

Update the structure of the workflow with a graph approach.

The workflow definition is now created based on a list of nodes and a list of edges that connects these nodes.
Every node need to be unique in the workflow and will be identifiable. With the node identifiable we can now store the current step when creating execution and can drop the height support that was not valid when a workflow was triggered by a result of the same workflow and the same task.

With this new stepID we can simply find the children of this node in the workflow definition graph and execute all the children's tasks.

• SDK of execution updated to accept the stepID
• API updated with the new workflow struct and execution stepID

TO TEST

Create and start the services with the cli or API.

// ServiceA
{"name":"serviceA","tasks":[{"inputs":[{"type":"String","key":"message"}],"outputs":[{"type":"String","key":"message"}],"key":"taskX"},{"key":"emits"}],"events":[{"data":[{"type":"String","key":"message"}],"key":"eventX"}],"configuration":{},"hash":"3kWX7CYtNwbeb1h35HiNAPctnh94MU8zstT7PSuusxyy","sid":"serviceA","source":"QmUojyMjo2JUvhktLU4royUBDEsVzU4z1onPovGpVQAFDW"}

// ServiceB
{"name":"serviceB","tasks":[{"inputs":[{"type":"String","key":"message"}],"outputs":[{"type":"String","key":"message"}],"key":"taskY"}],"configuration":{},"hash":"8ExNFweHXavpZSE9iTwjBFSDWzKcRpCXEc4BuX1UwFcg","sid":"serviceB","source":"Qmf2e92jNxxmYyjxEFrCBJdCNUS1LKNWet2ZQt3ByEUpAP"}

// ServiceC
{"name":"serviceC","tasks":[{"inputs":[{"type":"String","key":"message"}],"outputs":[{"type":"String","key":"message"}],"key":"taskZ"},{"inputs":[{"type":"String","key":"message"}],"outputs":[{"type":"String","key":"message"}],"key":"taskA"}],"configuration":{},"hash":"qTtisJG3Bso5rRxSHhnrQSkJpC3JGePD3ovvuEGVgeg","sid":"serviceC","source":"QmPuHX3LumnDkrxY3wUR8J2J9xmXGw8yTVy5jf7P24MprN"}

Create one or all the following workflow(s) with the API

eg: grpcurl -plaintext -d "$(cat result.json| jq -c .)" localhost:50052 api.Workflow/Create

// result.json
{"key":"result","trigger":{"type":2,"instanceHash":"A7712VMAiT2ehXV1nHHUnkvyTbZKVHeFxBZZj4Gx7MZk","key":"taskX","initilNode":"task1"},"nodes":[{"key":"task1","instanceHash":"DUto1J3ZAheVL6PGrwYTtGF4Bnw15CQocVdEmqkHSGbB","taskKey":"taskY"}, {"key":"task2","instanceHash":"FjFzXXme2dkLBsgPk2xgyiPD6dX5EN1dBLGqW6G9tStf","taskKey":"taskZ"}],"edges":[{"src":"task1","dst":"task2"}]}

// event.json
{"key":"event","trigger":{"type":1,"instanceHash":"A7712VMAiT2ehXV1nHHUnkvyTbZKVHeFxBZZj4Gx7MZk","key":"eventX","initialNode":"task1"},"nodes":[{"key":"task1","instanceHash":"DUto1J3ZAheVL6PGrwYTtGF4Bnw15CQocVdEmqkHSGbB","taskKey":"taskY"},{"key":"task2","instanceHash":"FjFzXXme2dkLBsgPk2xgyiPD6dX5EN1dBLGqW6G9tStf","taskKey":"taskZ"}],"edges":[{"src":"task1","dst":"task2"}]}

// result-from-other-workflow
{"key":"result-from-other-workflow","trigger":{"type":2,"instanceHash":"DUto1J3ZAheVL6PGrwYTtGF4Bnw15CQocVdEmqkHSGbB","key":"taskY","initialNode":"task1"},"nodes":[{"key":"task1","instanceHash":"FjFzXXme2dkLBsgPk2xgyiPD6dX5EN1dBLGqW6G9tStf","taskKey":"taskZ"}],"edges":[]}

Then you see the workflow by running either an event with

mesg-cli service:execute serviceA emits

or a task

mesg-cli service:execute serviceA taskX --data message=hello

[krhubert]

What do you think about moving the graph as a separate structure? In that case you can control and test all graph functions like searching path, walking order and more without Workflow and workflow will just use it like:

type Workflow struct {
    // ....
    g wgraph
}

[antho1404]

for now, the workflow is pretty much only the graph but yes I agree that could be nice. I would still keep that as a future refactoring if the package gets bigger or we need to reuse the graph

[krhubert]

I don't see a validation if the edge has proper src and dst.

[antho1404]

yes not yet, I wanted to add the validation of nodes/edges and also the instances on another pr

[NicolasMahe]

Fixed workflows to send to API:

// result.json
{"key":"result","trigger":{"type":2,"instanceHash":"HX6xrMxEZRgSN97qL6GUo6aAqpnrNDWoPiS1wae9Qi2s","key":"taskX","nodeKey":"task1"},"nodes":[{"key":"task1","instanceHash":"37Hmvhp2xmH9JPmipk1ES2ATbSVJPGt7z2cD3VDJ1dtf","taskKey":"taskY"}, {"key":"task2","instanceHash":"GbDa3xsmzY3V6HvF8XKL9gxvV6aFZoj5nNvCrFq8GDTd","taskKey":"taskZ"}],"edges":[{"src":"task1","dst":"task2"}]}

// event.json
{"key":"event","trigger":{"type":1,"instanceHash":"HX6xrMxEZRgSN97qL6GUo6aAqpnrNDWoPiS1wae9Qi2s","key":"eventX","nodeKey":"task1"},"nodes":[{"key":"task1","instanceHash":"37Hmvhp2xmH9JPmipk1ES2ATbSVJPGt7z2cD3VDJ1dtf","taskKey":"taskY"},{"key":"task2","instanceHash":"GbDa3xsmzY3V6HvF8XKL9gxvV6aFZoj5nNvCrFq8GDTd","taskKey":"taskZ"}],"edges":[{"src":"task1","dst":"task2"}]}

// result-from-other-workflow
{"key":"result-from-other-workflow","trigger":{"type":2,"instanceHash":"37Hmvhp2xmH9JPmipk1ES2ATbSVJPGt7z2cD3VDJ1dtf","key":"taskY","nodeKey":"task1"},"nodes":[{"key":"task1","instanceHash":"GbDa3xsmzY3V6HvF8XKL9gxvV6aFZoj5nNvCrFq8GDTd","taskKey":"taskZ"}],"edges":[]}