服务编排–Conductor 文档翻译 (介绍与基本概念)

• 2019 年 11 月 21 日
• 筆記

• 介绍
  • Conductor 优点
  • 为什么不进行点对点编排？
• 基本概念
  • 工作流定义
  • 任务定义
  • 系统任务
  • 工人任务
  • 工作流任务的生命周期
• 元数据定义
  • 任务定义
  • 重试逻辑
  • 超时政策
• 工作流定义
  • 工作流程中的任务
  • 连接输入和输出
  • $ {SOURCE.input / output.JSONPath}
  • JSON路径支持
• 系统任务的创建
  • (DYNAMIC) 动态任务定义
  • （DECIDE）决策任务定义
  • Fork 并行任务定义
  • Dynamic Fork （动态分支）
  • Join
  • Join 任务输出
  • 子工作流程
  • Wait
  • SQS队列
  • HTTP
    • 参数
  • Event (事件)
  • 支持的接收器
  • 事件任务输入
  • 事件任务输出

本文是对 Conductor 文档的简单翻译，建议你认真阅读，如果阅读后你仍然不知道如何使用，可以继续关注本博客，我会在后续的博客中更新 Conductor 实战

介绍

Conductor是一个微服务的编排引擎

Conductor 优点

Conductor，帮助我们协调基于微服务的流程，具有以下功能：

• 允许创建复杂的流程/业务流，其中由微服务实现单个任务。
• 基于JSON DSL的定义执行流程。
• 为这些流程提供可见性和可追溯性。
• 在暂停，恢复，重启等周围公开控制语义，以获得更好的devops体验。
• 允许更多地重用现有的微服务，为管理提供更容易的途径。
• 用户界面可视化流程。
• 能够在需要时同步处理所有任务。
• 能够扩展数百万个并发运行的流程。
• 由客户端提取的排队服务支持。
• 能够在HTTP或其他传输上运行，例如gRPC。

为什么不进行点对点编排？

通过点对点任务编排，我们发现随着业务需求和复杂性的增长难以扩展。发布/订阅模型适用于最简单的流程， 但很快就突出了与该方法相关的一些问题：

• 流程“嵌入”在多个应用程序的代码中。
• 通常，围绕输入/输出，SLA等存在紧密耦合和假设，使得更难以适应不断变化的需求。
• 几乎没有办法系统地回答“我们用过程X做了多少”？

基本概念

工作流定义

工作流是使用基于JSON的DSL定义的，包括一组作为工作流的一部分执行的任务。任务是在远程机器上执行的控制任务（fork，条件等）或应用程序任务（例如编码文件）。

任务定义

• 所有任务都需要在活动工作流程使用之前进行注册。
• 任务可以在多个工作流程中重复使用。工人任务分为两类：
  • 系统任务
  • 工人任务

系统任务

系统任务在Conductor服务器的JVM内执行，并由Conductor管理，以实现其可执行性和可扩展性。

Conductor提供了一个API来创建在与引擎相同的JVM中执行的用户定义任务。有关详细信息，请参阅WorkflowSystemTask接口。

工人任务

工作人员任务由应用程序实现，并在与Conductor不同的环境中运行。工作人员任务可以用任何语言实现。这些任务通过REST API端点与Conductor服务器通信，以轮询任务发现并执行，并在执行后更新其状态。

工作人员任务由计划中的任务类型SIMPLE标识。

工作流任务的生命周期

元数据定义

任务定义

Conductor维护着一个工作人员任务类型的注册表。在工作流程中使用之前必须注册任务类型。

例

{    "name": "encode_task",    "retryCount": 3,    "timeoutSeconds": 1200,    "inputKeys": [      "sourceRequestId",      "qcElementType"    ],    "outputKeys": [      "state",      "skipped",      "result"    ],    "timeoutPolicy": "TIME_OUT_WF",    "retryLogic": "FIXED",    "retryDelaySeconds": 600,    "responseTimeoutSeconds": 3600  }

重试逻辑

• FIXED ：重新安排任务后的任务 retryDelaySeconds
• EXPONENTIAL_BACKOFF：重新安排之后 retryDelaySeconds * attempNo

超时政策

• RETRY ：再次重试该任务
• TIME_OUT_WF：工作流程标记为TIMED_OUT并终止
• ALERT_ONLY：注册计数器（task_timeout）

工作流定义

使用基于JSON的DSL定义工作流。

例

{    "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}"    },    "schemaVersion": 2  }

工作流程中的任务

