zentol commented on a change in pull request #15165:
URL: https://github.com/apache/flink/pull/15165#discussion_r593048085
##########
File path:
flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporter.java
##########
@@ -56,43 +56,42 @@
private boolean deleteOnShutdown;
private Map<String, String> groupingKey;
- @Override
- public void open(MetricConfig config) {
- super.open(config);
-
- String host = config.getString(HOST.key(), HOST.defaultValue());
- int port = config.getInteger(PORT.key(), PORT.defaultValue());
- String configuredJobName = config.getString(JOB_NAME.key(),
JOB_NAME.defaultValue());
- boolean randomSuffix =
- config.getBoolean(
- RANDOM_JOB_NAME_SUFFIX.key(),
RANDOM_JOB_NAME_SUFFIX.defaultValue());
- deleteOnShutdown =
- config.getBoolean(DELETE_ON_SHUTDOWN.key(),
DELETE_ON_SHUTDOWN.defaultValue());
- groupingKey =
- parseGroupingKey(config.getString(GROUPING_KEY.key(),
GROUPING_KEY.defaultValue()));
-
- if (host == null || host.isEmpty() || port < 1) {
+ PrometheusPushGatewayReporter(
+ @Nullable final String hostConfig,
+ @Nullable final int portConfig,
+ @Nullable final String jobNameConfig,
+ @Nullable final boolean randomJobSuffixConfig,
+ @Nullable final boolean deleteOnShutdownConfig,
+ @Nullable final Map<String, String> groupingKeyConfig
+ ) {
+ deleteOnShutdown = deleteOnShutdownConfig
+ groupingKey = parseGroupingKey(groupingKeyConfig)
+
+ if (hostConfig == null || hostConfig.isEmpty() || portConfig < 1) {
throw new IllegalArgumentException(
- "Invalid host/port configuration. Host: " + host + " Port:
" + port);
+ "Invalid host/port configuration. Host: " + hostConfig + "
Port: " + portConfig);
}
- if (randomSuffix) {
- this.jobName = configuredJobName + new AbstractID();
+ if (randomJobSuffixConfig) {
+ this.jobName = jobNameConfig + new AbstractID();
} else {
- this.jobName = configuredJobName;
+ this.jobName = jobNameConfig;
}
- pushGateway = new PushGateway(host + ':' + port);
+ pushGateway = new PushGateway(hostConfig + ':' + portConfig);
log.info(
"Configured PrometheusPushGatewayReporter with {host:{},
port:{}, jobName:{}, randomJobNameSuffix:{}, deleteOnShutdown:{},
groupingKey:{}}",
- host,
- port,
- jobName,
- randomSuffix,
+ hostConfig,
+ portConfig,
+ jobNameConfig,
+ randomJobSuffixConfig,
deleteOnShutdown,
groupingKey);
}
+ @Override
+ public void open(MetricConfig config) { }
+
Map<String, String> parseGroupingKey(final String groupingKeyConfig) {
Review comment:
I'd move this into the factory and make it static; then we can also test
it without having to instantiate a reporter/factory.
##########
File path:
flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporter.java
##########
@@ -56,43 +56,42 @@
private boolean deleteOnShutdown;
private Map<String, String> groupingKey;
- @Override
- public void open(MetricConfig config) {
- super.open(config);
-
- String host = config.getString(HOST.key(), HOST.defaultValue());
- int port = config.getInteger(PORT.key(), PORT.defaultValue());
- String configuredJobName = config.getString(JOB_NAME.key(),
JOB_NAME.defaultValue());
- boolean randomSuffix =
- config.getBoolean(
- RANDOM_JOB_NAME_SUFFIX.key(),
RANDOM_JOB_NAME_SUFFIX.defaultValue());
- deleteOnShutdown =
- config.getBoolean(DELETE_ON_SHUTDOWN.key(),
DELETE_ON_SHUTDOWN.defaultValue());
- groupingKey =
- parseGroupingKey(config.getString(GROUPING_KEY.key(),
GROUPING_KEY.defaultValue()));
-
- if (host == null || host.isEmpty() || port < 1) {
+ PrometheusPushGatewayReporter(
+ @Nullable final String hostConfig,
+ @Nullable final int portConfig,
+ @Nullable final String jobNameConfig,
+ @Nullable final boolean randomJobSuffixConfig,
+ @Nullable final boolean deleteOnShutdownConfig,
+ @Nullable final Map<String, String> groupingKeyConfig
+ ) {
+ deleteOnShutdown = deleteOnShutdownConfig
+ groupingKey = parseGroupingKey(groupingKeyConfig)
+
+ if (hostConfig == null || hostConfig.isEmpty() || portConfig < 1) {
throw new IllegalArgumentException(
- "Invalid host/port configuration. Host: " + host + " Port:
" + port);
+ "Invalid host/port configuration. Host: " + hostConfig + "
Port: " + portConfig);
}
- if (randomSuffix) {
- this.jobName = configuredJobName + new AbstractID();
+ if (randomJobSuffixConfig) {
+ this.jobName = jobNameConfig + new AbstractID();
} else {
- this.jobName = configuredJobName;
+ this.jobName = jobNameConfig;
Review comment:
I'd move this logic to the factory, with the reporter just accepting the
jobName argument.
##########
File path:
flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporter.java
##########
@@ -56,43 +56,42 @@
private boolean deleteOnShutdown;
private Map<String, String> groupingKey;
- @Override
- public void open(MetricConfig config) {
- super.open(config);
-
- String host = config.getString(HOST.key(), HOST.defaultValue());
- int port = config.getInteger(PORT.key(), PORT.defaultValue());
- String configuredJobName = config.getString(JOB_NAME.key(),
JOB_NAME.defaultValue());
- boolean randomSuffix =
- config.getBoolean(
- RANDOM_JOB_NAME_SUFFIX.key(),
RANDOM_JOB_NAME_SUFFIX.defaultValue());
- deleteOnShutdown =
- config.getBoolean(DELETE_ON_SHUTDOWN.key(),
DELETE_ON_SHUTDOWN.defaultValue());
- groupingKey =
- parseGroupingKey(config.getString(GROUPING_KEY.key(),
GROUPING_KEY.defaultValue()));
-
- if (host == null || host.isEmpty() || port < 1) {
+ PrometheusPushGatewayReporter(
+ @Nullable final String hostConfig,
+ @Nullable final int portConfig,
+ @Nullable final String jobNameConfig,
+ @Nullable final boolean randomJobSuffixConfig,
+ @Nullable final boolean deleteOnShutdownConfig,
+ @Nullable final Map<String, String> groupingKeyConfig
+ ) {
+ deleteOnShutdown = deleteOnShutdownConfig
+ groupingKey = parseGroupingKey(groupingKeyConfig)
Review comment:
pretty sure this doesn't compile since `parseGroupingKey` expects a
`String`
##########
File path:
flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporter.java
##########
@@ -56,43 +56,42 @@
private boolean deleteOnShutdown;
private Map<String, String> groupingKey;
- @Override
- public void open(MetricConfig config) {
- super.open(config);
-
- String host = config.getString(HOST.key(), HOST.defaultValue());
- int port = config.getInteger(PORT.key(), PORT.defaultValue());
- String configuredJobName = config.getString(JOB_NAME.key(),
JOB_NAME.defaultValue());
- boolean randomSuffix =
- config.getBoolean(
- RANDOM_JOB_NAME_SUFFIX.key(),
RANDOM_JOB_NAME_SUFFIX.defaultValue());
- deleteOnShutdown =
- config.getBoolean(DELETE_ON_SHUTDOWN.key(),
DELETE_ON_SHUTDOWN.defaultValue());
- groupingKey =
- parseGroupingKey(config.getString(GROUPING_KEY.key(),
GROUPING_KEY.defaultValue()));
-
- if (host == null || host.isEmpty() || port < 1) {
+ PrometheusPushGatewayReporter(
+ @Nullable final String hostConfig,
Review comment:
with the below comments we can enforce that none of these may be null.
##########
File path:
flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporter.java
##########
@@ -56,43 +56,42 @@
private boolean deleteOnShutdown;
private Map<String, String> groupingKey;
- @Override
- public void open(MetricConfig config) {
- super.open(config);
-
- String host = config.getString(HOST.key(), HOST.defaultValue());
- int port = config.getInteger(PORT.key(), PORT.defaultValue());
- String configuredJobName = config.getString(JOB_NAME.key(),
JOB_NAME.defaultValue());
- boolean randomSuffix =
- config.getBoolean(
- RANDOM_JOB_NAME_SUFFIX.key(),
RANDOM_JOB_NAME_SUFFIX.defaultValue());
- deleteOnShutdown =
- config.getBoolean(DELETE_ON_SHUTDOWN.key(),
DELETE_ON_SHUTDOWN.defaultValue());
- groupingKey =
- parseGroupingKey(config.getString(GROUPING_KEY.key(),
GROUPING_KEY.defaultValue()));
-
- if (host == null || host.isEmpty() || port < 1) {
+ PrometheusPushGatewayReporter(
+ @Nullable final String hostConfig,
+ @Nullable final int portConfig,
Review comment:
I'd remove all the `Config` suffixes
##########
File path:
flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporter.java
##########
@@ -56,43 +56,42 @@
private boolean deleteOnShutdown;
private Map<String, String> groupingKey;
- @Override
- public void open(MetricConfig config) {
- super.open(config);
-
- String host = config.getString(HOST.key(), HOST.defaultValue());
- int port = config.getInteger(PORT.key(), PORT.defaultValue());
- String configuredJobName = config.getString(JOB_NAME.key(),
JOB_NAME.defaultValue());
- boolean randomSuffix =
- config.getBoolean(
- RANDOM_JOB_NAME_SUFFIX.key(),
RANDOM_JOB_NAME_SUFFIX.defaultValue());
- deleteOnShutdown =
- config.getBoolean(DELETE_ON_SHUTDOWN.key(),
DELETE_ON_SHUTDOWN.defaultValue());
- groupingKey =
- parseGroupingKey(config.getString(GROUPING_KEY.key(),
GROUPING_KEY.defaultValue()));
-
- if (host == null || host.isEmpty() || port < 1) {
+ PrometheusPushGatewayReporter(
+ @Nullable final String hostConfig,
+ @Nullable final int portConfig,
+ @Nullable final String jobNameConfig,
+ @Nullable final boolean randomJobSuffixConfig,
+ @Nullable final boolean deleteOnShutdownConfig,
+ @Nullable final Map<String, String> groupingKeyConfig
+ ) {
+ deleteOnShutdown = deleteOnShutdownConfig
+ groupingKey = parseGroupingKey(groupingKeyConfig)
+
+ if (hostConfig == null || hostConfig.isEmpty() || portConfig < 1) {
Review comment:
this check should be moved to the factory
##########
File path:
flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporter.java
##########
@@ -56,43 +56,42 @@
private boolean deleteOnShutdown;
private Map<String, String> groupingKey;
- @Override
- public void open(MetricConfig config) {
- super.open(config);
-
- String host = config.getString(HOST.key(), HOST.defaultValue());
- int port = config.getInteger(PORT.key(), PORT.defaultValue());
- String configuredJobName = config.getString(JOB_NAME.key(),
JOB_NAME.defaultValue());
- boolean randomSuffix =
- config.getBoolean(
- RANDOM_JOB_NAME_SUFFIX.key(),
RANDOM_JOB_NAME_SUFFIX.defaultValue());
- deleteOnShutdown =
- config.getBoolean(DELETE_ON_SHUTDOWN.key(),
DELETE_ON_SHUTDOWN.defaultValue());
- groupingKey =
- parseGroupingKey(config.getString(GROUPING_KEY.key(),
GROUPING_KEY.defaultValue()));
-
- if (host == null || host.isEmpty() || port < 1) {
+ PrometheusPushGatewayReporter(
+ @Nullable final String hostConfig,
+ @Nullable final int portConfig,
+ @Nullable final String jobNameConfig,
+ @Nullable final boolean randomJobSuffixConfig,
+ @Nullable final boolean deleteOnShutdownConfig,
Review comment:
these can never be null since they are primitives ;)
##########
File path:
flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusReporter.java
##########
@@ -50,11 +47,7 @@ int getPort() {
return port;
}
- @Override
- public void open(MetricConfig config) {
- super.open(config);
-
- String portsConfig = config.getString(ARG_PORT, DEFAULT_PORT);
+ PrometheusReporter(@Nullable final String portsConfig) {
Review comment:
This is not nullable, which we should also enforce via
`Preconditions.checkNotNull`
##########
File path:
flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporterFactory.java
##########
@@ -22,13 +22,38 @@
import java.util.Properties;
+import static
org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.DELETE_ON_SHUTDOWN;
+import static
org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.GROUPING_KEY;
+import static
org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.HOST;
+import static
org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.JOB_NAME;
+import static
org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.PORT;
+import static
org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.RANDOM_JOB_NAME_SUFFIX;
+
/** {@link MetricReporterFactory} for {@link PrometheusPushGatewayReporter}. */
@InterceptInstantiationViaReflection(
reporterClassName =
"org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter")
public class PrometheusPushGatewayReporterFactory implements
MetricReporterFactory {
@Override
public PrometheusPushGatewayReporter createMetricReporter(Properties
properties) {
- return new PrometheusPushGatewayReporter();
+ String hostConfig = properties.getString(HOST.key(),
HOST.defaultValue());
+ int portConfig = properties.getInteger(PORT.key(),
PORT.defaultValue());
+ String jobNameConfig = properties.getString(JOB_NAME.key(),
JOB_NAME.defaultValue())
Review comment:
```suggestion
String jobNameConfig = properties.getString(JOB_NAME.key(),
JOB_NAME.defaultValue());
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
