OpenTsdbTelnetFormatParser implements FormatParser { @Override public Iterable<MetricTimeSeries> parse(InputStream stream) throws FormatParseException { Map<Metric, MetricTimeSeries.Builder> metrics = new HashMap<>(); BufferedReader reader = new BufferedReader(new InputStreamReader(stream, UTF_8)); String line; try { while ((line = reader.readLine()) != null) { String[] parts = StringUtils.split(line, ' '); if (parts.length < 5) { throw new FormatParseException("Expected at least 5 parts, found " + parts.length + " in line '" + line + "'"); } if (!parts[0].equals("put")) { throw new FormatParseException("Expected first segment to be 'put', but was '" + parts[0] + "'"); } String metricName = getMetricName(parts); Instant timestamp = getMetricTimestamp(parts); double value = getMetricValue(parts); Map<String, String> tags = getMetricTags(parts); Metric metric = new Metric(metricName, tags); MetricTimeSeries.Builder metricBuilder = metrics.get(metric); if (metricBuilder == null) { metricBuilder = new MetricTimeSeries.Builder(metricName, METRIC_TYPE); for (Map.Entry<String, String> tagEntry : tags.entrySet()) { metricBuilder.attribute(tagEntry.getKey(), tagEntry.getValue()); } metrics.put(metric, metricBuilder); } metricBuilder.point(timestamp.toEpochMilli(), value); } } catch (IOException e) { throw new FormatParseException("IO exception while parsing OpenTSDB telnet format", e); } return metrics.values().stream().map(MetricTimeSeries.Builder::build).collect(Collectors.toList()); }  @Override Iterable<MetricTimeSeries> parse(InputStream stream);  }
@Test public void testParse() throws Exception { try (InputStream stream = GraphiteFormatParserTest.class.getResourceAsStream("/openTSDB-telnet.txt")) { assertNotNull(stream); List<MetricTimeSeries> series = Lists.newArrayList(sut.parse(stream)); assertThat(series.size(), is(3)); MetricTimeSeries cpu0series = findWithCpu(series, 0); assertThat(cpu0series.getName(), is("sys.cpu.user")); assertThat(cpu0series.getTimestamps().size(), is(2)); assertThat(cpu0series.getTimestamps().get(0), is(1356998400000L)); assertThat(cpu0series.getTimestamps().get(1), is(1356998401000L)); assertThat(cpu0series.getAttributesReference().get("cpu"), is("0")); assertThat(cpu0series.getAttributesReference().get("host"), is("webserver01")); MetricTimeSeries cpu1series = findWithCpu(series, 1); assertThat(cpu1series.getName(), is("sys.cpu.user")); assertThat(cpu1series.getTimestamps().size(), is(1)); assertThat(cpu1series.getTimestamps().get(0), is(1356998400000L)); assertThat(cpu1series.getAttributesReference().get("cpu"), is("1")); assertThat(cpu1series.getAttributesReference().get("host"), is("webserver01")); MetricTimeSeries cpu2series = findWithCpu(series, 2); assertThat(cpu2series.getName(), is("sys.cpu.user")); assertThat(cpu2series.getTimestamps().size(), is(1)); assertThat(cpu2series.getTimestamps().get(0), is(1356998400000L)); assertThat(cpu2series.getAttributesReference().get("cpu"), is("2")); assertThat(cpu2series.getAttributesReference().get("host"), is("webserver01")); } }