Add http client to retrieve metrics from other nodes.

4 files changed, 215 insertions, 1 deletions

diff --git a/metrics-proxy/pom.xml b/metrics-proxy/pom.xml
index e752d8a4803..f72ad75c6af 100644
--- a/metrics-proxy/pom.xml
+++ b/metrics-proxy/pom.xml
@@ -135,6 +135,18 @@
 
     <!-- test scope -->
 
+
+    <dependency>
+      <groupId>com.github.tomakehurst</groupId>
+      <artifactId>wiremock-standalone</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.yahoo.vespa</groupId>
+      <artifactId>testutil</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
diff --git a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/MetricsHandler.java b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/MetricsHandler.java
index 63a79d93b07..86f1260207c 100644
--- a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/MetricsHandler.java
+++ b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/MetricsHandler.java
@@ -30,7 +30,7 @@ import static com.yahoo.jdisc.Response.Status.OK;
 public class MetricsHandler extends HttpHandlerBase {
 
     public static final String V1_PATH = "/metrics/v1";
-    static final String VALUES_PATH = V1_PATH + "/values";
+    public static final String VALUES_PATH = V1_PATH + "/values";
 
     private final ValuesFetcher valuesFetcher;
 
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
new file mode 100644
index 00000000000..28130380c13
--- /dev/null
+++ b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/application/NodeMetricsClient.java
@@ -0,0 +1,99 @@
+/*
+ * Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+ */
+
+package ai.vespa.metricsproxy.http.application;
+
+import ai.vespa.metricsproxy.http.MetricsHandler;
+import ai.vespa.metricsproxy.metric.model.MetricsPacket;
+import ai.vespa.metricsproxy.metric.model.json.GenericJsonUtil;
+import com.yahoo.yolean.Exceptions;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.BasicResponseHandler;
+
+import java.io.IOException;
+import java.net.URI;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.logging.Logger;
+
+import static com.yahoo.log.LogLevel.DEBUG;
+import static java.util.Collections.emptyList;
+
+/**
+ * This class is used to retrieve metrics from a single Vespa node over http.
+ * Keeps and serves a snapshot of the node's metrics, with a fixed TTL, to
+ * avoid unnecessary load on metrics proxies.
+ *
+ * @author gjoranv
+ */
+public class NodeMetricsClient {
+    private static final Logger log = Logger.getLogger(NodeMetricsClient.class.getName());
+
+    static final Duration METRICS_TTL = Duration.ofSeconds(30);
+
+    private final HttpClient httpClient;
+    private final Node node;
+    private final Clock clock;
+
+    private List<MetricsPacket.Builder> metrics = emptyList();
+    private Instant metricsTimestamp = Instant.EPOCH;
+    private long snapshotsRetrieved = 0;
+
+    public NodeMetricsClient(HttpClient httpClient, Node node, Clock clock) {
+        this.httpClient = httpClient;
+        this.node = node;
+        this.clock = clock;
+    }
+
+    public List<MetricsPacket.Builder> getMetrics() {
+        if (Instant.now(clock).isAfter(metricsTimestamp.plus(METRICS_TTL))) {
+            retrieveMetrics();
+        }
+        return metrics;
+    }
+
+    private void retrieveMetrics() {
+        log.log(DEBUG, () -> "Retrieving metrics from host " + node.metricsUri);
+
+        try {
+            String metricsJson = httpClient.execute(new HttpGet(node.metricsUri), new BasicResponseHandler());
+            metrics = GenericJsonUtil.toMetricsPackets(metricsJson);
+            metricsTimestamp = Instant.now(clock);
+            snapshotsRetrieved ++;
+            log.log(DEBUG, () -> "Successfully retrieved " + metrics.size() + " metrics packets from " + node.metricsUri);
+
+        } catch (IOException e) {
+            log.warning("Unable to retrieve metrics from " + node.metricsUri + ": " + Exceptions.toMessageString(e));
+            metrics = emptyList();
+        }
+    }
+
+    long snapshotsRetrieved() {
+        return snapshotsRetrieved;
+    }
+
+    // TODO: move to separate file
+    static class Node {
+        final String configId;
+        final String host;
+        final int port;
+        final URI metricsUri;
+
+        public Node(String configId, String host, int port) {
+            this.configId = configId;
+            this.host = host;
+            this.port = port;
+            metricsUri = getMetricsUri(host, port);
+        }
+
+        private static URI getMetricsUri(String host, int port) {
+          return URI.create("http://" + host + ":" + port + MetricsHandler.VALUES_PATH);
+        }
+
+    }
+
+}
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
new file mode 100644
index 00000000000..b0852776a89
--- /dev/null
+++ b/metrics-proxy/src/test/java/ai/vespa/metricsproxy/http/application/NodeMetricsClientTest.java
@@ -0,0 +1,103 @@
+/*
+ * Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+ */
+
+package ai.vespa.metricsproxy.http.application;
+
+import ai.vespa.metricsproxy.http.application.NodeMetricsClient.Node;
+import ai.vespa.metricsproxy.metric.model.MetricsPacket;
+import com.github.tomakehurst.wiremock.junit.WireMockRule;
+import com.yahoo.test.ManualClock;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.List;
+
+import static ai.vespa.metricsproxy.TestUtil.getFileContents;
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
+import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.options;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * @author gjoranv
+ */
+public class NodeMetricsClientTest {
+
+    private static final String TEST_FILE = "generic-sample.json";
+    private static final String RESPONSE = getFileContents(TEST_FILE);
+    private static final int PORT = getAvailablePort();
+
+    private static final CloseableHttpClient httpClient = HttpClients.createDefault();
+
+    private final Node node = new Node("id", "localhost", PORT);
+    private ManualClock clock;
+    private NodeMetricsClient nodeMetricsClient;
+
+    @Before
+    public void setup() {
+        clock = new ManualClock();
+        nodeMetricsClient = new NodeMetricsClient(httpClient, node, clock);
+    }
+
+    @Rule
+    public WireMockRule wireMockRule = new WireMockRule(options().port(PORT));
+
+    @Test
+    public void metrics_are_not_retrieved_until_first_request() {
+        assertEquals(0, nodeMetricsClient.snapshotsRetrieved());
+    }
+
+    @Test
+    public void metrics_are_retrieved_upon_first_request() {
+        stubFor(get(urlEqualTo(node.metricsUri.getPath()))
+                        .willReturn(aResponse().withBody(RESPONSE)));
+
+        List<MetricsPacket.Builder> metrics = nodeMetricsClient.getMetrics();
+        assertEquals(1, nodeMetricsClient.snapshotsRetrieved());
+        assertEquals(4, metrics.size());
+   }
+
+    @Test
+    public void cached_metrics_are_used_when_ttl_has_not_expired() {
+        stubFor(get(urlEqualTo(node.metricsUri.getPath()))
+                        .willReturn(aResponse().withBody(RESPONSE)));
+
+        nodeMetricsClient.getMetrics();
+        assertEquals(1, nodeMetricsClient.snapshotsRetrieved());
+
+        clock.advance(NodeMetricsClient.METRICS_TTL.minusMillis(1));
+        nodeMetricsClient.getMetrics();
+        assertEquals(1, nodeMetricsClient.snapshotsRetrieved());
+    }
+
+    @Test
+    public void metrics_are_refreshed_when_ttl_has_expired() {
+        stubFor(get(urlEqualTo(node.metricsUri.getPath()))
+                        .willReturn(aResponse().withBody(RESPONSE)));
+
+        nodeMetricsClient.getMetrics();
+        assertEquals(1, nodeMetricsClient.snapshotsRetrieved());
+
+        clock.advance(NodeMetricsClient.METRICS_TTL.plusMillis(1));
+        nodeMetricsClient.getMetrics();
+        assertEquals(2, nodeMetricsClient.snapshotsRetrieved());
+    }
+
+    private static int getAvailablePort() {
+        try (ServerSocket socket = new ServerSocket(0)) {
+            socket.setReuseAddress(true);
+            return socket.getLocalPort();
+        } catch (IOException e) {
+            throw new RuntimeException("Could not find available port: ", e);
+        }
+    }
+
+}