Even if we do not filter by consumer on the services, we can do it early in the metrics-proxy.

This is especially benefiscal for the most frequent requests like autoscaling which only looks at a small set (1%) of the metrics.
This both reduces memory footprint and saves cpu.

4 files changed, 75 insertions, 26 deletions

diff --git a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/core/MetricsConsumers.java b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/core/MetricsConsumers.java
index 457e27a5896..895bf344266 100644
--- a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/core/MetricsConsumers.java
+++ b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/core/MetricsConsumers.java
@@ -16,6 +16,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collector;
+import java.util.stream.Collectors;
 
 import static com.yahoo.stream.CustomCollectors.toLinkedMap;
 import static java.util.Collections.unmodifiableSet;
@@ -31,6 +32,7 @@ public class MetricsConsumers {
 
     // All metrics for each consumer.
     private final Map<ConsumerId, List<ConfiguredMetric>> consumerMetrics;
+    private final Map<ConsumerId, Map<MetricId, ConfiguredMetric>> configuredMetricByMetricByConsumer;
 
     // All consumers for each metric (more useful than the opposite map).
     private final Map<ConfiguredMetric, Set<ConsumerId>> consumersByMetric;
@@ -42,6 +44,10 @@ public class MetricsConsumers {
         consumerMetrics = config.consumer().stream().collect(
                 toUnmodifiableLinkedMap(consumer -> ConsumerId.toConsumerId(consumer.name()), consumer -> convert(consumer.metric())));
 
+        configuredMetricByMetricByConsumer = new HashMap<>();
+        consumerMetrics.forEach((consumer, configuredList) ->
+            configuredMetricByMetricByConsumer.put(consumer,
+                    configuredList.stream().collect(Collectors.toMap(ConfiguredMetric::id, Function.identity()))));
         consumersByMetric = createConsumersByMetric(consumerMetrics);
         consumersByMetricByMetricId = new HashMap<>();
         consumersByMetric.forEach((configuredMetric, consumers) -> {
@@ -67,6 +73,10 @@ public class MetricsConsumers {
         return consumersByMetricByMetricId.get(id);
     }
 
+    public Map<MetricId, ConfiguredMetric> getMetricsForConsumer(ConsumerId consumerId) {
+        return configuredMetricByMetricByConsumer.get(consumerId);
+    }
+
     public Set<ConsumerId> getAllConsumers() {
         return unmodifiableSet(consumerMetrics.keySet());
     }
diff --git a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/core/MetricsManager.java b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/core/MetricsManager.java
index de8d2c62880..f9ecaa29153 100644
--- a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/core/MetricsManager.java
+++ b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/core/MetricsManager.java
@@ -81,7 +81,16 @@ public class MetricsManager {
      * @return metrics for all matching services
      */
     public List<MetricsPacket> getMetrics(List<VespaService> services, Instant startTime) {
-        MetricsPacket.Builder [] builderArray = getMetricsBuildersAsArray(services, startTime);
+        MetricsPacket.Builder [] builderArray = getMetricsBuildersAsArray(services, startTime, null);
+        List<MetricsPacket> metricsPackets = new ArrayList<>(builderArray.length);
+        for (int i = 0; i < builderArray.length; i++) {
+            metricsPackets.add(builderArray[i].build());
+            builderArray[i] = null; // Set null to be able to GC the builder when packet has been created
+        }
+        return metricsPackets;
+    }
+    public List<MetricsPacket> getMetrics(List<VespaService> services, Instant startTime, ConsumerId consumerId) {
+        MetricsPacket.Builder [] builderArray = getMetricsBuildersAsArray(services, startTime, consumerId);
         List<MetricsPacket> metricsPackets = new ArrayList<>(builderArray.length);
         for (int i = 0; i < builderArray.length; i++) {
             metricsPackets.add(builderArray[i].build());
@@ -90,8 +99,8 @@ public class MetricsManager {
         return metricsPackets;
     }
 
-    public MetricsPacket.Builder [] getMetricsBuildersAsArray(List<VespaService> services, Instant startTime) {
-        List<MetricsPacket.Builder> builders = getMetricsAsBuilders(services, startTime);
+    private MetricsPacket.Builder [] getMetricsBuildersAsArray(List<VespaService> services, Instant startTime, ConsumerId consumerId) {
+        List<MetricsPacket.Builder> builders = getMetricsAsBuilders(services, startTime, consumerId);
         return builders.toArray(new MetricsPacket.Builder[builders.size()]);
     }
 
@@ -99,13 +108,13 @@ public class MetricsManager {
      * Returns the metrics for the given services, in mutable state for further processing.
      * NOTE: Use {@link #getMetrics(List, Instant)} instead, unless further processing of the metrics is necessary.
      */
-    public List<MetricsPacket.Builder> getMetricsAsBuilders(List<VespaService> services, Instant startTime) {
+    public List<MetricsPacket.Builder> getMetricsAsBuilders(List<VespaService> services, Instant startTime, ConsumerId consumerId) {
         if (services.isEmpty()) return Collections.emptyList();
 
         log.log(FINE, () -> "Updating services prior to fetching metrics, number of services= " + services.size());
         vespaServices.updateServices(services);
 
-        List<MetricsPacket.Builder> result = vespaMetrics.getMetrics(services);
+        List<MetricsPacket.Builder> result = vespaMetrics.getMetrics(services, consumerId);
         log.log(FINE, () -> "Got " + result.size() + " metrics packets for vespa services.");
 
         purgeStaleMetrics();
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 93f8ec0440b..282068dc02f 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
@@ -65,7 +65,7 @@ public class VespaMetrics {
      * @param services the services to get metrics for
      * @return a list of metrics packet builders (to allow modification by the caller)
      */
-    public List<MetricsPacket.Builder> getMetrics(List<VespaService> services) {
+    public List<MetricsPacket.Builder> getMetrics(List<VespaService> services, ConsumerId consumerId) {
         List<MetricsPacket.Builder> metricsPackets = new ArrayList<>();
 
         for (VespaService service : services) {
@@ -74,7 +74,9 @@ public class VespaMetrics {
             systemCheck.ifPresent(metricsPackets::add);
 
             MetricAggregator aggregator = new MetricAggregator(service.getDimensions());
-            GetServiceMetricsConsumer metricsConsumer = new GetServiceMetricsConsumer(metricsConsumers, aggregator);
+            MetricsParser.Consumer metricsConsumer = (consumerId != null)
+                    ? new GetServiceMetricsConsumer(metricsConsumers, aggregator, consumerId)
+                    : new GetServiceMetricsConsumerForAll(metricsConsumers, aggregator);
             service.consumeMetrics(metricsConsumer);
 
             if (! aggregator.getAggregated().isEmpty()) {
@@ -115,24 +117,14 @@ 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 static class GetServiceMetricsConsumer implements MetricsParser.Consumer {
-        private final MetricAggregator aggregator;
-        private final MetricsConsumers metricsConsumers;
-        GetServiceMetricsConsumer(MetricsConsumers metricsConsumers, MetricAggregator aggregator) {
-            this.metricsConsumers = metricsConsumers;
+    private static abstract class GetServiceMetricsConsumerBase implements MetricsParser.Consumer {
+        protected final MetricAggregator aggregator;
+
+        GetServiceMetricsConsumerBase(MetricAggregator aggregator) {
             this.aggregator = aggregator;
         }
 
-        @Override
-        public void consume(Metric candidate) {
-            Map<ConfiguredMetric, Set<ConsumerId>> consumersByMetric = metricsConsumers.getConsumersByMetric(candidate.getName());
-            if (consumersByMetric != null) {
-                consumersByMetric.keySet().forEach(
-                        configuredMetric -> aggregator.aggregate(
-                                metricWithConfigProperties(candidate, configuredMetric, consumersByMetric.get(configuredMetric))));
-            }
-        }
-        private static Metric metricWithConfigProperties(Metric candidate,
+        protected static Metric metricWithConfigProperties(Metric candidate,
                                                          ConfiguredMetric configuredMetric,
                                                          Set<ConsumerId> consumers) {
             Metric metric = candidate.clone();
@@ -153,8 +145,46 @@ public class VespaMetrics {
         }
 
         private static Set<ConsumerId> extractConsumers(Set<ConsumerId> configuredConsumers) {
-            if (configuredConsumers != null) return configuredConsumers;
-            return Set.of();
+            return (configuredConsumers != null) ? configuredConsumers : Set.of();
+        }
+    }
+
+    private static class GetServiceMetricsConsumer extends GetServiceMetricsConsumerBase {
+        private final Map<MetricId, ConfiguredMetric> configuredMetrics;
+        private final Set<ConsumerId> consumerId;
+
+        GetServiceMetricsConsumer(MetricsConsumers metricsConsumers, MetricAggregator aggregator, ConsumerId consumerId) {
+            super(aggregator);
+            this.consumerId = Set.of(consumerId);
+            this.configuredMetrics = metricsConsumers.getMetricsForConsumer(consumerId);
+        }
+
+        @Override
+        public void consume(Metric candidate) {
+            ConfiguredMetric configuredMetric = configuredMetrics.get(candidate.getName());
+            if (configuredMetric != null) {
+                aggregator.aggregate(
+                        metricWithConfigProperties(candidate, configuredMetric, consumerId));
+            }
+        }
+    }
+
+    private static class GetServiceMetricsConsumerForAll extends GetServiceMetricsConsumerBase {
+        private final MetricsConsumers metricsConsumers;
+
+        GetServiceMetricsConsumerForAll(MetricsConsumers metricsConsumers, MetricAggregator aggregator) {
+            super(aggregator);
+            this.metricsConsumers = metricsConsumers;
+        }
+
+        @Override
+        public void consume(Metric candidate) {
+            Map<ConfiguredMetric, Set<ConsumerId>> consumersByMetric = metricsConsumers.getConsumersByMetric(candidate.getName());
+            if (consumersByMetric != null) {
+                consumersByMetric.keySet().forEach(
+                        configuredMetric -> aggregator.aggregate(
+                                metricWithConfigProperties(candidate, configuredMetric, consumersByMetric.get(configuredMetric))));
+            }
         }
     }
 
diff --git a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/ValuesFetcher.java b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/ValuesFetcher.java
index 8f7f1c8a779..df670addcd8 100644
--- a/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/ValuesFetcher.java
+++ b/metrics-proxy/src/main/java/ai/vespa/metricsproxy/http/ValuesFetcher.java
@@ -41,7 +41,7 @@ public class ValuesFetcher {
     public List<MetricsPacket> fetch(String requestedConsumer) throws JsonRenderingException {
         ConsumerId consumer = getConsumerOrDefault(requestedConsumer, metricsConsumers);
 
-        return fetchAllMetrics()
+        return metricsManager.getMetrics(vespaServices.getVespaServices(), Instant.now(), consumer)
                 .stream()
                 .filter(metricsPacket -> metricsPacket.consumers().contains(consumer))
                 .collect(Collectors.toList());
@@ -50,7 +50,7 @@ public class ValuesFetcher {
     public MetricsPacket.Builder [] fetchMetricsAsBuilders(String requestedConsumer) throws JsonRenderingException {
         ConsumerId consumer = getConsumerOrDefault(requestedConsumer, metricsConsumers);
 
-        List<MetricsPacket.Builder> builders = metricsManager.getMetricsAsBuilders(vespaServices.getVespaServices(), Instant.now())
+        List<MetricsPacket.Builder> builders = metricsManager.getMetricsAsBuilders(vespaServices.getVespaServices(), Instant.now(), consumer)
                 .stream()
                 .filter(builder -> builder.hasConsumer(consumer))
                 .collect(Collectors.toList());