Summary - APIs

[kymr]

Link

• https://netflix.github.io/conductor/runtime/

[kymr]

Task & Workflow Metadata

Endpoint	Description	Input
GET /metadata/taskdefs	Get all the task definitions	n/a
GET /metadata/taskdefs/{taskType}	Retrieve task definition	Task Name
POST /metadata/taskdefs	Register new task definitions	List of Task Definitions
PUT /metadata/taskdefs	Update a task definition	A Task Definition
DELETE /metadata/taskdefs/{taskType}	Delete a task definition	Task Name
GET /metadata/workflow	Get all the workflow definitions	n/a
POST /metadata/workflow	Register new workflow	Workflow Definition
PUT /metadata/workflow	Register/Update new workflows	List of Workflow Definition
GET /metadata/workflow/{name}?version=	Get the workflow definitions	workflow name, version (optional)

[kymr]

Start A Workflow

With Input only

POST /workflow/{name}?version=&correlationId=
{
   //JSON payload for workflow
}

Parameter	Description
version	Optional. If not specified uses the latest version of the workflow
correlationId	User supplied Id that can be used to retrieve workflows

Input

• JSON Payload to start the workflow.
• Mandatory.
• If workflow does not expect any input MUST pass an empty JSON like {}

Output

• Id of the workflow (GUID)

With Input and Task Domains

POST /workflow
{
   //JSON payload for Start workflow request
}

Start workflow request

• JSON for start workflow request

  {
  "name": "myWorkflow", // Name of the workflow
  "version": 1, // Version
  “correlatond”: “corr1” // correlation Id
  "input": {
  // Input map. 
  },
  "taskToDomain": {
  // Task to domain map
  }
  }

  Output

• Id of the workflow (GUID)

[kymr]

Retrieve Workflows

Endpoint	Description
GET /workflow/{workflowId}?includeTasks=true|false	Get Workflow State by workflow Id. If includeTasks is set, then also includes all the tasks executed and scheduled.
GET /workflow/running/{name}	Get all the running workflows of a given type
GET /workflow/running/{name}/correlated/{correlationId}?includeClosed=true|false&includeTasks=true|false	Get all the running workflows filtered by correlation Id. If includeClosed is set, also includes workflows that have completed running.
GET /workflow/search	Search for workflows. See Below.

[kymr]

Search for Workflows

• Conductor uses Elasticsearch for indexing workflow execution and is used by search APIs.

  GET /workflow/search?start=&size=&sort=&freeText=&query=

Parameter	Description
start	Page number. Defaults to 0
size	Number of results to return
sort	Sorting. Format is: ASC: or DESC: to sort in ascending or descending order by a field
freeText	Elasticsearch supported query. e.g. workflowType:"name_of_workflow"
query	SQL like where clause. e.g. workflowType = 'name_of_workflow'. Optional if freeText is provided.

Output

• Search result as described below

  {
  "totalHits": 0,
  "results": [
  {
    "workflowType": "string",
    "version": 0,
    "workflowId": "string",
    "correlationId": "string",
    "startTime": "string",
    "updateTime": "string",
    "endTime": "string",
    "status": "RUNNING",
    "input": "string",
    "output": "string",
    "reasonForIncompletion": "string",
    "executionTime": 0,
    "event": "string"
  }
  ]
  }

[kymr]

Manage Workflows

Endpoint	Description
PUT /workflow/{workflowId}/pause	Pause. No further tasks will be scheduled until resumed. Currently running tasks are not paused.
PUT /workflow/{workflowId}/resume	Resume normal operations after a pause.
POST /workflow/{workflowId}/rerun	See Below.
POST /workflow/{workflowId}/restart	Restart workflow execution from the start. Current execution history is wiped out.
POST /workflow/{workflowId}/retry	Retry the last failed task.
PUT /workflow/{workflowId}/skiptask/{taskReferenceName}	See below.
DELETE /workflow/{workflowId}	Terminates the running workflow.
DELETE /workflow/{workflowId}/remove	Deletes the workflow from system. Use with caution.

Rerun

• Re-runs a completed workflow from a specific task.

  POST /workflow/{workflowId}/rerun

  {
  "reRunFromWorkflowId": "string",
  "workflowInput": {},
  "reRunFromTaskId": "string",
  "taskInput": {}
  }

Skip Task

• Skips a task execution (specified as taskReferenceName parameter) in a running workflow and continues forward.
• Optionally updating task's input and output as specified in the payload.

  PUT /workflow/{workflowId}/skiptask/{taskReferenceName}?workflowId=&taskReferenceName

  {
  "taskInput": {},
  "taskOutput": {}
  }

[kymr]

Manage Tasks

Endpoint	Description
GET /tasks/{taskId}	Get task details.
GET /tasks/queue/all	List the pending task sizes.
GET /tasks/queue/all/verbose	Same as above, includes the size per shard
GET /tasks/queue/sizes?taskType=&taskType=&taskType	Return the size of pending tasks for given task types

[kymr]

Polling, Ack and Update Task

• These are critical endpoints used to poll for task, send ack (after polling) and finally updating the task result by worker.

Endpoint	Description
GET /tasks/poll/{taskType}?workerid=&domain=	Poll for a task. workerid identifies the worker that polled for the job and domain allows the poller to poll for a task in a specific domain
GET /tasks/poll/batch/{taskType}?count=&timeout=&workerid=&domain	Poll for a task in a batch specified by count. This is a long poll and the connection will wait until timeout or if there is at-least 1 item available, whichever comes first.workerididentifies the worker that polled for the job and domainallows the poller to poll for a task in a specific domain
POST /tasks	Update the result of task execution. See the schema below.
POST /tasks/{taskId}/ack	Acknowledges the task received AFTER poll by worker.

Schema for updating Task Result

{
    "workflowInstanceId": "Workflow Instance Id",
    "taskId": "ID of the task to be updated",
    "reasonForIncompletion" : "If failed, reason for failure",
    "callbackAfterSeconds": 0,
    "status": "IN_PROGRESS|FAILED|COMPLETED",
    "outputData": {
        //JSON document representing Task execution output     
    }

}

Acknowledging tasks after poll

• If the worker fails to ack the task after polling, the task is re-queued and put back in queue and is made available during subsequent poll.

[kymr]

Check it

Common

• Demo it.
• make sure you understand all those apis and operate conductor with it.
• Real workflows in production are really complex, it is going to be very hard to manage with text-based definitions.
  • Check out the UI module
  • Also, check out it support well

Retrieve and Search Workflows

• Getting all workflows that are running operation could be dangerous if those are many.

Manage Workflows

• Pausing workflow operation looks like stopping consume messages at event-driven. But it differs from stopping entire workflow, not stopping consumption from each server
• How does workflow's status change after rerun?
  • Does rerun operation init status and delete previous results?
• Restart operation seems dangerous, because every task is not idempotent.
  • use it carefully
• How retry works when many tasks are failed at the same time?
• SkipTask looks quite useful to many cases.

Manage Tasks

• Listing all queued task could be dangerous.
• What does sharded tasks mean?
  • It seems to mean that queues are created separately on each server in the cluster, but need to find out for sure.

Polling, Ack and Update Task

• It is a little bit difficult to implement Exactly-Once in a distributed system with polling style