tasks工作流中的属性定义要按该顺序执行的任务数组。以下是每项任务所需的强制性最低参数：

除了这些参数，需要进行具体的任务类型附加参数记录在这里

连接输入和输出

当触发新的执行时，客户端会为工作流提供输入。工作流输入是通过${workflow.input…}表达式提供的JSON有效负载。

基于inputParameters工作流定义中配置的模板，为工作流中的每个任务提供输入。 inputParameters是一个JSON片段，其值包含用于在执行期间映射工作流的输入或输出或其他任务的值的参数。

映射值的语法遵循以下模式：

$ {SOURCE.input / output.JSONPath}

JSON路径支持

Conductor支持JSONPath规范并从此处使用Java实现。

例

考虑一个任务，其输入配置为使用来自工作流的输入/输出参数和名为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"        }      }    }  }

请将以下内容视为工作流输入

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

并且loc_task的输出如下;

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

在安排任务时，Conductor将合并工作流输入和loc_task输出中的值，并按如下方式创建任务输入：

{    "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"      }    }  }

系统任务的创建

(DYNAMIC) 动态任务定义

参数：

例

{    "name": "user_task",    "taskReferenceName": "t1",    "inputParameters": {      "files": "${workflow.input.files}",      "taskToExecute": "${workflow.input.user_supplied_task}"    },    "type": "DYNAMIC",    "dynamicTaskNameParam": "taskToExecute"  }

如果使用输入参数user_supplied_task的值作为user_task_2启动工作流，则Conductor将在计划此动态任务时调度user_task_2。

（DECIDE）决策任务定义

决策任务类似于case…switch编程语言中的语句。该任务需要3个参数：

参数：

例

{    "name": "decide_task",    "taskReferenceName": "decide1",    "inputParameters": {      "case_value_param": "${workflow.input.movieType}"    },    "type": "DECISION",    "caseValueParam": "case_value_param",    "decisionCases": {      "Show": [        {          "name": "setup_episodes",          "taskReferenceName": "se1",          "inputParameters": {            "movieId": "${workflow.input.movieId}"          },          "type": "SIMPLE"        },        {          "name": "generate_episode_artwork",          "taskReferenceName": "ga",          "inputParameters": {            "movieId": "${workflow.input.movieId}"          },          "type": "SIMPLE"        }      ],      "Movie": [        {          "name": "setup_movie",          "taskReferenceName": "sm",          "inputParameters": {            "movieId": "${workflow.input.movieId}"          },          "type": "SIMPLE"        },        {          "name": "generate_movie_artwork",          "taskReferenceName": "gma",          "inputParameters": {            "movieId": "${workflow.input.movieId}"          },          "type": "SIMPLE"        }      ]    }  }

Fork 并行任务定义

Fork用于调度并行任务集。

参数：

例

{    "forkTasks": [      [        {          "name": "task11",          "taskReferenceName": "t11"        },        {          "name": "task12",          "taskReferenceName": "t12"        }      ],      [        {          "name": "task21",          "taskReferenceName": "t21"        },        {          "name": "task22",          "taskReferenceName": "t22"        }      ]    ]  }

执行时，task11和task21被安排在同一时间执行。

Dynamic Fork （动态分支）

Dynamic fork与FORK_JOIN任务相同。除了在运行时使用任务的输入提供要并行的任务列表。当并行的任务数量不固定并根据输入而变化时很有用。

例

{    "inputParameters": {       "dynamicTasks": "${taskA.output.dynamicTasksJSON}",       "dynamicTasksInput": "${taskA.output.dynamicTasksInputJSON}"    }    "type": "FORK_JOIN_DYNAMIC",    "dynamicForkTasksParam": "dynamicTasks",    "dynamicForkTasksInputParamName": "dynamicTasksInput"  }

将taskA的输出视为：

{    "dynamicTasksInputJSON": {      "forkedTask1": {        "width": 100,        "height": 100,        "params": {          "recipe": "jpg"        }      },      "forkedTask2": {        "width": 200,        "height": 200,        "params": {          "recipe": "jpg"        }      }    },    "dynamicTasksJSON": [      {        "name": "encode_task",        "taskReferenceName": "forkedTask1",        "type": "SIMPLE"      },      {        "name": "encode_task",        "taskReferenceName": "forkedTask2",        "type": "SIMPLE"      }    ]  }

执行时，Dynamic fork 任务将调度两个类型为“encode_task”的并行任务，引用名称为“forkedTask1”和“forkedTask2”，输入由_ dynamicTasksInputJSON_指定

