Workflow Definition

Workflow Definition¶

Workflows are defined using a JSON based DSL.

Example

{
  "name": "encode_and_deploy",
  "description": "Encodes a file and deploys to CDN",
  "version": 1,
  "tasks": [
    {
      "name": "encode",
      "taskReferenceName": "encode",
      "type": "SIMPLE",
      "inputParameters": {
        "fileLocation": "${workflow.input.fileLocation}"
      }
    },
    {
      "name": "deploy",
      "taskReferenceName": "d1",
      "type": "SIMPLE",
      "inputParameters": {
        "fileLocation": "${encode.output.encodeLocation}"
      }
    }
  ],
  "outputParameters": {
    "cdn_url": "${d1.output.location}"
  },
  "failureWorkflow": "cleanup_encode_resources",
  "restartable": true,
  "workflowStatusListenerEnabled": true,
  "schemaVersion": 2
}

field	description	Notes
name	Name of the workflow
description	Description of the workflow	optional
version	Numeric field used to identify the version of the schema. Use incrementing numbers	When starting a workflow execution, if not specified, the definition with highest version is used
tasks	An array of task definitions as described below.
inputParameters	List of input parameters. Used for documenting the required inputs to workflow	optional
outputParameters	JSON template used to generate the output of the workflow	If not specified, the output is defined as the output of the last executed task
failureWorkflow	String; Workflow to be run on current Workflow failure. Useful for cleanup or post actions on failure.	optional
schemaVersion	Current Conductor Schema version. schemaVersion 1 is discontinued.	Must be 2
restartable	Boolean flag to allow Workflow restarts	defaults to true
workflowStatusListenerEnabled	If true, every workflow that gets terminated or completed will send a notification. See below	optional (false by default)

Tasks within Workflow¶

tasks property in a workflow execution defines an array of tasks to be executed in that order.

field	description	Notes
name	Name of the task. MUST be registered as a task with Conductor before starting the workflow
taskReferenceName	Alias used to refer the task within the workflow. MUST be unique within workflow.
type	Type of task. SIMPLE for tasks executed by remote workers, or one of the system task types
description	Description of the task	optional
optional	true or false. When set to true - workflow continues even if the task fails. The status of the task is reflected as `COMPLETED_WITH_ERRORS`	Defaults to `false`
inputParameters	JSON template that defines the input given to the task	See Wiring Inputs and Outputs for details
domain	See Task Domains for more information.	optional

In addition to these parameters, System Tasks have their own parameters. Checkout System Tasks for more information.

Wiring Inputs and Outputs¶

Workflows are supplied inputs by client when a new execution is triggered. Workflow input is a JSON payload that is available via ${workflow.input...} expressions.

Each task in the workflow is given input based on the inputParameters template configured in workflow definition. inputParameters is a JSON fragment with value containing parameters for mapping values from input or output of a workflow or another task during the execution.

Syntax for mapping the values follows the pattern as:

${SOURCE.input/output.JSONPath}

field	description
SOURCE	can be either "workflow" or any of the task reference name
input/output	refers to either the input or output of the source
JSONPath	JSON path expression to extract JSON fragment from source's input/output

JSON Path Support

Conductor supports JSONPath specification and uses Java implementation from here.

Example

Consider a task with input configured to use input/output parameters from workflow and a task named loc_task.

{
  "inputParameters": {
    "movieId": "${workflow.input.movieId}",
    "url": "${workflow.input.fileLocation}",
    "lang": "${loc_task.output.languages[0]}",
    "http_request": {
      "method": "POST",
      "url": "http://example.com/${loc_task.output.fileId}/encode",
      "body": {
        "recipe": "${workflow.input.recipe}",
        "params": {
          "width": 100,
          "height": 100
        }
      },
      "headers": {
        "Accept": "application/json",
        "Content-Type": "application/json"
      }
    }
  }
}

Consider the following as the workflow input

{
  "movieId": "movie_123",
  "fileLocation":"s3://moviebucket/file123",
  "recipe":"png"
}

And the output of the loc_task as the following;

{
  "fileId": "file_xxx_yyy_zzz",
  "languages": ["en","ja","es"]
}

When scheduling the task, Conductor will merge the values from workflow input and loc_task's output and create the input to the task as follows:

{
  "movieId": "movie_123",
  "url": "s3://moviebucket/file123",
  "lang": "en",
  "http_request": {
    "method": "POST",
    "url": "http://example.com/file_xxx_yyy_zzz/encode",
    "body": {
      "recipe": "png",
      "params": {
        "width": 100,
        "height": 100
      }
    },
    "headers": {
        "Accept": "application/json",
        "Content-Type": "application/json"
    }
  }
}

Workflow notifications¶

Conductor can be configured to publish notifications to external systems upon completion/termination of workflows. See extending conductor for details.