How to Design Workflow Sytem

https://docs.aws.amazon.com/amazonswf/latest/awsflowguide/swf-timeout-types.html
https://docs.aws.amazon.com/amazonswf/latest/awsflowguide/awsflow-basics-reliable-execution.html

Ensuring that Results are Not Lost

Maintaining Workflow History

An activity that performs a data-mining operation on a petabyte of data might take hours to complete, and an activity that directs a human worker to perform a complex task might take days, or even weeks to complete!

To accommodate scenarios such as these, AWS Flow Framework workflows and activities can take arbitrarily long to complete: up to a limit of one year for a workflow execution. Reliably executing long running processes requires a mechanism to durably store the workflow's execution history on an ongoing basis.

The AWS Flow Framework handles this by depending on Amazon SWF, which maintains a running history of each workflow instance. The workflow's history provides a complete and authoritative record of the workflow's progress, including all the workflow and activity tasks that have been scheduled and completed, and the information returned by completed or failed activities.

AWS Flow Framework applications usually don't need to interact with the workflow history directly, although they can access it if necessary. For most purposes, applications can simply let the framework interact with the workflow history behind the scenes. For a full discussion of workflow history, see Workflow History in the Amazon Simple Workflow Service Developer Guide.

Stateless Execution

The execution history allows workflow workers to be stateless. If you have multiple instances of a workflow or activity worker, any worker can perform any task. The worker receives all the state information that it needs to perform the task from Amazon SWF.

This approach makes workflows more reliable. For example, if an activity worker fails, you don't have to restart the workflow. Just restart the worker and it will start polling the task list and processing whatever tasks are on the list, regardless of when the failure occurred. You can make your overall workflow fault-tolerant by using two or more workflow and activity workers, perhaps on separate systems. Then, if one of the workers fails, the others will continue to handle scheduled tasks without any interruption in workflow progress.

http://www.raychase.net/3758

最近工作中一直和SWF（Amazon的Simple Work Flow）打交道，在一个基于SWF的工作流框架上面开发和修bug。SWF的activity超时时间是5分钟，在activity task开始执行以后，activity worker需要主动发送心跳请求告知service端：“我还活着，我还在干活”，如果出现超过5分钟（可以配置）没有心跳，SWF的service端就认为，你已经挂了，我需要把这个activity安排到别的activity worker上来执行了。借用AWS官网的一张图：

每台机器上有若干个activity task在被执行。可以看到，在activity任务启动起来以后，需要用不断的心跳来告知service端任务还在进行，activity worker还活着。这个“汇报”需要activity worker所在的host主动进行，这也是SWF的service端无状态（几年前写过一点东西介绍它）的基本要求之一。任务都是由worker端去pull的，这些行为也都是worker端主动触发的。

这个机制描述起来很简单，但是实际在相关设计实现的时候，有许多有趣和值得琢磨的地方。

在我手头的这个workflow里面，心跳机制是这样实现的：

• 有两个queue，一个是main queue，是dequeue（双端队列）；另一个是backup queue，普通队列。二者都是用来存放需要发送心跳的activity信息（heartbeatable对象）。
• 每秒钟都尝试执行这样一个方法A：从main queue里面poll一个heartbeatable对象（如果queue为空就忽略本次执行），检查该心跳所代表的activity task是否还在工作，如果是，就发送一个心跳。发送成功以后，就把这个heartbeatable对象扔到backup queue里面去。这样，一秒一个，逐渐地，main queue的heartbeatable对象全部慢慢被转移到backup queue里去了。
• 每隔两分钟（称为一个cycle）执行方法B：把backup queue里面所有的heartbeatable对象全部转移到main queue里去，于是就又可以继续执行上面一步的逐个心跳逻辑。

这个机制的基本好处是，所有activity task的心跳统一管理，通常情况下保证了心跳不会过快（默认配置下是一秒一个，或者不发送），同时保证了没有谁会被遗漏：

但是，这里又会浮现好多好多问题：

为什么要使用两个queue？

首先，有这样一个事实：方法A在执行的时候，理论上每秒钟会执行一次，但是这里并没有强制的保证，使得前一秒的A执行一定会在这一秒的A开始之前完成。换言之，它们的理论启动时间是按序的，但是实际启动时间和实际的心跳执行时间是不定的，需要处理并发的情形。而到底最多可能存在多少个执行A的线程并行，取决于用于此心跳功能的线程池的配置。因此，在执行和判断的过程中，需要对当前poll出来的heartbeatable对象加锁。

