TIS基于Flink的实时状态监控及流控

实时任务状态监控及流控需求

过去，我们通过Flink-CDC部署整库同步方案，执行数据同步过程需要了解当前管道的基本执行指标，如：管道瞬时消息流量增大导致背压，需要立即实施有效应对措施，如：限流，暂停/启动，泄洪等操作。在生产环境有实际经验的 同学一定深有感触，DBA经常会在深夜大批量订正业务系统数据，碰到双11业务量激增，这些场景往往导致实时处理任务大量数据堆积导致背压，导致JVM频繁FullGC最终导致实时任务宕机。 为了应对这些场景，需要在Flink的实时流任务中实现状态反馈及响应机制，需要提供以下两方面能力：

1. 快速且方便地获取流任务执行状态，为下一步应对响应作为依据。
2. 能够在Flink任务不停机的状态下做出有效操作，能够第一时间处理系统的异常状态。

针对以上第2点，可以细化成三种功能，分别是：1. 限流、2. 启/停（Pause/Resume）3. 泄洪，以下是说明：

启/停（Pause/Resume）

试想这样的场景，你的 Flink 作业正在消费来自 Kafka/Pulsar 的消息。现在需要对这个 Kafka 集群进行计划内的维护（例如：版本升级、扩容 Broker、更换磁盘等）。维护期间，你希望 Flink 作业停止消费，以避免给 Kafka 带来压力，同时也避免因 Kafka 不可用而导致 Flink 作业持续抛出异常、产生大量错误日志。

应对该场景，如果使用传统的flink 的Stop with Savepoint操作的弊端：

1. 需要耗费短则几分钟的时间制作一个大的Savepoint，恢复也需要较长时间
2. 资源波动：作业取消后，TaskManager slot 资源会释放；恢复时又需要重新申请。在资源紧张或集群管理平台（如 K8s）上，这可能导致不必要的延迟。
3. 风险：Savepoint 的制作和恢复并非 100% 无风险，尤其是在复杂状态或自定义算子状态下。

Pause/Resume 的优势：

1. 瞬时操作：pause 命令会让 Source 立即停止消费（提交其偏移量），所有算子处理完当前数据后进入“静默”状态。整个过程是轻量级的，几乎是瞬时的。
2. 状态常驻内存：作业的整个状态（所有算子的状态）都保留在内存中，没有任何序列化/反序列化或磁盘 I/O 开销。
3. 无缝衔接：维护结束后，一发 resume 命令，Source 会从 exactly-once 的位点重新开始拉取数据，处理立即继续，延迟极低。

避免下游过载和背压（Backpressure）的传播

• 场景描述：Flink 作业的下游是一个数据库（如 MySQL、ClickHouse）或一个外部 API。该下游系统突然遇到性能瓶颈或开始限流。持续的写入失败会导致 Flink 作业产生背压，一直回溯到 Source，最终可能拖垮整个作业甚至消息队列。
• 为什么 pause/resume 是最佳方案？
  1. 传统做法的弊端：同样面临停止/重启的开销和风险。你可能需要设置复杂的故障重试和回退策略，但这通常会导致数据延迟堆积。
  2. pause/resume 的优势：
     1. 快速响应：监控系统检测到下游持续失败后，可以自动触发 pause，立即切断数据流，保护下游系统不被冲垮。
     2. 主动控制：这是一种优雅的、主动的“熔断”机制。等下游系统恢复后，再触发 resume，作业会继续处理，数据不会丢失。
     3. 避免雪崩：有效阻止了因为一个组件的故障在整个数据流中引发雪崩效应。

总结：Pause/Resume vs Stop/Restart

特性	运行时 Pause/Resume	停止并用 Savepoint 重启
速度	极快（近乎瞬时）	慢（依赖保存点大小和网络）
状态位置	内存中	磁盘（Savepoint）
资源占用	占用 Slot（但可协同缩容）	释放 Slot
操作风险	低（无序列化/持久化过程）	中（依赖保存点兼容性）
核心场景	短暂暂停、快速恢复	版本升级、代码变更、长期停止
数据延迟	恢复后延迟极低	恢复后需要“追赶”积压的数据

结论：Pause/Resume 功能的核心价值在于为流处理作业提供了一种 “战术性” 的、低开销的短暂冻结和即时恢复的能力。它非常适合应对生产环境中那些计划内或突发性的、预期持续时间不长的中断事件，是保障流处理作业高可用性、高可维护性和高资源效率的重要武器。

泄洪

