Flink配置Prometheus监控

准备Flink软件包

首先在Flink官网下载Flink的安装包。

复制opt/flink-metrics-prometheus-1.8.1.jar 到flink的lib目录下。

配置flink.yaml

metrics.reporter.promgateway.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter
metrics.reporter.promgateway.host: 10.180.210.172
metrics.reporter.promgateway.port: 9091

Prometheus配置

下载prometheus

Prometheus 的下载链接为：
https://prometheus.io/download/

这里所需Prometheus的组件为：

prometheus
pushgateway(Flink推送监控数据到此)
node_exporter(节点信息监控，CPU内存等，可以不安装)

将这些组件分别解压到任意目录。

配置Prometheus

修改Prometheus根目录prometheus.yml文件的scrape_config，如下图所示：

scrape_configs:
  # The job name is added as a label `job=` to any timeseries scraped from this config.
  - job_name: 'pushgateway'

    # metrics_path defaults to '/metrics'
    # scheme defaults to 'http'.

    static_configs:
    - targets: ['10.180.210.172:9091']
      labels:
        instance: 'pushgateway'
  - job_name: 'linux'
    static_configs:
    - targets: ['10.180.210.172:9100']
      labels:
        instance: 'linux'
        
  - job_name: prometheus
    static_configs:
    - targets: ['10.180.210.172:9090']
      labels:
        instance: prometheus

启动prometheus, node_exporter和pushgateway

在命令行启动prometheus, node_exporter和pushgateway。

通过访问组件对应的URL确定是否成功启动。
端口号：

prometheus: 9090
node_exporter: 9100
pushgateway: 9091

Prometheus主界面

pushgateway界面

node-exporter界面

查看metrics:
访问对应组件的URL，例如http://10.180.210.172:9091/metrics（初次安装这里为空白）

Prometheus的metrics界面

查看各个endpoint是否成功注册在prometheus：
访问prometheus首页（9090端口），打开status -> targets，观察各个endpoint的status是否为UP。

Prometheus的target状态页面

配置安装Grafana

下载并启动Grafana

下载Grafana解压，添加prometheus数据源。

Grafana的下载地址为: https://grafana.com/grafana/download

下载后解压至任意目录，使用默认配置。执行Grafana安装目录中的bin/grafana-server 即可启动Grafana。

配置Grafana

配置数据源

Grafana默认运行在3000端口。使用默认用户名密码（admin/admin）登录Grafana后，选择左侧菜单的Configuration -> Data Sources。

Data Source菜单

接下来点击右侧的Add data source按钮。在数据源类型选择中界面选择Prometheus。

数据源选择

最后，将Prometheus的地址端口号填写入URL这一栏，点击下方的Save & Test按钮。如果配置无误，会弹出数据源正常连接的提示信息。

数据源配置

配置Dashboard

选择左侧菜单的Create -> Dashboard，然后选择Add Query。
Dashboard的Add Panel，Add Query。会进入到指标查询配置页面。

监控指标配置

打开Query右侧的下拉列表，选择使用的数据源。

数据源选择框

接下来配置查询语句。有以下两种方式点击Metrics按钮，根据分类选择监控指标。或者是在文本框中输入指标的关键字。

查询语句配置

查询语句配置完毕后，页面上方会显示出监控数据，如图所示。

查询语句配置完毕

返回Dashboard后可以看到配置的监控图已经固定到了页面当中，方便以后查看。

至此，Grafana显示监控指标的基本配置已介绍完毕。Grafana的其他配置会在以后博客中更新。

Flink增加自定义监控指标

默认Flink仅提供了job manager和task manager的JVM和系统配置相关的监控数据。我们可以通过Flink提供的metrics API，在具体业务中添加自定义的监控指标。

Flink metrics的官网文档链接：https://ci.apache.org/projects/flink/flink-docs-stable/monitoring/metrics.html

下面我们举个例子。我们自定义一个监控指标，叫做my_counter，监控数据源（socket）读取到的数据条数。代码如下：

env.enableCheckpointing(500)

val stream = env.socketTextStream("10.180.210.172", 11000)

// 使用RichFunction可以获取到RuntimeContext
stream.map(new RichMapFunction[String, String] {
  var counter: Counter = _

  override def open(parameters: Configuration): Unit = {
    // 自定义指标名字为`my_counter`
    counter = getRuntimeContext.getMetricGroup.counter("my_counter")
  }

  override def map(in: String): String = {
    counter.inc()
    in
  }
})

// execute program
env.execute("Flink Custom Metrics Demo")

Flink运行该任务后，我们可以在Grafana监控配置界面中找到这个自定义指标。