使用两个queue，这主要是为了记录在本次cycle里面，能够很容易判断某一个heartbeatable对象是否已经完成心跳行为。还没有完成心跳的，都在main queue里；完成了的，都放到backup queue里。

如果使用一个queue，那么也是有解决方案的：

• 有一个公共计数器，每个cycle开始的时候，给计数器+1。
• 每一个heartbeatable对象自身需要携带一个私有计数器，用以标识当前这轮cycle的心跳是否已经完成。
• 每次完成的heartbeatable对象给自己的计数器+1以后扔到队尾；每次A取新的heartbeatable对象的时候从队首取。
• 如果取到的对象自己的计数器已经等于公共计数器的数值，说明整个queue里面的对象心跳都已经完成了。

当然，这种方法的弊端在于，判断是否还需要发送心跳这件事情，不仅需要从queue里取对象，还要判断对象的计数器数值，明显比两个queue的解决方案复杂和开销大。因为两个queue的解决方案下，只需要尝试从main queue里面取对象就好，取不到了就说明本次cycle里没有需要发送心跳的对象了。看起来是多了一个queue，但是方案其实还是简单一些。

心跳的频率保持在多久为好？

显然不是越高越好，不只是成本，因为心跳也是需要消耗资源的，比如CPU资源；而且，心跳在service端也有throttling，当前activity worker发起太频繁的心跳，当前心跳可能被拒，还可能会让别的activity worker的正常心跳被拒了。

我们要解决的最核心问题是，正常情况下，必须保持上限5分钟内能发起一次成功心跳就好。

要这么说，尽量增大cycle，那我设计一个每隔5分钟就执行一次的定时器就好了。但是问题没那么简单，首先要考虑心跳的发起不一定成功。如果在接近5分钟的时候才去尝试发起心跳，一旦失败了，也没有时间重试了。因此，要trade-off。比如，配置cycle为120秒，这样的好处是，5分钟的超时时间内，可以覆盖1~2个完整的cycle。如果cycle配置为3分钟，那么5分钟无法严格保证一定覆盖有一个完整的cycle。

确定心跳频率的有两个重要参数，一个是方法A的执行频率，一个则是一个cycle的时间长度。例如，前者为1 per second，后者为2分钟，那么在理想情况下，一个cycle 120秒，可以处理120个activity task，换言之，极限是120个activity task在这台机器上一起执行。超过了这个数，就意味着在一个cycle内，无法完成所有的心跳发送任务。

当然，实际情况没有那么理想，考虑到暂时性的网络问题，线程、CPU资源的竞争等等，实际可以并行的activity task要比这个数低不少。

异常处理和重试

在上图中，步骤③有三个箭头，表示了心跳出现不同种情形的处理：

• 有一些常规异常，比如表示资源不存在，或者任务已经cancel了，这种情况发生的时候，要把相应的activity task给cancel掉，同时，把自己这个heartbeatable对象永久移除出queue。
• 重试情形1：throttling导致的异常，这种异常发生的时候，把当前heartbeatable对象再addFirst回main queue，因为这不是当前有什么不可解决的或者不明原因的问题造成的，只需要简单重试即可。
• 重试情形2：其它未知原因的异常，这种情况当然需要重试（之前我们缺少这样的重试机制，导致下一次该activity task能够得到心跳的机会被推到了下一个cycle，这显然是不够合理的），但是，可以把heartbeatable对象放到queue尾部去重试（addLast），并且附上一个私有计数器，如果重试超过一定次数，就挪到下一个cycle（backup queue）去。这个放到queue尾部的办法，使得重试可以在当前cycle里进行，又可以使得这个重试能够尽量不影响其他heartbeatable对象的心跳及时发送。整个重试过程其实就是把当前失败对象再放回queue的过程，没有线程阻塞。

曾经遇到过一些这方面的问题，经过改进才有了上述的机制：

在CPU或者load达到一定程度的时候（比如这个时候有一个进程在call service，占用了大量的CPU资源），就很容易发生心跳无法及时进行的问题，比如有时候线程已经初始化了，但是会stuck若干时间，因为没有足够的资源去进行。等到某一时刻，资源被释放（比如这个call service 的进程结束），这个时候之前积攒的心跳任务会一下子爆发出来。不但这些心跳的顺序无法保证，而且严重的情况下会导致throttling。如果没有当前cycle内的重试机制，那么下一次该对象的心跳需要等到下一个cycle，很容易造成activity task的timeout。

下面再说一个和心跳异常有关的问题。

