KairosDbFormatParser implements FormatParser { @Override public Iterable<MetricTimeSeries> parse(InputStream stream) throws FormatParseException { Map<Metric, MetricTimeSeries.Builder> metrics = new HashMap<>(); List<KairosDbMetric> kairosMetrics = parseJson(stream); for (KairosDbMetric kairosMetric : kairosMetrics) { Metric metric = new Metric(kairosMetric.getName(), kairosMetric.getTags()); MetricTimeSeries.Builder metricBuilder = metrics.get(metric); if (metricBuilder == null) { metricBuilder = new MetricTimeSeries.Builder(kairosMetric.getName(), METRIC_TYPE); for (Map.Entry<String, String> tagEntry : kairosMetric.getTags().entrySet()) { metricBuilder.attribute(tagEntry.getKey(), tagEntry.getValue()); } metrics.put(metric, metricBuilder); } Instant timestamp = convertTimestamp(kairosMetric.getTimestamp()); metricBuilder.point(timestamp.toEpochMilli(), kairosMetric.getValue()); } return metrics.values().stream().map(MetricTimeSeries.Builder::build).collect(Collectors.toList()); }  @Override Iterable<MetricTimeSeries> parse(InputStream stream);  }
@Test public void test() throws Exception { try (InputStream stream = KairosDbFormatParserTest.class.getResourceAsStream("/kairosdb.json")) { assertNotNull(stream); List<MetricTimeSeries> series = Lists.newArrayList(sut.parse(stream)); assertThat(series.size(), is(2)); MetricTimeSeries first = series.get(0); assertThat(first.getName(), is("archive.file.tracked")); assertThat(first.getTimestamps().size(), is(1)); assertThat(first.getValues().get(0), is(123.0)); assertThat(first.getTimestamps().get(0), is(1349109376L)); assertThat(first.getAttributesReference().get("host"), is("test")); MetricTimeSeries second = series.get(1); assertThat(second.getName(), is("archive.file.search")); assertThat(second.getTimestamps().size(), is(1)); assertThat(second.getValues().get(0), is(32.1)); assertThat(second.getTimestamps().get(0), is(1349109665L)); assertThat(second.getAttributesReference().get("host"), is("test")); } }