货拉拉实时研发平台指标监控实践

背景

指标是系统可观测性的重要一环，对于系统的稳定性保障和性能优化，有着重要的作用。 flink 提供了原生的指标体系，但是flink的指标体系要结合可高度定制化的可视化看板和可高度定制化的告警功能，才能有效发挥其用途。飞流 flink 实时平台原有的指标体系是基于开源的prometheus 和 grafana，特别是告警功能和运维问题，限制指标体系的发挥。本文介绍了一种基于公司告警体系 lala-monitor，能实现指标快速接入，指标灵活控制，指标质量保证，指标可视化，配置灵活化，能拓展，能隔离的方案。

通过该文，你将能收获到以下几个方面内容：

1. Flink 的指标体系，以及如何自定义指标和自定义指标 report
2. 指标的通用处理方式
3. 一种指标接入公司级监控的方案
4. 基于指标的功能拓展

1 链路演进

原有链路

原有链路使用 flink 原生提供的 prometheus report，推送采集指标到一台中间 gateway 节点，prometheus 配置对 gateway 指标拉取，grafana 配置对 prometheus 数据进行可视化。原有链路存在以下问题：

1. 经常遇到的问题是 gateway 节点内存满，经常收到机器告警，需要专门的去运维
2. 由于内存满，新的指标发送不进来，整个链路堵塞 内存满的原因
3. 上报到 gateway 的指标没有经过任何处理，同任务不同运行实例的指标会重复存储，需要专门且准确的清理机制
4. 指标没有经过任何的处理，一些无用的 label 占用内存

新链路

针对原有链路的问题，我们接入公司的监控告警平台，来解决指标的存储展示以及告警配置问题。公司监控平台的最小粒度是项目级别，只采集项目机器上的指标；而飞流是一个任务管理平台，任务运行在容器里，新链接需要打通任务指标到飞流机器的流程，即metrics report模块。新链接指标上报到kafka，项目机器上部署消费kafka的进程，暴露数据给lala-monitor进行采集。新链路具有以下优点：

1. 高效解决了指标的存储展示以及告警配置
2. kafka 的数据可用于其他场景使用，比如降维度预处理来减少 lala-monitor 的采集量，如果用户有需求也可以消费处理
3. kafka 的数据，不会出现 prom 同维度不同时刻覆盖的场景，这样可以保留不同时刻的原始数据，可用于其他场景使用，比如 CEP 处理

2 链路总体设计

2.1 metrics-collect 模块介绍

模块定位：指标收集，指标收集的来源目前分成 4 部分:

1. flink 任务提供的原生指标
2. flink sql 任务，飞流平台的埋点指标
3. flink jar 任务，用户的埋点指标
4. 飞流系统自己采集的其他指标，主要是一些 flink 中不提供，或者不能直接使用的指标

2.2 Kafak 模块介绍

存储指标 topic，目前有 3 个：

1. 自采集指标 topic
2. flink 指标 topic
3. 预处理指标 topic

2.3 metrics-report 模块介绍

模块定位：消费 kafka 数据，进行预处理和控制，暴露数据给 lala-monitor 采集，暴露 2 类指标：

1. 任务里的业务指标
2. metrics-report 采集进程的过程指标

2.4 Metrics-alarm / Metrics-Display 模块介绍

模块定位：提供看版，提供告警

1. 从用户和管理员视角，提供多种看板
2. 提供任务维度，采集链路维度，系统维度告警

3 metrics-collect 实现解析

Metrics-collect 通过自定义的 kafka-report，对指标进行管理和一定的预处理，上报到 kafka topic。

3.1 Kafka-report

Flink 原生提供的8种report中，没有kafka report，我们自定义了kafka report，Kafka report里我们做了以下事项：

1. kafka topic record 规范

