本文首发于泊浮目的简书:https://www.jianshu.com/u/204b8aaab8ba

版本 日期 备注
1.0 2021.10.8 文章首发
1.1 2022.3.9 fix typo
1.2 2022.7.3 fix typo

0. 前言

前阵子笔者涉及了些许监控相关的开发工作，在开发过程中也碰到过些许问题，便翻读了Flink相关部分的代码，在读代码的过程中发现了一些好的设计，因此也是写成文章整理上来。

本文的源码基于Flink1.13.2。

1. 扩展插件化

在官网中，Flink社区自己提供了一些已接入的Reporter，如果我们有自己定制的Reporter，也可以根据它的规范去实现自己的Reporter。

在Flink的代码中，提供了反射机制实例化MetricReporter：要求MetricReporter的实现类必须是public的访问修饰符，不能是抽象类，必须有一个无参构造函数。

核心代码为ReporterSetup#getAllReporterFactories：

  1.     private static Iterator<MetricReporterFactory> getAllReporterFactories(
  2.             @Nullable PluginManager pluginManager) {
  3.         final Iterator<MetricReporterFactory> factoryIteratorSPI =
  4.                 ServiceLoader.load(MetricReporterFactory.class).iterator();
  5.         final Iterator<MetricReporterFactory> factoryIteratorPlugins =
  6.                 pluginManager != null
  7.                         ? pluginManager.load(MetricReporterFactory.class)
  8.                         : Collections.emptyIterator();
  10.         return Iterators.concat(factoryIteratorPlugins, factoryIteratorSPI);
  11.     }

该代码会通过Java的SPI机制来获取MetricReporter的相关实现类，本质上是通过ClassLoder来获取。

  1. |-- ReporterSetup
  2.      \-- fromConfiguration //当集群启动时，会从配置读取监控并初始化相关类
  3.          \-- loadAvailableReporterFactories // 加载有效的Reporter们
  4.              \-- getAllReporterFactories //  核心代码，通过SPI以及ClassLoader机制获取Reporter们

2. 内置松耦合

上文提到了社区会提供常见的一些监控Reporter。在代码中，本质是工厂模式的实现。

  1. /**
  2.  * {@link MetricReporter} factory.
  3.  *
  4.  * <p>Reporters that can be instantiated with a factory automatically qualify for being loaded as a
  5.  * plugin, so long as the reporter jar is self-contained (excluding Flink dependencies) and contains
  6.  * a {@code META-INF/services/org.apache.flink.metrics.reporter.MetricReporterFactory} file
  7.  * containing the qualified class name of the factory.
  8.  *
  9.  * <p>Reporters that previously relied on reflection for instantiation can use the {@link
  10.  * InstantiateViaFactory} annotation to redirect reflection-base instantiation attempts to the
  11.  * factory instead.
  12.  */
  13. public interface MetricReporterFactory {
  15.     /**
  16.      * Creates a new metric reporter.
  17.      *
  18.      * @param properties configured properties for the reporter
  19.      * @return created metric reporter
  20.      */
  21.     MetricReporter createMetricReporter(final Properties properties);
  22. }

每接入一个监控，只要实现相应的工厂方法即可。目前实现的有：

  • org.apache.flink.metrics.graphite.GraphiteReporterFactory
  • org.apache.flink.metrics.influxdb.InfluxdbReporterFactory
  • org.apache.flink.metrics.prometheus.PrometheusReporter
  • org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter
  • org.apache.flink.metrics.statsd.StatsDReporterFactory
  • org.apache.flink.metrics.datadog.DatadogHttpReporterFactory
  • org.apache.flink.metrics.slf4j.Slf4jReporterFactory

每当社区需要接入新的Reporter时，仅仅需要实现MetricReporterFactory即可，而上层能感知到的也仅仅是MetricReporter，和任何具体实现无关，这也是典型的一种防腐设计。

4207742-583abacf45d4c87f.jpeg

3. Fail safe

在流计算业务中，如果监控这种旁路逻辑发生问题，是否应该影响到主干逻辑呢？答案是不应该的。

在MetricRegistryImpl中（顾名思义，它会将所有的Reporter注册进这个类），构造函数会将相关的MetricReporter放到线程池中，定期的让它们上报数据。

  1. |-- MetricRegistryImpl
  2.   \-- constructor

在WebMonitorEndpoint中，也有线程池的身影。这个类提供了RestAPI来便于查询Metric。对于其他组件的请求通过Akka来异步发送，并通过线程池来处理这些回调的回复。

  1. |-- WebMonitorEndpoint
  2.   \-- start
  3.     \-- initializeHandlers
  4.       \--   new JobConfigHandler
  5. |-- AbstractExecutionGraphHandler
  6.   \-- handleRequest

这是典型Fail-safe的设计。

4. 不仅只支持Push

在Flink中，监控数据不仅支持Push，同时还实现了Pull，而实现也非常的简单。

MetricQueryService实现了MetricQueryServiceGateway，这意味着它可以被远程调用。

其监控数据来源代码追踪：

  1. |-- AbstractMetricGroup
  2.   \-- counter
  3.     |-- MetricRegistryImpl
  4.       \-- register
  5.         |-- MetricQueryService
  6.           \-- addMetric

上面提到的WebMonitorEndpoint也是一样，不过是基于RestAPI的实现，同样提供了Pull的策略。

5. 参考资料