Dynamic Fork and Join

Join任务必须遵循FORK_JOIN_DYNAMIC

工作流定义必须包含一个Join任务定义，后跟FORK_JOIN_DYNAMIC任务。但是，考虑到任务的动态特性，此Join不需要joinOn参数。在完成之前，连接将等待所有并行分支任务完成。

与FORK不同，FORK可以执行并行流，每个fork按顺序执行一系列任务，FORK_JOIN_DYNAMIC仅限于每个fork一个任务。但是，并行任务可以是子工作流，允许更复杂的执行流。

Join

Join任务用于等待fork任务生成的一个或多个任务的完成。

参数

例

{      "joinOn": ["taskRef1", "taskRef3"]  }

Join 任务输出

Fork任务的输出将是一个JSON对象，其中key是任务引用名称，value是fork任务的输出。

子工作流程

子工作流任务允许在另一个工作流中嵌套工作流。

参数

例

{    "name": "sub_workflow_task",    "taskReferenceName": "sub1",    "inputParameters": {      "requestId": "${workflow.input.requestId}",      "file": "${encode.output.location}"    },    "type": "SUB_WORKFLOW",    "subWorkflowParam": {      "name": "deployment_workflow",      "version": 1    }  }

执行时，deployment_workflow使用两个输入参数requestId和file执行a 。生成的工作流程完成后，任务标记为已完成。如果子工作流终止或失败，则任务被标记为失败并在配置时重试。

Wait

Wait 任务被实现为保持在IN_PROGRESS状态的门，除非标记为外部触发器COMPLETED或FAILED由外部触发器标记。要使用Wait任务，请将任务类型设置为WAIT

参数 没有要求

Wait 任务的外部触发器

任务资源端点可用于将任务的状态更新为终止状态。

Contrib模块提供SQS集成，外部系统可以将消息放入服务器侦听的预配置队列中。当消息到达时，它们被标记为COMPLETED或FAILED。

SQS队列

• 可以使用以下API检索服务器用于更新任务状态的SQS队列：

GET /queue

• 更新任务状态时，消息需要符合以下规范：
  • 消息必须是有效的JSON字符串。
  • 消息JSON应包含一个名为key的键externalId，该值是一个包含以下键的JSONified字符串：
  • workflowId：工作流程的ID
  • taskRefName：应更新的任务引用名称。
  • 每个队列代表一个特定的任务状态，并相应地标记任务。例如，发送到COMPLETED队列的消息将任务状态标记为COMPLETED。
  • 任务的输出随消息更新。

示例SQS有效负载：

{    "some_key": "valuex",    "externalId": "{"taskRefName":"TASK_REFERENCE_NAME","workflowId":"WORKFLOW_ID"}"  }

HTTP

HTTP任务用于通过HTTP调用另一个微服务。

参数

该任务需要一个输入参数http_request，该参数作为任务输入的一部分，具有以下详细信息：

HTTP任务输出

例

任务使用vipAddress输入有效负载

{    "http_request": {      "vipAddress": "examplevip-prod",      "uri": "/",      "method": "GET",      "accept": "text/plain"    }  }

任务使用绝对URL输入

{    "http_request": {      "uri": "http://example.com/",      "method": "GET",      "accept": "text/plain"    }  }

该任务被标记为FAILED无法完成请求或远程服务器返回非成功的状态代码。

注意

HTTP任务当前仅支持Content-Type作为application / json，并且能够解析文本以及JSON响应。目前不支持XML输入/输出。但是，如果无法将响应解析为JSON或Text，则将字符串表示形式存储为文本值。

Event (事件)

事件任务提供将事件（消息）发布到Conductor或外部事件系统（如SQS）的功能。事件任务对于为工作流和任务创建基于事件的依赖项非常有用。

参数

例

{      "sink": 'sqs:example_sqs_queue_name'  }

使用Conductor作为接收器生成事件时，事件名称遵循以下结构： conductor::

对于SQS，请使用队列的名称而不是URI。Conductor根据名称查找URI。

警告

使用SQS时，将ContribsModule添加到部署中。需要使用AWSCredentialsProvider为Conductor配置模块，以便能够使用AWS API。

支持的接收器

• Conductor
• SQS

事件任务输入

给予事件任务的输入可作为有效负载用于已发布的消息。例如，如果消息被放入SQS队列（接收器是sqs），则消息有效负载将是任务的输入。

事件任务输出

event_produced 生成的事件的名称。