Revert "Revert "Consume and process metrics as they are parsed.""

8 files changed, 165 insertions, 147 deletions

diff --git a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/core/VespaMetrics.java b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/core/VespaMetrics.java
index f112acd6e05..a8ef79d827e 100644
--- a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/core/VespaMetrics.java
+++ b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/core/VespaMetrics.java
@@ -12,6 +12,7 @@ import ai.vespa.metricsproxy.metric.model.Dimension;
 import ai.vespa.metricsproxy.metric.model.DimensionId;
 import ai.vespa.metricsproxy.metric.model.MetricId;
 import ai.vespa.metricsproxy.metric.model.MetricsPacket;
+import ai.vespa.metricsproxy.service.MetricsParser;
 import ai.vespa.metricsproxy.service.VespaService;
 
 import java.util.ArrayList;
@@ -75,23 +76,20 @@ public class VespaMetrics {
             Optional<MetricsPacket.Builder> systemCheck = getSystemMetrics(service);
             systemCheck.ifPresent(metricsPackets::add);
 
-            Metrics allServiceMetrics = service.getMetrics();
+            MetricAggregator aggregator = new MetricAggregator(service.getDimensions());
+            GetServiceMetricsConsumer metricsConsumer = new GetServiceMetricsConsumer(consumersByMetric, aggregator);
+            service.consumeMetrics(metricsConsumer);
 
-            if (! allServiceMetrics.getMetrics().isEmpty()) {
-                Metrics serviceMetrics = getServiceMetrics(allServiceMetrics, consumersByMetric);
+            if (! aggregator.getAggregated().isEmpty()) {
 
                 // One metrics packet per set of metrics that share the same dimensions+consumers
-                // TODO: Move aggregation into MetricsPacket itself?
-                MetricAggregator aggregator = new MetricAggregator(service.getDimensions());
-                serviceMetrics.getMetrics().forEach(metric -> aggregator.aggregate(metric));
-
                 aggregator.getAggregated().forEach((aggregationKey, metrics) -> {
                     MetricsPacket.Builder builder = new MetricsPacket.Builder(service.getMonitoringName())
                             .putMetrics(metrics)
                             .putDimension(METRIC_TYPE_DIMENSION_ID, "standard")
                             .putDimension(INSTANCE_DIMENSION_ID, service.getInstanceName())
                             .putDimensions(aggregationKey.getDimensions());
-                    setMetaInfo(builder, serviceMetrics.getTimeStamp());
+                    setMetaInfo(builder, metrics.get(0).getTimeStamp());
                     builder.addConsumers(aggregationKey.getConsumers());
                     metricsPackets.add(builder);
                 });
@@ -120,15 +118,20 @@ public class VespaMetrics {
      * In order to include a metric, it must exist in the given map of metric to consumers.
      * Each returned metric will contain a collection of consumers that it should be routed to.
      */
-    private Metrics getServiceMetrics(Metrics allServiceMetrics, Map<ConfiguredMetric, Set<ConsumerId>> consumersByMetric) {
-        Metrics configuredServiceMetrics = new Metrics();
-        configuredServiceMetrics.setTimeStamp(getMostRecentTimestamp(allServiceMetrics));
-        for (Metric candidate : allServiceMetrics.getMetrics()) {
+    private class GetServiceMetricsConsumer implements MetricsParser.Consumer {
+        private final MetricAggregator aggregator;
+        private final Map<ConfiguredMetric, Set<ConsumerId>> consumersByMetric;
+        GetServiceMetricsConsumer(Map<ConfiguredMetric, Set<ConsumerId>> consumersByMetric, MetricAggregator aggregator) {
+            this.consumersByMetric = consumersByMetric;
+            this.aggregator = aggregator;
+        }
+
+        @Override
+        public void consume(Metric candidate) {
             getConfiguredMetrics(candidate.getName(), consumersByMetric.keySet()).forEach(
-                    configuredMetric -> configuredServiceMetrics.add(
+                    configuredMetric -> aggregator.aggregate(
                             metricWithConfigProperties(candidate, configuredMetric, consumersByMetric)));
         }
-        return configuredServiceMetrics;
     }
 
     private Map<DimensionId, String> extractDimensions(Map<DimensionId, String> dimensions, List<Dimension> configuredDimensions) {
@@ -185,16 +188,6 @@ public class VespaMetrics {
         return Optional.of(builder);
     }
 
-    private long getMostRecentTimestamp(Metrics metrics) {
-        long mostRecentTimestamp = 0L;
-        for (Metric metric : metrics.getMetrics()) {
-            if (metric.getTimeStamp() > mostRecentTimestamp) {
-                mostRecentTimestamp = metric.getTimeStamp();
-            }
-        }
-        return mostRecentTimestamp;
-    }
-
     private static class MetricAggregator {
         private final Map<AggregationKey, List<Metric>> aggregated = new HashMap<>();
         private final Map<DimensionId, String> serviceDimensions;
@@ -218,15 +211,6 @@ public class VespaMetrics {
         }
     }
 
-    private Map<AggregationKey, List<Metric>> aggregateMetrics(Map<DimensionId, String> serviceDimensions, Metrics metrics) {
-        MetricAggregator aggregator = new MetricAggregator(serviceDimensions);
-
-        for (Metric metric :  metrics.getMetrics() ) {
-            aggregator.aggregate(metric);
-        }
-        return aggregator.getAggregated();
-    }
-
     private List<ConfiguredMetric> getMetricDefinitions(ConsumerId consumer) {
         if (metricsConsumers == null) return Collections.emptyList();
 
@@ -240,75 +224,100 @@ public class VespaMetrics {
                 .statusMessage("Data collected successfully");
     }
 
+    private class MetricStringBuilder implements MetricsParser.Consumer {
+        private final StringBuilder sb = new StringBuilder();
+        private VespaService service;
+        @Override
+        public void consume(Metric metric) {
+            MetricId key = metric.getName();
+            MetricId alias = key;
+
+            boolean isForwarded = false;
+            for (ConfiguredMetric metricConsumer : getMetricDefinitions(vespaMetricsConsumerId)) {
+                if (metricConsumer.id().equals(key)) {
+                    alias = metricConsumer.outputname();
+                    isForwarded = true;
+                }
+            }
+            if (isForwarded) {
+                sb.append(formatter.format(service, alias.id, metric.getValue())).append(" ");
+            }
+        }
+
+        @Override
+        public String toString() {
+            return sb.toString();
+        }
+    }
     /**
      * Returns a string representation of metrics for the given services;
      * a space separated list of key=value.
      */
     public String getMetricsAsString(List<VespaService> services) {
-        StringBuilder b = new StringBuilder();
-        for (VespaService s : services) {
-            for (Metric metric : s.getMetrics().getMetrics()) {
-                MetricId key = metric.getName();
-                MetricId alias = key;
-
-                boolean isForwarded = false;
-                for (ConfiguredMetric metricConsumer : getMetricDefinitions(vespaMetricsConsumerId)) {
-                    if (metricConsumer.id().equals(key)) {
-                        alias = metricConsumer.outputname();
-                        isForwarded = true;
+        MetricStringBuilder msb = new MetricStringBuilder();
+        for (VespaService service : services) {
+            msb.service = service;
+            service.consumeMetrics(msb);
+        }
+        return msb.toString();
+    }
+
+    private class MetricNamesBuilder implements MetricsParser.Consumer {
+        private final StringBuilder bufferOn = new StringBuilder();
+        private final StringBuilder bufferOff = new StringBuilder();
+        private final ConsumerId consumer;
+        MetricNamesBuilder(ConsumerId consumer) {
+            this.consumer = consumer;
+        }
+        @Override
+        public void consume(Metric m) {
+            String description = m.getDescription();
+            MetricId alias = MetricId.empty;
+            boolean isForwarded = false;
+
+            for (ConfiguredMetric metric : getMetricDefinitions(consumer)) {
+                if (metric.id().equals(m.getName())) {
+                    alias = metric.outputname();
+                    isForwarded = true;
+                    if (description.isEmpty()) {
+                        description = metric.description();
                     }
                 }
-                if (isForwarded) {
-                    b.append(formatter.format(s, alias.id, metric.getValue())).append(" ");
-                }
             }
+
+            String message = "OFF";
+            StringBuilder buffer = bufferOff;
+            if (isForwarded) {
+                buffer = bufferOn;
+                message = "ON";
+            }
+            buffer.append(m.getName()).append('=').append(message);
+            if (!description.isEmpty()) {
+                buffer.append(";description=").append(description);
+            }
+            if (!alias.id.isEmpty()) {
+                buffer.append(";output-name=").append(alias);
+            }
+            buffer.append(',');
         }
-        return b.toString();
-    }
 
+        @Override
+        public String toString() {
+            return bufferOn.append(bufferOff).toString();
+        }
+    }
     /**
      * Get all metric names for the given services
      *
      * @return String representation
      */
     public String getMetricNames(List<VespaService> services, ConsumerId consumer) {
-        StringBuilder bufferOn = new StringBuilder();
-        StringBuilder bufferOff = new StringBuilder();
-        for (VespaService s : services) {
-
-            for (Metric m : s.getMetrics().getMetrics()) {
-                String description = m.getDescription();
-                MetricId alias = MetricId.empty;
-                boolean isForwarded = false;
-
-                for (ConfiguredMetric metric : getMetricDefinitions(consumer)) {
-                    if (metric.id().equals(m.getName())) {
-                        alias = metric.outputname();
-                        isForwarded = true;
-                        if (description.isEmpty()) {
-                            description = metric.description();
-                        }
-                    }
-                }
-
-                String message = "OFF";
-                StringBuilder buffer = bufferOff;
-                if (isForwarded) {
-                    buffer = bufferOn;
-                    message = "ON";
-                }
-                buffer.append(m.getName()).append('=').append(message);
-                if (!description.isEmpty()) {
-                    buffer.append(";description=").append(description);
-                }
-                if (!alias.id.isEmpty()) {
-                    buffer.append(";output-name=").append(alias);
-                }
-                buffer.append(',');
-            }
+        MetricNamesBuilder metricNamesBuilder = new MetricNamesBuilder(consumer);
+        for (VespaService service : services) {
+            service.consumeMetrics(metricNamesBuilder);
         }
 
-        return bufferOn.append(bufferOff).toString();
+        return metricNamesBuilder.toString();
     }
 
 }
diff --git a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/DummyMetricsFetcher.java b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/DummyMetricsFetcher.java
index b304e5d74d3..e9fbc942e6e 100644
--- a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/DummyMetricsFetcher.java
+++ b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/DummyMetricsFetcher.java
@@ -1,8 +1,6 @@
 // Copyright 2020 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package ai.vespa.metricsproxy.service;
 
-import ai.vespa.metricsproxy.metric.Metrics;
-
 /**
  * Dummy class used for getting health status for a vespa service that has no HTTP service
  * for getting metrics
@@ -21,7 +19,6 @@ public class DummyMetricsFetcher extends RemoteMetricsFetcher {
     /**
      * Connect to remote service over http and fetch metrics
      */
-    public Metrics getMetrics(int fetchCount) {
-        return new Metrics();
+    public void getMetrics(MetricsParser.Consumer consumer, int fetchCount) {
     }
 }
diff --git a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/MetricsParser.java b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/MetricsParser.java
index 48621b9abab..22812c07b78 100644
--- a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/MetricsParser.java
+++ b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/MetricsParser.java
@@ -2,7 +2,6 @@
 package ai.vespa.metricsproxy.service;
 
 import ai.vespa.metricsproxy.metric.Metric;
-import ai.vespa.metricsproxy.metric.Metrics;
 import ai.vespa.metricsproxy.metric.model.DimensionId;
 import ai.vespa.metricsproxy.metric.model.MetricId;
 import com.fasterxml.jackson.core.JsonParser;
@@ -25,59 +24,61 @@ import static ai.vespa.metricsproxy.metric.model.DimensionId.toDimensionId;
  * @author Jo Kristian Bergum
  */
 public class MetricsParser {
+    public interface Consumer {
+        void consume(Metric metric);
+    }
 
     private static final ObjectMapper jsonMapper = new ObjectMapper();
 
-    static Metrics parse(String data) throws IOException {
-        return parse(jsonMapper.createParser(data));
+    static void parse(String data, Consumer consumer) throws IOException {
+        parse(jsonMapper.createParser(data), consumer);
     }
 
-    static Metrics parse(InputStream data) throws IOException {
-        return parse(jsonMapper.createParser(data));
+    static void parse(InputStream data, Consumer consumer) throws IOException {
+        parse(jsonMapper.createParser(data), consumer);
     }
-    private static Metrics parse(JsonParser parser) throws IOException {
+    private static void parse(JsonParser parser, Consumer consumer) throws IOException {
         if (parser.nextToken() != JsonToken.START_OBJECT) {
             throw new IOException("Expected start of object, got " + parser.currentToken());
         }
 
-        Metrics metrics  = new Metrics();
         for (parser.nextToken(); parser.getCurrentToken() != JsonToken.END_OBJECT; parser.nextToken()) {
             String fieldName = parser.getCurrentName();
             JsonToken token = parser.nextToken();
             if (fieldName.equals("metrics")) {
-                metrics = parseMetrics(parser);
+                parseMetrics(parser, consumer);
             } else {
                 if (token == JsonToken.START_OBJECT || token == JsonToken.START_ARRAY) {
                     parser.skipChildren();
                 }
             }
         }
-        return metrics;
     }
-
-    static private Metrics parseSnapshot(JsonParser parser) throws IOException {
+    private static long secondsSince1970UTC() {
+        return System.currentTimeMillis() / 1000L;
+    }
+    static private long parseSnapshot(JsonParser parser) throws IOException {
         if (parser.getCurrentToken() != JsonToken.START_OBJECT) {
             throw new IOException("Expected start of 'snapshot' object, got " + parser.currentToken());
         }
-        Metrics metrics = new Metrics();
+        long timestamp = secondsSince1970UTC();
         for (parser.nextToken(); parser.getCurrentToken() != JsonToken.END_OBJECT; parser.nextToken()) {
             String fieldName = parser.getCurrentName();
             JsonToken token = parser.nextToken();
             if (fieldName.equals("to")) {
-                long timestamp = parser.getLongValue();
+                timestamp = parser.getLongValue();
                 long now = System.currentTimeMillis() / 1000;
                 timestamp = Metric.adjustTime(timestamp, now);
-                metrics = new Metrics(timestamp);
             } else {
                 if (token == JsonToken.START_OBJECT || token == JsonToken.START_ARRAY) {
                     parser.skipChildren();
                 }
             }
         }
-        return metrics;
+        return timestamp;
     }
 
-    static private void parseValues(JsonParser parser, Metrics metrics) throws IOException {
+    static private void parseValues(JsonParser parser, long timestamp, Consumer consumer) throws IOException {
         if (parser.getCurrentToken() != JsonToken.START_ARRAY) {
             throw new IOException("Expected start of 'metrics:values' array, got " + parser.currentToken());
         }
@@ -87,34 +88,34 @@ public class MetricsParser {
             // read everything from this START_OBJECT to the matching END_OBJECT
             // and return it as a tree model ObjectNode
             JsonNode value = jsonMapper.readTree(parser);
-            handleValue(value, metrics.getTimeStamp(), metrics, uniqueDimensions);
+            handleValue(value, timestamp, consumer, uniqueDimensions);
 
             // do whatever you need to do with this object
         }
     }
 
-    static private Metrics parseMetrics(JsonParser parser) throws IOException {
+    static private void parseMetrics(JsonParser parser, Consumer consumer) throws IOException {
         if (parser.getCurrentToken() != JsonToken.START_OBJECT) {
             throw new IOException("Expected start of 'metrics' object, got " + parser.currentToken());
         }
-        Metrics metrics = new Metrics();
+        long timestamp = System.currentTimeMillis() / 1000L;
         for (parser.nextToken(); parser.getCurrentToken() != JsonToken.END_OBJECT; parser.nextToken()) {
             String fieldName = parser.getCurrentName();
             JsonToken token = parser.nextToken();
             if (fieldName.equals("snapshot")) {
-                metrics = parseSnapshot(parser);
+                timestamp = parseSnapshot(parser);
             } else if (fieldName.equals("values")) {
-                parseValues(parser, metrics);
+                parseValues(parser, timestamp, consumer);
             } else {
                 if (token == JsonToken.START_OBJECT || token == JsonToken.START_ARRAY) {
                     parser.skipChildren();
                 }
             }
         }
-        return metrics;
     }
 
-    static private void handleValue(JsonNode metric, long timestamp, Metrics metrics, Map<String, Map<DimensionId, String>> uniqueDimensions) {
+    static private void handleValue(JsonNode metric, long timestamp, Consumer consumer,
+                                    Map<String, Map<DimensionId, String>> uniqueDimensions) {
         String name = metric.get("name").textValue();
         String description = "";
 
@@ -155,7 +156,7 @@ public class MetricsParser {
                 throw new IllegalArgumentException("Value for aggregator '" + aggregator + "' is not a number");
             }
             String metricName = new StringBuilder().append(name).append(".").append(aggregator).toString();
-            metrics.add(new Metric(MetricId.toMetricId(metricName), value, timestamp, dim, description));
+            consumer.consume(new Metric(MetricId.toMetricId(metricName), value, timestamp, dim, description));
         }
     }
 }
diff --git a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/RemoteMetricsFetcher.java b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/RemoteMetricsFetcher.java
index f2cb5c4e8b3..8acaa0fb58e 100644
--- a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/RemoteMetricsFetcher.java
+++ b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/RemoteMetricsFetcher.java
@@ -23,32 +23,25 @@ public class RemoteMetricsFetcher extends HttpMetricFetcher {
     /**
      * Connect to remote service over http and fetch metrics
      */
-    public Metrics getMetrics(int fetchCount) {
+    public void getMetrics(MetricsParser.Consumer consumer, int fetchCount) {
         try {
-            return createMetrics(getJson(), fetchCount);
+            createMetrics(getJson(), consumer, fetchCount);
         } catch (IOException | InterruptedException | ExecutionException e) {
-            return new Metrics();
         }
     }
 
-    Metrics createMetrics(String data, int fetchCount) {
-        Metrics remoteMetrics = new Metrics();
+    void createMetrics(String data, MetricsParser.Consumer consumer, int fetchCount) {
         try {
-            remoteMetrics = MetricsParser.parse(data);
+            MetricsParser.parse(data, consumer);
         } catch (Exception e) {
             handleException(e, data, fetchCount);
         }
-
-        return remoteMetrics;
     }
-    Metrics createMetrics(InputStream data, int fetchCount) {
-        Metrics remoteMetrics = new Metrics();
+    private void createMetrics(InputStream data, MetricsParser.Consumer consumer, int fetchCount) {
         try {
-            remoteMetrics = MetricsParser.parse(data);
+            MetricsParser.parse(data, consumer);
         } catch (Exception e) {
             handleException(e, data, fetchCount);
         }
-
-        return remoteMetrics;
     }
 }
diff --git a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/VespaService.java b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/VespaService.java
index 9d165b2d5a9..1b37abde3c8 100644
--- a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/VespaService.java
+++ b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/service/VespaService.java
@@ -2,6 +2,7 @@
 package ai.vespa.metricsproxy.service;
 
 import ai.vespa.metricsproxy.metric.HealthMetric;
+import ai.vespa.metricsproxy.metric.Metric;
 import ai.vespa.metricsproxy.metric.Metrics;
 import ai.vespa.metricsproxy.metric.model.DimensionId;
 import ai.vespa.metricsproxy.metric.model.ServiceId;
@@ -137,10 +138,22 @@ public class VespaService implements Comparable<VespaService> {
      *
      * @return the non-system metrics
      */
-    public Metrics getMetrics() {
-        Metrics remoteMetrics = remoteMetricsFetcher.getMetrics(metricsFetchCount.get());
+    public void consumeMetrics(MetricsParser.Consumer consumer) {
+        remoteMetricsFetcher.getMetrics(consumer, metricsFetchCount.get());
         metricsFetchCount.getAndIncrement();
-        return remoteMetrics;
+    }
+
+    private static class CollectMetrics implements MetricsParser.Consumer {
+        private final Metrics metrics = new Metrics();
+        @Override
+        public void consume(Metric metric) {
+            metrics.add(metric);
+        }
+    }
+    public final Metrics getMetrics() {
+        CollectMetrics collector = new CollectMetrics();
+        consumeMetrics(collector);
+        return collector.metrics;
     }
 
     /**
diff --git a/metrics-proxy/src/test/java/ai/vespa/metricsproxy/service/DownService.java b/metrics-proxy/src/test/java/ai/vespa/metricsproxy/service/DownService.java
index d39cabcbf93..72277f54d88 100644
--- a/metrics-proxy/src/test/java/ai/vespa/metricsproxy/service/DownService.java
+++ b/metrics-proxy/src/test/java/ai/vespa/metricsproxy/service/DownService.java
@@ -18,8 +18,7 @@ public class DownService extends VespaService {
     }
 
     @Override
-    public Metrics getMetrics() {
-        return new Metrics();
+    public void consumeMetrics(MetricsParser.Consumer consumer) {
     }
 
    @Override
diff --git a/metrics-proxy/src/test/java/ai/vespa/metricsproxy/service/DummyService.java b/metrics-proxy/src/test/java/ai/vespa/metricsproxy/service/DummyService.java
index ccbb237ae37..28b4db12855 100644
--- a/metrics-proxy/src/test/java/ai/vespa/metricsproxy/service/DummyService.java
+++ b/metrics-proxy/src/test/java/ai/vespa/metricsproxy/service/DummyService.java
@@ -2,7 +2,6 @@
 package ai.vespa.metricsproxy.service;
 
 import ai.vespa.metricsproxy.metric.Metric;
-import ai.vespa.metricsproxy.metric.Metrics;
 import ai.vespa.metricsproxy.metric.model.MetricId;
 
 /**
@@ -21,14 +20,10 @@ public class DummyService extends VespaService {
     }
 
     @Override
-    public Metrics getMetrics() {
-        Metrics m = new Metrics();
-
+    public void consumeMetrics(MetricsParser.Consumer consumer) {
         long timestamp = System.currentTimeMillis() / 1000;
-        m.add(new Metric(MetricId.toMetricId(METRIC_1), 5 * num + 1, timestamp));
-        m.add(new Metric(MetricId.toMetricId(METRIC_2), 1.3 * num + 1.05, timestamp));
-
-        return m;
+        consumer.consume(new Metric(MetricId.toMetricId(METRIC_1), 5 * num + 1, timestamp));
+        consumer.consume(new Metric(MetricId.toMetricId(METRIC_2), 1.3 * num + 1.05, timestamp));
     }
 
 }
diff --git a/metrics-proxy/src/test/java/ai/vespa/metricsproxy/service/MetricsFetcherTest.java b/metrics-proxy/src/test/java/ai/vespa/metricsproxy/service/MetricsFetcherTest.java
index defcf4bfaaa..ed3e8ef3d56 100644
--- a/metrics-proxy/src/test/java/ai/vespa/metricsproxy/service/MetricsFetcherTest.java
+++ b/metrics-proxy/src/test/java/ai/vespa/metricsproxy/service/MetricsFetcherTest.java
@@ -2,6 +2,7 @@
 package ai.vespa.metricsproxy.service;
 
 import ai.vespa.metricsproxy.TestUtil;
+import ai.vespa.metricsproxy.metric.Metric;
 import ai.vespa.metricsproxy.metric.Metrics;
 import ai.vespa.metricsproxy.metric.model.MetricId;
 import org.junit.Test;
@@ -15,11 +16,24 @@ public class MetricsFetcherTest {
 
     private static int port = 9;  //port number is not used in this test
 
+    private class MetricsConsumer implements MetricsParser.Consumer {
+        Metrics metrics = new Metrics();
+        @Override
+        public void consume(Metric metric) {
+            metrics.add(metric);
+        }
+    }
+    Metrics fetch(String data) {
+        RemoteMetricsFetcher fetcher = new RemoteMetricsFetcher(new DummyService(0, "dummy/id/0"), port);
+        MetricsConsumer consumer = new MetricsConsumer();
+        fetcher.createMetrics(data, consumer, 0);
+        return consumer.metrics;
+    }
+
     @Test
     public void testStateFormatMetricsParse() {
         String jsonData = TestUtil.getFileContents("metrics-state.json");
-        RemoteMetricsFetcher fetcher = new RemoteMetricsFetcher(new DummyService(0, "dummy/id/0"), port);
-        Metrics metrics = fetcher.createMetrics(jsonData, 0);
+        Metrics metrics = fetch(jsonData);
         assertThat(metrics.size(), is(10));
         assertThat(metrics.getMetric(MetricId.toMetricId("query_hits.count")).getValue().intValue(), is(28));
         assertThat(metrics.getMetric(MetricId.toMetricId("queries.rate")).getValue().doubleValue(), is(0.4667));
@@ -29,8 +43,7 @@ public class MetricsFetcherTest {
     @Test
     public void testEmptyJson() {
         String  jsonData = "{}";
-        RemoteMetricsFetcher fetcher = new RemoteMetricsFetcher(new DummyService(0, "dummy/id/0"), port);
-        Metrics metrics = fetcher.createMetrics(jsonData, 0);
+        Metrics metrics = fetch(jsonData);
         assertThat("Wrong number of metrics", metrics.size(), is(0));
     }
 
@@ -39,10 +52,8 @@ public class MetricsFetcherTest {
         String jsonData;
         Metrics metrics;
 
-        RemoteMetricsFetcher fetcher = new RemoteMetricsFetcher(new DummyService(0, "dummy/id/0"), port);
-
         jsonData = "";
-        metrics = fetcher.createMetrics(jsonData, 0);
+        metrics = fetch(jsonData);
         assertThat("Wrong number of metrics", metrics.size(), is(0));
 
         jsonData = "{\n" +
@@ -51,7 +62,7 @@ public class MetricsFetcherTest {
                 "  \"message\" : \"Everything ok here\"\n" +
                 "}\n" +
                 "}";
-        metrics = fetcher.createMetrics(jsonData, 0);
+        metrics = fetch(jsonData);
         assertThat("Wrong number of metrics", metrics.size(), is(0));
 
         jsonData = "{\n" +
@@ -80,7 +91,7 @@ public class MetricsFetcherTest {
                 "}\n" +
                 "}";
 
-        metrics = fetcher.createMetrics(jsonData, 0);
+        metrics = fetch(jsonData);
         assertThat("Wrong number of metrics", metrics.size(), is(0));
     }
 }