滴滴 Flink 指标系统的架构设计与实践

毫不夸张地说，Flink 指标是洞察 Flink 任务健康状况的关键工具，它们如同 Flink 任务的眼睛一般至关重要。简而言之，这些指标可以被理解为滴滴数据开发平台实时运维系统的数据图谱。在实时计算领域，Flink 指标扮演着举足轻重的角色，例如，实时任务的消费延迟和检查点失败的警报都是基于对 Flink 报告的指标进行监控而触发的；同时，许多实时任务智能诊断的关键决策点也是依 Flink 指标来制定的。

鉴于 Flink 指标系统的重要性，深入理解其工作原理显得尤为必要，这是灵活运用 Flink 指标系统的前提。作为一名平台工程师，我尝试对 Flink 的原理进行一次剖析，如果存在任何不准确之处，敬请各位指正。

Flink 指标系统的核心概念

接下来我们将探讨一些核心概念，它们是理解 Flink 指标系统不可或缺的基础。

Metric Reporters

Metric Reporter 是 Flink 用于导出指标数据的接口，通过 flink-conf.yaml 文件可以轻松配置所需的 MetricReporter。Flink 提供了多种 MetricReporter 的实现，包括 Prometheus、Datadog 等，以满足不同的监控需求。

值得注意的是，尽管 Flink 提供了众多 MetricReporter 的实现，但它如何根据需要动态加载这些实现呢？我们将在后文关于弹性设计的讨论中深入分析这一机制，现在先留个悬念。

MetricReporter 支持两种指标上报方式：Push 和 Pull。具体不赘述了，我们直接引用官方文档中的描述：

Metrics are exported either via pushes or pulls.

Push-based reporters usually implement the Scheduled interface and periodically send a summary of current metrics to an external system.

Pull-based reporters are queried from an external system instead.

滴滴内部的Metric Reporters

滴滴内部没有采用社区的MetricReporter，而是根据滴滴内部实际情况，自研了flink-metrics-kafka。简单来讲，采用push的方式，

滴滴并未使用社区提供的 MetricReporter，而是根据自身需求自主研发了 flink-metrics-kafka。简单来说，该系统采用推送 Push 方式，周期性将 Flink 计算的指标推送到 kafka topic 当中。下文中的实现原理，也是基于 flink-metrics-kafka 介绍的。

Metrics type

Flink 社区对指标进行了高度抽象，定义了四种主要的指标类型：Counter、Gauge、Histogram 和 Meter。

Counter：这是一种简单的计数器，用于记录事件的数量。
Gauge：这种指标非常灵活，可以返回任意类型的统计数据。Gauge 可以看作是 Counter 的泛化版本。
Histogram：正如其名，Histogram 表示的是一系列长整型数值的分布情况，常用于绘制直方图。
Meter：这种指标用于度量平均吞吐量，适用于评估系统的处理能力。

滴滴内部的 Metered

在滴滴内部，我们独创了一种名为 Metered 的指标类型（请注意与 Meter 区分），专门用于衡量一段时间内的指标数值。这包括了时间段内的平均值、最大值、最小值以及计数等。实际上，我们内部使用的许多指标都是基于 Metered 进行计算的。

View

另一个关键概念是 View，它通常与 Metric 结合使用。让我们先来了解一下 View 的定义：

An interface for metrics which should be updated in regular intervals by a background thread.

换句话说，View 为 Metric 提供了一种定时刷新的能力。

Scope

Scope 可以简单地理解为命名空间，它允许在指标名称前添加一系列前缀。我个人认为 Scope 的主要作用是为了区分不同来源或类型的指标，以便于管理和识别。

系统指标

Flink 社区提供了丰富的开箱即用的指标，主要集中在系统性能方面，比如CPU、内存等。这部分内容不是本文重点，下文我们会重点讲解一个滴滴内部非常重要的指标：消费延迟。

Flink 指标系统的弹性设计

