You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2021/09/01 17:38:48 UTC
[flink] branch master updated: [FLINK-17495][metrics] Add support
for configuring additional variables
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 869b486 [FLINK-17495][metrics] Add support for configuring additional variables
869b486 is described below
commit 869b48635826bca49a8a861c7fcb2eb88263ae4b
Author: Mason Chen <ma...@berkeley.edu>
AuthorDate: Wed Sep 1 10:38:22 2021 -0700
[FLINK-17495][metrics] Add support for configuring additional variables
---
.../content.zh/docs/deployment/metric_reporters.md | 2 +
docs/content/docs/deployment/metric_reporters.md | 2 +
.../flink/configuration/ConfigConstants.java | 3 ++
.../metrics/prometheus/PrometheusReporterTest.java | 2 +-
.../flink/runtime/metrics/MetricRegistryImpl.java | 3 +-
.../flink/runtime/metrics/ReporterSetup.java | 46 +++++++++++++++++++---
.../runtime/metrics/groups/FrontMetricGroup.java | 14 ++++++-
.../metrics/groups/ReporterScopedSettings.java | 15 ++++++-
.../flink/runtime/metrics/ReporterSetupTest.java | 34 ++++++++++++++++
.../metrics/groups/FrontMetricGroupTest.java | 23 ++++++++++-
10 files changed, 130 insertions(+), 14 deletions(-)
diff --git a/docs/content.zh/docs/deployment/metric_reporters.md b/docs/content.zh/docs/deployment/metric_reporters.md
index 70ea0a5..fd405e4 100644
--- a/docs/content.zh/docs/deployment/metric_reporters.md
+++ b/docs/content.zh/docs/deployment/metric_reporters.md
@@ -42,6 +42,7 @@ reporters will be instantiated on each job and task manager when they are starte
- `metrics.reporter.<name>.scope.delimiter`: The delimiter to use for the identifier (default value use `metrics.scope.delimiter`) for the reporter named `<name>`.
- `metrics.reporter.<name>.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.<name>.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).
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`.
@@ -55,6 +56,7 @@ metrics.reporters: my_jmx_reporter,my_other_reporter
metrics.reporter.my_jmx_reporter.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
metrics.reporter.my_jmx_reporter.port: 9020-9040
metrics.reporter.my_jmx_reporter.scope.variables.excludes:job_id;task_attempt_num
+metrics.reporter.my_jmx_reporter.scope.variables.additional: cluster_name:my_test_cluster,tag_name:tag_value
metrics.reporter.my_other_reporter.class: org.apache.flink.metrics.graphite.GraphiteReporter
metrics.reporter.my_other_reporter.host: 192.168.1.1
diff --git a/docs/content/docs/deployment/metric_reporters.md b/docs/content/docs/deployment/metric_reporters.md
index ae36b6f..983e1a7 100644
--- a/docs/content/docs/deployment/metric_reporters.md
+++ b/docs/content/docs/deployment/metric_reporters.md
@@ -41,6 +41,7 @@ reporters will be instantiated on each job and task manager when they are starte
- `metrics.reporter.<name>.interval`: The reporter interval to use for the reporter named `<name>`.
- `metrics.reporter.<name>.scope.delimiter`: The delimiter to use for the identifier (default value use `metrics.scope.delimiter`) for the reporter named `<name>`.
- `metrics.reporter.<name>.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.<name>.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.
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.
@@ -55,6 +56,7 @@ metrics.reporters: my_jmx_reporter,my_other_reporter
metrics.reporter.my_jmx_reporter.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
metrics.reporter.my_jmx_reporter.port: 9020-9040
metrics.reporter.my_jmx_reporter.scope.variables.excludes:job_id;task_attempt_num
+metrics.reporter.my_jmx_reporter.scope.variables.additional: cluster_name:my_test_cluster,tag_name:tag_value
metrics.reporter.my_other_reporter.class: org.apache.flink.metrics.graphite.GraphiteReporter
metrics.reporter.my_other_reporter.host: 192.168.1.1
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 6576e0f..d3c4d71 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
@@ -1134,6 +1134,9 @@ public final class ConfigConstants {
/** The set of variables that should be excluded. */
public static final String METRICS_REPORTER_EXCLUDED_VARIABLES = "scope.variables.excludes";
+ /** The map of additional variables that should be included. */
+ public static final String METRICS_REPORTER_ADDITIONAL_VARIABLES = "scope.variables.additional";
+
/** @deprecated Use {@link MetricOptions#SCOPE_DELIMITER} instead. */
@Deprecated public static final String METRICS_SCOPE_DELIMITER = "metrics.scope.delimiter";
diff --git a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java
index 6f043ae..5e254d2 100644
--- a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java
+++ b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java
@@ -414,6 +414,6 @@ public class PrometheusReporterTest extends TestLogger {
}
private static ReporterScopedSettings createReporterScopedSettings() {
- return new ReporterScopedSettings(0, ',', Collections.emptySet());
+ return new ReporterScopedSettings(0, ',', Collections.emptySet(), Collections.emptyMap());
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryImpl.java
index 26669f4..f1150d7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryImpl.java
@@ -170,7 +170,8 @@ public class MetricRegistryImpl implements MetricRegistry {
new ReporterScopedSettings(
reporters.size(),
delimiterForReporter.charAt(0),
- reporterSetup.getExcludedVariables())));
+ reporterSetup.getExcludedVariables(),
+ reporterSetup.getAdditionalVariables())));
} catch (Throwable t) {
LOG.error(
"Could not instantiate metrics reporter {}. Metrics might not be exposed/reported.",
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 7d73837..e8b1924 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,6 +20,8 @@ package org.apache.flink.runtime.metrics;
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;
@@ -46,6 +48,7 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Optional;
import java.util.ServiceConfigurationError;
import java.util.ServiceLoader;
@@ -81,15 +84,25 @@ public final class ReporterSetup {
+ Pattern.quote(ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX)
+ ')');
+ private static final ConfigOption<Map<String, String>> 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;
+ private final Map<String, String> additionalVariables;
public ReporterSetup(
- final String name, final MetricConfig configuration, MetricReporter reporter) {
+ final String name,
+ final MetricConfig configuration,
+ MetricReporter reporter,
+ final Map<String, String> additionalVariables) {
this.name = name;
this.configuration = configuration;
this.reporter = reporter;
+ this.additionalVariables = additionalVariables;
}
public Optional<String> getDelimiter() {
@@ -116,6 +129,10 @@ public final class ReporterSetup {
}
}
+ public Map<String, String> getAdditionalVariables() {
+ return additionalVariables;
+ }
+
public String getName() {
return name;
}
@@ -131,20 +148,24 @@ public final class ReporterSetup {
@VisibleForTesting
public static ReporterSetup forReporter(String reporterName, MetricReporter reporter) {
- return createReporterSetup(reporterName, new MetricConfig(), reporter);
+ return createReporterSetup(
+ reporterName, new MetricConfig(), reporter, Collections.emptyMap());
}
@VisibleForTesting
public static ReporterSetup forReporter(
String reporterName, MetricConfig metricConfig, MetricReporter reporter) {
- return createReporterSetup(reporterName, metricConfig, reporter);
+ return createReporterSetup(reporterName, metricConfig, reporter, Collections.emptyMap());
}
private static ReporterSetup createReporterSetup(
- String reporterName, MetricConfig metricConfig, MetricReporter reporter) {
+ String reporterName,
+ MetricConfig metricConfig,
+ MetricReporter reporter,
+ Map<String, String> additionalVariables) {
reporter.open(metricConfig);
- return new ReporterSetup(reporterName, metricConfig, reporter);
+ return new ReporterSetup(reporterName, metricConfig, reporter, additionalVariables);
}
public static List<ReporterSetup> fromConfiguration(
@@ -284,12 +305,25 @@ public final class ReporterSetup {
try {
Optional<MetricReporter> metricReporterOptional =
loadReporter(reporterName, reporterConfig, reporterFactories);
+
+ // massage user variables keys into scope format for parity to variable exclusion
+ Map<String, String> additionalVariables =
+ reporterConfig.get(ADDITIONAL_VARIABLES).entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ e -> ScopeFormat.asVariable(e.getKey()),
+ Entry::getValue));
+
metricReporterOptional.ifPresent(
reporter -> {
MetricConfig metricConfig = new MetricConfig();
reporterConfig.addAllToProperties(metricConfig);
reporterSetups.add(
- createReporterSetup(reporterName, metricConfig, reporter));
+ createReporterSetup(
+ reporterName,
+ metricConfig,
+ reporter,
+ additionalVariables));
});
} catch (Throwable t) {
LOG.error(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroup.java
index 0e730d5..e1a8a5a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroup.java
@@ -23,6 +23,9 @@ import org.apache.flink.metrics.CharacterFilter;
import org.apache.flink.metrics.LogicalScopeProvider;
import org.apache.flink.metrics.MetricGroup;
+import org.apache.commons.collections.map.CompositeMap;
+
+import java.util.Collections;
import java.util.Map;
/**
@@ -72,8 +75,15 @@ public class FrontMetricGroup<P extends AbstractMetricGroup<?>> extends ProxyMet
@Override
public Map<String, String> getAllVariables() {
- return parentMetricGroup.getAllVariables(
- this.settings.getReporterIndex(), this.settings.getExcludedVariables());
+ Map<String, String> allVariables =
+ parentMetricGroup.getAllVariables(
+ this.settings.getReporterIndex(), this.settings.getExcludedVariables());
+
+ if (!this.settings.getAdditionalVariables().isEmpty()) {
+ allVariables = new CompositeMap(allVariables, this.settings.getAdditionalVariables());
+ }
+
+ return Collections.unmodifiableMap(allVariables);
}
/** @deprecated work against the LogicalScopeProvider interface instead. */
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ReporterScopedSettings.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ReporterScopedSettings.java
index ab2887a..5fbea42 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ReporterScopedSettings.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ReporterScopedSettings.java
@@ -19,6 +19,7 @@ package org.apache.flink.runtime.metrics.groups;
import org.apache.flink.util.Preconditions;
+import java.util.Map;
import java.util.Set;
/** Encapsulates all settings that are defined per reporter. */
@@ -28,14 +29,20 @@ public class ReporterScopedSettings {
private final char delimiter;
- private Set<String> excludedVariables;
+ private final Set<String> excludedVariables;
+
+ private final Map<String, String> additionalVariables;
public ReporterScopedSettings(
- int reporterIndex, char delimiter, Set<String> excludedVariables) {
+ int reporterIndex,
+ char delimiter,
+ Set<String> excludedVariables,
+ Map<String, String> additionalVariables) {
this.excludedVariables = excludedVariables;
Preconditions.checkArgument(reporterIndex >= 0);
this.reporterIndex = reporterIndex;
this.delimiter = delimiter;
+ this.additionalVariables = additionalVariables;
}
public int getReporterIndex() {
@@ -49,4 +56,8 @@ public class ReporterScopedSettings {
public Set<String> getExcludedVariables() {
return excludedVariables;
}
+
+ public Map<String, String> getAdditionalVariables() {
+ return additionalVariables;
+ }
}
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 4591e1a..98d6977 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
@@ -40,6 +40,7 @@ import java.util.Optional;
import java.util.Properties;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.collection.IsMapContaining.hasEntry;
import static org.hamcrest.core.IsCollectionContaining.hasItems;
import static org.hamcrest.core.IsInstanceOf.instanceOf;
import static org.junit.Assert.assertEquals;
@@ -432,6 +433,39 @@ public class ReporterSetupTest extends TestLogger {
assertTrue(metricReporter.createdByFactory);
}
+ @Test
+ public void testAdditionalVariablesParsing() {
+ final String tag1 = "foo";
+ final String tagValue1 = "bar";
+ final String tag2 = "fizz";
+ final String tagValue2 = "buzz";
+ final Configuration config = new Configuration();
+ config.setString(
+ ConfigConstants.METRICS_REPORTER_PREFIX
+ + "test."
+ + ConfigConstants.METRICS_REPORTER_FACTORY_CLASS_SUFFIX,
+ TestReporterFactory.class.getName());
+ config.setString(
+ ConfigConstants.METRICS_REPORTER_PREFIX
+ + "test."
+ + ConfigConstants.METRICS_REPORTER_ADDITIONAL_VARIABLES,
+ String.join(",", tag1 + ":" + tagValue1, tag2 + ":" + tagValue2));
+
+ final List<ReporterSetup> reporterSetups = ReporterSetup.fromConfiguration(config, null);
+
+ assertEquals(1, reporterSetups.size());
+
+ final ReporterSetup reporterSetup = reporterSetups.get(0);
+
+ assertThat(
+ reporterSetup.getAdditionalVariables(),
+ hasEntry(ScopeFormat.asVariable(tag1), tagValue1));
+
+ assertThat(
+ reporterSetup.getAdditionalVariables(),
+ hasEntry(ScopeFormat.asVariable(tag2), tagValue2));
+ }
+
/** Factory that exposed the last provided metric config. */
public static class ConfigExposingReporterFactory implements MetricReporterFactory {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroupTest.java
index ede4e20..a1208fb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroupTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroupTest.java
@@ -23,6 +23,8 @@ import org.apache.flink.runtime.metrics.scope.ScopeFormat;
import org.apache.flink.runtime.metrics.scope.ScopeFormats;
import org.apache.flink.runtime.metrics.util.TestingMetricRegistry;
+import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap;
+
import org.junit.Test;
import java.util.Collections;
@@ -45,7 +47,8 @@ public class FrontMetricGroupTest {
final FrontMetricGroup<?> frontMetricGroup =
new FrontMetricGroup<>(
- new ReporterScopedSettings(0, delimiter, Collections.emptySet()),
+ new ReporterScopedSettings(
+ 0, delimiter, Collections.emptySet(), Collections.emptyMap()),
new ProcessMetricGroup(
TestingMetricRegistry.builder()
.setScopeFormats(ScopeFormats.fromConfig(config))
@@ -75,7 +78,8 @@ public class FrontMetricGroupTest {
final FrontMetricGroup<?> frontMetricGroup =
new FrontMetricGroup<>(
- new ReporterScopedSettings(0, delimiter, Collections.emptySet()),
+ new ReporterScopedSettings(
+ 0, delimiter, Collections.emptySet(), Collections.emptyMap()),
new ProcessMetricGroup(
TestingMetricRegistry.builder()
.setScopeFormats(ScopeFormats.fromConfig(config))
@@ -95,4 +99,19 @@ public class FrontMetricGroupTest {
// context where the delimiter matters
assertThat(frontMetricGroup.getAllVariables(), hasEntry(ScopeFormat.SCOPE_HOST, hostName));
}
+
+ @Test
+ public void testGetAllVariablesWithAdditionalVariables() {
+ final FrontMetricGroup<?> frontMetricGroup =
+ new FrontMetricGroup<>(
+ new ReporterScopedSettings(
+ 0,
+ '.',
+ Collections.emptySet(),
+ ImmutableMap.of(ScopeFormat.asVariable("foo"), "bar")),
+ new ProcessMetricGroup(TestingMetricRegistry.builder().build(), "host"));
+
+ assertThat(
+ frontMetricGroup.getAllVariables(), hasEntry(ScopeFormat.asVariable("foo"), "bar"));
+ }
}