@Override
    public CompletableFuture<MetricsResponse> fetchMetrics(ApplicationId application) {
        NodeList applicationNodes = nodeRepository.nodes().list().owner(application).state(Node.State.active);

        Optional<Node> metricsV2Container = applicationNodes.container()
                .matching(this::expectedUp)
                .stream()
                .filter(node -> ! newNode(node)) // Skip newly added nodes, as they may not be reachable
                .findFirst();

        if (metricsV2Container.isEmpty()) {
            return CompletableFuture.completedFuture(MetricsResponse.empty());
        }
        else {
            // Collector '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));
        }
    }
@Test
    public void testMetricsFetch() throws Exception {
        NodeResources resources = new NodeResources(1, 10, 100, 1);
        ProvisioningTester tester = new ProvisioningTester.Builder().build();
        OrchestratorMock orchestrator = new OrchestratorMock();
        MockHttpClient httpClient = new MockHttpClient();
        MetricsV2MetricsFetcher fetcher = new MetricsV2MetricsFetcher(tester.nodeRepository(), orchestrator, httpClient);

        tester.makeReadyNodes(4, resources); // Creates (in order) host-1.yahoo.com, host-2.yahoo.com, host-3.yahoo.com, host-4.yahoo.com
        tester.clock().advance(Duration.ofMinutes(5)); // Make sure these are not considered new nodes (metrics will not be fetched for them)
        tester.activateTenantHosts();

        ApplicationId application1 = ProvisioningTester.applicationId();
        ApplicationId application2 = ProvisioningTester.applicationId();
        tester.deploy(application1, Capacity.from(new ClusterResources(2, 1, resources))); // host-1.yahoo.com, host-2.yahoo.com
        tester.deploy(application2, Capacity.from(new ClusterResources(2, 1, resources))); // host-4.yahoo.com, host-3.yahoo.com

        orchestrator.suspend(new HostName("host-4.yahoo.com"));

        {
            httpClient.cannedResponse = cannedResponseForApplication1;
            List<Pair<String, NodeMetricSnapshot>> values = new ArrayList<>(fetcher.fetchMetrics(application1).get().nodeMetrics());
            assertEquals("http://host-1.yahoo.com:4080/metrics/v2/values?consumer=autoscaling",
                         httpClient.requestsReceived.get(0));
            assertEquals(2, values.size());

            assertEquals("host-1.yahoo.com", values.get(0).getFirst());
            assertEquals(0.162, values.get(0).getSecond().load().cpu(), delta);
            assertEquals(0.231, values.get(0).getSecond().load().memory(), delta);
            assertEquals(0.820, values.get(0).getSecond().load().disk(), delta);

            assertEquals("host-2.yahoo.com", values.get(1).getFirst());
            assertEquals(0.0,  values.get(1).getSecond().load().cpu(), delta);
            assertEquals(0.35, values.get(1).getSecond().load().memory(), delta);
            assertEquals(0.45,  values.get(1).getSecond().load().disk(), delta);
            assertEquals(45.0, values.get(1).getSecond().queryRate(), delta);
        }

        {
            httpClient.cannedResponse = cannedResponseForApplication2;
            List<Pair<String, NodeMetricSnapshot>> values = new ArrayList<>(fetcher.fetchMetrics(application2).get().nodeMetrics());
            assertEquals("http://host-3.yahoo.com:4080/metrics/v2/values?consumer=autoscaling",
                         httpClient.requestsReceived.get(1));
            assertEquals(1, values.size());
            assertEquals("host-3.yahoo.com", values.get(0).getFirst());
            assertEquals(0.10, values.get(0).getSecond().load().cpu(), delta);
            assertEquals(0.15, values.get(0).getSecond().load().memory(), delta);
            assertEquals(0.20, values.get(0).getSecond().load().disk(), delta);
            assertEquals(3, values.get(0).getSecond().generation(), delta);
            assertFalse(values.get(0).getSecond().inService());
            assertTrue(values.get(0).getSecond().stable());
        }

        { // read response 2 when unstable
            httpClient.cannedResponse = cannedResponseForApplication2;
            try (Mutex lock = tester.nodeRepository().applications().lock(application1)) {
                tester.nodeRepository().nodes().write(tester.nodeRepository().nodes().list(Node.State.active).owner(application2)
                        .first().get().retire(tester.clock().instant()), lock);
            }
            List<Pair<String, NodeMetricSnapshot>> values = new ArrayList<>(fetcher.fetchMetrics(application2).get().nodeMetrics());
            assertFalse(values.get(0).getSecond().stable());
        }

        {
            httpClient.cannedResponse = cannedResponseForApplication3;
            List<Pair<String, NodeMetricSnapshot>> values = new ArrayList<>(fetcher.fetchMetrics(application2).get().nodeMetrics());
            assertEquals("http://host-3.yahoo.com:4080/metrics/v2/values?consumer=autoscaling",
                         httpClient.requestsReceived.get(1));
            assertEquals(1, values.size());
            assertEquals("host-3.yahoo.com", values.get(0).getFirst());
            assertEquals(0.13, values.get(0).getSecond().load().cpu(), delta);
            assertEquals(0.9375, values.get(0).getSecond().load().memory(), delta);
            assertEquals(0.13, values.get(0).getSecond().load().gpu(), delta);
            assertEquals(0.9375, values.get(0).getSecond().load().gpuMemory(), delta);
            assertFalse("Unstable because buckets are being merged", values.get(0).getSecond().stable());
        }

    }