为运行中的 Flink 作业提供一种 “泄洪”（或称为“熔断”、“旁路”） 机制——即主动、有选择地丢弃上游数据而不进行任何处理或状态更新——这听起来似乎与流处理的“数据不丢不重”原则相悖，但在特定的生产危机场景下，这是一个非常重要的最后防线和止损策略。

它与 pause 的区别在于：pause 是停止消费，数据会堆积在上游消息队列（如 Kafka）；而“泄洪”是消费后直接丢弃，目的是用数据的损失来换取整个作业乃至集群的存活。

以下是几个必须使用此功能的实际生产场景：

1. 应对下游不可用导致的灾难性背压（Catastrophic Backpressure）

   这是最核心、最关键的场景。

   • 场景描述：Flink 作业的下游是一个外部系统（如数据库、存储服务、外部 API），该下游系统完全不可用（例如：网络分区、凭证失效、磁盘写满）或严重过载。作业无法向下游写入任何数据。

     • 由于 Sink 无法输出，会产生严重的背压（Backpressure）。
     • 背压会迅速沿着数据流反向传播，导致整个作业的所有算子都停止处理。
     • 最终，Source 也会停止从上游消息队列（如 Kafka）消费。
     • 结果：数据不再流动，整个作业“僵死”。更糟糕的是，背压会导致 Checkpoint 超时失败，作业无法创建成功的检查点，精确一次（exactly-once）的保障被打破。如果此时作业失败，将无法从有效的检查点恢复。

   • 为什么需要“泄洪”功能？

     • 作为最后的熔断机制：在监控系统检测到下游持续失败且背压已导致作业停滞时，可以自动或手动触发“泄洪”模式。
     • 作用：在 Sink 算子之前（或直接在 Sink 算子内）将数据全部丢弃。这样会立即消除背压，让数据流重新开始流动。

   • 好处：

     1. 保住作业和集群：Source 可以继续消费，算子可以继续处理（直到丢弃点），Checkpoint 能够继续成功完成。作业保持“ Alive and Healthy”的状态，避免了因背压导致的作业全面崩溃和重启循环。
     2. 为上维修复争取时间：运维团队可以在这个“泄洪”期间，从容地修复下游系统的问题，而不用担心整个流处理平台被一个故障组件拖垮。
     3. 避免数据积压：虽然丢弃了数据，但避免了 Kafka 中数据的无限堆积，防止问题蔓延到消息队列本身。

2. 处理“毒药消息”或无法预见的数据风暴

   • 场景描述：上游数据流中突然出现一种极端异常的消息（“毒药消息”）或流量出现极其异常的超高峰值（例如：Due to a bug, a service starts logging the entire HTTP request payload instead of just a summary）。

     • 这种消息可能导致某个算子的状态急剧膨胀（例如，导致一个 MapState 或 ValueState 巨大无比）。
     • 或者导致窗口算子创建海量的窗口实例，耗尽内存。
     • 最终结果同样是作业崩溃。

   • 为什么需要“泄洪”功能？

     • 选择性丢弃：“泄洪”功能可以设计得更智能，例如基于正则表达式匹配消息内容，或者基于某些特征（如来源IP、消息类型）来选择性丢弃问题数据流，而不是丢弃所有数据。
     • 作用：在问题数据影响到关键状态之前将其丢弃，保护作业的核心业务逻辑和状态不受污染。
     • 好处：牺牲一部分数据的完整性，保全整个作业。这比让作业彻底崩溃、从头恢复（并且可能再次被同样的“毒药消息”击垮）要好得多。

3. 成本控制与差异化服务保障

   • 场景描述：在一个多租户或处理多种业务数据的 Flink 集群中，某个非核心业务的数据流量突然激增（例如，一个临时性的、调试用的日志流），占用了大量计算资源，威胁到了核心业务作业的稳定性和延迟。

   • 为什么需要“泄洪”功能？

     • 资源调度：可以触发对该非核心数据流的“泄洪”，主动丢弃其数据。
     • 作用：立即为核心业务作业释放出宝贵的 CPU、内存和网络资源，确保核心业务的低延迟和高可用。
     • 好处：实现了业务级别的服务质量（QoS）保障和成本控制。在资源受限的情况下，优先保障最重要的业务。

4. 安全与合规应急响应

   • 场景描述：安全团队发现某个数据源可能包含了敏感信息泄露（例如，Due to a misconfiguration, PII data is being streamed into a Kafka topic that shouldn‘t have it）。在问题修复前，必须立即停止处理这些数据以符合合规要求。

   • 为什么需要“泄洪”功能？

     • 快速响应：相比于停止整个作业（可能影响其他正常数据），或者紧急开发、部署一个过滤逻辑的新版本，“泄洪”功能提供了一个即时生效的开关。
     • 作用：立即阻止有问题的数据被进一步处理、存储或分发到下游，满足安全应急响应的要求。
     • 好处：实现了秒级的合规性控制，避免了在漫长的作业重启和发布过程中持续发生数据泄露。

