Pure graph implementation for the orchestration system

[antho1404]

This PR changes the way we designed the workflow.
Instead of having a graph only for the tasks and trigger this graph of tasks when a trigger is matched. We now include the trigger as a node of this graph and we process all the dependencies for this node (this even allows a multi-tree of tasks based on the same trigger or event multiple triggers).

The mapping has also been considered as a node in order to be able to chain them and reduce the complexity of a task node and providing more flexibility in the future.

Benefits:
With this structure where everything is a node, we can easily switch to a graph database where different objects type are different nodes in the database and so having better performances on the lookup of the nodes to resolve. It also provides the flexibility to add new types of processing.

Here is the graph structure:

Nodes:

• Task: A definition that will create a new execution when processed.
• Event: A definition of an event to listen. Previously called trigger. Only one trigger is allowed in the graph.
• Result: A definition of a result to listen. Previously called trigger. Only one trigger is allowed in the graph. A result can be from a different workflow where connections between tasks are in the same workflow.
• Map: A definition to map data. If a nodeKey is execution, it will find the matching outputs in the execution's graph, otherwise, it will map based on the previous current data (from a trigger or another mapping)
• Filter: A definition of a filter based on the data. This filter will stop the discovery in the graph if not valid. (will be implemented later)

Edges:

Classic edges composed of src and dst that can connect any kind of node

To Test:

3 services:

// serviceA
{"name":"serviceA","tasks":[{"inputs":[{"type":"String","key":"message"}],"outputs":[{"type":"String","key":"value"}],"key":"taskX"},{"key":"emits"}],"events":[{"data":[{"type":"String","key":"message"}],"key":"eventX"}],"configuration":{},"hash":"F8v5pwfTcFK8BDUXAbRUGyjJZ88Xd2d8J7p4RgKZdczg","sid":"serviceA","source":"QmT4DzLGD8GHsvJkfQuzmabpbDCVZAA7NgeRYz3y485kW9"}
// serviceB
{"name":"serviceB","tasks":[{"inputs":[{"type":"String","key":"inputA"}],"outputs":[{"type":"String","key":"res"}],"key":"taskY"}],"configuration":{},"hash":"7GJiYp6XRtZGgQKbQrnvTvDWeTPhAmjSMs9T5S83aPiY","sid":"serviceB","source":"QmRzco9eDNjyyjagd5qd1c1nDJ6eZupg71fYXdyU9JXVpL"}
// 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":"F3qDSqUwn2mxBpRv3LUzL6vTNqfvTFvoQnv1KhGnuTsn","sid":"serviceC","source":"QmPuHX3LumnDkrxY3wUR8J2J9xmXGw8yTVy5jf7P24MprN"}

2 types of workflow:

serviceA#eventX => serviceA#taskX => map(inputA: serviceA#taskX.value) => serviceB#taskY

{
  "key": "Event",
  "nodes": [
    {
      "event": {
        "key": "trigger:event",
        "instanceHash": "6hDyE3NurqLtxgV4AtXMpLemNMEzvEiTgDDwtxjZzdPR",
        "eventKey": "eventX"
      }
    },
    {
      "task": {
        "key": "node1",
        "taskKey": "taskX",
        "instanceHash": "6hDyE3NurqLtxgV4AtXMpLemNMEzvEiTgDDwtxjZzdPR"
      }
    },
    {
      "map": {
        "key": "map1",
        "outputs": [{
          "key": "inputA",
          "ref": {
            "nodeKey": "node1",
            "key": "value"
          }
        }]
      }
    },
    {
      "task": {
        "key": "node2",
        "taskKey": "taskY",
        "instanceHash": "2iLAvYeJu78wi4Y4sHBKmVYnM3PVVqP8gpCocUnnLJBH"
      }
    }
  ],
  "edges": [
    {
      "src": "trigger:event",
      "dst": "node1"
    },
    {
      "src": "node1",
      "dst": "map1"
    },
    {
      "src": "map1",
      "dst": "node2"
    }
  ]
}

serviceB#taskY => map(message: serviceB#taskY.res) => serviceC#taskZ

{
  "key": "result-from-other-workflow",
  "nodes": [
    {
      "result": {
        "key": "trigger:result",
        "instanceHash": "2iLAvYeJu78wi4Y4sHBKmVYnM3PVVqP8gpCocUnnLJBH",
        "taskKey": "taskY"
      }
    },
    {
      "task": {
        "key": "task1",
        "instanceHash": "FXGpKzcEHMnyr2BQrH44Dq2SoAYtjw1C7qNRVGbxTNpa",
        "taskKey": "taskZ"
      }
    },
    {
      "map": {
        "key": "map",
        "outputs": [
          {
            "key": "message",
            "ref": {
              "nodeKey": "trigger:result",
              "key": "res"
            }
          }
        ]
      }
    }
  ],
  "edges": [
    {
      "src": "trigger:result",
      "dst": "map"
    },
    {
      "src": "map",
      "dst": "task1"
    }
  ]
}

Happy hacking :)

[antho1404]

keep this in comment to reuse it to manage the filters on another PR

[antho1404]

This is basically a file renaming from algorithm.go to graph.go but with a new graph struct

[krhubert]

This package got bigger with PR and still, there is not a single test :/

[antho1404]

this actually makes the scheduler smaller and yes I agree it needs testing, this is based on the sdk and don't have any mock for it so it's hard to reproduce behaviors. But yes I will add some tests on another PR

[krhubert]

and don't have any mock for it so it's hard to reproduce behaviors.

Please don't use mocks. Rather then mocks try to organize the code so it could be unit tested.

But yes I will add some tests on another PR

This functionality is in this PR so I think the tests should also be here. It's not the first pr about graph without testing and PR about filter has already been started. I'm ok to add them in PR about filters, but not later.

[NicolasMahe]

manual test are working good 👍
do you want to do the modif in this PR or in another one?

[NicolasMahe]

@antho1404 can you update the workflow in description so they can work with the latest modif?

[antho1404]

description updated