文|作业帮大数据团队（阮文俊、孙建业）
背 景
基于 Apache DolphinScheduler （以下简称DolphinScheduler）搭建的 UDA 任务调度平台有效支撑了公司的业务数据开发需求，处理着日均百万级别的任务量。
整个 UDA 的架构如下图所示，其中我们的引擎层主要基于 DolphinScheduler 的 Master 和 Worker 服务搭建，充分利用了 Master 节点优秀的高并发调度能力、worker 节点的分组隔离能力。我们通过一层适配层对 DolphinScheduler 对外暴露的接口进行了封装和增强来适配我们的业务场景。
随着使用的深入，我们发现 DolphinScheduler3.0.0 版本中存在一些难以通过运维手段规避的问题，这些问题影响数据平台的稳定性，导致隔一段时间需要重启服务来使集群恢复正常，并且核心组件对外暴露的可观测性指标十分有限，导致问题的排查定位过程十分繁琐。我们认为一个可以稳定运行的调度引擎应该具备以下能力
-
反压与过载保护机制：当系统负载比较高的时候，能够自动推迟任务执行，以保护自己。
-
任务分配均衡：保证任务分配的均衡性，避免任务倾斜问题。
-
无副作用运行：调度引擎在运行过程中不应对自身产生“副作用”，确保能够持续的长时间稳定运行。
-
集群可观测性：具备全面的可观测性指标，能够通过这些指标评估集群的健康状态。
在这些能力上，开源版 3.0.0 的 Apache DolphinScheduler 尚存在一些问题，对此，我们进行了系列优化改造和修复，同时积累了丰富的运维经验。
优化实践
2.1 问题修复
2.1.1 HadoopUtils 引发的线程泄漏问题
在某次巡检的过程中，我们发现服务节点的线程数在过去一段时间呈明显的上升态势，根据经验判断，应该是程序中存在线程泄漏的地方，结合 metrics 发现泄漏速度为恒定速率，并且与任务并发量无关。
通过堆栈发现泄漏的线程主要是与 HDFS 相关，进一步将代码范围缩小至 HadoopUtils 之后，我们发现此处存在引发线程泄漏的代码逻辑。在 DS HadoopUtils 中存在一个 Cache，会以恒定的速率不断生成新的 HadoopUtils 实例，并放入 Cache，在 HadoopUtils 实例化的过程中会创建 HDFS FileSystem，但是却不会关闭原有的 FileSystem。
开源的 DS 在使用 HDFS FileSystem 时泄漏速度比较慢，不易发现。我们在生产环境中使用的是腾讯云提供的 CosFileSystem 插件，该插件中会使用多线程来加速文件的上传 / 下载操作。通过插件中的线程数目比对，与我们泄漏的线程数完全一致。
至此，我们确定线程泄漏的原因是 HadoopUtils 在更新 Cache 的时候没有关闭 FileSystem，于是我们在更新 Cache 的时候关闭 FileSystem，并通过读写锁保证不会由于异步关闭导致文件操作失败，成功的解决了线程泄漏的问题。对比上线前后的 JVM 线程指标，修复之后线程数保持在一个小恒定范围内。
2.1.2 TaskExecuteRunnable 内存泄漏引发 CPU 飙高问题
在生产环境中，我们发现在任务量没有发生明显变化时，Master 服务随着运行时间越长，其 CPU 使用率出现增长趋势。
通过分析火焰图，我们发现 Master 中存在一处代码逻辑，随着程序的运行时间越久，这段代码逻辑对 CPU 的消耗会越来越高。
通过梳理代码逻辑，我们发现 DolphinScheduler 的 Master 服务在运行工作流时采用事件驱动的方式，每个任务实例在运行过程中会生成一个对应的 TaskExecuteRunnable 对象，任务在运行过程中产生的生命周期事件会存放在 TaskExecuteRunnable 对象中。会有一个后台线程轮询当前服务中存活的 TaskExecuteRunnable 对象，然后提交事件处理任务到事件线程池。
不过，当任务运行结束之后，TaskExecuteRunnable 并不会被释放，还会存放在任务事件线程池中，这就会导致任务事件处理线程空转时间越来越长。
通过分析堆栈，我们的判断得到了验证，TaskExecuteRunnable 的确会泄漏，不过由于 TaskExecuteRunnable 占用内存很少，因此很难从内存中反应出来。我们的集群中有 4 台 Master，任务实例数一天百万左右，因此对于单台 Master 一天会泄漏大约 25w TaskExecuteRunnable，随着时间的积累会拖慢引擎的事件处理。
于是我们进行了代码修复，在任务执行结束之后，移除内存中的 TaskExecuteRunnable 对象。对比修复前后的 JVM CPU 指标，修复之后，Master 的 CPU 指标随着运行时间始终维持在一个小的恒定范围内。
2.1.3 Master 执行逻辑任务重复提交事件，导致事件堆积问题
DolphinScheduler 中任务分为两类，分别为以 dependent 为代表的逻辑任务和以 Shell 为代表带的物理任务，其中逻辑任务在 Master 中执行，物理任务在 Worker 中执行。不管是逻辑任务还是物理任务在 Master 处理过程中都会经历以下阶段。
逻辑任务和物理任务的区别在于 Dispatch 和 Run 阶段的实现不同，对于逻辑任务不需要触发真正的 Dispatch，Run 阶段运行在 Master 中。而物理任务在 Dispatch 阶段会将任务分发给 Worker，Run 阶段运行在 Worker 中。
无论是哪种任务，当 Dispatch 阶段执行成功之后，会注册到 StateWheelThread 中，该组件会定时的每隔 5 秒钟为每个任务生成一个 TaskStateChangeEvent，提交到任务的事件队列中，TaskStateChangeEvent 被处理的时候会触发 Run 任务，对于逻辑任务会不断的通过 TaskStateChangeEvent 触发执行。
这里是一个典型的生产消费模型，生产者以固定的速率（每隔 5 秒）生成事件写入队列，消费者异步的从队列中消费事件。
因此当消费者处理的速度小于生产者生产的速度时，这里就会出现事件堆积。
而实际情况下，由于生产者生产事件的时候是纯内存计算，没有任何 io 阻塞，而消费者处理事件的时候需要多次查询 db。对于 Dependent 这类逻辑任务的运行时间通常都很长，因此如果达到一定的并发量，这里极大概率会出现事件积压，导致整个 master 中所有任务的状态事件处理出现延迟、增加数据库压力，严重的话还会导致 Master OOM 服务宕机。
在我们的测试环境中，单台 Master 服务，事件线程池大小为 100，Dependent 并发数超过 500，此时就 Master 中的 StateEvent 就会出现堆积的情况。
我们发现堆积的事件都是用来触发逻辑任务 Run 阶段，并且对于同一任务实例存在多个重复的触发事件，我们通过对事件去重从而修复堆积的情况。在修复之后，事件的堆积情况得到解决，一旦事件的消费速度低于增长速度，事件的堆积量最多为任务的并发数，不会出现一直积累的情况。
2.1.4 Master 任务调度不均匀
Master 在分配任务给 Worker 的时候，会使用负载均衡策略，使任务的分配尽量均衡。默认的均衡策略是 LOW_WEIGHT，该策略会通过 Worker 的心跳信息来计算一个负载量，会将任务分配给负载量最低的 worker。
在实际的使用过程中我们发现在大多数情况下，这种负载策略会出现严重的任务分配不均衡的情况，在同一个 WorkerGroup 下，不同的 Worker 被分配到的任务量可能会相差几十倍。
究其原因我们发现主要是由两个方面导致
- 负载计算的值无法代表负载情况
在计算 Worker 节点的负载时，Master 会对 Worker 的 CPU、内存、Load、等待任务数分别加一个权重来做归一化，但是针对各资源加权值和归一化算法表达不严谨。导致计算出来的负载值实际上并不能正确的反应 Worker 的真实负载情况，并且实际生产很难通过调节权重得到一个真实的值。
- 负载计算不是实时的
Worker 的心跳上报是定时上报，Master 在分发任务时使用的 Worker 心跳数据并不能反映当前 Worker 的真实情况，这会导致某个时刻一旦出现一个负载量偏低的 worker，master 在接下来一段时间中可能会将大量的任务都发送给这台 worker，从而导致任务倾斜。
分析完原因之后，我们决定使用 RANDOM 策略来分发任务，保证 Master 在分发时绝对均衡，然后由 Worker 自己通过自身负载决定是否要接受 Master 的分发请求。对比修复前后同一个 worker 分组下不同 Pod 接收的任务，发现修复后不同的 pod 接收的任务变得均衡，不再出现任务倾斜的情况。
2.1.5 Master 事件处理卡住问题
在生产环境中，我们发现 Master 的 CPU 持续升高，通过服务日志发现 Master 一直在处理某个事件，并且伴随异常，我们猜测此时出现了事件死循环的情况。通过研究 DolphinScheduler3.0.0 中 Master 事件驱动流程我们发现在该版本中存在三类事件。
- WorkflowStartEvent
WorkflowStartEvent 是工作流启动事件，该事件是由一个单独的后台线程产生，并且由一个单独的后台线程处理，用于启动工作流，工作流的元数据出现异常时会导致 WorkflowStartEvent 执行出现异常，此时异常的 WorkflowStartEvent 会一直重试并阻塞后面其他事件，直到在数据库中对元数据进行修复。
- StateEvent
StateEvent 是工作流和任务执行相关的事件，用于驱动 DAG 拓扑执行。
StateEvent 有以下事件类型：工作流状态变更、任务状态变更、工作流超时、任务超时、TaskGroup 中的任务被唤醒、任务重试、工作流阻塞。
对于一个工作流来说，里面所有的 StateEvent 都存储在一个队列中，事件按照进入队列的先后顺序被执行。并且采用的是 DFS 的方式被线程池消费，即一个队列被 fire 的时候会被分配给一个线程，该线程直到处理完队列中的所有事件才会退出，如果一旦有某个队列在处理时无法退出，那么线程会被一直占用。
- TaskEvent
当任务实例在运行时发生了变化会生成 TaskEvent，即该事件是由 Worker 发送的任务数据所转换而来，以下情况都会生成 TaskEvent，TaskEvent 处理流程和 StateEvent 类似。
-
任务实例被分发成功了，那么会触发 Dispatch 类型的 TaskEvent。
-
任务实例延迟执行了，那么会触发 Delay 类型的 TaskEvent
-
任务实例开始运行了，那么会触发 Running 的 TaskEvent
-
任务实例运行结束了，会触发 Result 类型的 TaskEvent
值得注意的是以上三种事件都采用死信队列的方式存放，即只有当事件被处理成功才会将事件从队列移除，社区最初这么设计是希望在某些情况下由于基础设施故障，例如 db 抖动等不会影响到事件的处理，但实际上有很多其他的意外情况会导致事件处理失败，例如数据库存在非正常数据，事件发送过程中出现乱序等。
我们认为对于引擎来说，需要避免由于某一个工作流事件处理出现问题，从而影响到引擎的稳定性。因此，我们移除了这里的死信队列，当事件处理失败的时候，会直接抛弃事件，并将工作流快速置为失败，由上层进行重试，并结合 Metrics 监控各类事件的处理情况。修复后，Master CPU 保持稳定，服务日志也不再出现一直重复处理某个事件。
2.2 稳定性优化
2.2.1 工作流实例健康检查
目前 DolphinScheduler 中 Master 执行工作流的时候会将工作流实例的元数据存储在内存，然后通过事件驱动的方式去进行状态流转，直至工作流中所有的任务都结束然后将工作流实例从内存卸载。在某些情况，例如网络原因导致事件丢失，或者事件在处理过程中由于状态机 bug 处理失败从而丢失，此时会导致工作流实例处理流程卡住，从而导致工作流实例成为孤儿实例，即永远不可能结束。
此时如果发现了可以在上层通过 kill 的方式去停止工作流实例，从而卸载，不过这种方式存在两个问题。一是依赖业务方自行检测，需要业务方定期的巡检整个系统，当业务方发现问题时往往业务已经受到了影响。二是处理的方式很繁琐，一旦运行的工作流实例数比较多的时候，逐个操作成本比较高。我们希望调度引擎能够有自检功能，能够自己检测工作流实例是否已经变成僵尸实例，并且自动上报，自动做恢复操作。
对此，我们进行了优化，在 Master 中添加一个组件 WorkflowInstanceHealthCoordinator，该组件用于定期对当前 Master 运行中的每个工作流实例执行健康检查。在健康检查的时候会通过 HeartbeatEmitter 去触发工作流实例的心跳检测，当连续多次的心跳检测失败之后，会通过 DeadWorkflowInstanceHandler 去清除该工作流实例，并上报 metrics。
整个检测主要是由 WorkflowInstanceHealthCoordinator 负责，该组件在每个 Master 中采用单例的形式，里面包含一个后台线程，和 EventExecuteService 工作模式类似，当一个工作流实例被加载到 Cache 之后，会同时在 WorkflowInstanceHealthCoordinator 中注册自己，当工作流实例执行结束从 Cache 中移除的时候也会同步从 WorkflowInstanceHealthCoordinator 移除。
WorkflowInstanceHealthCoordinator 中有一个后台线程会定期的（默认 5min 一次，可自定义配置检测间隔）对注册进来的工作流实例做健康检查。
健康检查的方式是通过对工作流实例中所有未结束的任务做心跳探测，如果探测成功，则表明该工作流实例是存活的，如果探测失败，则表明该工作流实例可能已经出现了异常。对一个工作流实例如果探测失败的次数超过了阈值，我们认为该工作流实例已经成为僵尸工作流实例，我们目前会进行告警，由运维同学介入，当前我们尚未实现自动故障恢复，因为此类僵尸实例发生的情况不会很多，后续我们会考虑实现对僵尸实例自动运维。
2.2.1 Worker 中任务事件 TTL
DolphinScheduler 中 Worker 主要职责是接受任务，执行任务，上报任务事件。
其中任务事件在上报的时候存放在内存中的一个死信队列中。
整个过程为
-
任务执行过程中生成任务事件，并将事件提交到死信队列，每个任务会有一个单独的死信队列
-
Worker 中有一个后台线程会定期轮询死信队列，当事件达到重试间隔之后会重新发送事件给 master
-
Master 在处理完事件之后会发送对应事件 ACK 给 worker，worker 收到 ACK 之后会清除事件，当死信队列中所有事件都为空，并且任务执行结束，此时会卸载死信队列。
这样做的好处是能够避免因为网络抖动或者 master 因为故障而导致某段时间内事件上报不成功从而丢失事件，不过这样也会导致可能出现内存泄漏的问题。例如，如果 master 发生容错，那么会进行工作流容错，容错的时候会先 kill 任务，然后重新提交，在 kill 的时候如果发送 RPC 给 worker 失败了，此时 worker 中的任务事件将永远不会被清除，并且由于工作流实例发生了容错，此时某些任务事件可能无法发送给容错后的 master，即会一直重试，变成僵尸消息。
即一旦消息的目的地发生了变化，但是 worker 感知不到，那么会导致消息泄漏到内存。
一旦发生泄漏，可能会导致重试线程中堆积大量的无效事件，这会占用线程资源，导致有效事件发送出现延迟，并且这类无效事件永远不会被释放，会造成内存泄漏，影响服务稳定性。
对此，我们在事件中添加了 TTL，每个事件在创建的时候会带有 createTime，如果 currentTime-createTime>ttl，那么表明事件在给定的时间内没有发送成功，此时说明事件可能已经出现了泄漏，会在 prometheus 中打点，并自动从死信队列队列中清除。
2.2.2 历史数据保留策略
随着系统使用时间的增长，数据库和磁盘数据逐渐积累，会影响服务运行和数据库稳定性。实际运维中我们发现，数据库中增长的主要是一些实例元数据，这些数据的积累会导致数据库压力越来越大，同时会伴随慢查询越来越多。磁盘中增长的主要是任务实例日志和任务实例工作目录，这些数据的积累会导致磁盘可用容量和 inode 变得越来越少。
我们希望程序能够自动的清理无用的历史数据。例如，在数据库中仅保留最近一个月的运行实例数据。磁盘上保留最近一周的临时文件，超出保留期限的数据则自动删除，以减少人工运维的工作。
由于 DolphinScheduler 原生的删除接口在做数据清理的时候是按照工作流实例的维度，即清理历史数据的时候需要先找出工作流实例下的任务实例，然后分别清理每个任务实例的数据，这个过程涉及大量的数据库操作和 RPC 操作，并且执行批量删除操作的时候会给服务带来很大的压力，不适用大批量数据的清理。为此，我们分别在 Master 和 Worker 中添加了 InstanceDBPurgerThread 和 TaskFilePurgerThread 两个后台线程组件，分别负责数据库实例数据和磁盘文件的定期检查、上报与清除工作。
数据库方面主要清理工作流实例、任务实例和告警事件等数据。磁盘方面，主要是清理 exec 目录下的临时文件夹，log 目录下的任务实例日志等数据。同时，通过暴露相关的 metrics 对这些数据量进行监控。将数据库和磁盘的清理分开可以极大的加速历史数据的清除速度。
2.3 巡检流程
为了确保系统稳定运行，我们不仅配置了大量告警，还定期进行日常、周、月巡检。通过巡检，我们能够提前发现潜在问题，不断的完善自动化运维流程。我们目前发现的大多数问题都是通过巡检提前发现，避免了对业务造成实际影响。
同时，巡检也帮助我们不断优化我们的监控大盘和告警项。目前我们从集群、项目、WorkerGroup 和 Pod 等维度搭建了监控面板和告警，以辅助巡检工作。
在日常巡检中，我们主要关注集群、项目、WorkerGroup 三个维度下的指标。在集群维度上，关注 Master Slot 变化、集群水位、并发量、资源使用等稳定性指标。在项目和 WorkerGroup 维度，关注异常任务、任务量的同比变化，WorkerGroup 下 Slot 使用率及业务运行情况。Pod 维度则用于周巡检和问题排查。
未来规划
目前 DolphinScheduler3.0.0 已经在我们的生产环境中稳定运行，我们针对使用场景中发现的问题，在不进行大规模架构调整的前提下做出了修复和优化，并且沉淀出了一套适用于当前业务场景的运维手段。社区在后续版本中对某些问题进行了更完善的修复，如针对逻辑任务事件阻塞的问题，重构了整个逻辑任务执行流程；针对状态机卡住问题，重构了状态机模型等。未来，随着业务量和使用场景的扩展，我们会考虑版本升级到 3.2+ 版本，以尽可能与社区保持同步，并将我们所做的一些优化项反馈至社区。
本文由 白鲸开源 提供发布支持！
