Dispatch tasks

[joshliberty]

Overview

The task manager is the subcomponent responsible for executing workflow tasks. It's responsible for triggering the task plug-ins, mapping input and output, providing meta data and updating meta data, updating task status, and notifying the workflow manager after a task has been completed.

This task is concerned with the initial basic functionality of dispatching a task that has been added to the queue. Please see the following tickets for additional functionality:

• Implement task callbacks #47
• Metadata store: #50

Dispatching tasks

Tasks are picked up from a queue in a RabbitMQ.
Queue name: md.tasks.dispatch
Event schema:

{
  "ExecutionId": $execution_id,
  "WorkflowInstanceId": $workflow_instance_id,
  "StorageInformation": Object
}

Additional task information is then retrieved from MongoDB based on the workflow instance ID and the execution ID.

Handling a task

1. Retrieve task information from MongoDB
2. Check the task type and instantiate the relevant plug-in
3. Provide the following information to the specific plug-in:
   1. Artifact dictionary
   2. The task_plugin_arguments property in the task's object in MongoDB
   3. Storage engine credentials
   4. Output directory path
4. Call the plug-in's "run" method
5. Update task status in MongoDB to submitted
6. Acknowledge the task dispatch event so that it is removed from RabbitMQ

Plug-in errors

A plug-in can fail either because the event doesn't contain the information it needs, or because of an external reason such as an app server not being available. To facilitate error handling plug-in should be able to raise one of the following exceptions:

• InvalidTask: to be used when the task does not contain the necessary information, for example when the task plug-in arguments don't contain the required fields for that plug-in
• UpstreamError: to be used when a plug-in can communicate with an external system.

Acceptance criteria

• Given a valid event in RabbitMQ, when it is picked up by the event broker adapter it should trigger the task executer
• Given an event doesn't include all of the required information, when it is picked up by the event broker adapter it should be rejected with requeue=false
• Given an event references a non-existent task, when it is picked up it should be rejected with requeue=false
• When the plug-in throws an upstream error exception, the message should be rejected with requeue=true
• When the plug-in throws an invalid event exception, the message should be rejected with requeue=false

[mocsharp]

from abc import abstractclassmethod

Here's a rough idea of how the Clinical Workflow Manager (CWM) & Task Manager (TM) would work together:

Clinical Workflow Manager workflow for requesting a task to be done:

• Publishes task requests for md.task.request.{plugin-name} to the Task Exchange
• Create a Task Payload (TP) (maybe a bucket?) for the task to access (a few options)
  • A single TP for R&W of input and output data
  • A read-only input plus and writable output

Task Manager:

• Subscribes to md.task.request.* to capture all task request events.
• Decides how incoming requests are scheduled based on some configurations provided by each plugin. E.g. number of concurrent jobs allowed. Or, based on the resource required by each task and the available resources provided by the plugin.
  (Another option is to offline the scheduling to the plugin)
• Each plugin registers a specific topic with the Task Manager. E.g. md.task.request.argo

When TM receives a task request:

• Retrieves data from TP
• Makes data in TP available to the request (could be an HTTP upload or mapping a local volume to argo)
• Executes the task
• Monitors the status of the task, which can be implemented in a few ways:
  • Event-driven
  • Callback (RESTful API)
  • Polling
• Notifies task manager upon completion of the task

The abstract (base) class of the TaskPlugin may look like this:

class AbstractTaskFactory(metaclass = ABCMeta):
   @abstractmethod
   def new(self):
       """ kicks off a new job"""
       pass
       
   @abstractmethod
   def available_resources(self):
       """ returns available resource"""
       pass
    
   @abstractmethod
   def task_status(self, task):
       """ gets the status of a task"""
       pass
       
class AbstractTask(metaclass = ABCMeta):
   @abstractmethod
   def register(self, topic):
      """ registers the plugin and subscribes to task requests for the 
      specified topic.
      """
      pass
   
   @property
   @abstractmethod
   def payload(self):
      """ access to the payload metadata: location, filenames, etc...
      """
      pass
   
   @abstractmethod
   def complete(self, task_id, delivery_tag, status, results):
      """ completes a task, sends a `md.task.complete` message with final status and results (structure to be defined)
      """
      pass

[mocsharp]

If the plan is to eventually break the Workflow Manager & Task Runner (Manager), I suggest we don't give TR access to the database.

Imagine TRs are deployed on different environments, combinations of cloud and on-prem settings if we can contain all necessary data to execute a task within an md.tasks.dispatch message; then, the Task Runner wouldn't need access to the MongoDB. This limits database access to only the Workflow Manager.

As @joshliberty defined in the Handling a task section above, the TR needs the information for which task to execute, task_plugin_arguments, the artifact dictionary, and credentials (if generated by the WM or let TR make the call to generate one on the fly).

Upon success, TR can emit the status back to WM.

If a task fails to execute, task runner crashes, or a task plug-in crashes, WM can emit another md.tasks.dispatch message to retry the same task.

[joshliberty]

When we discussed it a few weeks ago, this was one of the arguments for keeping them in the same process- because they will be coupled anyway by using the same database.

The task executer also needs to update the task status and power the task API- things will get noticeably more complex if we try to maintain two databases for that. I think we can do that in the future, when we really separate them- for now it feels like premature optimisation to me.

[mocsharp]

@DanielNuttall-AD, @JoeBatt1989, @JHancox, @dbericat, @MMelQin

As we've discussed this morning, we can still keep both Workload Manager (WM) & Task Manager (TM) in the same process for the first couple of releases. However, in the future, if we have a deployment as in the diagram, it's best to keep TM from accessing the database. We can do this now to reduce the amount of refactoring when the time comes to separate the two and reduce technical debts. This also simplifies database access controls at deploy time.

When a TM completes a task, it could simply emit an md.tasks.complete message along with the status.
When WM dispatches a new task, any data required, as defined by @joshliberty above, can be included in the message. If, for some reason, TM crashes, WM can also emit the same message for another TM to pick up the work.

Diagram illustrates 2 TMs + WM + IG running in different environments but all have access to the same 🐰 for message exchange.

[JHancox]

Thanks Victor – I see what you mean now and it seems to make sense to take this approach. Best Jonny