指标以 json 的形式上报。指标增加 2 个字段规范，1 torrent_task_id, 2 report_timestamp。torrent_task_id 用于定位任务和运行实例的关系，用 torrent_task_id 不用 torrent_task_name，是因为 taskid 信息更少，传输信息更少，至于其他需要的一些信息，在 metrics-report 环节进行补全，这里补全的数据，能根据实际值进行变化，实现了任务除 taskid 之前的其他属性可以修改，而不用重启任务上报新的指标来适配。report_timestamp 用来定义事件时间，用来跟踪端到端延迟，也可以用来处理 flink 时间语意。对于 report_timestamp，我们对于同一个 TM 的同一采集批次指标，进行同样的 timestamp 赋值，这个符合指标采集的含义，不会造成时间飘逸，影响基于时间的预聚合逻辑。

2. 指标记录拆分

原始的指标会把属于同一个 tm 的数据进行聚合，一个 tm 上只有一条数据，这样的数据如果直接发送 kafka，会出现单条记录偏大的情况，大记录对 kafka 不友好。同时 metircs-report 模块处理，或者其他预处理，都需要把各个维度进行拆开，才能按照不同的维度进行组合计算，在上报端就拆开，可以减少重复的解析流程。记录拆分提供了 task+metrics 更细粒度的 topic-key 制定，否则只能指定到 task 级别。更细的 key 指定，可以防止复杂任务都聚集发往某个分区，导致分区分配存储不均以及后续的处理不均问题。

3. kafka topic record key 指定

不管是哪类指标，在发送 kakfa 的时候都要指定 key 进行发送。统一采集用 task+metricname 的组合进行指定。这个是 kafka+lala-monitor 采集存储原理决定。指定 key 能避免同一 label 维度的指标由于随机分布导致在不同的 topic，metrics-report 实例消费到不同 key 的数据，都进行暴露，导致指标采集覆盖，采集重复问题。同时也可以避免数据不会在 topic 不同分区之间乱序，导致指标含义在不同时刻的乱序。

4. 指标预处理

对于 flink 指标里原生的 taskname 属性，由于这个含义和用户 sql 写法对应，有的写法，比如有很多个 sql 字段，这个名字就会很长，导致单条记录就会很大。针对 taskname，我们做了预处理，对于我们目前不关心的非 source 和 sink 的中间算子指标，我们用原生的 taskid 代替，taskid 是长度固定，但是被 hash 的字段，可以起到区分维度的作用。对于 source 和 sink 算子，由于我们要做统计 source 和 sink 的指标，换了一种与处理方式，解析出表名进行替换，保留了语意的同时也起到了缩短内容的作用。

5. 增加自定义的维度

Flink 原生指标里只提供一些 host，taskname 等这些属性，我们通过 scope 和 variables，增加了诸如 cp 编号，库表等维度。

6. kafka client

slot 共享下，一个 tm 实例，一般会有 2 个 kafka client，我们优化成一个 tm 只会有一个 kafka client，减少 kafka server 的链接数。发送支持 lz4 压缩，减少发送和存储的大小。

3.2 指标管控

指标管控能避免自定义指标和原生指标的重复，对指标含义的影响。定义统一的指标命名规范，命名规范中能直接区分出是哪类指标。控制指标发送路由不同的topic和集群，实现隔离和拓展。对某个任务是否上报指标，某个指标是否上报进行控制，运行上报控制必要性包含以下几点：

1. 减少送往 kakfa 的数据量；因为 flink 提供的指标很多，不是所有指标都是核心指标，没必要的数据不发送 kakfa。原生指标例子分为 Operate/ Task/机器/ 任务几类，从前到后，指标量级增长，主要开启部分 task 级别和机器/任务级别的指标
2. 降级，如果发现整体链路有问题，可以通过控制，优先保证核心链路的指标。
3. 管控，一些没有经过管控的指标，直接发送，如果同名是会对其他指标照成影响。

4 metrics-report 实现解析

模块定位：消费 kafka 数据，进行转换和控制，暴露数据给 lala-monitor 采集。

4.1 预处理

metrics-collect采集的原生的指标，我们没有做太多的维度和维度内容的丢弃处理，是为了后续处理时，能够提供丰富的信息，这些维度和内容如果直接暴露给lala-monitor采集的话，会照成label和label下维度的爆炸，对采集和存储都有影响。预处理是为控制label和label下的维度，目前有2种方式

