流式计算 Flink 版支持通过 PushGateway 将任务指标推送到您的 Prometheus 监控体系。除了 Flink 内置指标，您还可以推送在任务代码中注册的业务指标，例如处理记录数、队列长度和处理耗时，用于统一查询指标、查看业务变化和配置告警。

Prometheus 是开源监控系统。PushGateway 是接收指标推送的中间服务，Flink 的指标上报组件（Reporter）定期向其发送指标，再由 Prometheus 拉取。火山引擎托管 Prometheus（Volcengine Managed Service for Prometheus，VMP）提供兼容 PushGateway 协议的推送接口，也可以作为推送目标。

该功能适用于以下场景：

• 将 Flink 任务指标接入自建 Prometheus 监控体系。
• 将 Flink 任务指标推送到火山引擎托管 Prometheus（VMP）工作区。
• 向已启用 Basic Auth（Basic Authentication，基本认证）的推送目标发送指标，使用用户名和密码对推送请求进行鉴权。

• Basic Auth 仅支持 Flink 1.17-volcano 及以上版本。低版本不支持 username 和 password 参数。
• 推送目标必须兼容 Prometheus PushGateway 协议。
• 任务运行环境能够访问目标推送地址。

metrics.reporters: promvolc,promgateway

准备推送地址 #

根据目标服务准备 hostUrl：

• 火山引擎托管 Prometheus（VMP）：获取目标工作区的 Push Gateway URL，具体方法参见使用 Pushgateway 推送数据。使用工作区提供的完整地址，保留其中的工作区标识和 /api/v1/push 路径。
• 自建 PushGateway：准备 PushGateway，并配置 Prometheus 拉取其指标。获取 PushGateway 的访问地址，例如 http://pushgateway.example.com:9091。其中 http:// 是协议，pushgateway.example.com 是主机名，9091 是该服务监听的端口；请替换为实际服务地址。

将目标服务提供的完整 PushGateway 地址填入 hostUrl，保留其中的协议、主机、端口和路径。

目标服务启用 Basic Auth 时，需同时配置有效的 username 和 password。

在控制台配置任务参数 #

1. 登录流式计算 Flink 版控制台，在顶部选择目标地域。
2. 进入目标空间，在左侧导航栏选择任务开发 > 任务开发。
3. 在任务列表中打开需要配置指标推送的任务。
4. 单击任务编辑区域右侧的参数配置，打开配置参数抽屉。
   
5. 在配置参数抽屉中展开自定义参数，选择脚本模式或 KV 模式。
   
6. 参考下文的参数说明和配置示例填写参数，并保留已有的其他任务配置。脚本模式下，按 参数名: 参数值 的格式逐行填写；KV（Key-Value，键值对）模式下，分别填写参数名和参数值，例如键为 metrics.reporters，值为 promvolc,promgateway。
7. 单击抽屉右下角的保存。
8. 单击任务编辑区域顶部的保存，保存任务配置，然后单击上线，重新上线任务，并在应用新配置后启动任务。已运行的任务需在新配置下重新启动。

<名称> 为 metrics.reporters 中追加的自定义采集器名称，例如 promgateway；填写参数时，将 <名称> 替换为该名称。

以下示例使用脚本模式，已将 <名称> 替换为对应的采集器名称。使用前请替换推送地址、分组信息和认证信息，并将示例中的 metrics.reporters 与任务已有列表合并。

KV 模式下，分别填写完整参数名和值，例如：

推送到托管 Prometheus，启用 Basic Auth #

适用于火山引擎托管 Prometheus（VMP）工作区已启用 Basic Auth 的场景，任务引擎需为 Flink 1.17-volcano 及以上版本。请使用目标工作区实际提供的 Push Gateway URL；示例地址中的地域和工作区标识不能直接复用。

# 保留平台默认 Reporter；如已有其他名称，也需保留。
metrics.reporters: promvolc,promgateway

# 配置 PushGateway Reporter 和目标工作区的完整推送地址。
metrics.reporter.promgateway.factory.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterFactory
metrics.reporter.promgateway.hostUrl: "http://push.prometheus-cn-beijing.ivolces.com/workspaces/<workspace-id>/api/v1/push"
metrics.reporter.promgateway.jobName: flink-vmp-demo
metrics.reporter.promgateway.groupingKey: "env=prod;service=payment"

# 配置随机后缀、任务停止时的删除行为和推送间隔。
metrics.reporter.promgateway.randomJobNameSuffix: false
metrics.reporter.promgateway.deleteOnShutdown: false
metrics.reporter.promgateway.interval: 10 SECONDS

