Implement task callbacks

[joshliberty]

Task Update & Task Callback Events

When a task is completed by the app servers (external orchestration services), it needs to notify the Task Manager of its status via the md.tasks.callback event. Task Manager then queries & retrieves details statuses, metadata and output artifacts.

The Task Manager communicates any changes to an accepted task through the md.tasks.update event. For example, when a md.tasks.dispatch is accepted, a md.tasks.update is emitted with status=Accepted.

Task Callbacks - md.tasks.callback

The event is emitted by external orchestration services to notify Task Manager that a task has been completed.

Queue name: md.tasks.callback
Type: TaskCallbackEvent
Event schema:

{
  "correlation_id": "d1ffa435-1582-4aaa-8cab-912d86fb7e38",
  "execution_id": "91a70c2a-4a97-4103-a9a5-c77aed9f4d2a",
  "identity": "md-list-input-artifacts-template-xfl6v",
  "metadata": {},
  "task_id": "caa10114-db70-4655-8b5d-273e16833c03",
  "workflow_id": "65666477-a775-4914-825b-5ecc33d0d6b4"
}

This event includes the identity of the task on the remote orchestration service for the Task Manager and the plug-in to query the status. When the Workflow Manager picks up that event, the following needs to happen:

1. Validate incoming message
2. Call the plug-in's GetStatus() method
3. Emit a md.tasks.update event
4. Release allocated resource.

Task Update - md.tasks.update

The md.tasks.update is emitted by the Task Manger to notify any changes in the status of the accepted tasks.

Queue name: md.tasks.update
Type: TaskUpdateEvent
Event schema:

{
  "correlation_id": "d1ffa435-1582-4aaa-8cab-912d86fb7e38",
  "execution_id": "91a70c2a-4a97-4103-a9a5-c77aed9f4d2a",
  "message": "Argo status = Running. Messages = .",
  "metadata": {
    "JobIdentity": "md-list-input-artifacts-template-xfl6v"
  },
  "reason": "None",
  "status": "Succeeded",
  "task_id": "caa10114-db70-4655-8b5d-273e16833c03",
  "workflow_id": "65666477-a775-4914-825b-5ecc33d0d6b4"
}

The event includes the status of a task, failure reasons, and an optional failure message.
Metadata generated by the task and any artifacts registered by the task are returned in the event.

The format of the callback event or API call can differ between app servers. We need to define an overall wrapping object with the minimum set of information that we need, for example, the task ID-but otherwise the content of this object will eventually be passed to the plug-in itself, so the main interface needs to be between the plug-in and its app server.

Task statuses

• Unknown
• Created
• Dispatched
• Accepted
• Succeeded*
• Failed*
• Canceled*

Tasks can finish in status denoted with *.

Note: Inference operations can have more complex statuses, for example, the task may finish successfully but fail to reach a decision due to a lack of statistical significance. This is still represented as a success for the purposes of the workflow manager. To allow for these more complex nuanced statuses the metadata dictionary may be used by plug-ins to indicate the result in further detail.

Acceptance Criteria

• GIVEN an app server finished running the task, WHEN a callback event is published to RabbitMQ THEN it is picked up by the task executer for processing
• GIVEN the callback event has been picked up by the task executer, WHEN the task executer completes the task THEN a task update event is published to RabbitMQ
• GIVEN the callback event is invalid WHEN an error occurs while processing the callback event THEN the event is rejected with requeue=false
• GIVEN the callback event is valid WHEN the task cannot be found (in memory for now) THEN the event is rejected with requeue=false
• GIVEN the callback event that is valid WHEN the GetStatus call throw exceptions (which it shouldn't have) THEN the event is rejected with requeue=false
• GIVEN a valid callback event WHEN status is retrieved from the plug-in THEN the Task Manager shall include any metadata and output artifacts in the task update event.
• GIVEN a valid callback event WHEN status is retrieved from the plug-in THEN the Task Manager shall include any error messages in the task update event.

[joshliberty]

@mocsharp please take a look at the above. This is what I had in mind, and I'm not sure how it will work if the task executer doesn't have access to the database – let me know what you think.

[mocsharp]

There are 3 events used to communicate between the Workflow Manager <-> Task Manager & Task Manager <-> Task Plug-ins (including external services). TaskDispatchEvent.cs, TaskUpdateEvent.cs and TaskCallbackEvent.cs.

• TaskDispatchEvent is sent from WM to TM to start a task execution
• TaskUpdateEvent is sent from TM to WM to indicate a change in the task, including acceptance of a task as defined in TaskStatus.cs.
• TaskCallbackEvent is sent from either the task plug-in or the task execution service (external) to indicate completion of the task.

As discussed earlier, we will share the same database but in different collections (mongo) for the POC release and separate them in later releases.

In future releases, once TM & WM are separated, each instance of TM can have its own instance of DB and the TaskCallbackEvent can be extended from md.task.callback to md.task.callback.tm_cloud1 where tm_cloud1 is the name of a collection of (one or more) TMs running an environment.