diff --git a/docs/content.zh/docs/deployment/metric_reporters.md b/docs/content.zh/docs/deployment/metric_reporters.md index d9666628be7fc..7c432e5c6f599 100644 --- a/docs/content.zh/docs/deployment/metric_reporters.md +++ b/docs/content.zh/docs/deployment/metric_reporters.md @@ -35,18 +35,12 @@ For more information about Flink's metric system go to the [metric system docume Metrics can be exposed to an external system by configuring one or several reporters in `conf/flink-conf.yaml`. These reporters will be instantiated on each job and task manager when they are started. -- `metrics.reporter..`: Generic setting `` for the reporter named ``. -- `metrics.reporter..class`: The reporter class to use for the reporter named ``. -- `metrics.reporter..factory.class`: The reporter factory class to use for the reporter named ``. -- `metrics.reporter..interval`: The reporter interval to use for the reporter named ``. -- `metrics.reporter..scope.delimiter`: The delimiter to use for the identifier (default value use `metrics.scope.delimiter`) for the reporter named ``. -- `metrics.reporter..scope.variables.excludes`: (optional) A semi-colon (;) separate list of variables that should be ignored by tag-based reporters (e.g., Prometheus, InfluxDB). -- `metrics.reporters`: (optional) A comma-separated include list of reporter names. By default all configured reporters will be used. -- `metrics.reporter..scope.variables.additional`: (optional) A comma separated map of variables and their values, which are separated by a colon (:). These mappings are added to the variable map by tag-based reporters (e.g. Prometheux, InfluxDB). +Below is a list of parameters that are generally applicable to all reporters. All properties are configured by setting `metrics.reporter..` in the configuration. Reporters may additionally offer implementation-specific parameters, which are documented in the respective reporter's section. + +{{< include_reporter_config "layouts/shortcodes/generated/metric_reporters_section.html" >}} All reporters must at least have either the `class` or `factory.class` property. Which property may/should be used depends on the reporter implementation. See the individual reporter configuration sections for more information. Some reporters (referred to as `Scheduled`) allow specifying a reporting `interval`. -Below more settings specific to each reporter will be listed. Example reporter configuration that specifies multiple reporters: diff --git a/docs/content/docs/deployment/metric_reporters.md b/docs/content/docs/deployment/metric_reporters.md index a78acef737f2f..c327b76935a15 100644 --- a/docs/content/docs/deployment/metric_reporters.md +++ b/docs/content/docs/deployment/metric_reporters.md @@ -35,18 +35,12 @@ For more information about Flink's metric system go to the [metric system docume Metrics can be exposed to an external system by configuring one or several reporters in `conf/flink-conf.yaml`. These reporters will be instantiated on each job and task manager when they are started. -- `metrics.reporter..`: Generic setting `` for the reporter named ``. -- `metrics.reporter..class`: The reporter class to use for the reporter named ``. -- `metrics.reporter..factory.class`: The reporter factory class to use for the reporter named ``. -- `metrics.reporter..interval`: The reporter interval to use for the reporter named ``. -- `metrics.reporter..scope.delimiter`: The delimiter to use for the identifier (default value use `metrics.scope.delimiter`) for the reporter named ``. -- `metrics.reporter..scope.variables.excludes`: (optional) A semi-colon (;) separate list of variables that should be ignored by tag-based reporters (e.g., Prometheus, InfluxDB). -- `metrics.reporter..scope.variables.additional`: (optional) A comma separated map of variables and their values, which are separated by a colon (:). These mappings are added to the variable map by tag-based reporters (e.g. Prometheux, InfluxDB). -- `metrics.reporters`: (optional) A comma-separated include list of reporter names. By default all configured reporters will be used. +Below is a list of parameters that are generally applicable to all reporters. All properties are configured by setting `metrics.reporter..` in the configuration. Reporters may additionally offer implementation-specific parameters, which are documented in the respective reporter's section. + +{{< include_reporter_config "layouts/shortcodes/generated/metric_reporters_section.html" >}} All reporters must at least have either the `class` or `factory.class` property. Which property may/should be used depends on the reporter implementation. See the individual reporter configuration sections for more information. Some reporters (referred to as `Scheduled`) allow specifying a reporting `interval`. -Below more settings specific to each reporter will be listed. Example reporter configuration that specifies multiple reporters: diff --git a/docs/layouts/shortcodes/generated/metric_configuration.html b/docs/layouts/shortcodes/generated/metric_configuration.html index c0fbc8690544b..4fabd043fd93a 100644 --- a/docs/layouts/shortcodes/generated/metric_configuration.html +++ b/docs/layouts/shortcodes/generated/metric_configuration.html @@ -62,12 +62,36 @@ String The reporter class to use for the reporter named <name>. + +
metrics.reporter.<name>.factory.class
+ (none) + String + The reporter factory class to use for the reporter named <name>. +
metrics.reporter.<name>.interval
10 s Duration The reporter interval to use for the reporter named <name>. + +
metrics.reporter.<name>.scope.delimiter
+ "." + String + The delimiter used to assemble the metric identifier for the reporter named <name>. + + +
metrics.reporter.<name>.scope.variables.additional
+ + Map + The map of additional variables that should be included for the reporter named <name>. Only applicable to tag-based reporters (e.g., PRometheus, InfluxDB). + + +
metrics.reporter.<name>.scope.variables.excludes
+ "." + String + The set of variables that should be excluded for the reporter named <name>. Only applicable to tag-based reporters (e.g., PRometheus, InfluxDB). +
metrics.reporters
(none) diff --git a/docs/layouts/shortcodes/generated/metric_reporters_section.html b/docs/layouts/shortcodes/generated/metric_reporters_section.html new file mode 100644 index 0000000000000..bc4068ce743e5 --- /dev/null +++ b/docs/layouts/shortcodes/generated/metric_reporters_section.html @@ -0,0 +1,54 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
KeyDefaultTypeDescription
metrics.reporter.<name>.class
(none)StringThe reporter class to use for the reporter named <name>.
metrics.reporter.<name>.factory.class
(none)StringThe reporter factory class to use for the reporter named <name>.
metrics.reporter.<name>.interval
10 sDurationThe reporter interval to use for the reporter named <name>.
metrics.reporter.<name>.scope.delimiter
"."StringThe delimiter used to assemble the metric identifier for the reporter named <name>.
metrics.reporter.<name>.scope.variables.additional
MapThe map of additional variables that should be included for the reporter named <name>. Only applicable to tag-based reporters (e.g., PRometheus, InfluxDB).
metrics.reporter.<name>.scope.variables.excludes
"."StringThe set of variables that should be excluded for the reporter named <name>. Only applicable to tag-based reporters (e.g., PRometheus, InfluxDB).
metrics.reporter.<name>.<parameter>
(none)StringConfigures the parameter <parameter> for the reporter named <name>.
diff --git a/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java b/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java index e8b150fd7df39..7bdb196e66185 100644 --- a/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java +++ b/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java @@ -103,6 +103,8 @@ public static final class Sections { public static final String DEPRECATED_FILE_SINKS = "deprecated_file_sinks"; + public static final String METRIC_REPORTERS = "metric_reporters"; + private Sections() {} } diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java index fab7cd67ecb15..c4b0735532477 100644 --- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java +++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java @@ -21,6 +21,7 @@ import org.apache.flink.configuration.ConfigConstants; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.MemorySize; +import org.apache.flink.configuration.MetricOptions; import org.apache.flink.configuration.TaskManagerOptions; import org.apache.flink.metrics.jmx.JMXReporter; import org.apache.flink.runtime.client.JobExecutionException; @@ -133,7 +134,7 @@ public static Configuration getFlinkConfiguration() { flinkConfig.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "my_reporter." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), JMXReporter.class.getName()); return flinkConfig; } diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java index 48b014e46a68a..29c80ca7fb549 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java @@ -1113,28 +1113,24 @@ public final class ConfigConstants { */ public static final String METRICS_REPORTER_PREFIX = "metrics.reporter."; - /** The class of the reporter to use. This is used as a suffix in an actual reporter config */ - public static final String METRICS_REPORTER_CLASS_SUFFIX = "class"; + /** @deprecated use {@link MetricOptions#REPORTER_CLASS} */ + @Deprecated public static final String METRICS_REPORTER_CLASS_SUFFIX = "class"; - /** - * The class of the reporter factory to use. This is used as a suffix in an actual reporter - * config - */ - public static final String METRICS_REPORTER_FACTORY_CLASS_SUFFIX = "factory.class"; + /** @deprecated use {@link MetricOptions#REPORTER_FACTORY_CLASS} */ + @Deprecated public static final String METRICS_REPORTER_FACTORY_CLASS_SUFFIX = "factory.class"; - /** The interval between reports. This is used as a suffix in an actual reporter config */ - public static final String METRICS_REPORTER_INTERVAL_SUFFIX = "interval"; + /** @deprecated use {@link MetricOptions#REPORTER_INTERVAL} */ + @Deprecated public static final String METRICS_REPORTER_INTERVAL_SUFFIX = "interval"; - /** - * The delimiter used to assemble the metric identifier. This is used as a suffix in an actual - * reporter config. - */ - public static final String METRICS_REPORTER_SCOPE_DELIMITER = "scope.delimiter"; + /** @deprecated use {@link MetricOptions#REPORTER_SCOPE_DELIMITER} */ + @Deprecated public static final String METRICS_REPORTER_SCOPE_DELIMITER = "scope.delimiter"; - /** The set of variables that should be excluded. */ + /** @deprecated use {@link MetricOptions#REPORTER_EXCLUDED_VARIABLES} */ + @Deprecated public static final String METRICS_REPORTER_EXCLUDED_VARIABLES = "scope.variables.excludes"; - /** The map of additional variables that should be included. */ + /** @deprecated use {@link MetricOptions#REPORTER_ADDITIONAL_VARIABLES} */ + @Deprecated public static final String METRICS_REPORTER_ADDITIONAL_VARIABLES = "scope.variables.additional"; /** @deprecated Use {@link MetricOptions#SCOPE_DELIMITER} instead. */ diff --git a/flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java index 70d49035ac85c..19e7bb9fc5a85 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java @@ -25,7 +25,9 @@ import org.apache.flink.configuration.description.TextElement; import java.time.Duration; +import java.util.Collections; import java.util.List; +import java.util.Map; import static org.apache.flink.configuration.ConfigOptions.key; import static org.apache.flink.configuration.description.TextElement.text; @@ -34,6 +36,9 @@ @PublicEvolving public class MetricOptions { + private static final String NAMED_REPORTER_CONFIG_PREFIX = + ConfigConstants.METRICS_REPORTER_PREFIX + ""; + /** * An optional list of reporter names. If configured, only reporters whose name matches any of * the names in the list will be started. Otherwise, all reporters that could be found in the @@ -60,20 +65,62 @@ public class MetricOptions { + " any of the names in the list will be started. Otherwise, all reporters that could be found in" + " the configuration will be started."); + @Documentation.SuffixOption(NAMED_REPORTER_CONFIG_PREFIX) + @Documentation.Section(value = Documentation.Sections.METRIC_REPORTERS, position = 1) public static final ConfigOption REPORTER_CLASS = - key("metrics.reporter..class") + key("class") .stringType() .noDefaultValue() .withDescription("The reporter class to use for the reporter named ."); + @Documentation.SuffixOption(NAMED_REPORTER_CONFIG_PREFIX) + @Documentation.Section(value = Documentation.Sections.METRIC_REPORTERS, position = 1) + public static final ConfigOption REPORTER_FACTORY_CLASS = + key("factory.class") + .stringType() + .noDefaultValue() + .withDescription( + "The reporter factory class to use for the reporter named ."); + + @Documentation.SuffixOption(NAMED_REPORTER_CONFIG_PREFIX) + @Documentation.Section(value = Documentation.Sections.METRIC_REPORTERS, position = 2) public static final ConfigOption REPORTER_INTERVAL = - key("metrics.reporter..interval") + key("interval") .durationType() .defaultValue(Duration.ofSeconds(10)) .withDescription("The reporter interval to use for the reporter named ."); + @Documentation.SuffixOption(NAMED_REPORTER_CONFIG_PREFIX) + @Documentation.Section(value = Documentation.Sections.METRIC_REPORTERS, position = 2) + public static final ConfigOption REPORTER_SCOPE_DELIMITER = + key("scope.delimiter") + .stringType() + .defaultValue(".") + .withDescription( + "The delimiter used to assemble the metric identifier for the reporter named ."); + + @Documentation.SuffixOption(NAMED_REPORTER_CONFIG_PREFIX) + @Documentation.Section(value = Documentation.Sections.METRIC_REPORTERS, position = 3) + public static final ConfigOption> REPORTER_ADDITIONAL_VARIABLES = + key("scope.variables.additional") + .mapType() + .defaultValue(Collections.emptyMap()) + .withDescription( + "The map of additional variables that should be included for the reporter named . Only applicable to tag-based reporters (e.g., PRometheus, InfluxDB)."); + + @Documentation.SuffixOption(NAMED_REPORTER_CONFIG_PREFIX) + @Documentation.Section(value = Documentation.Sections.METRIC_REPORTERS, position = 3) + public static final ConfigOption REPORTER_EXCLUDED_VARIABLES = + key("scope.variables.excludes") + .stringType() + .defaultValue(".") + .withDescription( + "The set of variables that should be excluded for the reporter named . Only applicable to tag-based reporters (e.g., PRometheus, InfluxDB)."); + + @Documentation.SuffixOption(NAMED_REPORTER_CONFIG_PREFIX) + @Documentation.Section(value = Documentation.Sections.METRIC_REPORTERS, position = 4) public static final ConfigOption REPORTER_CONFIG_PARAMETER = - key("metrics.reporter..") + key("") .stringType() .noDefaultValue() .withDescription( diff --git a/flink-end-to-end-tests/flink-metrics-reporter-prometheus-test/src/test/java/org/apache/flink/metrics/prometheus/tests/PrometheusReporterEndToEndITCase.java b/flink-end-to-end-tests/flink-metrics-reporter-prometheus-test/src/test/java/org/apache/flink/metrics/prometheus/tests/PrometheusReporterEndToEndITCase.java index 70c43cf77855d..0b08ac9737bce 100644 --- a/flink-end-to-end-tests/flink-metrics-reporter-prometheus-test/src/test/java/org/apache/flink/metrics/prometheus/tests/PrometheusReporterEndToEndITCase.java +++ b/flink-end-to-end-tests/flink-metrics-reporter-prometheus-test/src/test/java/org/apache/flink/metrics/prometheus/tests/PrometheusReporterEndToEndITCase.java @@ -20,6 +20,7 @@ import org.apache.flink.configuration.ConfigConstants; import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.MetricOptions; import org.apache.flink.metrics.prometheus.PrometheusReporter; import org.apache.flink.metrics.prometheus.PrometheusReporterFactory; import org.apache.flink.tests.util.AutoClosableProcess; @@ -179,14 +180,14 @@ private static Configuration getFlinkConfig(TestParams.InstantiationType instant config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "prom." - + ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX, + + MetricOptions.REPORTER_FACTORY_CLASS.key(), PrometheusReporterFactory.class.getName()); break; case REFLECTION: config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "prom." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), PrometheusReporter.class.getCanonicalName()); } diff --git a/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/runtime/jobmanager/JMXJobManagerMetricTest.java b/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/runtime/jobmanager/JMXJobManagerMetricTest.java index 9da584dfdca2c..025a1f0571fff 100644 --- a/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/runtime/jobmanager/JMXJobManagerMetricTest.java +++ b/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/runtime/jobmanager/JMXJobManagerMetricTest.java @@ -72,7 +72,7 @@ private static Configuration getConfiguration() { flinkConfiguration.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), JMXReporter.class.getName()); flinkConfiguration.setString(MetricOptions.SCOPE_NAMING_JM_JOB, "jobmanager."); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/ReporterSetup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/ReporterSetup.java index e8b19240a7af4..c1e48dfca5774 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/ReporterSetup.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/ReporterSetup.java @@ -20,8 +20,6 @@ import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.ConfigConstants; -import org.apache.flink.configuration.ConfigOption; -import org.apache.flink.configuration.ConfigOptions; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.DelegatingConfiguration; import org.apache.flink.configuration.MetricOptions; @@ -79,16 +77,11 @@ public final class ReporterSetup { // classes "([\\S&&[^.]]*)\\." + '(' - + Pattern.quote(ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX) + + Pattern.quote(MetricOptions.REPORTER_CLASS.key()) + '|' - + Pattern.quote(ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX) + + Pattern.quote(MetricOptions.REPORTER_FACTORY_CLASS.key()) + ')'); - private static final ConfigOption> ADDITIONAL_VARIABLES = - ConfigOptions.key(ConfigConstants.METRICS_REPORTER_ADDITIONAL_VARIABLES) - .mapType() - .defaultValue(Collections.emptyMap()); - private final String name; private final MetricConfig configuration; private final MetricReporter reporter; @@ -107,17 +100,17 @@ public ReporterSetup( public Optional getDelimiter() { return Optional.ofNullable( - configuration.getString(ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, null)); + configuration.getString(MetricOptions.REPORTER_SCOPE_DELIMITER.key(), null)); } public Optional getIntervalSettings() { return Optional.ofNullable( - configuration.getString(ConfigConstants.METRICS_REPORTER_INTERVAL_SUFFIX, null)); + configuration.getString(MetricOptions.REPORTER_INTERVAL.key(), null)); } public Set getExcludedVariables() { String excludedVariablesList = - configuration.getString(ConfigConstants.METRICS_REPORTER_EXCLUDED_VARIABLES, null); + configuration.getString(MetricOptions.REPORTER_EXCLUDED_VARIABLES.key(), null); if (excludedVariablesList == null) { return Collections.emptySet(); } else { @@ -308,7 +301,8 @@ private static List setupReporters( // massage user variables keys into scope format for parity to variable exclusion Map additionalVariables = - reporterConfig.get(ADDITIONAL_VARIABLES).entrySet().stream() + reporterConfig.get(MetricOptions.REPORTER_ADDITIONAL_VARIABLES).entrySet() + .stream() .collect( Collectors.toMap( e -> ScopeFormat.asVariable(e.getKey()), @@ -341,11 +335,8 @@ private static Optional loadReporter( final Map reporterFactories) throws ClassNotFoundException, IllegalAccessException, InstantiationException { - final String reporterClassName = - reporterConfig.getString(ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, null); - final String factoryClassName = - reporterConfig.getString( - ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX, null); + final String reporterClassName = reporterConfig.get(MetricOptions.REPORTER_CLASS); + final String factoryClassName = reporterConfig.get(MetricOptions.REPORTER_FACTORY_CLASS); if (factoryClassName != null) { return loadViaFactory( @@ -431,9 +422,9 @@ private static Optional loadViaReflection( + " Please configure a factory class instead: '{}{}.{}: {}' to ensure that the configuration" + " continues to work with future versions.", reporterName, - ConfigConstants.METRICS_REPORTER_PREFIX, + MetricOptions.REPORTER_CLASS.key(), reporterName, - ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX, + MetricOptions.REPORTER_FACTORY_CLASS.key(), alternativeFactoryClassName); return loadViaFactory( alternativeFactoryClassName, reporterName, reporterConfig, reporterFactories); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/MetricRegistryImplTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/MetricRegistryImplTest.java index 5062c92428392..9a5e0097c225f 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/MetricRegistryImplTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/MetricRegistryImplTest.java @@ -142,7 +142,7 @@ public void open(MetricConfig config) { public void testReporterScheduling() throws Exception { MetricConfig config = new MetricConfig(); config.setProperty("arg1", "hello"); - config.setProperty(ConfigConstants.METRICS_REPORTER_INTERVAL_SUFFIX, "50 MILLISECONDS"); + config.setProperty(MetricOptions.REPORTER_INTERVAL.key(), "50 MILLISECONDS"); MetricRegistryImpl registry = new MetricRegistryImpl( @@ -180,7 +180,7 @@ public void testReporterIntervalParsingErrorFallsBackToDefaultValue() throws Exc MetricConfig config = new MetricConfig(); // in a prior implementation the time amount was applied even if the time unit was invalid // in this case this would imply using 1 SECOND as the interval (seconds is the default) - config.setProperty(ConfigConstants.METRICS_REPORTER_INTERVAL_SUFFIX, "1 UNICORN"); + config.setProperty(MetricOptions.REPORTER_INTERVAL.key(), "1 UNICORN"); final ManuallyTriggeredScheduledExecutorService manuallyTriggeredScheduledExecutorService = new ManuallyTriggeredScheduledExecutorService(); @@ -220,12 +220,12 @@ public void testReporterNotifications() throws Exception { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test1." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), TestReporter6.class.getName()); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test2." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), TestReporter7.class.getName()); MetricRegistryImpl registry = @@ -345,13 +345,13 @@ public void testConfigurableDelimiter() throws Exception { @Test public void testConfigurableDelimiterForReporters() throws Exception { MetricConfig config1 = new MetricConfig(); - config1.setProperty(ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, "_"); + config1.setProperty(MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "_"); MetricConfig config2 = new MetricConfig(); - config2.setProperty(ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, "-"); + config2.setProperty(MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "-"); MetricConfig config3 = new MetricConfig(); - config3.setProperty(ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, "AA"); + config3.setProperty(MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "AA"); MetricRegistryImpl registry = new MetricRegistryImpl( @@ -375,13 +375,13 @@ public void testConfigurableDelimiterForReporters() throws Exception { public void testConfigurableDelimiterForReportersInGroup() throws Exception { String name = "C"; MetricConfig config1 = new MetricConfig(); - config1.setProperty(ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, "_"); + config1.setProperty(MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "_"); MetricConfig config2 = new MetricConfig(); - config2.setProperty(ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, "-"); + config2.setProperty(MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "-"); MetricConfig config3 = new MetricConfig(); - config3.setProperty(ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, "AA"); + config3.setProperty(MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "AA"); Configuration config = new Configuration(); config.setString(MetricOptions.SCOPE_NAMING_TM, "A.B"); @@ -389,37 +389,37 @@ public void testConfigurableDelimiterForReportersInGroup() throws Exception { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test1." - + ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, + + MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "_"); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test1." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), CollectingMetricsReporter.class.getName()); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test2." - + ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, + + MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "-"); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test2." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), CollectingMetricsReporter.class.getName()); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test3." - + ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, + + MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "AA"); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test3." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), CollectingMetricsReporter.class.getName()); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test4." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), CollectingMetricsReporter.class.getName()); List reporterConfigurations = @@ -445,7 +445,7 @@ public void testConfigurableDelimiterForReportersInGroup() throws Exception { for (ReporterSetup cfg : reporterConfigurations) { String delimiter = cfg.getConfiguration() - .getProperty(ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER); + .getProperty(MetricOptions.REPORTER_SCOPE_DELIMITER.key()); if (delimiter == null || delimiter.equals("AA")) { // test3 reporter: 'AA' - not correct // for test4 reporter use global delimiter diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/ReporterSetupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/ReporterSetupTest.java index 98d69773c74fe..00eff56da2a18 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/ReporterSetupTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/ReporterSetupTest.java @@ -125,7 +125,7 @@ public void testReporterSetupSupplier() throws Exception { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "reporter1." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), TestReporter1.class.getName()); final List reporterSetups = ReporterSetup.fromConfiguration(config, null); @@ -145,17 +145,17 @@ public void testMultipleReporterInstantiation() throws Exception { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test1." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), TestReporter11.class.getName()); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test2." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), TestReporter12.class.getName()); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test3." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), TestReporter13.class.getName()); List reporterSetups = ReporterSetup.fromConfiguration(config, null); @@ -201,7 +201,7 @@ private static void configureReporter1(Configuration config) { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "reporter1." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), TestReporter1.class.getName()); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "reporter1.arg1", "value1"); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "reporter1.arg2", "value2"); @@ -220,7 +220,7 @@ private static void configureReporter2(Configuration config) { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "reporter2." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), TestReporter2.class.getName()); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "reporter2.arg1", "value1"); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "reporter2.arg3", "value3"); @@ -242,12 +242,12 @@ public void testVariableExclusionParsing() throws Exception { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test." - + ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX, + + MetricOptions.REPORTER_FACTORY_CLASS.key(), TestReporterFactory.class.getName()); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test." - + ConfigConstants.METRICS_REPORTER_EXCLUDED_VARIABLES, + + MetricOptions.REPORTER_EXCLUDED_VARIABLES.key(), excludedVariable1 + ";" + excludedVariable2); final List reporterSetups = ReporterSetup.fromConfiguration(config, null); @@ -270,7 +270,7 @@ public void testFactoryParsing() throws Exception { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test." - + ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX, + + MetricOptions.REPORTER_FACTORY_CLASS.key(), TestReporterFactory.class.getName()); final List reporterSetups = ReporterSetup.fromConfiguration(config, null); @@ -292,12 +292,12 @@ public void testFactoryPrioritization() throws Exception { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test." - + ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX, + + MetricOptions.REPORTER_FACTORY_CLASS.key(), InstantiationTypeTrackingTestReporterFactory.class.getName()); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), InstantiationTypeTrackingTestReporter.class.getName()); final List reporterSetups = ReporterSetup.fromConfiguration(config, null); @@ -318,12 +318,12 @@ public void testFactoryFailureIsolation() throws Exception { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test." - + ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX, + + MetricOptions.REPORTER_FACTORY_CLASS.key(), TestReporterFactory.class.getName()); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "fail." - + ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX, + + MetricOptions.REPORTER_FACTORY_CLASS.key(), FailingFactory.class.getName()); final List reporterSetups = ReporterSetup.fromConfiguration(config, null); @@ -338,12 +338,12 @@ public void testMixedSetupsFactoryParsing() throws Exception { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test1." - + ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX, + + MetricOptions.REPORTER_FACTORY_CLASS.key(), InstantiationTypeTrackingTestReporterFactory.class.getName()); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test2." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), InstantiationTypeTrackingTestReporter.class.getName()); final List reporterSetups = ReporterSetup.fromConfiguration(config, null); @@ -367,7 +367,7 @@ public void testFactoryArgumentForwarding() throws Exception { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test." - + ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX, + + MetricOptions.REPORTER_FACTORY_CLASS.key(), ConfigExposingReporterFactory.class.getName()); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "test.arg", "hello"); @@ -387,7 +387,7 @@ public void testFactoryAnnotation() { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), InstantiationTypeTrackingTestReporter2.class.getName()); final List reporterSetups = ReporterSetup.fromConfiguration(config, null); @@ -411,7 +411,7 @@ public void testReflectionInterception() { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), InstantiationTypeTrackingTestReporter.class.getName()); final List reporterSetups = @@ -443,12 +443,12 @@ public void testAdditionalVariablesParsing() { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test." - + ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX, + + MetricOptions.REPORTER_FACTORY_CLASS.key(), TestReporterFactory.class.getName()); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test." - + ConfigConstants.METRICS_REPORTER_ADDITIONAL_VARIABLES, + + MetricOptions.REPORTER_ADDITIONAL_VARIABLES.key(), String.join(",", tag1 + ":" + tagValue1, tag2 + ":" + tagValue2)); final List reporterSetups = ReporterSetup.fromConfiguration(config, null); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java index a76b192fcc025..b728d9f2a11dd 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java @@ -152,30 +152,30 @@ public void testScopeCachingForMultipleReporters() throws Exception { config.setString(MetricOptions.SCOPE_NAMING_TM, "A.B.C.D"); MetricConfig metricConfig1 = new MetricConfig(); - metricConfig1.setProperty(ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, "-"); + metricConfig1.setProperty(MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "-"); MetricConfig metricConfig2 = new MetricConfig(); - metricConfig2.setProperty(ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, "!"); + metricConfig2.setProperty(MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "!"); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test1." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), CollectingMetricsReporter.class.getName()); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test1." - + ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, + + MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "-"); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test2." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), CollectingMetricsReporter.class.getName()); config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test2." - + ConfigConstants.METRICS_REPORTER_SCOPE_DELIMITER, + + MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "!"); CollectingMetricsReporter reporter1 = new CollectingMetricsReporter(FILTER_B); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java index 15b9a27f66e2b..754766d1d3a19 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java @@ -21,6 +21,7 @@ import org.apache.flink.api.common.JobID; import org.apache.flink.configuration.ConfigConstants; import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.MetricOptions; import org.apache.flink.metrics.CharacterFilter; import org.apache.flink.metrics.Gauge; import org.apache.flink.metrics.Metric; @@ -243,7 +244,7 @@ public void testLogicalScopeShouldIgnoreValueGroupName() throws Exception { config.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "test." - + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, + + MetricOptions.REPORTER_CLASS.key(), TestReporter.class.getName()); MetricRegistryImpl registry = diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/InMemoryReporter.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/InMemoryReporter.java index 1a992e8394499..a2716cd3e6951 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/InMemoryReporter.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/InMemoryReporter.java @@ -21,6 +21,7 @@ import org.apache.flink.api.common.JobID; import org.apache.flink.configuration.ConfigConstants; import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.MetricOptions; import org.apache.flink.metrics.LogicalScopeProvider; import org.apache.flink.metrics.Metric; import org.apache.flink.metrics.MetricConfig; @@ -245,7 +246,7 @@ public Configuration addToConfiguration(Configuration configuration) { configuration.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "mini_cluster_resource_reporter." - + ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX, + + MetricOptions.REPORTER_FACTORY_CLASS.key(), InMemoryReporter.Factory.class.getName()); configuration.setString( ConfigConstants.METRICS_REPORTER_PREFIX + "mini_cluster_resource_reporter." + ID,