Followup: address PR feedback

- Keep `randomJobNameSuffix` config as deprecated
- When `randomJobNameSuffix` is disabled (by default), group metrics using random reporter id

Author: qinghui-xu

flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporter.java

@@ -58,6 +58,7 @@ public class PrometheusPushGatewayReporter extends AbstractPrometheusReporter im
     PrometheusPushGatewayReporter(
             URL hostUrl,
             String jobName,
+            boolean metricsGroupingByReporter,
             Map<String, String> groupingKey,
             final boolean deleteOnShutdown,
             @Nullable String username,
@@ -72,7 +73,10 @@ public class PrometheusPushGatewayReporter extends AbstractPrometheusReporter im
             this.basicAuthEnabled = false;
         }
         this.jobName = Preconditions.checkNotNull(jobName);
-        this.groupingKey = new GroupingKeyMap(Preconditions.checkNotNull(groupingKey));
+        this.groupingKey =
+                metricsGroupingByReporter
+                        ? new GroupingKeyMap(Preconditions.checkNotNull(groupingKey))
+                        : Preconditions.checkNotNull(groupingKey);
         this.deleteOnShutdown = deleteOnShutdown;
     }
 

flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporterFactory.java

@@ -20,6 +20,7 @@
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.metrics.MetricConfig;
 import org.apache.flink.metrics.reporter.MetricReporterFactory;
+import org.apache.flink.util.AbstractID;
 import org.apache.flink.util.StringUtils;
 
 import org.slf4j.Logger;
@@ -37,6 +38,7 @@
 import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.HOST_URL;
 import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.JOB_NAME;
 import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.PASSWORD;
+import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.RANDOM_JOB_NAME_SUFFIX;
 import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.USERNAME;
 
 /** {@link MetricReporterFactory} for {@link PrometheusPushGatewayReporter}. */
@@ -49,6 +51,9 @@ public class PrometheusPushGatewayReporterFactory implements MetricReporterFacto
     public PrometheusPushGatewayReporter createMetricReporter(Properties properties) {
         MetricConfig metricConfig = (MetricConfig) properties;
         String configuredJobName = metricConfig.getString(JOB_NAME.key(), JOB_NAME.defaultValue());
+        boolean randomSuffix =
+                metricConfig.getBoolean(
+                        RANDOM_JOB_NAME_SUFFIX.key(), RANDOM_JOB_NAME_SUFFIX.defaultValue());
         boolean deleteOnShutdown =
                 metricConfig.getBoolean(
                         DELETE_ON_SHUTDOWN.key(), DELETE_ON_SHUTDOWN.defaultValue());
@@ -63,6 +68,10 @@ public PrometheusPushGatewayReporter createMetricReporter(Properties properties)
         }
 
         String jobName = configuredJobName;
+        if (randomSuffix) {
+            jobName = configuredJobName + new AbstractID();
+        }
+
         String username = metricConfig.getString(USERNAME.key(), USERNAME.defaultValue());
         String password = metricConfig.getString(PASSWORD.key(), PASSWORD.defaultValue());
 
@@ -78,15 +87,17 @@ public PrometheusPushGatewayReporter createMetricReporter(Properties properties)
                     new PrometheusPushGatewayReporter(
                             new URL(hostUrl),
                             jobName,
+                            !randomSuffix,
                             groupingKey,
                             deleteOnShutdown,
                             username,
                             password);
 
             LOG.info(
-                    "Configured PrometheusPushGatewayReporter with {hostUrl:{}, jobName:{}, deleteOnShutdown:{}, groupingKey:{}, basicAuth:{}}",
+                    "Configured PrometheusPushGatewayReporter with {hostUrl:{}, jobName:{}, randomJobNameSuffix:{}, deleteOnShutdown:{}, groupingKey:{}, basicAuth:{}}",
                     hostUrl,
                     jobName,
+                    randomSuffix,
                     deleteOnShutdown,
                     groupingKey,
                     reporter.basicAuthEnabled);

flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporterOptions.java

@@ -45,6 +45,16 @@ public class PrometheusPushGatewayReporterOptions {
                     .defaultValue("")
                     .withDescription("The job name under which metrics will be pushed");
 
+    @Deprecated
+    public static final ConfigOption<Boolean> RANDOM_JOB_NAME_SUFFIX =
+            ConfigOptions.key("randomJobNameSuffix")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Specifies whether random suffixing `job` label (the old way) to avoid metric collision among reporters from taskamangers."
+                                    + " When disabled (now default) , metrics will be grouped under a random reporter id while job name is faithful to configuration."
+                                    + " This option is deprecated and it is recommended to rely on the reporter id grouping key.");
+
     public static final ConfigOption<Boolean> DELETE_ON_SHUTDOWN =
             ConfigOptions.key("deleteOnShutdown")
                     .booleanType()

flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporterTest.java

@@ -30,6 +30,7 @@
 import java.util.Map;
 
 import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter.REPORTER_ID_GROUPPING_KEY;
+import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.GROUPING_KEY;
 import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.HOST_URL;
 import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.PASSWORD;
 import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.USERNAME;
@@ -80,6 +81,21 @@ void testGroupingKeysIteratorEndsWithReporterId() {
                                 REPORTER_ID_GROUPPING_KEY, groupingKey.reporterId()));
     }
 
+    @Test
+    void testRejectReporterIdInUserGroupingKey() {
+        MetricConfig metricConfig = new MetricConfig();
+        metricConfig.setProperty(HOST_URL.key(), "http://localhost:8080");
+        metricConfig.setProperty(GROUPING_KEY.key(), "k1=v1;" + REPORTER_ID_GROUPPING_KEY + "=123");
+        assertThatThrownBy(
+                        () ->
+                                new PrometheusPushGatewayReporterFactory()
+                                        .createMetricReporter(metricConfig))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining(
+                        "Grouping keys must not contain the reserved key: "
+                                + REPORTER_ID_GROUPPING_KEY);
+    }
+
     @Test
     void testConnectToPushGatewayUsingHostUrl() {
         PrometheusPushGatewayReporterFactory factory = new PrometheusPushGatewayReporterFactory();