工作流系统的设计

几年前曾经写过一点点对于缓存框架设计的体会，这大半年和工作流系统打交道颇为丰富，因此想总结一点关于工作流系统的设计。

首先，明确工作流（workflow）系统的定义。 维基百科 上有极其简单的介绍。我记得以前在文章里面说过，作为大公司里面的小team，为了做一些有趣的东西，从而更好的招人，通常有几个众人皆知的突破口：比如一个更符合业务需求的storage，再比如一个自定义的工作流系统（workflow）。在Amazon内部，我接触过好多个workflow，而且大多以 Amazon SWF 为原型（当时学习的时候还写了一点体会，link 1和link 2），于是宏观上看，60%的东西是一样的，大同小异；但是也有很多重要的元素大不相同，而它们被放到一起比较也是常事。几次折腾之后，我也慢慢在思考，如何去设计一个工作流系统，其中都有哪些重要的需要考虑到的方面。

Scalability

基本上随便设计什么基础设施，扩展性都是重要的考虑内容。作为workflow来讲，基本上工作节点的水平扩展是考量扩展性的最重要标志。既然工作节点可以水平扩展，那么这就意味着任务（task）必须是以pull的方式由工作节点主动去获取，而不是由pull的方式从调度节点来分配（曾经非常简单地比较过pull和push，但其实二者差异远不止文中内容之浅显）。任务的分配上，需要考虑这样的事情：如果有多个工作节点尝试来pull任务，该分配给谁？具体来说，比如这样的例子：如果每一个task节点允许同时执行5个任务，而现在可同时执行的总任务数只有5个，总共的task节点也有5个，最理想的状态应当是这5个被均匀分配到这5个节点去，但是采用简单的pull机制并不能保证这一点，有可能这5个任务全部跑到一台机器上去了，因为这并不超过一个节点可同时执行任务数量的上限。

另一方面，通常来讲，所有任务都应当是idempotent的，即可以重复提交执行，执行若干次和执行一次的结果是一样的。工作节点的任务执行可以在任意一步发生错误，随着节点数量的增加，这样的错误更多地成为一种常态，而不是“异常”。工作节点的健康状态需要由某种方式来维系和通知，最典型和廉价有效的方式就是“心跳”，我曾经写过一篇文章详细介绍一种心跳系统的设计，感兴趣的话，欢迎移步阅读。

功能性解耦

• 资源管理和任务管理解耦。这一点我只在少数workflow里面见到。任务管理几乎是所有workflow都具备的，但是单独的资源管理则不是。举例来说，我可以写一个task去执行 EMR 上的任务，你也可以写一个task去EMR上执行，EMR的执行管理逻辑，可以以代码的方式被我们共用——但是这种架构下，你的task和我的task很难安全高效地共享同一个EMR资源，无论是资源的创建、销毁，状态的查询，还是throttling，都变得很麻烦。类似的例子还有，数据库的共享，打印机的共享，甚至另外一个工作流系统的共享。当有开销较大的资源，我们经常需要workflow层面被统一管理起来，管理一份或者几份资源，但是共享给数量众多的task。
• 业务逻辑和调度逻辑解耦。这基本上在所有workflow里面都具备，调度逻辑是业务无关的，也是相对来说“死”的东西，管理工作流的状态，和每个task的成功失败。但是业务逻辑则是组成workflow和其中的task“活生生”的血肉。我还没有见过哪个workflow把业务代码和调度逻辑写到一起。
• 状态查询和调度系统的解耦。一个完善的工作流系统，调度只能是其中核心的一个方面，如果没有一个好的状态查询系统，维护的工作量将是巨大的。而这二者，必须解耦开。举例来说，工作流和任务执行的状态，必然是持久化在某种存储介质中，比如关系数据库，比如NoSQL的数据库，比如磁盘日志文件等等。这个时候，调度系统可以说是这些信息写入存储系统的最主要来源，而这些信息的读取，则可能从调度系统读取，也可能从状态查询系统读取。这个存储的格式或者说schema，必须相对稳定。这个存储的一致性和可用性，将是整个系统一致性和可用性的核心组成部分。
• 事件系统和监听系统解耦。涉及这个的工作流只占少数。很多工作流系统都有内部的事件系统，比如某个task分配给某个节点了，某个task执行失败了等等，但是这样事件的监听系统，却没有独立出来，导致后续针对特殊事件要执行特定逻辑变得困难。