1. metrics-report 预处理

metrics-report 根据业务含义和展示告警需求进行预处理，处理分成 2 个方面，一方面是丢弃一些维度，另外一个方面是对 label 属性进行处理。

label 丢弃的场景 比如 flink_task_id 和 flink_task_name 是一个含义的 2 个表示方式，我们选择有业务含义的 flink_task_name，丢弃 flink_task_id。比如 tm_id / subtask_index / host，我们保留了 subtask_index 这个最细的粒度，保证指标不会覆盖；与之类似的 host ，我们有进行保留，为了的后续基于机器维度的进一步分析，比如热点问题等。

label 属性处理的场景 根据 label 的含义，对 flink_task_name 进行处理，主要是减少内容，同时又保留含义，我们的处理是只保留算子含义的 source 和 sink 算子（我们在定义算子的时候有对算子进行命名）。不同 sink 的命名规则保持统一。对于不同任务中，同样是 source 同一个表，但是由于 sql 逻辑不一样，导致的算子原始 taskName 不一样，通过统一的处理之后，可以进行表维度的数据统计。同时能够把不同 sql 下本来属于同一 source 统计但是算子内容不同的高基维变 label 成低基维 label。

2. flink sql 预处理

由于 TM-task 级别的指标量会很多，但是我们实际使用时，最终结果是基于任务的进行展示和告警，通过 flink-sql 任务消费 kafka topic，进行同维度处理聚合，把数据放到一个新的 topic，而不是放到原先的 topic，避免放进去的 topic 重复消费问题。kafka-report 中指定的同一时间，就是为了解决这里的聚合含义。

4.2 指标控制

metrics-collect阶段已经做了控制指标是否上报，metrics-report也做了同样的控制逻辑，原因有以下几点：

1. metrics-report 的控制作为一种兜底。
2. metrics-collect 目前没有实现实时控制，想要控制某些指标不上报，需要重启生效。
3. 如果 metrics-collect 不是唯一的输入源，metrics-report 也可以进行控制。

控制场景

1. 在功能除上线阶段，可能存在一些未考虑到的场景，控制是为了能够减少对 flink 任务/kafka/lala-monitor 的影响。
2. 后续任务太多，指标太多的阶段，我们的进程需要扩容，同时我们的采集可能会被服务端限速，我们需要根据需求，进行指标是否上报的控制。

指标治理

1. metrics-rport 采集丢弃数。含义是消费到了数据，但是发现没注册指标或者指标是不允许上报的状态，这些消费到的指标，会被丢弃；虽然快速丢弃能减少 prom 指标的采集量，但是 kafka 环节，还是存了这些无用指标。丢弃比例过多，就说明 metrics-collect 阶段采集了一些不需要的数据，这些数据可能是用户自己增加没在飞流系统注册管理的指标等，我们需要进行处理，减少 kafka 的无用 qps，以支持跟多有用的指标。
2. flink/torrent 指标-prom 指标数据量统计，用于分析某类指标是否过大，开启是否合理，根据情况进行处理，减少对整体系统的影响。
3. 任务-prom 指标数据量统计，用于分析某个任务指标是否过大，是否合理，根据情况进行处理，减少对整体系统的影响。

4.3 指标数据质量等级保证

指标重复或者丢失，都会影响指标数据的统计准确性。

4.3.1 指标出现丢失的场景

这里主要说 metrics-report 侧的丢失问题，出现丢失的场景就比较多，问题的核心是指标推送采集端口，然后等服务端采集的流程没有 ack 机制。

正常场景： 对于不能丢失的数据，我们是先 WAL，暴露数据之后，在 commit offset。异常场景：这个时候暴露的数据，有可能在采集间隔内，由于进程挂了，数据还没采集到服务端，但是数据已经 commit，这个时候恢复，我们从 wal 恢复，并且只恢复属于本消费者的分区数据，避免其他分区重复暴露，采集重复。