总结：“泄洪”功能是一种用空间换时间（丢弃数据以保住作业）、用部分换整体（牺牲非核心业务保全核心业务） 的高级容错和稳定性保障机制。它不是为了日常使用而设计，而是在系统遇到极端异常情况时，为避免全盘崩溃而设置的紧急安全阀。

基于Flink实现任务状态可视化及流控

为了实现上文提到的实时任务状态反馈及响应机制需要，为了最大限度地复用Flink架构中提供的工具栈，我们先要对Flink架构中已有工具栈进行考察。

实时任务状态监控

打开Flink的控制面板，查看metric控制台指标。 但通过直接查看metric控制台指标存在以下短板：

1. 指标项繁杂，用户需要找到目标metric指标并不容易

2. 通过查看metric指标，对终端用户来说指标的可观测性大打折扣

   例如，guage类型的metric指标，在flink内部实质是一个计数累加器，拿我们关注的binlog处理量来说（需要定位到一个具体的指标），源端每处理一条增量消息就会将该指标递增1。要实现可观测性，用户需要将flink metric指标通过flink内部MetricReporter组件将 metirc推送到外部的系统监控和警报工具包Prometheus组件中，经过该组件处理再对接外部数据可视化UI工具Grafana，用户通过Grafana可视化面板可以直观地观测到flink内部各种状态。要实现以上flink metric

实时任务流控实现

flink在最近的版本中提供了org.apache.flink.api.connector.source.util.ratelimit.RateLimiter接口 ，用户可使用Flink框架的默认实现（如GatedRateLimiter）或者对RateLimiter进行扩展即可方便地实现Flink实时Job的流控。

经过考察，GatedRateLimiter的功能上只提供限流功能，而且限流的阈值在Flink Job任务启动的时已经写死，Job任务运行期不能动态改变。

另外，要启用这套流控机制需进行二次开发，对开发者来说有一定门槛。

TIS实现基于Flink的一站式实时状态监控及流控方案

基于以上提到的两点需求，TIS提供了自己独特的实现方式。提供一站式开箱即用，为终端用户最大限度地屏蔽底层技术细节。以下是针对这两点需求提供的具体实施方案：

实时任务状态监控

优化Flink实施方案中工具栈链路（MetricReporter->Prometheus->可视化UI工具Grafana）过长，如需在生产环境中部署这条链路需要聘请熟悉Prometheus和Grafana的运维工程师，进行部署及联调需要额外增加服务器资源且费时费力。 所以，TIS在构建了一套内嵌的实时状态可视化方案，可以让用户方便、快捷地观测到实时任务处理流量。以下是实现方案中相关的组件架构图：

1. 构建TISPBReporter

   通过扩展Flink MetricReporter，通过主动推送的方式将流任务中相关的Metric指标（类型Gauge）推送到TIS自建指标处理中心。

2. 构建指标处理中心（角色类似Prometheus）

   TISPBReporter通过Grpc协议将flink各个Task Manager中的相关指标收集。指标处理中心针对推送的指标进行集中处理，按照秒为时间片，计算每个时间片中的增量值，为后续前端实时流量柱状图显示提供基础数据。

3. 在TIS中提供实时指标可视化

   在TIS增量实例控制台页面，连接指标处理中心提供的REST API，将收集到的增量的Gauge Metric指标计算转化成前端柱状图可视化组件可以接受的消费数据，通过WebSocket协议主动推送到前端。

实时任务流控实现

前文提到Flink默认提供的限流模块GatedRateLimiter是不支持运行期动态调整限流阈值的，因此，TIS基于org.apache.flink.api.connector.source.util.ratelimit.RateLimiter接口 构建了一个支持运行期动态调整限流阈值的限流器ResettableGuavaRateLimiter，上游通过构建HttpSource组件通过轮询的方式访问TIS Console组件限流阈值配置Rest API，一旦发现限流阈值变更了， HttpSource 就会获取到最新的限流阈值，并通过Broadcat方式向下游的BroadcastProcessFunction算子广播阈值，最终实现实时更新限流阈值。

通过这条消息链路，我们扩展Broadcat消息体，使在原先流控功能基础上扩展了另外两个生产环境中同样非常重要的功能，那就是：