Merge pull request #16400 from vespa-engine/bratseth/request-metrics-async

Request metrics async

8 files changed, 132 insertions, 70 deletions

diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsFetcher.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsFetcher.java
index 70ac915f792..ba833519b0c 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsFetcher.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsFetcher.java
@@ -6,6 +6,7 @@ import com.yahoo.config.provision.ApplicationId;
 
 import java.time.Instant;
 import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * Interface to retrieve metrics on (tenant) nodes.
@@ -15,11 +16,10 @@ import java.util.Collection;
 public interface MetricsFetcher {
 
     /**
-     * Fetches metrics for all hosts of an application. This call may be expensive.
+     * Fetches metrics asynchronously for all hosts of an application. This call may be expensive.
      *
      * @param application the application to fetch metrics from
-     * @return a metric snapshot for each hostname of this application
      */
-    Collection<Pair<String, MetricSnapshot>> fetchMetrics(ApplicationId application);
+    CompletableFuture<MetricsResponse> fetchMetrics(ApplicationId application);
 
 }
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsResponse.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsResponse.java
index 963ab85c5a0..341a2028f05 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsResponse.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsResponse.java
@@ -22,26 +22,32 @@ import java.util.Map;
 import java.util.Optional;
 
 /**
- * Consumes a response from the metrics/v2 API and populates the fields of this with the resulting values
+ * A response containing metrics for a collection of nodes.
  *
  * @author bratseth
  */
 public class MetricsResponse {
 
-    private final Collection<Pair<String, MetricSnapshot>> nodeMetrics = new ArrayList<>();
+    private final Collection<Pair<String, MetricSnapshot>> nodeMetrics;
 
+    /** Creates this from a metrics/V2 response */
     public MetricsResponse(String response, NodeList applicationNodes, NodeRepository nodeRepository) {
         this(SlimeUtils.jsonToSlime(response), applicationNodes, nodeRepository);
     }
 
-    public Collection<Pair<String, MetricSnapshot>> metrics() { return nodeMetrics; }
+    public MetricsResponse(Collection<Pair<String, MetricSnapshot>> metrics) {
+        this.nodeMetrics = metrics;
+    }
 
     private MetricsResponse(Slime response, NodeList applicationNodes, NodeRepository nodeRepository) {
+        nodeMetrics = new ArrayList<>();
         Inspector root = response.get();
         Inspector nodes = root.field("nodes");
         nodes.traverse((ArrayTraverser)(__, node) -> consumeNode(node, applicationNodes, nodeRepository));
     }
 
+    public Collection<Pair<String, MetricSnapshot>> metrics() { return nodeMetrics; }
+
     private void consumeNode(Inspector node, NodeList applicationNodes, NodeRepository nodeRepository) {
         String hostname = node.field("hostname").asString();
         consumeNodeMetrics(hostname, node.field("node"), applicationNodes, nodeRepository);
@@ -83,6 +89,8 @@ public class MetricsResponse {
         item.field("values").traverse((ObjectTraverser)(name, value) -> values.put(name, value.asDouble()));
     }
 
+    public static MetricsResponse empty() { return new MetricsResponse(List.of()); }
+
     /** The metrics this can read */
     private enum Metric {
 
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsV2MetricsFetcher.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsV2MetricsFetcher.java
index 4afc876056a..961c1393550 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsV2MetricsFetcher.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsV2MetricsFetcher.java
@@ -1,9 +1,8 @@
 // Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vespa.hosted.provision.autoscale;
 
-import ai.vespa.util.http.VespaHttpClientBuilder;
+import ai.vespa.util.http.VespaAsyncHttpClientBuilder;
 import com.google.inject.Inject;
-import com.yahoo.collections.Pair;
 import com.yahoo.component.AbstractComponent;
 import com.yahoo.config.provision.ApplicationId;
 import com.yahoo.vespa.applicationmodel.HostName;
@@ -12,15 +11,14 @@ import com.yahoo.vespa.hosted.provision.NodeList;
 import com.yahoo.vespa.hosted.provision.NodeRepository;
 import com.yahoo.vespa.orchestrator.HostNameNotFoundException;
 import com.yahoo.vespa.orchestrator.Orchestrator;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.impl.client.BasicResponseHandler;
-import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.hc.client5.http.async.methods.SimpleHttpRequest;
+import org.apache.hc.client5.http.async.methods.SimpleHttpResponse;
+import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
+import org.apache.hc.core5.concurrent.FutureCallback;
 
 import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.util.Collection;
-import java.util.Collections;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -37,32 +35,37 @@ public class MetricsV2MetricsFetcher extends AbstractComponent implements Metric
 
     private final NodeRepository nodeRepository;
     private final Orchestrator orchestrator;
-    private final HttpClient httpClient;
+    private final AsyncHttpClient httpClient;
 
     @Inject
     @SuppressWarnings("unused")
     public MetricsV2MetricsFetcher(NodeRepository nodeRepository, Orchestrator orchestrator) {
-        this(nodeRepository, orchestrator, new ApacheHttpClient());
+        this(nodeRepository, orchestrator, new AsyncApacheHttpClient());
     }
 
-    public MetricsV2MetricsFetcher(NodeRepository nodeRepository, Orchestrator orchestrator, HttpClient httpClient) {
+    public MetricsV2MetricsFetcher(NodeRepository nodeRepository, Orchestrator orchestrator, AsyncHttpClient httpClient) {
         this.nodeRepository = nodeRepository;
         this.orchestrator = orchestrator;
         this.httpClient = httpClient;
     }
 
     @Override
-    public Collection<Pair<String, MetricSnapshot>> fetchMetrics(ApplicationId application) {
+    public CompletableFuture<MetricsResponse> fetchMetrics(ApplicationId application) {
         NodeList applicationNodes = nodeRepository.list(application).state(Node.State.active);
 
         Optional<Node> metricsV2Container = applicationNodes.container()
                                                             .matching(node -> expectedUp(node))
                                                             .stream()
                                                             .findFirst();
-        if (metricsV2Container.isEmpty()) return Collections.emptyList();
-        // Consumer 'autoscaling' defined in com.yahoo.vespa.model.admin.monitoring.MetricConsumer
-        String url = "http://" + metricsV2Container.get().hostname() + ":" + 4080 + apiPath + "?consumer=autoscaling";
-        return new MetricsResponse(httpClient.get(url), applicationNodes, nodeRepository).metrics();
+        if (metricsV2Container.isEmpty()) {
+            return CompletableFuture.completedFuture(MetricsResponse.empty());
+        }
+        else {
+            // Consumer 'autoscaling' defined in com.yahoo.vespa.model.admin.monitoring.MetricConsumer
+            String url = "http://" + metricsV2Container.get().hostname() + ":" + 4080 + apiPath + "?consumer=autoscaling";
+            return httpClient.get(url)
+                             .thenApply(response -> new MetricsResponse(response, applicationNodes, nodeRepository));
+        }
     }
 
     @Override
@@ -79,27 +82,28 @@ public class MetricsV2MetricsFetcher extends AbstractComponent implements Metric
         }
     }
 
-    /** The simplest possible http client interface */
-    public interface HttpClient {
+    /** A simple async HTTP client */
+    public interface AsyncHttpClient {
 
-        String get(String url);
+        CompletableFuture<String> get(String url);
         void close();
 
     }
 
-    /** Implements the HttpClient interface by delegating to an Apache HTTP client */
-    public static class ApacheHttpClient implements HttpClient {
+    /** Implements the AsyncHttpClient interface by delegating to an Apache HTTP client */
+    public static class AsyncApacheHttpClient implements AsyncHttpClient {
 
-        private final CloseableHttpClient httpClient = VespaHttpClientBuilder.createWithBasicConnectionManager().build();
+        private final CloseableHttpAsyncClient httpClient = VespaAsyncHttpClientBuilder.create().build();
+
+        public AsyncApacheHttpClient() {
+            httpClient.start();
+        }
 
         @Override
-        public String get(String url) {
-            try {
-                return httpClient.execute(new HttpGet(url), new BasicResponseHandler());
-            }
-            catch (IOException e) {
-                throw new UncheckedIOException("Could not get " + url, e);
-            }
+        public CompletableFuture<String> get(String url) {
+            CompletableFuture<String> callback = new CompletableFuture<>();
+            httpClient.execute(new SimpleHttpRequest("GET", url), new CallbackAdaptor(callback));
+            return callback;
         }
 
         @Override
@@ -112,6 +116,31 @@ public class MetricsV2MetricsFetcher extends AbstractComponent implements Metric
             }
         }
 
+        private static class CallbackAdaptor implements FutureCallback<SimpleHttpResponse> {
+
+            private final CompletableFuture<String> callback;
+
+            public CallbackAdaptor(CompletableFuture<String> callback) {
+                this.callback = callback;
+            }
+
+            @Override
+            public void completed(SimpleHttpResponse simpleHttpResponse) {
+                callback.complete(simpleHttpResponse.getBodyText());
+            }
+
+            @Override
+            public void failed(Exception e) {
+                callback.completeExceptionally(e);
+            }
+
+            @Override
+            public void cancelled() {
+                callback.cancel(true);
+            }
+
+        }
+
     }
 
 }
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/NodeMetricsDbMaintainer.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/NodeMetricsDbMaintainer.java
index 017e1264f1c..b8548c4c3f4 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/NodeMetricsDbMaintainer.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/maintenance/NodeMetricsDbMaintainer.java
@@ -3,12 +3,15 @@ package com.yahoo.vespa.hosted.provision.maintenance;
 
 import com.yahoo.config.provision.ApplicationId;
 import com.yahoo.jdisc.Metric;
+import com.yahoo.lang.MutableInteger;
 import com.yahoo.vespa.hosted.provision.NodeRepository;
 import com.yahoo.vespa.hosted.provision.autoscale.MetricsFetcher;
 import com.yahoo.vespa.hosted.provision.autoscale.MetricsDb;
+import com.yahoo.vespa.hosted.provision.autoscale.MetricsResponse;
 import com.yahoo.yolean.Exceptions;
 
 import java.time.Duration;
+import java.util.Set;
 import java.util.logging.Level;
 
 /**
@@ -35,23 +38,44 @@ public class NodeMetricsDbMaintainer extends NodeRepositoryMaintainer {
 
     @Override
     protected boolean maintain() {
-        int warnings = 0;
-        for (ApplicationId application : activeNodesByApplication().keySet()) {
-            try {
-                metricsDb.add(metricsFetcher.fetchMetrics(application));
-            }
-            catch (Exception e) {
-                // TODO: Don't warn if this only happens occasionally
-                if (warnings++ < maxWarningsPerInvocation)
-                    log.log(Level.WARNING, "Could not update metrics for " + application + ": " + Exceptions.toMessageString(e), e);
+        try {
+            var warnings = new MutableInteger(0);
+            Set<ApplicationId> applications = activeNodesByApplication().keySet();
+            if (applications.isEmpty()) return true;
+
+            long pauseMs = interval().toMillis() / applications.size() - 1; // spread requests over interval
+            int done = 0;
+            for (ApplicationId application : applications) {
+                metricsFetcher.fetchMetrics(application)
+                              .whenComplete((metricsResponse, exception) -> handleResponse(metricsResponse,
+                                                                                           exception,
+                                                                                           warnings,
+                                                                                           application));
+                if (++done < applications.size())
+                    Thread.sleep(pauseMs);
             }
-        }
-        metricsDb.gc();
+            metricsDb.gc();
 
-        // Suppress failures for manual zones for now to avoid noise
-        if (nodeRepository().zone().environment().isManuallyDeployed()) return true;
+            // Suppress failures for manual zones for now to avoid noise
+            return nodeRepository().zone().environment().isManuallyDeployed() || warnings.get() == 0;
+        }
+        catch (InterruptedException e) {
+            return false;
+        }
+    }
 
-        return warnings == 0;
+    private void handleResponse(MetricsResponse response,
+                                Throwable exception,
+                                MutableInteger warnings,
+                                ApplicationId application) {
+        if (exception != null) {
+            if (warnings.get() < maxWarningsPerInvocation)
+                log.log(Level.WARNING, "Could not update metrics for " + application, exception);
+            warnings.add(1);
+        }
+        else if (response != null) {
+            metricsDb.add(response.metrics());
+        }
     }
 
 }
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/testutils/MockMetricsFetcher.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/testutils/MockMetricsFetcher.java
index ed2d7eed9e4..da5fa67425f 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/testutils/MockMetricsFetcher.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/testutils/MockMetricsFetcher.java
@@ -1,13 +1,11 @@
 // Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vespa.hosted.provision.testutils;
 
-import com.yahoo.collections.Pair;
 import com.yahoo.config.provision.ApplicationId;
-import com.yahoo.vespa.hosted.provision.autoscale.MetricSnapshot;
 import com.yahoo.vespa.hosted.provision.autoscale.MetricsFetcher;
+import com.yahoo.vespa.hosted.provision.autoscale.MetricsResponse;
 
-import java.util.ArrayList;
-import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * @author bratseth
@@ -15,8 +13,8 @@ import java.util.Collection;
 public class MockMetricsFetcher implements MetricsFetcher {
 
     @Override
-    public Collection<Pair<String, MetricSnapshot>>  fetchMetrics(ApplicationId application) {
-        return new ArrayList<>();
+    public CompletableFuture<MetricsResponse> fetchMetrics(ApplicationId application) {
+        return CompletableFuture.completedFuture(MetricsResponse.empty());
     }
 
 }
diff --git a/node-repository/src/test/java/com/yahoo/vespa/hosted/provision/autoscale/AutoscalingIntegrationTest.java b/node-repository/src/test/java/com/yahoo/vespa/hosted/provision/autoscale/AutoscalingIntegrationTest.java
index 0d423333ce1..8ef2fd72d08 100644
--- a/node-repository/src/test/java/com/yahoo/vespa/hosted/provision/autoscale/AutoscalingIntegrationTest.java
+++ b/node-repository/src/test/java/com/yahoo/vespa/hosted/provision/autoscale/AutoscalingIntegrationTest.java
@@ -14,6 +14,7 @@ import org.junit.Test;
 
 import java.time.Duration;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 
 import static org.junit.Assert.assertEquals;
@@ -45,7 +46,7 @@ public class AutoscalingIntegrationTest {
 
         for (int i = 0; i < 1000; i++) {
             tester.clock().advance(Duration.ofSeconds(10));
-            tester.nodeMetricsDb().add(fetcher.fetchMetrics(application1));
+            fetcher.fetchMetrics(application1).whenComplete((r, e) -> tester.nodeMetricsDb().add(r.metrics()));
             tester.clock().advance(Duration.ofSeconds(10));
             tester.nodeMetricsDb().gc();
         }
@@ -63,7 +64,7 @@ public class AutoscalingIntegrationTest {
         assertTrue(scaledResources.isPresent());
     }
 
-    private static class MockHttpClient implements MetricsV2MetricsFetcher.HttpClient {
+    private static class MockHttpClient implements MetricsV2MetricsFetcher.AsyncHttpClient {
 
         private final ManualClock clock;
 
@@ -116,7 +117,9 @@ public class AutoscalingIntegrationTest {
                 "}\n";
 
         @Override
-        public String get(String url) { return cannedResponse.replace("[now]", String.valueOf(clock.millis())); }
+        public CompletableFuture<String> get(String url) {
+            return CompletableFuture.completedFuture(cannedResponse.replace("[now]",
+                                                                            String.valueOf(clock.millis()))); }
 
         @Override
         public void close() { }
diff --git a/node-repository/src/test/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsV2MetricsFetcherTest.java b/node-repository/src/test/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsV2MetricsFetcherTest.java
index cc0450ec2ea..14626a40070 100644
--- a/node-repository/src/test/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsV2MetricsFetcherTest.java
+++ b/node-repository/src/test/java/com/yahoo/vespa/hosted/provision/autoscale/MetricsV2MetricsFetcherTest.java
@@ -15,6 +15,7 @@ import org.junit.Test;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -25,7 +26,7 @@ public class MetricsV2MetricsFetcherTest {
     private static final double delta = 0.00000001;
 
     @Test
-    public void testMetricsFetch() {
+    public void testMetricsFetch() throws Exception {
         NodeResources resources = new NodeResources(1, 10, 100, 1);
         ProvisioningTester tester = new ProvisioningTester.Builder().build();
         OrchestratorMock orchestrator = new OrchestratorMock();
@@ -44,7 +45,7 @@ public class MetricsV2MetricsFetcherTest {
 
         {
             httpClient.cannedResponse = cannedResponseForApplication1;
-            List<Pair<String, MetricSnapshot>> values = new ArrayList<>(fetcher.fetchMetrics(application1));
+            List<Pair<String, MetricSnapshot>> values = new ArrayList<>(fetcher.fetchMetrics(application1).get().metrics());
             assertEquals("http://host-1.yahoo.com:4080/metrics/v2/values?consumer=autoscaling",
                          httpClient.requestsReceived.get(0));
             assertEquals(2, values.size());
@@ -62,7 +63,7 @@ public class MetricsV2MetricsFetcherTest {
 
         {
             httpClient.cannedResponse = cannedResponseForApplication2;
-            List<Pair<String, MetricSnapshot>> values = new ArrayList<>(fetcher.fetchMetrics(application2));
+            List<Pair<String, MetricSnapshot>> values = new ArrayList<>(fetcher.fetchMetrics(application2).get().metrics());
             assertEquals("http://host-3.yahoo.com:4080/metrics/v2/values?consumer=autoscaling",
                          httpClient.requestsReceived.get(1));
             assertEquals(1, values.size());
@@ -80,21 +81,21 @@ public class MetricsV2MetricsFetcherTest {
                 tester.nodeRepository().write(tester.nodeRepository().getNodes(application2, Node.State.active)
                         .get(0).retire(tester.clock().instant()), lock);
             }
-            List<Pair<String, MetricSnapshot>> values = new ArrayList<>(fetcher.fetchMetrics(application2));
+            List<Pair<String, MetricSnapshot>> values = new ArrayList<>(fetcher.fetchMetrics(application2).get().metrics());
             assertFalse(values.get(0).getSecond().stable());
         }
     }
 
-    private static class MockHttpClient implements MetricsV2MetricsFetcher.HttpClient {
+    private static class MockHttpClient implements MetricsV2MetricsFetcher.AsyncHttpClient {
 
         List<String> requestsReceived = new ArrayList<>();
 
         String cannedResponse = null;
 
         @Override
-        public String get(String url) {
+        public CompletableFuture<String> get(String url) {
             requestsReceived.add(url);
-            return cannedResponse;
+            return CompletableFuture.completedFuture(cannedResponse);
         }
 
         @Override
diff --git a/node-repository/src/test/java/com/yahoo/vespa/hosted/provision/maintenance/NodeMetricsDbMaintainerTest.java b/node-repository/src/test/java/com/yahoo/vespa/hosted/provision/maintenance/NodeMetricsDbMaintainerTest.java
index 4f0b0d55742..0a12a255f30 100644
--- a/node-repository/src/test/java/com/yahoo/vespa/hosted/provision/maintenance/NodeMetricsDbMaintainerTest.java
+++ b/node-repository/src/test/java/com/yahoo/vespa/hosted/provision/maintenance/NodeMetricsDbMaintainerTest.java
@@ -16,6 +16,7 @@ import java.time.Duration;
 import java.time.Instant;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 
 import static org.junit.Assert.assertEquals;
@@ -23,7 +24,6 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 /**
- * @author bratseth
  */
 public class NodeMetricsDbMaintainerTest {
 
@@ -56,7 +56,7 @@ public class NodeMetricsDbMaintainerTest {
         assertTrue(allSnapshots.stream().anyMatch(snapshot -> ! snapshot.inService()));
     }
 
-    private static class MockHttpClient implements MetricsV2MetricsFetcher.HttpClient {
+    private static class MockHttpClient implements MetricsV2MetricsFetcher.AsyncHttpClient {
 
         final String cannedResponse =
                 "{\n" +
@@ -107,8 +107,8 @@ public class NodeMetricsDbMaintainerTest {
                 "}\n";
 
         @Override
-        public String get(String url) {
-            return cannedResponse;
+        public CompletableFuture<String> get(String url) {
+            return CompletableFuture.completedFuture(cannedResponse);
         }
 
         @Override
@@ -116,5 +116,4 @@ public class NodeMetricsDbMaintainerTest {
 
     }
 
-
 }