Flink Python自定义Metrics无法在Datadog展示，求问题排查

问题场景 #

我通过Python的FlatMapFunction生成了自定义Metrics，代码如下：

class OccupancyEventFlatMap(FlatMapFunction):
    def open(self, runtime_context: RuntimeContext):
        mg = runtime_context.get_metrics_group()
        self.counter_sum = mg.counter("my_counter_sum")
        self.counter_total = mg.counter("my_counter_total")
    def flat_map(self, value):
        self.counter_sum.inc(10)
        self.counter_total.inc()

通过Flink REST API可以正常查询到自定义指标数值：

请求地址：http://localhost:43491/jobs/9a376e28a1bb022b45c127d75fb1b447/vertices/5239a5f0e3e9cdca6a88500e58b5759e/metrics?get=0.FlatMap.my_counter_sum

返回结果：

[{"id":"0.FlatMap.my_counter_sum","value":"28201"}]

但Datadog中无法看到任何自定义Metrics，Flink标准指标却能正常显示。我的Flink Datadog exporter配置如下：

# Datadog
metrics.reporter.dghttp.class: org.apache.flink.metrics.datadog.DatadogHttpReporter
metrics.reporter.dghttp.dataCenter: US
metrics.reporter.dghttp.apikey: ${datadog_api_key}
metrics.reporter.dghttp.tags: env:development

# Metrics scope configuration
metrics.scope.jm: flink.jobmanager
metrics.scope.jm.job: flink.jobmanager.job
metrics.scope.tm: flink.taskmanager
metrics.scope.tm.job: flink.taskmanager.job
metrics.scope.task: flink.task
metrics.scope.operator: flink.operator

排查与解决方案 #

以下是几个可能的问题点及对应解决方法：

• 搜索完整指标名称
  Flink会为自定义指标拼接作用域前缀，你的my_counter_sum最终上报的完整指标名应为flink.operator.0.FlatMap.my_counter_sum。直接在Datadog搜索框输入这个完整名称，避免因命名空间过滤导致无法找到。

• 添加Reporter过滤规则
  默认情况下Datadog Reporter会上报所有Metrics，但如果存在全局过滤规则可能排除自定义指标。显式添加规则确保自定义指标被包含：

  metrics.reporter.dghttp.filter: *
  

  或精准匹配自定义指标模式：

  metrics.reporter.dghttp.filter: my_counter_*
  

• 配置上报间隔
  显式设置上报间隔，确保指标定时推送：

  metrics.reporter.dghttp.interval: 10 SECONDS
  

• 检查Flink日志
  查看JobManager或TaskManager日志，搜索DatadogHttpReporter相关内容，确认是否存在指标发送失败、格式错误等异常信息。

• 验证Datadog集成设置
  确认Datadog的Flink集成未开启"仅收集默认指标"选项，同时检查Datadog的指标过滤规则，确保flink.operator前缀的指标未被排除。

内容的提问来源于stack exchange，提问作者Adrian Chang Alcover