@Test public void testAddingMetrics() throws Exception { Configuration configuration = new Configuration(); String taskName = "testTask"; String jobName = "testJob:-!ax..?"; String hostname = "local::host:"; String taskManagerId = "tas:kMana::ger"; String counterName = "testCounter"; configuration.setString(MetricOptions.SCOPE_NAMING_TASK, "<host>.<tm_id>.<job_name>"); configuration.setString(MetricOptions.SCOPE_DELIMITER, "_"); MetricRegistryImpl metricRegistry = new MetricRegistryImpl( MetricRegistryConfiguration.fromConfiguration(configuration), Collections.singletonList(ReporterSetup.forReporter("test", new TestingStatsDReporter()))); char delimiter = metricRegistry.getDelimiter(); TaskManagerMetricGroup tmMetricGroup = new TaskManagerMetricGroup(metricRegistry, hostname, taskManagerId); TaskManagerJobMetricGroup tmJobMetricGroup = new TaskManagerJobMetricGroup(metricRegistry, tmMetricGroup, new JobID(), jobName); TaskMetricGroup taskMetricGroup = new TaskMetricGroup(metricRegistry, tmJobMetricGroup, new JobVertexID(), new AbstractID(), taskName, 0, 0); SimpleCounter myCounter = new SimpleCounter(); taskMetricGroup.counter(counterName, myCounter); List<MetricReporter> reporters = metricRegistry.getReporters(); assertTrue(reporters.size() == 1); MetricReporter metricReporter = reporters.get(0); assertTrue("Reporter should be of type StatsDReporter", metricReporter instanceof StatsDReporter); TestingStatsDReporter reporter = (TestingStatsDReporter) metricReporter; Map<Counter, String> counters = reporter.getCounters(); assertTrue(counters.containsKey(myCounter)); String expectedCounterName = reporter.filterCharacters(hostname) + delimiter + reporter.filterCharacters(taskManagerId) + delimiter + reporter.filterCharacters(jobName) + delimiter + reporter.filterCharacters(counterName); assertEquals(expectedCounterName, counters.get(myCounter)); metricRegistry.shutdown().get(); }
@Override public String filterCharacters(String input) { char[] chars = null; final int strLen = input.length(); int pos = 0; for (int i = 0; i < strLen; i++) { final char c = input.charAt(i); switch (c) { case ':': if (chars == null) { chars = input.toCharArray(); } chars[pos++] = '-'; break; default: if (chars != null) { chars[pos] = c; } pos++; } } return chars == null ? input : new String(chars, 0, pos); }
StatsDReporter extends AbstractReporter implements Scheduled { @Override public String filterCharacters(String input) { char[] chars = null; final int strLen = input.length(); int pos = 0; for (int i = 0; i < strLen; i++) { final char c = input.charAt(i); switch (c) { case ':': if (chars == null) { chars = input.toCharArray(); } chars[pos++] = '-'; break; default: if (chars != null) { chars[pos] = c; } pos++; } } return chars == null ? input : new String(chars, 0, pos); } }
StatsDReporter extends AbstractReporter implements Scheduled { @Override public String filterCharacters(String input) { char[] chars = null; final int strLen = input.length(); int pos = 0; for (int i = 0; i < strLen; i++) { final char c = input.charAt(i); switch (c) { case ':': if (chars == null) { chars = input.toCharArray(); } chars[pos++] = '-'; break; default: if (chars != null) { chars[pos] = c; } pos++; } } return chars == null ? input : new String(chars, 0, pos); }  }
StatsDReporter extends AbstractReporter implements Scheduled { @Override public String filterCharacters(String input) { char[] chars = null; final int strLen = input.length(); int pos = 0; for (int i = 0; i < strLen; i++) { final char c = input.charAt(i); switch (c) { case ':': if (chars == null) { chars = input.toCharArray(); } chars[pos++] = '-'; break; default: if (chars != null) { chars[pos] = c; } pos++; } } return chars == null ? input : new String(chars, 0, pos); }  @Override void open(MetricConfig config); @Override void close(); @Override void report(); @Override String filterCharacters(String input); }
StatsDReporter extends AbstractReporter implements Scheduled { @Override public String filterCharacters(String input) { char[] chars = null; final int strLen = input.length(); int pos = 0; for (int i = 0; i < strLen; i++) { final char c = input.charAt(i); switch (c) { case ':': if (chars == null) { chars = input.toCharArray(); } chars[pos++] = '-'; break; default: if (chars != null) { chars[pos] = c; } pos++; } } return chars == null ? input : new String(chars, 0, pos); }  @Override void open(MetricConfig config); @Override void close(); @Override void report(); @Override String filterCharacters(String input); static final String ARG_HOST; static final String ARG_PORT; }