同步与异步任务

事实上，当考虑到了独立的资源管理功能，异步和同步任务的划分就变得自然而然。

• 有很多任务是需要在当前的工作节点上执行的。比如需要在工作节点上下载一个文件，然后经过处理以后写到数据库里去，这些任务消耗大量的内存和CPU，需要分配独立的专属的线程去完成，是同步任务。
• 还有一些任务，工作节点并非实际的工作执行者，而是针对某一个资源系统的客户端，只负责提交任务到该系统内，并且负责管理和监控。比如打印任务，向打印机提交打印请求，然后只需要不断地向打印机查询任务的状态，以及根据需要作出删除任务和重新提交等操作即可。这些任务通常不需要长期占有线程，一个线程可以在一个周期内处理多个任务。它们是异步任务。另外，举一个特例，工作流的嵌套，即工作流调用子工作流，那么对于子工作流状态的查询这个行为来说，必然是异步任务。异步任务就涉及到事件的通知和监听机制，后文有提到。

分布式锁

在某些情况下，分布式锁变成一个必选项。比如前面提到的资源管理。有许多资源是要求操作是独占的，换言之，不支持两个操作并发调用，期间可能出现不可以预料的问题；另一方面，一个节点在对资源进行操作时，它需要和别的节点进行协作，从而两个工作节点的操作是有序和正确的，不至于发生冲突。

举个例子来说，工作节点A要查询当前EMR的状态，如果已经空闲10分钟，就要执行操作结束掉这个EMR资源；而工作节点B则查询该EMR的状态，如果没有被结束掉，就要往上面提交新的计算任务。这时候，如果没有分布式锁的协作，问题就来了，可能B节点先查询发现EMR状态还活着，就这这一瞬间，A节点结束了它，可是B不知道，接着提交了一个计算任务到这个已经结束了的（死了的）EMR资源上，于是这个提交的计算任务必然执行失败了。

有很多分布式锁的实现方式，简单的有强一致性的存储系统，当然也有更高效的实现，比如一些专门的分布式锁系统。

功能的可扩展性

之前讲到了性能架构上的可扩展性，在功能层面亦然。

• 自定义任务。这是几乎所有工作流系统都会考虑的事情，这也是业务逻辑和调度逻辑解耦的必然。因为工作流系统设计的时候，必然没法预知所有的任务类型，用户是可以定义自己的执行逻辑的。
• 自定义资源。有了资源管理，就有自定义资源的必要。
• 自定义事件监听。事件管理通常在工作流系统中是很容易被忽视的内容，比如我希望在某一个task超时的时候发送一个特殊的消息通知我，这就需要给这个事件监听提供扩展的可能性。
• 运行时的工作流任务执行条件。通常workflow都会有一个定义如何执行的文件（meta file），但是有一些执行的参数和条件，是在运行时才能够确定的，甚至依赖于上一步执行的结果，或者需要执行一些逻辑才能得到。

可用性和可靠性

大多数workflow，都采用了去中心节点的设计，保证不存在任何单点故障问题。所有的子系统都是。也保证在业务压力增加的情况下，标志着可用性的latency在预期范围之内。其它的内容不展开，介绍这方面的文章到处都是。

生命周期管理

这里既指workflow一次执行的生命周期管理，也指单个task的生命周期管理。

谈论这些必然涉及到这样几个问题：

• workflow definition和workflow的分离，task definition和task execution的分离。其中definition定义执行的逻辑，而execution才真正和执行的环境、时间、参数等等相关。逻辑通常可以只有一份（但这也不一定，要看workflow是否支持多版本，后文有提到），但是execution随着重试的发生，会保存多份。
• workflow重试时，参数变化的处理。有些参数的变化，是不会影响已完成任务的，但是有的参数则不是。
• workflow重试时，对于已完成任务的处理。有的情况我们希望已完成任务也要重新执行，而又的情况我们则希望这些已完成任务被跳过。
• task的重试次数，以及重试时back off的策略。比如第一次重试需要等5分钟，第二次重试需要等10分钟，最多重试2次。
• 如何礼貌地结束工作节点上的任务执行。在很多情况下我们不得不中断并结束某个节点上的任务执行，比如这个工作节点需要重启，这并不能算作业务代码导致的任务执行失败，而更像是一种“resource termination”。这种情况下，任务通常需要被分配到另外活着的节点去，而这里有牵涉到这个reallocation的策略，前面已经提到过。