作为一款成功的分布式实时计算引擎，Flink 在指标系统设计上展现出了其独特之处。它不仅对指标上报进行了标准化定义和设计，还充分考虑了指标上报实现的多样性和可扩展性问题，因此我将之称为指标系统的弹性设计。下面，我们将详细分析 Flink 是如何实现这一设计的。

首先，我们来看看 Flink 对 MetricReporter 接口的定义。这个接口的定义非常简洁，其中包含三个关键信息：

1、open 和 close 方法分别在 MetricReporter 初始化和关闭时被调用。以滴滴内部的 KafkaReporter 实现为例，其主要功能是 Kafka 客户端的初始化和关闭操作。

2、notifyOfAddedMetric 和 notifyOfRemovedMetric 方法的主要作用是在 MetricGroup 添加或删除指标时，通知 MetricReporter，使其能够感知需要发送哪些指标。Flink 对 MetricReporter 进行了基础实现 AbstractReporter，我们可以看到在调用 notifyOfAddedMetric/notifyOfRemovedMetric 方法时，主要是在内存中维护了对指标的引用。

3、Flink 框架是如何加载并初始化 MetricReporter 的呢？源码中的注释给了我们提示：

>Reporters are instantiated via a {@link MetricReporterFactory}.

这意味着，MetricReporter 是通过 MetricReporterFactory 创建的（当然，在实际实现中，如果用户没有定义 MetricReporterFactory 的实现，也可以通过反射的方式初始化 MetricReporter）。

MetricReporterFactory 的定义是典型的工厂模式，注释中也包含了丰富的信息。简单来说，MetricReporterFactory 的实现类是通过 Java SPI 机制加载和实例化的。这里实际上是 Flink 指标系统实现弹性设计的关键所在。

也就是说，Flink 框架能够通过 Java SPI 机制按需加载 MetricReporterFactory 的实现类，再通过 MetricReporterFactory 实例化各种 MetricReporter。

MetricReporter 是 Flink 对指标系统的规范，MetricReporterFactory 的 SPI 加载机制为框架提供了灵活性。用户可以直接使用社区已经实现的指标系统，也可以自定义指标系统（比如，滴滴的 KafkaReporter）。本质上来说，这是一种遵循开闭原则的设计思想。

类似地，如果业务系统中有明确的业务逻辑，平台可以进行沉淀，同时将不同类型的需求通过 SPI 接口暴露出来，由业务方自行实现和维护，平台则按需加载业务方实现的 Jar 包即可，这就是插件思想的体现。

下面，我们将在源码层面详细分析弹性设计背后的实现细节：
当 Flink 启动 TaskManager 时，会触发 ReporterSetup 的初始化，其中的秘密都隐藏在这个类里。

让我们来详细分析一下 ReporterSetup 的 fromConfiguration 方法。简而言之，这个方法的核心步骤有两个：

第一步是通过 SPI 方式加载 MetricReporterFactory。
第二步是实例化 MetricReporter。

接下来，我们来分析通过 SPI 方式加载 MetricReporterFactory 的过程。ServiceLoader.load 是关键所在，我们可以看到它并没有传入 ClassLoader，也就是说它默认使用了 AppClassLoader（Flink 还支持 PluginManager 加载 MetricReporterFactory，本文不展开讨论，但其本质也是通过自定义类加载器加载不同指标系统实现的 jar 包）。

然后，我们再来分析一下 MetricReporter 的实例化过程。

最后，让我们来看一个真实的 MetricReporterFactory 实现类，以便对整个过程有更具体的了解。

Flink 指标注册和周期上报

接下来，我们将详细分析 Flink 指标注册与周期上报指标的实现原理。

指标注册

Flink 通过 MetricGroup 接口注册指标，这一功能的实现最终委托给了抽象类 AbstractMetricGroup 的 addMetric 方法。AbstractMetricGroup 将指标最终注册到了 MetricRegistry。MetricRegistry 是 MetricReporter 和指标之间的桥梁，负责跟踪所有已注册的指标。整体流程如下图所示：