有这样一个例子，在这个工作流框架内，我们需要管理EMR资源，有一个activity把EMR cluster初始化完成，另一个activity把实际执行的steps提交上去。但是发现在实际运行时有如下的问题：EMR cluster已经初始化完成，但是steps迟迟没有办法提交上去，导致了这个cluster空闲太长时间，被框架内的monitor认为已经没有人使用了，需要回收，于是EMR cluster就被terminate了。但是这之后，steps才被提交上去，但是这时候cluster已经处于terminating状态了，自然这个step提交就失败了。而经过分析，造成这个EMR cluster非预期的termination，包括这样几个原因：

• decision task timeout。在EMR cluster创建好之后，SWF会问decider下一步该干嘛，这时候如果因为CPU高负荷等各种原因，导致decision task timeout，SWF就会一直等在那里，而如果这个timeout的时间配得太长，这段超时就足以让上面的这个EMR cluster空闲过长时间导致被误回收了。
• 判断EMR cluster空闲到一定时间就要回收的逻辑有问题。我们以前的实现是，每隔2分钟执行一次“EMR资源操作”，包括检查资源状态，进行资源操作，然后如果发现该EMR资源创建后经过了4次资源操作，依然没有step提交上去，就认为空闲时间过长，需要回收（2 x 4 = 8分钟）。但是问题在于，实际由于种种原因（和心跳的执行间隔实际时间不确定的原理一样），间隔执行EMR资源操作并不能严格保证每隔2分钟一次，有时一段时间都得不到执行，而有时候会迎来一次集中爆发，这个时候就可能实际EMR资源空闲了远远不到8分钟就被回收了。因此，这个逻辑最好是能够用绝对的“空闲时间”来判断，例如EMR资源创建时记录好时间，之后每次检查时都用当前时间去和创建时间比较，空闲超过8分钟再回收。
• 由于之前提到过的心跳无法按时完成导致activity task timeout，于是这个EMR cluster创建的任务实际已经完成了，但是被当做超时给无视了。

最后，我想说的是。设计一个好的工作流框架，还是有很多困难的地方，需要尤其考虑周全的地方。即便是基于SWF这样现有的workflow来搭积木和叠加功能，也有很多不易和有趣的地方。

http://www.raychase.net/3998

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的策略，前面已经提到过。
• 任务的权重。或者叫做优先级，这项功能我只在少数workflow中看到。在考虑到资源分配时，某些更重要任务可具备更高优先级，而无关紧要的任务失败甚至可以不影响workflow的状态。

任务DAG的设计和表达

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

另外，workflow的状态和执行情况，还有对其的归档和管理，也需要一个整合工具来协助。这方面几乎所有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，应当怎样处理，是更新还是保持原样（通常都是保持原样，因为更新带来的复杂问题非常多）。

http://www.raychase.net/244
观察者模式中，消息采用推和拉方式来传递的比较

更新自己。

现在要说的分歧在这里：

“推”的方式是指，Subject维护一份观察者的列表，每当有更新发生，Subject会把更新消息主动推送到各个Observer去。

“拉”的方式是指，各个Observer维护各自所关心的Subject列表，自行决定在合适的时间去Subject获取相应的更新数据。

“推”的好处包括：

1、高效。如果没有更新发生，不会有任何更新消息推送的动作，即每次消息推送都发生在确确实实的更新事件之后，都是有意义的。

2、实时。事件发生后的第一时间即可触发通知操作。

3、可以由Subject确立通知的时间，可以避开一些繁忙时间。

4、可以表达出不同事件发生的先后顺序。

“拉”的好处包括：

1、如果观察者众多，Subject来维护订阅者的列表，可能困难，或者臃肿，把订阅关系解脱到Observer去完成。

2、Observer可以不理会它不关心的变更事件，只需要去获取自己感兴趣的事件即可。

3、Observer可以自行决定获取更新事件的时间。

4、拉的形式可以让Subject更好地控制各个Observer每次查询更新的访问权限。

事实上“推”和“拉”可以比较的内容太多了，比如：

客户端通常是不稳定的，服务端是稳定的，如果消息由客户端主动发起去获取，它很容易找到服务端的地址，可以比较容易地做到权限控制（集中在服务端一处），服务端也可以比较容易地跟踪客户端的位置和状态，反之则不行；

互联网页面的访问就是一个最好的“拉”的模式的例子；

通常我们希望把压力分散到各个客户端上去，服务端只做最核心的事情，只提供内容，不管理分发列表；

……

还有一个idea是关于“推”和“拉”结合的形式，例如，服务端只负责通知某一些数据已经准备好，至于是否需要获取和什么时候客户端来获取这些数据，完全由客户端自行确定。