[FLINK-6900] [metrics] Limit size of metric name components

docs/monitoring/metrics.md

@@ -632,6 +632,7 @@ Parameters:
 - `host` - the Graphite server host
 - `port` - the Graphite server port
 - `protocol` - protocol to use (TCP/UDP)
+- `maxComponentLength` - limits the length of each scope component
 
 Example configuration:
 
@@ -641,6 +642,7 @@ metrics.reporter.grph.class: org.apache.flink.metrics.graphite.GraphiteReporter
 metrics.reporter.grph.host: localhost
 metrics.reporter.grph.port: 2003
 metrics.reporter.grph.protocol: TCP
+metrics.reporter.grph.maxComponentLength: 80
 
 {% endhighlight %}
 
@@ -708,6 +710,7 @@ Parameters:
 
 - `host` - the StatsD server host
 - `port` - the StatsD server port
+- `maxComponentLength` - limits the length of each scope component
 
 Example configuration:
 
@@ -716,6 +719,7 @@ Example configuration:
 metrics.reporter.stsd.class: org.apache.flink.metrics.statsd.StatsDReporter
 metrics.reporter.stsd.host: localhost
 metrics.reporter.stsd.port: 8125
+metrics.reporter.stsd.maxComponentLength: 80
 
 {% endhighlight %}
 

flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/ScheduledDropwizardReporter.java

@@ -61,6 +61,7 @@ public abstract class ScheduledDropwizardReporter implements MetricReporter, Sch
 	public static final String ARG_PREFIX = "prefix";
 	public static final String ARG_CONVERSION_RATE = "rateConversion";
 	public static final String ARG_CONVERSION_DURATION = "durationConversion";
+	public static final String ARG_MAX_COMPONENT_LENGTH = "maxComponentLength";
 
 	// ------------------------------------------------------------------------
 
@@ -73,6 +74,8 @@ public abstract class ScheduledDropwizardReporter implements MetricReporter, Sch
 	private final Map<Histogram, String> histograms = new HashMap<>();
 	private final Map<Meter, String> meters = new HashMap<>();
 
+	private int maxComponentLength = 80;
+
 	// ------------------------------------------------------------------------
 
 	protected ScheduledDropwizardReporter() {
@@ -109,6 +112,7 @@ Map<Histogram, String> getHistograms() {
 
 	@Override
 	public void open(MetricConfig config) {
+		this.maxComponentLength = config.getInteger(ARG_MAX_COMPONENT_LENGTH, 80);
 		this.reporter = getReporter(config);
 	}
 
@@ -184,7 +188,11 @@ public void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup
 	@Override
 	public String filterCharacters(String metricName) {
 		char[] chars = null;
-		final int strLen = metricName.length();
+		int strLen = metricName.length();
+		if (strLen > maxComponentLength) {
+			log.warn("The metric name component {} exceeded the {} characters length limit and was truncated.", metricName, maxComponentLength);
+			strLen = maxComponentLength;
+		}
 		int pos = 0;
 
 		for (int i = 0; i < strLen; i++) {

flink-metrics/flink-metrics-dropwizard/src/test/java/org/apache/flink/dropwizard/ScheduledDropwizardReporterTest.java

@@ -44,7 +44,6 @@
 import com.codahale.metrics.ScheduledReporter;
 import org.junit.Test;
 
-import java.lang.reflect.InvocationTargetException;
 import java.util.List;
 import java.util.Map;
 
@@ -57,7 +56,28 @@
 public class ScheduledDropwizardReporterTest {
 
 	@Test
-	public void testInvalidCharacterReplacement() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
+	public void testNameTruncating() {
+		final MetricConfig config = new MetricConfig();
+		config.setProperty(ScheduledDropwizardReporter.ARG_MAX_COMPONENT_LENGTH, "10");
+
+		final ScheduledDropwizardReporter reporter = new ScheduledDropwizardReporter() {
+			@Override
+			public ScheduledReporter getReporter(MetricConfig config) {
+				return null;
+			}
+		};
+
+		try {
+			reporter.open(config);
+
+			assertEquals("0123456789", reporter.filterCharacters("0123456789DEADBEEF"));
+		} finally {
+			reporter.close();
+		}
+	}
+
+	@Test
+	public void testInvalidCharacterReplacement() {
 		ScheduledDropwizardReporter reporter = new ScheduledDropwizardReporter() {
 			@Override
 			public ScheduledReporter getReporter(MetricConfig config) {

flink-metrics/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporter.java

@@ -55,12 +55,15 @@ public class StatsDReporter extends AbstractReporter implements Scheduled {
 
 	public static final String ARG_HOST = "host";
 	public static final String ARG_PORT = "port";
+	public static final String ARG_MAX_COMPONENT_LENGTH = "maxComponentLength";
 
 	private boolean closed = false;
 
 	private DatagramSocket socket;
 	private InetSocketAddress address;
 
+	private int maxComponentLength = 80;
+
 	@Override
 	public void open(MetricConfig config) {
 		String host = config.getString(ARG_HOST, null);
@@ -77,7 +80,9 @@ public void open(MetricConfig config) {
 		} catch (SocketException e) {
 			throw new RuntimeException("Could not create datagram socket. ", e);
 		}
-		log.info("Configured StatsDReporter with {host:{}, port:{}}", host, port);
+
+		maxComponentLength = config.getInteger(ARG_MAX_COMPONENT_LENGTH, 80);
+		log.info("Configured StatsDReporter with {host:{}, port:{}, maxComponentLength:{}}", host, port, maxComponentLength);
 	}
 
 	@Override
@@ -193,7 +198,11 @@ private void send(final String name, final String value) {
 	@Override
 	public String filterCharacters(String input) {
 		char[] chars = null;
-		final int strLen = input.length();
+		int strLen = input.length();
+		if (strLen > maxComponentLength) {
+			log.warn("The metric name component {} exceeded the {} characters length limit and was truncated.", input, maxComponentLength);
+			strLen = maxComponentLength;
+		}
 		int pos = 0;
 
 		for (int i = 0; i < strLen; i++) {

flink-metrics/flink-metrics-statsd/src/test/java/org/apache/flink/metrics/statsd/StatsDReporterTest.java

@@ -61,7 +61,24 @@
 public class StatsDReporterTest extends TestLogger {
 
 	@Test
-	public void testReplaceInvalidChars() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
+	public void testNameTruncating() {
+		final MetricConfig config = new MetricConfig();
+		config.setProperty(StatsDReporter.ARG_HOST, "localhost");
+		config.setProperty(StatsDReporter.ARG_PORT, "12345");
+		config.setProperty(StatsDReporter.ARG_MAX_COMPONENT_LENGTH, "10");
+
+		final StatsDReporter reporter = new StatsDReporter();
+		try {
+			reporter.open(config);
+
+			assertEquals("0123456789", reporter.filterCharacters("0123456789DEADBEEF"));
+		} finally {
+			reporter.close();
+		}
+	}
+
+	@Test
+	public void testReplaceInvalidChars() {
 		StatsDReporter reporter = new StatsDReporter();
 
 		assertEquals("", reporter.filterCharacters(""));