WAL 的清理机制： 由于采集是由另外一个进程通过 report 进程提供的端口暴露的进行采集，没有任何的回调机制和锁机制，所以只能找能确定数据已经被采集的标志。

采集 lag 标记： 我们是在机器采集数据中增加一个指标，该指标的含义是 report 赋值时系统的当前时间，这条指标会和其他待采集指标一起作为一批数据被采集上去，如果这个指标的 value 和在服务端和当前时间差值超多 N 分钟，就说明采集延迟了 N 分钟；利用这个延迟的时间作为清理标记，进行 WAL 的清理，和一些不更新数据的 TTL 的清理。

4.3.2 指标出现重复的场景

指标重复场景

1. topic reblance，造成这个 reblance 的场景可能有进程停止，增加进程，进程 crash，机器 crash 等。topic reblance 会导致不继续属于本消费者的数据还继续暴露在本机上，这些暴露的数据和新分配在其他机器上消费者的数据，会指标重复，影响指标的准确性。
2. 飞流任务不同的运行实例，会形成不同的 flink 任务，同一个飞流任务 2 次运行实例的指标是有可能不同的，比如分布的 yarn 机器不同，这些不同会导致前后 2 次实例暴露出来的数据由于 yarn host 不同而重复，也会影响指标的准确性。
3. 主动停止进程，暴露的数据已经采集，但是 WAL 的数据没及时清理，从 WAL 恢复，这些数据会被在暴露一遍。

处理手段：主动清理+主动 TTL

1. 对于重复场景 1 – 主动清理：利用 kakfa 分区切换时候的钩子，获取上次消费的分区和本次消费的分区，主动剔出之前消费但是现在不属于消费者的分区的数据。
2. 对于重复场景 2 – 主动 TTL：对于不同运行实例，任务上一次运行实例停止之后，就不会生产数据，这些数据在多久没更新之后，就可以被清理掉，目前使用的定时 TTL，后续可以优化成事件 TTL。TTL 还有其他用途：长周期采集上报的指标，比如 10 分钟采集一次这些指标，在这个 10 分钟期间是没有更新的，采集是重复数据的采集，是无效采集，TTL 可以清理掉这种无效采集，减少采集数据量，但是 TTL 在服务端会有数据掉坑现象，我们配置告警有进行优化，同时后续服务端也支持掉坑数据的连线展示，而不是掉坑展示。
3. 对于重复场景 3 – 后续看服务端是否能提供采集指标的查询接口，查询 lag 指标，做到采集，暴露，WAL 清理的精准控制。

4.3.3 数据等级划分

为了应对不同的业务场景和实际运行场景，我们把指标分成 4 个等级

1. 不做任何保证
2. 不能丢失
3. 不能重复
4. 不能丢失，不能重复

不同等级适用于不同场景的指标数据，使用不同的的机制来保障

1. 不做任何保证等级：上报频率很快的指标，比如 flink 任务中的指标，默认 10 秒一次，即使这次丢失了，下次还会快速补充回来。
2. 不能丢失等级：长报周期数据，比如我们采集的 lag 数据，如果丢失一次，需要在等一次采集周期，一个采集周期目前是 10 分钟，那丢失一次，告警就延迟到 20 分钟。
3. 不能重复等级 ：比如统计类型的指标，如果重复，统计指标含义就会有其他含义。
4. 不能丢失，不能重复等级：比如对账指标，如果其中一个指标丢失或者重复，多个环节的数据对不齐全；目前的指标体系还不能完全支持，因为指标在 flink 上报层面存在还没采集任务就挂的场景，我们使用 event 体系来支撑，以后会有介绍。

5 Metrics-alarm / Metrics-Display 实现解析

5.1 看板

针对不同的用户，提供多看板；对于普通用户：提供任务维度，任务分组维度，我们定制的看板的原则是简化。

