[FLINK-28997][datadog] Add switch to use logical identifier

docs/content.zh/docs/deployment/metric_reporters.md

@@ -261,6 +261,8 @@ metrics.reporter.stsd.interval: 60 SECONDS
 
 使用 Datadog 时，Flink 运行指标中的任何变量，例如 `<host>`、`<job_name>`、 `<tm_id>`、 `<subtask_index>`、`<task_name>`、 `<operator_name>`，都会被当作 `host:localhost`、`job_name:myjobname` 这样的 tag 发送。
 
+<span class="label label-danger">Note</span> For legacy reasons the reporter uses _both_ the metric identifier _and_ tags. This redundancy can be avoided by enabling `useLogicalIdentifier`.
+
 <span class="label label-info">注意</span> 按照 Datedog 的 Histograms 命名约定，Histograms 类的运行指标会作为一系列 gauges 显示（`<metric_name>.<aggregation>`）。
 默认情况下 `min` 即最小值被发送到 Datedog，`sum` 不会被发送。
 与 Datadog 提供的 Histograms 相比，Histograms 类的运行指标不会按照指定的发送间隔进行聚合计算。
@@ -273,6 +275,7 @@ metrics.reporter.stsd.interval: 60 SECONDS
 - `proxyPort` - （可选的）发送到 Datadog 时使用的代理端口，默认为 8080。
 - `dataCenter` - （可选的）要连接的数据中心（`EU`/`US`），默认为 `US`。
 - `maxMetricsPerRequest` - （可选的）每次请求携带的最大运行指标个数，默认为 2000。
+- `useLogicalIdentifier` -> (optional) Whether the reporter uses a logical metric identifier, defaults to `false`.
 
 配置示例:
 
@@ -285,6 +288,7 @@ metrics.reporter.dghttp.proxyPort: 8080
 metrics.reporter.dghttp.dataCenter: US
 metrics.reporter.dghttp.maxMetricsPerRequest: 2000
 metrics.reporter.dghttp.interval: 60 SECONDS
+metrics.reporter.dghttp.useLogicalIdentifier: true
 ```
 
 <a name="slf4j"></a>

docs/content/docs/deployment/metric_reporters.md

@@ -245,6 +245,8 @@ Type: push/tags
 Note any variables in Flink metrics, such as `<host>`, `<job_name>`, `<tm_id>`, `<subtask_index>`, `<task_name>`, and `<operator_name>`,
 will be sent to Datadog as tags. Tags will look like `host:localhost` and `job_name:myjobname`.
 
+<span class="label label-danger">Note</span> For legacy reasons the reporter uses _both_ the metric identifier _and_ tags. This redundancy can be avoided by enabling `useLogicalIdentifier`.
+
 <span class="label label-info">Note</span> Histograms are exposed as a series of gauges following the naming convention of Datadog histograms (`<metric_name>.<aggregation>`).
 The `min` aggregation is reported by default, whereas `sum` is not available.
 In contrast to Datadog-provided Histograms the reported aggregations are not computed for a specific reporting interval.
@@ -257,6 +259,7 @@ Parameters:
 - `proxyPort` - (optional) The proxy port to use when sending to Datadog, defaults to 8080.
 - `dataCenter` - (optional) The data center (`EU`/`US`) to connect to, defaults to `US`.
 - `maxMetricsPerRequest` - (optional) The maximum number of metrics to include in each request, defaults to 2000.
+- `useLogicalIdentifier` -> (optional) Whether the reporter uses a logical metric identifier, defaults to `false`.
 
 Example configuration:
 
@@ -269,6 +272,7 @@ metrics.reporter.dghttp.proxyPort: 8080
 metrics.reporter.dghttp.dataCenter: US
 metrics.reporter.dghttp.maxMetricsPerRequest: 2000
 metrics.reporter.dghttp.interval: 60 SECONDS
+metrics.reporter.dghttp.useLogicalIdentifier: true
 ```
 
 

flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java

@@ -18,9 +18,11 @@
 
 package org.apache.flink.metrics.datadog;
 
+import org.apache.flink.metrics.CharacterFilter;
 import org.apache.flink.metrics.Counter;
 import org.apache.flink.metrics.Gauge;
 import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.LogicalScopeProvider;
 import org.apache.flink.metrics.Meter;
 import org.apache.flink.metrics.Metric;
 import org.apache.flink.metrics.MetricConfig;
@@ -56,6 +58,7 @@ public class DatadogHttpReporter implements MetricReporter, Scheduled {
  private final DatadogHttpClient client;
  private final List<String> configTags;
  private final int maxMetricsPerRequestValue;
+ private final boolean useLogicalIdentifier;
 
  private final Clock clock = () -> System.currentTimeMillis() / 1000L;
 
@@ -65,8 +68,10 @@ public DatadogHttpReporter(
  int proxyPort,
  int maxMetricsPerRequestValue,
  DataCenter dataCenter,
- String tags) {
+ String tags,
+ boolean useLogicalIdentifier) {
  this.maxMetricsPerRequestValue = maxMetricsPerRequestValue;
+ this.useLogicalIdentifier = useLogicalIdentifier;
  this.client = new DatadogHttpClient(apiKey, proxyHost, proxyPort, dataCenter, true);
  this.configTags = getTagsFromConfig(tags);
 
@@ -81,7 +86,13 @@ public DatadogHttpReporter(
 
  @Override
  public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
- final String name = group.getMetricIdentifier(metricName);
+ final String name =
+ this.useLogicalIdentifier
+ ? ((LogicalScopeProvider) group)
+ .getLogicalScope(CharacterFilter.NO_OP_FILTER)
+ + "."
+ + metricName
+ : group.getMetricIdentifier(metricName);
 
  List<String> tags = new ArrayList<>(configTags);
  tags.addAll(getTagsFromMetricGroup(group));

flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporterFactory.java

@@ -32,6 +32,7 @@ public class DatadogHttpReporterFactory implements MetricReporterFactory {
  private static final String DATA_CENTER = "dataCenter";
  private static final String TAGS = "tags";
  private static final String MAX_METRICS_PER_REQUEST = "maxMetricsPerRequest";
+ private static final String USE_LOGICAL_IDENTIFIER = "useLogicalIdentifier";
 
  @Override
  public MetricReporter createMetricReporter(Properties config) {
@@ -43,13 +44,16 @@ public MetricReporter createMetricReporter(Properties config) {
  Integer.valueOf(config.getProperty(MAX_METRICS_PER_REQUEST, "2000"));
  final DataCenter dataCenter = DataCenter.valueOf(rawDataCenter);
  final String tags = config.getProperty(TAGS, "");
+ final boolean useLogicalIdentifier =
+ Boolean.parseBoolean(config.getProperty(USE_LOGICAL_IDENTIFIER, "false"));
 
  return new DatadogHttpReporter(
  apiKey,
  proxyHost,
  proxyPort,
  maxMetricsPerRequestValue,
  dataCenter,
- tags);
+ tags,
+ useLogicalIdentifier);
  }
 }