[FLINK-23539][metrics][influxdb] Filter '\n' in tags

mbalassi · Aug 4, 2021 · 1c7100a · 1c7100a
1 parent 2ca9178
commit 1c7100a
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 1 deletion.
diff --git a/...ics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MeasurementInfoProvider.java b/...ics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MeasurementInfoProvider.java
@@ -29,6 +29,7 @@
 
 class MeasurementInfoProvider implements MetricInfoProvider<MeasurementInfo> {
  @VisibleForTesting static final char SCOPE_SEPARATOR = '_';
+ private static final String POINT_DELIMITER = "\n";
 
  private static final CharacterFilter CHARACTER_FILTER =
  new CharacterFilter() {
@@ -52,7 +53,9 @@ private static Map<String, String> getTags(MetricGroup group) {
  Map<String, String> tags = new HashMap<>();
  for (Map.Entry<String, String> variable : group.getAllVariables().entrySet()) {
  String name = variable.getKey();
- tags.put(name.substring(1, name.length() - 1), variable.getValue());
+ tags.put(
+ normalize(name.substring(1, name.length() - 1)),
+ normalize(variable.getValue()));
  }
  return tags;
  }
@@ -65,4 +68,8 @@ private static String getLogicalScope(MetricGroup group) {
  return LogicalScopeProvider.castFrom(group)
  .getLogicalScope(CHARACTER_FILTER, SCOPE_SEPARATOR);
  }
+
+ private static String normalize(String value) {
+ return value.replace(POINT_DELIMITER, "");
+ }
 }
diff --git a/...influxdb/src/test/java/org/apache/flink/metrics/influxdb/MeasurementInfoProviderTest.java b/...influxdb/src/test/java/org/apache/flink/metrics/influxdb/MeasurementInfoProviderTest.java
@@ -61,4 +61,16 @@ public void simpleTestGetMetricInfo() {
  assertThat(info.getTags(), hasEntry("C", "c"));
  assertEquals(3, info.getTags().size());
  }
+
+ @Test
+ public void testNormalizingTags() {
+ Map<String, String> variables = new HashMap<>();
+ variables.put("<A\n>", "a\n");
+
+ final MetricGroup metricGroup =
+ TestMetricGroup.newBuilder().setVariables(variables).build();
+
+ MeasurementInfo info = provider.getMetricInfo("m1", metricGroup);
+ assertThat(info.getTags(), hasEntry("A", "a"));
+ }
 }