1. 每个仪表盘每个任务只用一条曲线进行展示，这里需要对一些原始的指标进行聚合，比如不展示到flink任务并发度细粒度的指标；
2. 比率化；比如tm内存使用，原始指标是2个指标，used内存和max内存，但是这个内存对于设置不同内存大小的任务，max也是变化的，所以我们进行一个内存使用率的转化；比如gc变化，原始指标是累计值，对于运行很久的任务，累计值就是很大，所以我们进行一个变化率的转化；比如source波动，原始的指标含义一般是qps，但是qps的变化对于不同任务，不同运行时间段，不同运行环境下，都是因它而异的，所以我们转换成波动率，这样用户只需要配置一个浮动比率即可。
3. 时间化，对于一些原始指标，含义是数值，但是数值对于不同任务，不同运行时间段，不同运行环境下，都是因它而异的；所以我们把这些含义不好确定的转换成用户好理解的时间，用户只需要配置可以容忍的一个时间，比如kakfa lag 需要恢复的时间。
4. 即使我们对指标进行了比率化，但是没有水位线+历史曲线，用户不知道什么情况需要处理，一个我们有对应的告警，而是含义清晰的水位线和面板描述，减少了用户使用成本。

对于管理员：提供系统维度，metrics-report采集维度，我们定制看板的原则是专业化和宏观化

5.2 告警种类

告警也是针对不同用户进行设计，目前分成 3 类，告警任务，告警采集，告警系统

5.2.1 告警原理

飞流使用 lala-monitor 和其他业务接入 lala-monitor 的一个区别是，飞流是一个任务管理系统，我们的告警主要是给任务的管理员发送，同时飞流管理员也作为兜底接受告警。这在没有使用到服务端提供的api之前，我们使用的新监控服务的动态订阅组功能。

动态订阅组：是可以把数据发送给一个指定的变量化的订阅组，如果订阅组不存在，对应的Appid订阅组会作为兜底，这个比较符合我们的需求。如果只以单个任务负责人作为告警发送对象，会存在告警接受处理单点问题，同时任务负责人会有离职或者项目变动的场景，我们目前以任务tag作为告警组的最小粒度，后续会增加项目信息进来一起维护。

tag是同类任务/同项目下的任务的交集，需要把tag 对应任务的负责人并集增加到订阅组。我们配置了通用的告警配置，发送到
${torrent_tag} / torrent / system / x 这个订阅组，只要根据任务配置了告警组，比如 tag1/torrent/system/p2，就能接收到tag 下任务的告警。如果觉得我们配置的阈值的告警等级不合理，可以禁用${torrent_tag}/torrent/system/x 告警组，然后自己配置订阅组，比如${torrent_id}/torrent/custom/x +自己复制我们提供告警模版，或者自己配置，进行告警。基于告警组进行配置，只是我们提供的通用解决方案，用户完全可以自行配置。

5.2.2 误告警/告警准确性/告警延迟等问题

• 告警泛滥：1.比如一些告警在高峰期容易出现告警，我们设置退避告警间隔；2.设置合理的阈值；3.同任务告警合并
• 误告警：对于数据掉坑/数据高低峰/数据突发带来的误告警，使用平均值；
• 告警延迟：监控端到端延迟在一定范围之内

6 部署 实现解析

• metrics-report 属于无中心架构，可以横向拓展
• Kafka 分区可以增加分区，metrics-report 自动发现新分区，并从新分区最早数据消费
• 可以增加 topic，用于隔离和拓展

7 建设成果

通过指标告警体系的搭建，系统的解决了指标的存储展示以及告警配置问题。告警时效性从原来分钟级别 -> 10秒级别，且可配置；多次发现了任务和系统的异常情况；辅助进行任务资源成本治理；基于任务告警，远程进行运维任务；实现告警数，告警恢复时间等统计。

8 规划

目前的指标监控告警体系，只是搭建了一个框架和进行一些场景优化，在告警实时性和告警的有效率上还有提升的空间。对指标的一些深度应用，目前还缺失，比如结合任务日志和其他服务的告警，进行根因分析，基于指标的自动运维和资源调优等，这些都是后续需要补齐的方向。

作者：王世涛，货拉拉大数据技术与产品部-数据平台组-实时研发平台负责人，负责flink实时研发平台产品演进及迭代。