[FLINK-11251][metrics] Exclude variable values from logical scope of …

…generic groups
deepfield · Jan 9, 2019 · 9742ef7 · 9742ef7
1 parent 9ab04da
commit 9742ef7
Show file tree

Hide file tree

Showing 3 changed files with 35 additions and 2 deletions.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
@@ -172,7 +172,7 @@ String getLogicalScope(CharacterFilter filter, char delimiter, int reporterIndex
  }
  }
 
- private String createLogicalScope(CharacterFilter filter, char delimiter) {
+ protected String createLogicalScope(CharacterFilter filter, char delimiter) {
  final String groupName = getGroupName(filter);
  return parent == null
  ? groupName

diff --git a/...untime/src/main/java/org/apache/flink/runtime/metrics/groups/GenericValueMetricGroup.java b/...untime/src/main/java/org/apache/flink/runtime/metrics/groups/GenericValueMetricGroup.java
@@ -51,7 +51,7 @@ protected void putVariables(Map<String, String> variables) {
  }
 
  @Override
- public String getLogicalScope(CharacterFilter filter, char delimiter) {
+ protected String createLogicalScope(CharacterFilter filter, char delimiter) {
  return parent.getLogicalScope(filter, delimiter);
  }
 }
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
@@ -19,6 +19,8 @@
 package org.apache.flink.runtime.metrics.groups;
 
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.metrics.CharacterFilter;
 import org.apache.flink.metrics.Gauge;
 import org.apache.flink.metrics.Metric;
@@ -31,17 +33,21 @@
 import org.apache.flink.runtime.metrics.dump.QueryScopeInfo;
 import org.apache.flink.runtime.metrics.scope.ScopeFormat;
 import org.apache.flink.runtime.metrics.util.DummyCharacterFilter;
+import org.apache.flink.runtime.metrics.util.TestReporter;
 import org.apache.flink.util.AbstractID;
 import org.apache.flink.util.TestLogger;
 
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.not;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -216,6 +222,33 @@ public void testNameCollisionAfterKeyValueGroup() {
  assertFalse("Value is present in logical scope.", logicalScope.contains(value));
  }
 
+ /**
+ * Verifies that calling {@link AbstractMetricGroup#getLogicalScope(CharacterFilter, char, int)} on {@link GenericValueMetricGroup}
+ * should ignore value as well.
+ */
+ @Test
+ public void testLogicalScopeShouldIgnoreValueGroupName() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "test." + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, TestReporter.class.getName());
+
+ MetricRegistryImpl registry = new MetricRegistryImpl(MetricRegistryConfiguration.fromConfiguration(config));
+ try {
+ GenericMetricGroup root = new GenericMetricGroup(registry, new DummyAbstractMetricGroup(registry), "root");
+
+ String key = "key";
+ String value = "value";
+
+ MetricGroup group = root.addGroup(key, value);
+
+ String logicalScope = ((AbstractMetricGroup) group)
+ .getLogicalScope(new DummyCharacterFilter(), registry.getDelimiter(), 0);
+ assertThat("Key is missing from logical scope.", logicalScope, containsString(key));
+ assertThat("Value is present in logical scope.", logicalScope, not(containsString(value)));
+ } finally {
+ registry.shutdown().get();
+ }
+ }
+
  @Test
  public void closedGroupDoesNotRegisterMetrics() {
  GenericMetricGroup group = new GenericMetricGroup(