Merge pull request #13116 from vespa-engine/gjoranv/concurrency-bugfix

Gjoranv/concurrency bugfix

5 files changed, 34 insertions, 45 deletions

diff --git a/container-core/src/main/java/com/yahoo/container/handler/metrics/ErrorResponse.java b/container-core/src/main/java/com/yahoo/container/handler/metrics/ErrorResponse.java
index 321f7b3994a..1fcde746878 100644
--- a/container-core/src/main/java/com/yahoo/container/handler/metrics/ErrorResponse.java
+++ b/container-core/src/main/java/com/yahoo/container/handler/metrics/ErrorResponse.java
@@ -18,7 +18,7 @@ public class ErrorResponse extends JsonResponse {
     private static ObjectMapper objectMapper = new ObjectMapper();
 
     public ErrorResponse(int code, String message) {
-        super(code, asErrorJson(message));
+        super(code, asErrorJson(message != null ? message : "<null>"));
     }
 
     static String asErrorJson(String message) {
diff --git a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/application/ApplicationMetricsHandler.java b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/application/ApplicationMetricsHandler.java
index d9303e80dcd..c3231daab5f 100644
--- a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/application/ApplicationMetricsHandler.java
+++ b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/application/ApplicationMetricsHandler.java
@@ -4,8 +4,6 @@ package ai.vespa.metricsproxy.http.application;
 
 import ai.vespa.metricsproxy.core.MetricsConsumers;
 import ai.vespa.metricsproxy.metric.model.ConsumerId;
-import ai.vespa.metricsproxy.metric.model.MetricsPacket;
-import ai.vespa.metricsproxy.metric.model.processing.MetricsProcessor;
 import com.google.inject.Inject;
 import com.yahoo.container.handler.metrics.ErrorResponse;
 import com.yahoo.container.handler.metrics.HttpHandlerBase;
@@ -14,19 +12,15 @@ import com.yahoo.container.jdisc.HttpResponse;
 import com.yahoo.restapi.Path;
 
 import java.net.URI;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.Executor;
 import java.util.logging.Level;
 
 import static ai.vespa.metricsproxy.http.ValuesFetcher.getConsumerOrDefault;
 import static ai.vespa.metricsproxy.metric.model.json.GenericJsonUtil.toGenericApplicationModel;
-import static ai.vespa.metricsproxy.metric.model.processing.MetricsProcessor.applyProcessors;
 import static com.yahoo.jdisc.Response.Status.INTERNAL_SERVER_ERROR;
 import static com.yahoo.jdisc.Response.Status.OK;
-import static java.util.stream.Collectors.toList;
 
 /**
  * Http handler that returns metrics for all nodes in the Vespa application.
@@ -38,8 +32,6 @@ public class ApplicationMetricsHandler extends HttpHandlerBase {
     public static final String V1_PATH = "/applicationmetrics/v1";
     public static final String VALUES_PATH = V1_PATH + "/values";
 
-    private static final int MAX_DIMENSIONS = 10;
-
     private final ApplicationMetricsRetriever metricsRetriever;
     private final MetricsConsumers metricsConsumers;
 
@@ -62,32 +54,14 @@ public class ApplicationMetricsHandler extends HttpHandlerBase {
     private JsonResponse applicationMetricsResponse(String requestedConsumer) {
         try {
             ConsumerId consumer = getConsumerOrDefault(requestedConsumer, metricsConsumers);
-            var buildersByNode =  metricsRetriever.getMetrics(consumer);
-            var metricsByNode = processAndBuild(buildersByNode,
-                                                new ServiceIdDimensionProcessor(),
-                                                new ClusterIdDimensionProcessor(),
-                                                new PublicDimensionsProcessor(MAX_DIMENSIONS));
+            var metricsByNode =  metricsRetriever.getMetrics(consumer);
 
             return new JsonResponse(OK, toGenericApplicationModel(metricsByNode).serialize());
+
         } catch (Exception e) {
             log.log(Level.WARNING, "Got exception when retrieving metrics:", e);
             return new ErrorResponse(INTERNAL_SERVER_ERROR, e.getMessage());
         }
     }
 
-    private static Map<Node, List<MetricsPacket>> processAndBuild(Map<Node, List<MetricsPacket.Builder>> buildersByNode,
-                                                                  MetricsProcessor... processors) {
-        var metricsByNode = new HashMap<Node, List<MetricsPacket>>();
-
-        buildersByNode.forEach((node, builders) -> {
-            var metrics = builders.stream()
-                    .map(builder -> applyProcessors(builder, processors))
-                    .map(MetricsPacket.Builder::build)
-                    .collect(toList());
-
-            metricsByNode.put(node, metrics);
-        });
-        return metricsByNode;
-    }
-
 }
diff --git a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/application/ApplicationMetricsRetriever.java b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/application/ApplicationMetricsRetriever.java
index 40011a0dc72..5c7e64c4ed1 100644
--- a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/application/ApplicationMetricsRetriever.java
+++ b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/application/ApplicationMetricsRetriever.java
@@ -65,11 +65,11 @@ public class ApplicationMetricsRetriever extends AbstractComponent {
         super.deconstruct();
     }
 
-    public Map<Node, List<MetricsPacket.Builder>> getMetrics() {
+    public Map<Node, List<MetricsPacket>> getMetrics() {
         return getMetrics(DEFAULT_PUBLIC_CONSUMER_ID);
     }
 
-    public Map<Node, List<MetricsPacket.Builder>> getMetrics(ConsumerId consumer) {
+    public Map<Node, List<MetricsPacket>> getMetrics(ConsumerId consumer) {
         log.log(Level.FINE, () -> "Retrieving metrics from " + clients.size() + " nodes.");
         var forkJoinTask = forkJoinPool.submit(() -> clients.parallelStream()
                 .map(client -> getNodeMetrics(client, consumer))
@@ -87,7 +87,7 @@ public class ApplicationMetricsRetriever extends AbstractComponent {
         }
     }
 
-    private Map.Entry<Node, List<MetricsPacket.Builder>> getNodeMetrics(NodeMetricsClient client, ConsumerId consumer) {
+    private Map.Entry<Node, List<MetricsPacket>> getNodeMetrics(NodeMetricsClient client, ConsumerId consumer) {
         try {
             return new AbstractMap.SimpleEntry<>(client.node, client.getMetrics(consumer));
         } catch (Exception e) {
diff --git a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/application/NodeMetricsClient.java b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/application/NodeMetricsClient.java
index a93b5eb31c0..78bc3e96322 100644
--- a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/application/NodeMetricsClient.java
+++ b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/application/NodeMetricsClient.java
@@ -4,6 +4,7 @@ package ai.vespa.metricsproxy.http.application;
 import ai.vespa.metricsproxy.metric.model.ConsumerId;
 import ai.vespa.metricsproxy.metric.model.MetricsPacket;
 import ai.vespa.metricsproxy.metric.model.json.GenericJsonUtil;
+import ai.vespa.metricsproxy.metric.model.processing.MetricsProcessor;
 import com.yahoo.yolean.Exceptions;
 import org.apache.http.client.HttpClient;
 import org.apache.http.client.methods.HttpGet;
@@ -13,15 +14,15 @@ import java.io.IOException;
 import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
-import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.logging.Logger;
 
-import static java.util.logging.Level.FINE;
+import static ai.vespa.metricsproxy.metric.model.processing.MetricsProcessor.applyProcessors;
 import static java.util.Collections.emptyList;
+import static java.util.logging.Level.FINE;
+import static java.util.stream.Collectors.toList;
 
 /**
  * Retrieves metrics from a single Vespa node over http. To avoid unnecessary load on metrics
@@ -38,6 +39,7 @@ public class NodeMetricsClient {
     private static final Logger log = Logger.getLogger(NodeMetricsClient.class.getName());
 
     static final Duration METRICS_TTL = Duration.ofSeconds(30);
+    private static final int MAX_DIMENSIONS = 10;
 
     final Node node;
     private final HttpClient httpClient;
@@ -52,7 +54,7 @@ public class NodeMetricsClient {
         this.clock = clock;
     }
 
-    public List<MetricsPacket.Builder> getMetrics(ConsumerId consumer) {
+    public List<MetricsPacket> getMetrics(ConsumerId consumer) {
         var currentSnapshot = snapshots.get(consumer);
         if (currentSnapshot == null || currentSnapshot.isStale(clock) || currentSnapshot.metrics.isEmpty()) {
             Snapshot snapshot = retrieveMetrics(consumer);
@@ -69,16 +71,29 @@ public class NodeMetricsClient {
 
         try {
             String metricsJson = httpClient.execute(new HttpGet(metricsUri), new BasicResponseHandler());
-            var newMetrics = GenericJsonUtil.toMetricsPackets(metricsJson);
+            var metricsBuilders = GenericJsonUtil.toMetricsPackets(metricsJson);
+            var metrics = processAndBuild(metricsBuilders,
+                                          new ServiceIdDimensionProcessor(),
+                                          new ClusterIdDimensionProcessor(),
+                                          new PublicDimensionsProcessor(MAX_DIMENSIONS));
             snapshotsRetrieved ++;
-            log.log(FINE, () -> "Successfully retrieved " + newMetrics.size() + " metrics packets from " + metricsUri);
-            return new Snapshot(Instant.now(clock), newMetrics);
+            log.log(FINE, () -> "Successfully retrieved " + metrics.size() + " metrics packets from " + metricsUri);
+
+            return new Snapshot(Instant.now(clock), metrics);
         } catch (IOException e) {
             log.warning("Unable to retrieve metrics from " + metricsUri + ": " + Exceptions.toMessageString(e));
             return new Snapshot(Instant.now(clock), emptyList());
         }
     }
 
+    private static List<MetricsPacket> processAndBuild(List<MetricsPacket.Builder> builders,
+                                                       MetricsProcessor... processors) {
+        return builders.stream()
+                    .map(builder -> applyProcessors(builder, processors))
+                    .map(MetricsPacket.Builder::build)
+                    .collect(toList());
+    }
+
     long snapshotsRetrieved() {
         return snapshotsRetrieved;
     }
@@ -89,9 +104,9 @@ public class NodeMetricsClient {
     static class Snapshot {
 
         final Instant timestamp;
-        final List<MetricsPacket.Builder> metrics;
+        final List<MetricsPacket> metrics;
 
-        Snapshot(Instant timestamp, List<MetricsPacket.Builder> metrics) {
+        Snapshot(Instant timestamp, List<MetricsPacket> metrics) {
             this.timestamp = timestamp;
             this.metrics = metrics;
         }
diff --git a/metrics-proxy/src/test/java/ai/vespa/metricsproxy/http/application/NodeMetricsClientTest.java b/metrics-proxy/src/test/java/ai/vespa/metricsproxy/http/application/NodeMetricsClientTest.java
index c3e34920163..d8443ece8e8 100644
--- a/metrics-proxy/src/test/java/ai/vespa/metricsproxy/http/application/NodeMetricsClientTest.java
+++ b/metrics-proxy/src/test/java/ai/vespa/metricsproxy/http/application/NodeMetricsClientTest.java
@@ -85,7 +85,7 @@ public class NodeMetricsClientTest {
 
     @Test
     public void metrics_are_retrieved_upon_first_request() {
-        List<MetricsPacket.Builder> metrics = nodeMetricsClient.getMetrics(DEFAULT_PUBLIC_CONSUMER_ID);
+        List<MetricsPacket> metrics = nodeMetricsClient.getMetrics(DEFAULT_PUBLIC_CONSUMER_ID);
         assertEquals(1, nodeMetricsClient.snapshotsRetrieved());
         assertEquals(4, metrics.size());
    }
@@ -112,15 +112,15 @@ public class NodeMetricsClientTest {
 
     @Test
     public void metrics_for_different_consumers_are_cached_separately() {
-        List<MetricsPacket.Builder> defaultMetrics = nodeMetricsClient.getMetrics(DEFAULT_PUBLIC_CONSUMER_ID);
+        List<MetricsPacket> defaultMetrics = nodeMetricsClient.getMetrics(DEFAULT_PUBLIC_CONSUMER_ID);
         assertEquals(1, nodeMetricsClient.snapshotsRetrieved());
         assertEquals(4, defaultMetrics.size());
 
-        List<MetricsPacket.Builder> customMetrics = nodeMetricsClient.getMetrics(toConsumerId(CUSTOM_CONSUMER));
+        List<MetricsPacket> customMetrics = nodeMetricsClient.getMetrics(toConsumerId(CUSTOM_CONSUMER));
         assertEquals(2, nodeMetricsClient.snapshotsRetrieved());
         assertEquals(4, customMetrics.size());
 
-        MetricsPacket replacedCpuMetric = customMetrics.get(0).build();
+        MetricsPacket replacedCpuMetric = customMetrics.get(0);
         assertTrue(replacedCpuMetric.metrics().containsKey(toMetricId(REPLACED_CPU_METRIC)));
     }
 }