# 同时填写目标服务的 Basic Auth 用户名和密码。
metrics.reporter.promgateway.username: "<basic-auth-username>"
metrics.reporter.promgateway.password: "<basic-auth-password>"

推送到自建 PushGateway，无需认证 #

适用于自建 PushGateway 未启用 Basic Auth 的场景。省略 username 和 password，不要填写空值。替换 <pushgateway-host> 和端口，并确认 Prometheus 已配置对该 PushGateway 的指标拉取。

# 保留已有 Reporter，追加自定义推送组件。
metrics.reporters: promvolc,promgateway

# 替换为自建 PushGateway 的实际地址。
metrics.reporter.promgateway.factory.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterFactory
metrics.reporter.promgateway.hostUrl: "http://<pushgateway-host>:9091"
metrics.reporter.promgateway.jobName: flink-pushgateway-demo

# 配置随机后缀、任务停止时的删除行为和推送间隔。
metrics.reporter.promgateway.randomJobNameSuffix: false
metrics.reporter.promgateway.deleteOnShutdown: false
metrics.reporter.promgateway.interval: 10 SECONDS

# 无需 Basic Auth，省略 username 和 password。

同时推送到多个目标 #

需要同时向多个服务推送时，在 metrics.reporters 中追加不同名称，并为每个名称分别配置参数。以下示例同时配置需要 Basic Auth 的火山引擎托管 Prometheus（VMP）和无需认证的自建 PushGateway。

# 保留原有列表，再追加两个不同名称。
metrics.reporters: promvolc,promgateway-auth,promgateway-noauth

# 目标一：托管 Prometheus，启用 Basic Auth。
metrics.reporter.promgateway-auth.factory.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterFactory
metrics.reporter.promgateway-auth.hostUrl: "http://push.prometheus-cn-beijing.ivolces.com/workspaces/<workspace-id>/api/v1/push"
metrics.reporter.promgateway-auth.jobName: flink-vmp-demo
metrics.reporter.promgateway-auth.randomJobNameSuffix: false
metrics.reporter.promgateway-auth.deleteOnShutdown: false
metrics.reporter.promgateway-auth.username: "<basic-auth-username>"
metrics.reporter.promgateway-auth.password: "<basic-auth-password>"
metrics.reporter.promgateway-auth.interval: 10 SECONDS

# 目标二：自建 PushGateway，无需 Basic Auth。
metrics.reporter.promgateway-noauth.factory.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterFactory
metrics.reporter.promgateway-noauth.hostUrl: "http://<pushgateway-host>:9091"
metrics.reporter.promgateway-noauth.jobName: flink-pushgateway-demo
metrics.reporter.promgateway-noauth.randomJobNameSuffix: false
metrics.reporter.promgateway-noauth.deleteOnShutdown: false
metrics.reporter.promgateway-noauth.interval: 10 SECONDS

控制台参数用于配置指标推送；需要观测业务自定义数据时，还需在任务代码中注册指标。以下以 Java DataStream 任务为例，使用 Flink 指标接口注册业务指标。完整说明参见 Apache Flink 指标文档。

在继承 RichFunction 的用户函数中，通过 getRuntimeContext().getMetricGroup() 获取指标组 MetricGroup，通常在 open() 中注册指标，再在业务处理逻辑中更新指标。可使用 addGroup() 将相关指标归入同一业务组。

Java 示例 #

以下示例使用 Flink 1.17 API（Application Programming Interface，应用程序编程接口），在 RichMapFunction 中注册已处理记录数 processed_records 和最近一条字符串记录的长度 last_record_length，统一归入 business 指标组。

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.MetricGroup;

public class BusinessMetricsMapper extends RichMapFunction<String, String> {
    private transient Counter processedRecords;
    private transient volatile int lastRecordLength;

    @Override
    public void open(Configuration parameters) {
        // 在函数初始化时创建业务指标组并注册指标。
        MetricGroup metrics = getRuntimeContext()
                .getMetricGroup()
                .addGroup("business");
        processedRecords = metrics.counter("processed_records");
        metrics.gauge("last_record_length",
                (Gauge<Integer>) () -> lastRecordLength);
    }

    @Override
    public String map(String value) {
        // 每处理一条记录，更新计数器和最近一条记录的长度。
        processedRecords.inc();
        lastRecordLength = value.length();
        return value;
    }
}

将该函数接入已有的非空字符串数据流，并连接任务实际使用的下游输出：

// stream 为已有的 DataStream<String>；后续接入任务实际使用的下游。
stream.map(new BusinessMetricsMapper());

每处理一条记录，processed_records 增加 1，last_record_length 更新为该字符串的 length() 返回值。

• 托管 Prometheus：使用 Pushgateway 推送数据
• Apache Flink 1.17：Metrics