任务 DAG 的设计和表达

这是workflow执行的流程图，也是所有task之间依赖关系的表述。我见过多种表达方式的，有XML的，也有JSON的，还有一些不知名的自己定义的格式的。有些workflow的定义可以以一个图形化工具来协助完成这个流程图。这个DSL的设计，一定程度上决定了workflow的使用是不是能够易于理解。另外提一句，这里提到的这个可选的图形化工具，毕竟只是一个辅助，它不是workflow的核心（你可以说这个DSL是核心的一部分，但这个帮助完成的工具显然不是）——我见过一个团队，workflow整体设计得不怎么样，跑起来一堆问题，但是这个工具花了大量的时间精力去修缮，本末倒置。

输入输出的管理

这也是一个nice-to-have的东西，对于每一个task，都存在input和output，它们可以完全交给用户自己来实现，比如用户把它们存储到文件里面，或者写到数据库里面，而workflow根本不管，每个task内部自己去读取相应的用户文件即可。但是更好的方法是，对于一些常用和简单的input、output，是可以随着execution一起持久化到workflow和task的状态里面去的。这样也便于workflow的definition里面，放置一些根据前一步task执行结果来决策后续执行的表达式。

另外，还有一个稍微冷门的use case，就是input和output的管理。通常workflow是重复执行的，而每次执行的input和output的数据规模往往是很多人关心的内容。关于这部分，我还没有见到任何一个workflow提供这样的功能。许多用户自己写工具和脚本来获取这样的信息。

独立的metrics和日志系统

对于metrics，核心的内容也无非节点的健康状况、CPU、内存，task执行时间分布，失败率等等几项。有些情况下用户还希望自行扩展。

关于日志，则主要指的是归档和合并。归档，指的是历史日志不丢失，或者在一定时间内不丢失，过期日志可以被覆写，从而不引起磁盘容量的问题；而合并，指的是日志能被以更统一的视角进行查询和浏览，出了问题不至于到每台机器上去手动查找。缺少这个功能，有时候会很麻烦。在工作中我遇到过一个资源被异常终止的问题，为了找到那个终止资源的节点，我查阅了几十个节点的日志，痛苦不堪。

版本控制和平滑部署

把这两个放一起是因为，代码升级是不可避免且经常要发生的。为了保证平滑部署，显然通常情况下，节点上的代码不能同时更新，需要一部分一部分进行。比如，先终止50%的节点，部署代码后，激活并确保成功，再进行剩下那50%的节点。但是在这期间存在新老代码并存的问题，这通常会带来很多奇形怪状的问题。对于这种问题，我见过这样两个解决方式：

• 一个是全部节点同时部署，这种情况下所有节点全部失活，有可能出现因为这个失活导致的task超时，甚至导致workflow执行失败。但是workflow的生命周期由单独的调度系统管理，因此除去超时外多数情况不受影响。
• 还有一种是一部分一部分部署，最平滑，但是这种情况下需要管理多版本共存问题，也对代码质量提出了新的要求——向后兼容。

无论选用哪一种，这种方式实现起来相对简单，但是也有不少问题，比如这种情况下，外部资源怎么处理？例如在外部EMR资源上执行Spark任务，但是已经有老代码被放到EMR上去执行了，这时候工作节点更新，这些EMR上正在执行的任务怎样处理？是作废还是保留，如果保留的话这些执行可还是依仗着老代码的，其结果的后续处理是否会和刚部署的新代码产生冲突。再比如对于workflow有定义上的改变（比如DAG的改变），对于现有的execution，应当怎样处理，是更新还是保持原样（通常都是保持原样，因为更新带来的复杂问题非常多）。

先说那么多吧，思路不够清晰，欢迎讨论并一起完善。