InfluxDbFormatParser 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 < 2) { throw new FormatParseException("Expected at least 2 parts, found " + parts.length + " in line '" + line + "'"); } String metricName = getMetricName(parts); Map<String, String> tags = getMetricTags(parts); double value = getMetricValue(parts); Instant timestamp = getMetricTimestamp(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()); } InfluxDbFormatParser(Clock clock); @Override Iterable<MetricTimeSeries> parse(InputStream stream);  }
@Test public void testParse() throws Exception { try (InputStream stream = InfluxDbFormatParserTest.class.getResourceAsStream("/influxdb.txt")) { assertNotNull(stream); List<MetricTimeSeries> series = Lists.newArrayList(sut.parse(stream)); assertThat(series.size(), is(3)); MetricTimeSeries server01 = findWithHost(series, "server01"); assertThat(server01.getName(), is("cpu_load_short")); assertThat(server01.getTimestamps().size(), is(1)); assertThat(server01.getTimestamps().get(0), is(NOW.toEpochMilli())); assertThat(server01.getValues().size(), is(1)); assertThat(server01.getValues().get(0), is(0.67)); assertThat(server01.getAttributesReference().get("host"), is("server01")); MetricTimeSeries server02 = findWithHost(series, "server02"); assertThat(server02.getName(), is("cpu_load_short")); assertThat(server02.getTimestamps().size(), is(1)); assertThat(server02.getTimestamps().get(0), is(1422568543702900L)); assertThat(server02.getValues().size(), is(1)); assertThat(server02.getValues().get(0), is(0.55)); assertThat(server02.getAttributesReference().get("host"), is("server02")); MetricTimeSeries server03 = findWithHost(series, "server03"); assertThat(server03.getName(), is("cpu_load_long")); assertThat(server03.getTimestamps().size(), is(2)); assertThat(server03.getTimestamps().get(0), is(1422568543702900L)); assertThat(server03.getTimestamps().get(1), is(1422568544702900L)); assertThat(server03.getValues().size(), is(2)); assertThat(server03.getValues().get(0), is(2.0)); assertThat(server03.getValues().get(1), is(3.0)); assertThat(server03.getAttributesReference().get("host"), is("server03")); assertThat(server03.getAttributesReference().get("direction"), is("in")); assertThat(server03.getAttributesReference().get("region"), is("us-west")); } }