Make it possible to specify feed block resource limits in content cluster tuning.

These will be used by the cluster controller to decide whether
external feed should be blocked and stopped by the distributors.
Content node (proton) hard limits are derived from the new limits if not specified, and the other way around.
This way we are backwards compatible when the new functionality in cluster controller is enabled by default.

11 files changed, 385 insertions, 69 deletions

diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/ClusterControllerConfig.java b/config-model/src/main/java/com/yahoo/vespa/model/content/ClusterControllerConfig.java
index 8d6127970c8..66ec0d81947 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/content/ClusterControllerConfig.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/content/ClusterControllerConfig.java
@@ -19,12 +19,14 @@ import org.w3c.dom.Element;
 public class ClusterControllerConfig extends AbstractConfigProducer<ClusterControllerConfig> implements FleetcontrollerConfig.Producer {
 
     public static class Builder extends VespaDomBuilder.DomConfigProducerBuilder<ClusterControllerConfig> {
-        String clusterName;
-        ModelElement clusterElement;
+        private final String clusterName;
+        private final ModelElement clusterElement;
+        private final ResourceLimits resourceLimits;
 
-        public Builder(String clusterName, ModelElement clusterElement) {
+        public Builder(String clusterName, ModelElement clusterElement, ResourceLimits resourceLimits) {
             this.clusterName = clusterName;
             this.clusterElement = clusterElement;
+            this.resourceLimits = resourceLimits;
         }
 
         @Override
@@ -51,27 +53,29 @@ public class ClusterControllerConfig extends AbstractConfigProducer<ClusterContr
                         tuning.childAsDouble("min-storage-up-ratio"),
                         bucketSplittingMinimumBits,
                         minNodeRatioPerGroup,
-                        enableClusterFeedBlock);
+                        enableClusterFeedBlock,
+                        resourceLimits);
             } else {
                 return new ClusterControllerConfig(ancestor, clusterName,
                         null, null, null, null, null, null,
                         bucketSplittingMinimumBits,
                         minNodeRatioPerGroup,
-                        enableClusterFeedBlock);
+                        enableClusterFeedBlock, resourceLimits);
             }
         }
     }
 
-    String clusterName;
-    Duration initProgressTime;
-    Duration transitionTime;
-    Long maxPrematureCrashes;
-    Duration stableStateTimePeriod;
-    Double minDistributorUpRatio;
-    Double minStorageUpRatio;
-    Integer minSplitBits;
-    private Double minNodeRatioPerGroup;
-    private boolean enableClusterFeedBlock = false;
+    private final String clusterName;
+    private final Duration initProgressTime;
+    private final Duration transitionTime;
+    private final Long maxPrematureCrashes;
+    private final Duration stableStateTimePeriod;
+    private final Double minDistributorUpRatio;
+    private final Double minStorageUpRatio;
+    private final Integer minSplitBits;
+    private final Double minNodeRatioPerGroup;
+    private final boolean enableClusterFeedBlock;
+    private final ResourceLimits resourceLimits;
 
     // TODO refactor; too many args
     private ClusterControllerConfig(AbstractConfigProducer parent,
@@ -84,7 +88,8 @@ public class ClusterControllerConfig extends AbstractConfigProducer<ClusterContr
                                     Double minStorageUpRatio,
                                     Integer minSplitBits,
                                     Double minNodeRatioPerGroup,
-                                    boolean enableClusterFeedBlock) {
+                                    boolean enableClusterFeedBlock,
+                                    ResourceLimits resourceLimits) {
         super(parent, "fleetcontroller");
 
         this.clusterName = clusterName;
@@ -97,6 +102,7 @@ public class ClusterControllerConfig extends AbstractConfigProducer<ClusterContr
         this.minSplitBits = minSplitBits;
         this.minNodeRatioPerGroup = minNodeRatioPerGroup;
         this.enableClusterFeedBlock = enableClusterFeedBlock;
+        this.resourceLimits = resourceLimits;
     }
 
     @Override
@@ -139,18 +145,7 @@ public class ClusterControllerConfig extends AbstractConfigProducer<ClusterContr
             builder.min_node_ratio_per_group(minNodeRatioPerGroup);
         }
         builder.enable_cluster_feed_block(enableClusterFeedBlock);
-        setDefaultClusterFeedBlockLimits(builder);
+        resourceLimits.getConfig(builder);
     }
 
-    private static void setDefaultClusterFeedBlockLimits(FleetcontrollerConfig.Builder builder) {
-        // TODO: Override these based on resource-limits in services.xml (if they are specified).
-        // TODO: Choose other defaults when this is default enabled.
-        // Note: The resource categories must match the ones used in host info reporting
-        // between content nodes and cluster controller:
-        // storage/src/vespa/storage/persistence/filestorage/service_layer_host_info_reporter.cpp
-        builder.cluster_feed_block_limit.put("memory", 0.79);
-        builder.cluster_feed_block_limit.put("disk", 0.79);
-        builder.cluster_feed_block_limit.put("attribute-enum-store", 0.89);
-        builder.cluster_feed_block_limit.put("attribute-multi-value", 0.89);
-    }
 }
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/ClusterResourceLimits.java b/config-model/src/main/java/com/yahoo/vespa/model/content/ClusterResourceLimits.java
new file mode 100644
index 00000000000..5324ee171ec
--- /dev/null
+++ b/config-model/src/main/java/com/yahoo/vespa/model/content/ClusterResourceLimits.java
@@ -0,0 +1,103 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.model.content;
+
+import com.yahoo.vespa.model.builder.xml.dom.ModelElement;
+import com.yahoo.vespa.model.content.cluster.DomResourceLimitsBuilder;
+
+import java.util.Optional;
+import java.util.function.Consumer;
+
+/**
+ * Class tracking the feed block resource limits for a content cluster.
+ *
+ * This includes the limits used by the cluster controller and the content nodes (proton).
+ *
+ * @author geirst
+ */
+public class ClusterResourceLimits {
+
+    private final ResourceLimits clusterControllerLimits;
+    private final ResourceLimits contentNodeLimits;
+
+    private ClusterResourceLimits(Builder builder) {
+        clusterControllerLimits = builder.ctrlBuilder.build();
+        contentNodeLimits = builder.nodeBuilder.build();
+    }
+
+    public ResourceLimits getClusterControllerLimits() {
+        return clusterControllerLimits;
+    }
+
+    public ResourceLimits getContentNodeLimits() {
+        return contentNodeLimits;
+    }
+
+    public static class Builder {
+
+        private ResourceLimits.Builder ctrlBuilder = new ResourceLimits.Builder();
+        private ResourceLimits.Builder nodeBuilder = new ResourceLimits.Builder();
+
+        public ClusterResourceLimits build(ModelElement clusterElem) {
+
+            ModelElement tuningElem = clusterElem.childByPath("tuning");
+            if (tuningElem != null) {
+                ctrlBuilder = DomResourceLimitsBuilder.createBuilder(tuningElem);
+            }
+
+            ModelElement protonElem = clusterElem.childByPath("engine.proton");
+            if (protonElem != null) {
+                nodeBuilder = DomResourceLimitsBuilder.createBuilder(protonElem);
+            }
+
+            deriveLimits();
+            return new ClusterResourceLimits(this);
+        }
+
+        public void setClusterControllerBuilder(ResourceLimits.Builder builder) {
+            ctrlBuilder = builder;
+        }
+
+        public void setContentNodeBuilder(ResourceLimits.Builder builder) {
+            nodeBuilder = builder;
+        }
+
+        public ClusterResourceLimits build() {
+            deriveLimits();
+            return new ClusterResourceLimits(this);
+        }
+
+        private void deriveLimits() {
+            deriveClusterControllerLimit(ctrlBuilder.getDiskLimit(), nodeBuilder.getDiskLimit(), ctrlBuilder::setDiskLimit);
+            deriveClusterControllerLimit(ctrlBuilder.getMemoryLimit(), nodeBuilder.getMemoryLimit(), ctrlBuilder::setMemoryLimit);
+
+            deriveContentNodeLimit(nodeBuilder.getDiskLimit(), ctrlBuilder.getDiskLimit(), nodeBuilder::setDiskLimit);
+            deriveContentNodeLimit(nodeBuilder.getMemoryLimit(), ctrlBuilder.getMemoryLimit(), nodeBuilder::setMemoryLimit);
+        }
+
+        private void deriveClusterControllerLimit(Optional<Double> clusterControllerLimit,
+                                                  Optional<Double> contentNodeLimit,
+                                                  Consumer<Double> setter) {
+            if (!clusterControllerLimit.isPresent()) {
+                contentNodeLimit.ifPresent(limit ->
+                        // TODO: emit warning when using cluster controller resource limits are default enabled.
+                        setter.accept(limit));
+            }
+        }
+
+        private void deriveContentNodeLimit(Optional<Double> contentNodeLimit,
+                                            Optional<Double> clusterControllerLimit,
+                                            Consumer<Double> setter) {
+            if (!contentNodeLimit.isPresent()) {
+                clusterControllerLimit.ifPresent(limit ->
+                        setter.accept(calcContentNodeLimit(limit)));
+            }
+        }
+
+        private double calcContentNodeLimit(double clusterControllerLimit) {
+            // Note that validation in the range [0.0-1.0] is handled by the rnc schema.
+            return clusterControllerLimit + ((1.0 - clusterControllerLimit) / 2);
+        }
+
+    }
+
+}
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
index dd29df61f35..d7df62d56cf 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
@@ -79,13 +79,15 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
         private final Map<String, NewDocumentType> documentDefinitions;
         private final Set<NewDocumentType> globallyDistributedDocuments;
         private final boolean combined;
+        private final ResourceLimits resourceLimits;
 
         public Builder(Map<String, NewDocumentType> documentDefinitions,
                        Set<NewDocumentType> globallyDistributedDocuments,
-                       boolean combined) {
+                       boolean combined, ResourceLimits resourceLimits) {
             this.documentDefinitions = documentDefinitions;
             this.globallyDistributedDocuments = globallyDistributedDocuments;
             this.combined = combined;
+            this.resourceLimits = resourceLimits;
         }
 
         @Override
@@ -106,10 +108,7 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
             if (tuning != null) {
                 search.setTuning(new DomSearchTuningBuilder().build(deployState, search, tuning.getXml()));
             }
-            ModelElement protonElem = clusterElem.childByPath("engine.proton");
-            if (protonElem != null) {
-                search.setResourceLimits(DomResourceLimitsBuilder.build(protonElem));
-            }
+            search.setResourceLimits(resourceLimits);
 
             buildAllStreamingSearchClusters(deployState, clusterElem, clusterName, search);
             buildIndexedSearchCluster(deployState, clusterElem, clusterName, search);
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/ResourceLimits.java b/config-model/src/main/java/com/yahoo/vespa/model/content/ResourceLimits.java
index 28e8c36d202..e96ba47c6b3 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/content/ResourceLimits.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/content/ResourceLimits.java
@@ -1,16 +1,17 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vespa.model.content;
 
+import com.yahoo.vespa.config.content.FleetcontrollerConfig;
 import com.yahoo.vespa.config.search.core.ProtonConfig;
 
 import java.util.Optional;
 
 /**
- * Class tracking resource limits for a content cluster with engine proton.
+ * Class tracking feed block resource limits used by a component in a content cluster (e.g. cluster controller or content node).
  *
  * @author geirst
  */
-public class ResourceLimits implements ProtonConfig.Producer {
+public class ResourceLimits implements FleetcontrollerConfig.Producer, ProtonConfig.Producer {
 
     private final Optional<Double> diskLimit;
     private final Optional<Double> memoryLimit;
@@ -20,6 +21,26 @@ public class ResourceLimits implements ProtonConfig.Producer {
         this.memoryLimit = builder.memoryLimit;
     }
 
+    public Optional<Double> getDiskLimit() {
+        return diskLimit;
+    }
+
+    public Optional<Double> getMemoryLimit() {
+        return memoryLimit;
+    }
+
+    @Override
+    public void getConfig(FleetcontrollerConfig.Builder builder) {
+        // TODO: Choose other defaults when this is default enabled.
+        // Note: The resource categories must match the ones used in host info reporting
+        // between content nodes and cluster controller:
+        // storage/src/vespa/storage/persistence/filestorage/service_layer_host_info_reporter.cpp
+        builder.cluster_feed_block_limit.put("memory", memoryLimit.orElse(0.79));
+        builder.cluster_feed_block_limit.put("disk", diskLimit.orElse(0.79));
+        builder.cluster_feed_block_limit.put("attribute-enum-store", 0.89);
+        builder.cluster_feed_block_limit.put("attribute-multi-value", 0.89);
+    }
+
     @Override
     public void getConfig(ProtonConfig.Builder builder) {
         if (diskLimit.isPresent()) {
@@ -39,11 +60,19 @@ public class ResourceLimits implements ProtonConfig.Producer {
             return new ResourceLimits(this);
         }
 
+        public Optional<Double> getDiskLimit() {
+            return diskLimit;
+        }
+
         public Builder setDiskLimit(double diskLimit) {
             this.diskLimit = Optional.of(diskLimit);
             return this;
         }
 
+        public Optional<Double> getMemoryLimit() {
+            return memoryLimit;
+        }
+
         public Builder setMemoryLimit(double memoryLimit) {
             this.memoryLimit = Optional.of(memoryLimit);
             return this;
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/cluster/ContentCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/content/cluster/ContentCluster.java
index a627e030156..44de4a1abec 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/content/cluster/ContentCluster.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/content/cluster/ContentCluster.java
@@ -38,6 +38,7 @@ import com.yahoo.vespa.model.container.Container;
 import com.yahoo.vespa.model.container.ContainerCluster;
 import com.yahoo.vespa.model.container.ContainerModel;
 import com.yahoo.vespa.model.content.ClusterControllerConfig;
+import com.yahoo.vespa.model.content.ClusterResourceLimits;
 import com.yahoo.vespa.model.content.ContentSearch;
 import com.yahoo.vespa.model.content.ContentSearchCluster;
 import com.yahoo.vespa.model.content.DistributionBitCalculator;
@@ -134,11 +135,14 @@ public class ContentCluster extends AbstractConfigProducer implements
             ContentCluster c = new ContentCluster(context.getParentProducer(), getClusterId(contentElement), documentDefinitions,
                                                   globallyDistributedDocuments, routingSelection,
                                                   deployState.zone(), deployState.isHosted());
-            c.clusterControllerConfig = new ClusterControllerConfig.Builder(getClusterId(contentElement), contentElement).build(deployState, c, contentElement.getXml());
+            var resourceLimits = new ClusterResourceLimits.Builder().build(contentElement);
+            c.clusterControllerConfig = new ClusterControllerConfig.Builder(getClusterId(contentElement),
+                    contentElement,
+                    resourceLimits.getClusterControllerLimits()).build(deployState, c, contentElement.getXml());
             c.search = new ContentSearchCluster.Builder(documentDefinitions,
-                                                        globallyDistributedDocuments,
-                                                        isCombined(getClusterId(contentElement), containers))
-                               .build(deployState, c, contentElement.getXml());
+                    globallyDistributedDocuments,
+                    isCombined(getClusterId(contentElement), containers),
+                    resourceLimits.getContentNodeLimits()).build(deployState, c, contentElement.getXml());
             c.persistenceFactory = new EngineFactoryBuilder().build(contentElement, c);
             c.storageNodes = new StorageCluster.Builder().build(deployState, c, w3cContentElement);
             c.distributorNodes = new DistributorCluster.Builder(c).build(deployState, c, w3cContentElement);
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/cluster/DomResourceLimitsBuilder.java b/config-model/src/main/java/com/yahoo/vespa/model/content/cluster/DomResourceLimitsBuilder.java
index 8e91f14238e..210f062f9b2 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/content/cluster/DomResourceLimitsBuilder.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/content/cluster/DomResourceLimitsBuilder.java
@@ -5,17 +5,17 @@ import com.yahoo.vespa.model.builder.xml.dom.ModelElement;
 import com.yahoo.vespa.model.content.ResourceLimits;
 
 /**
- * Builder for resource limits for a content cluster with engine proton.
+ * Builder for feed block resource limits.
  *
  * @author geirst
  */
 public class DomResourceLimitsBuilder {
 
-    public static ResourceLimits build(ModelElement contentXml) {
+    public static ResourceLimits.Builder createBuilder(ModelElement contentXml) {
         ResourceLimits.Builder builder = new ResourceLimits.Builder();
         ModelElement resourceLimits = contentXml.child("resource-limits");
         if (resourceLimits == null) {
-            return builder.build();
+            return builder;
         }
         if (resourceLimits.child("disk") != null) {
             builder.setDiskLimit(resourceLimits.childAsDouble("disk"));
@@ -23,7 +23,7 @@ public class DomResourceLimitsBuilder {
         if (resourceLimits.child("memory") != null) {
             builder.setMemoryLimit(resourceLimits.childAsDouble("memory"));
         }
-        return builder.build();
+        return builder;
     }
 
 }
diff --git a/config-model/src/main/resources/schema/content.rnc b/config-model/src/main/resources/schema/content.rnc
index 5646bc72056..a48d38b9f2c 100644
--- a/config-model/src/main/resources/schema/content.rnc
+++ b/config-model/src/main/resources/schema/content.rnc
@@ -98,7 +98,8 @@ ClusterTuning = element tuning {
    ClusterControllerTuning? &
    Maintenance? &
    PersistenceThreads? &
-   MinNodeRatioPerGroup?
+   MinNodeRatioPerGroup? &
+   ResourceLimits?
 }
 
 Content = element content {
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/content/ClusterResourceLimitsTest.java b/config-model/src/test/java/com/yahoo/vespa/model/content/ClusterResourceLimitsTest.java
new file mode 100644
index 00000000000..bc830c079d0
--- /dev/null
+++ b/config-model/src/test/java/com/yahoo/vespa/model/content/ClusterResourceLimitsTest.java
@@ -0,0 +1,101 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.model.content;
+
+import org.junit.Test;
+
+import java.util.Optional;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+/**
+ * @author geirst
+ */
+public class ClusterResourceLimitsTest {
+
+    private static class Fixture {
+        ResourceLimits.Builder ctrlBuilder = new ResourceLimits.Builder();
+        ResourceLimits.Builder nodeBuilder = new ResourceLimits.Builder();
+
+        public Fixture ctrlDisk(double limit) {
+            ctrlBuilder.setDiskLimit(limit);
+            return this;
+        }
+        public Fixture ctrlMemory(double limit) {
+            ctrlBuilder.setMemoryLimit(limit);
+            return this;
+        }
+        public Fixture nodeDisk(double limit) {
+            nodeBuilder.setDiskLimit(limit);
+            return this;
+        }
+        public Fixture nodeMemory(double limit) {
+            nodeBuilder.setMemoryLimit(limit);
+            return this;
+        }
+        public ClusterResourceLimits build() {
+            var builder = new ClusterResourceLimits.Builder();
+            builder.setClusterControllerBuilder(ctrlBuilder);
+            builder.setContentNodeBuilder(nodeBuilder);
+            return builder.build();
+        }
+    }
+
+    @Test
+    public void content_node_limits_are_derived_from_cluster_controller_limits_if_not_set() {
+        assertLimits(0.6, 0.7, 0.8, 0.85,
+                new Fixture().ctrlDisk(0.6).ctrlMemory(0.7));
+        assertLimits(0.6, null, 0.8, null,
+                new Fixture().ctrlDisk(0.6));
+        assertLimits(null, 0.7, null, 0.85,
+                new Fixture().ctrlMemory(0.7));
+    }
+
+    @Test
+    public void content_node_limits_can_be_set_explicit() {
+        assertLimits(0.6, 0.7, 0.9, 0.95,
+                new Fixture().ctrlDisk(0.6).ctrlMemory(0.7).nodeDisk(0.9).nodeMemory(0.95));
+        assertLimits(0.6, null, 0.9, null,
+                new Fixture().ctrlDisk(0.6).nodeDisk(0.9));
+        assertLimits(null, 0.7, null, 0.95,
+                new Fixture().ctrlMemory(0.7).nodeMemory(0.95));
+    }
+
+    @Test
+    public void cluster_controller_limits_are_equal_to_content_node_limits_if_not_set() {
+        assertLimits(0.9, 0.95, 0.9, 0.95,
+                new Fixture().nodeDisk(0.9).nodeMemory(0.95));
+        assertLimits(0.9, null, 0.9, null,
+                new Fixture().nodeDisk(0.9));
+        assertLimits(null, 0.95, null, 0.95,
+                new Fixture().nodeMemory(0.95));
+    }
+
+    @Test
+    public void limits_are_derived_from_the_other_if_not_set() {
+        assertLimits(0.6, 0.95, 0.8, 0.95,
+                new Fixture().ctrlDisk(0.6).nodeMemory(0.95));
+        assertLimits(0.9, 0.7, 0.9, 0.85,
+                new Fixture().ctrlMemory(0.7).nodeDisk(0.9));
+    }
+
+    private void assertLimits(Double expCtrlDisk, Double expCtrlMemory, Double expNodeDisk, Double expNodeMemory, Fixture f) {
+        var limits = f.build();
+        assertLimits(expCtrlDisk, expCtrlMemory, limits.getClusterControllerLimits());
+        assertLimits(expNodeDisk, expNodeMemory, limits.getContentNodeLimits());
+    }
+
+    private void assertLimits(Double expDisk, Double expMemory, ResourceLimits limits) {
+        assertLimit(expDisk, limits.getDiskLimit());
+        assertLimit(expMemory, limits.getMemoryLimit());
+    }
+
+    private void assertLimit(Double expLimit, Optional<Double> actLimit) {
+        if (expLimit == null) {
+            assertFalse(actLimit.isPresent());
+        } else {
+            assertEquals(expLimit, actLimit.get(), 0.00001);
+        }
+    }
+
+}
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/content/ContentSearchClusterTest.java b/config-model/src/test/java/com/yahoo/vespa/model/content/ContentSearchClusterTest.java
index 3415044b088..bc60908e268 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/content/ContentSearchClusterTest.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/content/ContentSearchClusterTest.java
@@ -69,15 +69,30 @@ public class ContentSearchClusterTest {
     }
 
     private static ProtonConfig getProtonConfig(ContentCluster cluster) {
-        ProtonConfig.Builder protonCfgBuilder = new ProtonConfig.Builder();
-        cluster.getSearch().getConfig(protonCfgBuilder);
-        return new ProtonConfig(protonCfgBuilder);
+        var builder = new ProtonConfig.Builder();
+        cluster.getSearch().getConfig(builder);
+        return new ProtonConfig(builder);
     }
 
-    private static void assertProtonResourceLimits(double expDiskLimit, double expMemoryLimits, String clusterXml) throws Exception {
-        ProtonConfig cfg = getProtonConfig(createCluster(clusterXml));
+    private static void assertProtonResourceLimits(double expDiskLimit, double expMemoryLimit, String clusterXml) throws Exception {
+        assertProtonResourceLimits(expDiskLimit, expMemoryLimit, createCluster(clusterXml));
+    }
+
+    private static void assertProtonResourceLimits(double expDiskLimit, double expMemoryLimit, ContentCluster cluster) {
+        var cfg = getProtonConfig(cluster);
         assertEquals(expDiskLimit, cfg.writefilter().disklimit(), EPSILON);
-        assertEquals(expMemoryLimits, cfg.writefilter().memorylimit(), EPSILON);
+        assertEquals(expMemoryLimit, cfg.writefilter().memorylimit(), EPSILON);
+    }
+
+    private static void assertClusterControllerResourceLimits(double expDiskLimit, double expMemoryLimit, String clusterXml) throws Exception {
+        assertClusterControllerResourceLimits(expDiskLimit, expMemoryLimit, createCluster(clusterXml));
+    }
+
+    private static void assertClusterControllerResourceLimits(double expDiskLimit, double expMemoryLimit, ContentCluster cluster) {
+        var limits = getFleetcontrollerConfig(cluster).cluster_feed_block_limit();
+        assertEquals(4, limits.size());
+        assertEquals(expDiskLimit, limits.get("disk"), EPSILON);
+        assertEquals(expMemoryLimit, limits.get("memory"), EPSILON);
     }
 
     @Test
@@ -105,6 +120,19 @@ public class ContentSearchClusterTest {
     }
 
     @Test
+    public void cluster_controller_resource_limits_can_be_set() throws Exception {
+        assertClusterControllerResourceLimits(0.92, 0.93,
+                new ContentClusterBuilder().clusterControllerDiskLimit(0.92).clusterControllerMemoryLimit(0.93).getXml());
+    }
+
+    @Test
+    public void resource_limits_are_derived_from_the_other_if_not_specified() throws Exception {
+        var cluster = createCluster(new ContentClusterBuilder().clusterControllerDiskLimit(0.5).protonMemoryLimit(0.95).getXml());
+        assertProtonResourceLimits(0.75, 0.95, cluster);
+        assertClusterControllerResourceLimits(0.5, 0.95, cluster);
+    }
+
+    @Test
     public void requireThatGloballyDistributedDocumentTypeIsTaggedAsSuch() throws Exception {
         ProtonConfig cfg = getProtonConfig(createClusterWithGlobalType());
         assertEquals(2, cfg.documentdb().size());
@@ -149,8 +177,9 @@ public class ContentSearchClusterTest {
     }
 
     private static FleetcontrollerConfig getFleetcontrollerConfig(ContentCluster cluster) {
-        FleetcontrollerConfig.Builder builder = new FleetcontrollerConfig.Builder();
+        var builder = new FleetcontrollerConfig.Builder();
         cluster.getConfig(builder);
+        cluster.getClusterControllerConfig().getConfig(builder);
         builder.cluster_name("unknown");
         builder.index(0);
         builder.zookeeper_server("unknown");
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/content/FleetControllerClusterTest.java b/config-model/src/test/java/com/yahoo/vespa/model/content/FleetControllerClusterTest.java
index 01bbffce360..3a59f35ce2e 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/content/FleetControllerClusterTest.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/content/FleetControllerClusterTest.java
@@ -10,6 +10,7 @@ import com.yahoo.vespa.model.builder.xml.dom.ModelElement;
 import org.junit.Test;
 import org.w3c.dom.Document;
 
+import static com.yahoo.config.model.test.TestUtil.joinLines;
 import static org.junit.Assert.assertEquals;
 
 public class FleetControllerClusterTest {
@@ -19,8 +20,11 @@ public class FleetControllerClusterTest {
         var deployState = new DeployState.Builder().properties(
                 new TestProperties().enableFeedBlockInDistributor(enableFeedBlockInDistributor)).build();
         MockRoot root = new MockRoot("", deployState);
-        return new ClusterControllerConfig.Builder("storage", new ModelElement(doc.getDocumentElement())).build(root.getDeployState(), root,
-                new ModelElement(doc.getDocumentElement()).getXml());
+        var clusterElement = new ModelElement(doc.getDocumentElement());
+        return new ClusterControllerConfig.Builder("storage",
+                clusterElement,
+                new ClusterResourceLimits.Builder().build(clusterElement).getClusterControllerLimits()).
+                build(root.getDeployState(), root, clusterElement.getXml());
     }
 
     private ClusterControllerConfig parse(String xml) {
@@ -94,15 +98,43 @@ public class FleetControllerClusterTest {
         assertEquals(0.0, config.min_node_ratio_per_group(), 0.01);
     }
 
+
     @Test
     public void default_cluster_feed_block_limits_are_set() {
-        var config = getConfigForBasicCluster();
+        assertLimits(0.79, 0.79, getConfigForBasicCluster());
+    }
+
+    @Test
+    public void resource_limits_can_be_set_in_tuning() {
+        assertLimits(0.6, 0.7, getConfigForResourceLimitsTuning(0.6, 0.7));
+        assertLimits(0.6, 0.79, getConfigForResourceLimitsTuning(0.6, null));
+        assertLimits(0.79, 0.7, getConfigForResourceLimitsTuning(null, 0.7));
+    }
+
+    private static double DELTA = 0.00001;
+
+    private void assertLimits(double expDisk, double expMemory, FleetcontrollerConfig config) {
         var limits = config.cluster_feed_block_limit();
         assertEquals(4, limits.size());
-        assertEquals(0.79, limits.get("memory"), 0.0001);
-        assertEquals(0.79, limits.get("disk"), 0.0001);
-        assertEquals(0.89, limits.get("attribute-enum-store"), 0.0001);
-        assertEquals(0.89, limits.get("attribute-multi-value"), 0.0001);
+        assertEquals(expDisk, limits.get("disk"), DELTA);
+        assertEquals(expMemory, limits.get("memory"), DELTA);
+        assertEquals(0.89, limits.get("attribute-enum-store"), DELTA);
+        assertEquals(0.89, limits.get("attribute-multi-value"), DELTA);
+    }
+
+    private FleetcontrollerConfig getConfigForResourceLimitsTuning(Double diskLimit, Double memoryLimit) {
+        FleetcontrollerConfig.Builder builder = new FleetcontrollerConfig.Builder();
+        parse(joinLines("<cluster id=\"test\">",
+                "<documents/>",
+                "<tuning>",
+                "  <resource-limits>",
+                (diskLimit != null ? ("    <disk>" + diskLimit + "</disk>") : ""),
+                (memoryLimit != null ? ("    <memory>" + memoryLimit + "</memory>") : ""),
+                "  </resource-limits>",
+                "</tuning>" +
+                "</cluster>")).
+                getConfig(builder);
+        return new FleetcontrollerConfig(builder);
     }
 
     @Test
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/content/utils/ContentClusterBuilder.java b/config-model/src/test/java/com/yahoo/vespa/model/content/utils/ContentClusterBuilder.java
index 866c03d82f0..491326fdc9c 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/content/utils/ContentClusterBuilder.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/content/utils/ContentClusterBuilder.java
@@ -26,6 +26,8 @@ public class ContentClusterBuilder {
     private Optional<String> dispatchXml = Optional.empty();
     private Optional<Double> protonDiskLimit = Optional.empty();
     private Optional<Double> protonMemoryLimit = Optional.empty();
+    private Optional<Double> clusterControllerDiskLimit = Optional.empty();
+    private Optional<Double> clusterControllerMemoryLimit = Optional.empty();
 
     public ContentClusterBuilder() {
     }
@@ -67,13 +69,23 @@ public class ContentClusterBuilder {
         return this;
     }
 
-    public ContentClusterBuilder protonDiskLimit(double diskLimit) {
-        protonDiskLimit = Optional.of(diskLimit);
+    public ContentClusterBuilder protonDiskLimit(double limit) {
+        protonDiskLimit = Optional.of(limit);
         return this;
     }
 
-    public ContentClusterBuilder protonMemoryLimit(double memoryLimit) {
-        protonMemoryLimit = Optional.of(memoryLimit);
+    public ContentClusterBuilder protonMemoryLimit(double limit) {
+        protonMemoryLimit = Optional.of(limit);
+        return this;
+    }
+
+    public ContentClusterBuilder clusterControllerDiskLimit(double limit) {
+        clusterControllerDiskLimit = Optional.of(limit);
+        return this;
+    }
+
+    public ContentClusterBuilder clusterControllerMemoryLimit(double limit) {
+        clusterControllerMemoryLimit = Optional.of(limit);
         return this;
     }
 
@@ -88,14 +100,17 @@ public class ContentClusterBuilder {
                "  <engine>",
                "    <proton>",
                "      <searchable-copies>" + searchableCopies + "</searchable-copies>",
-               getResourceLimitsXml("      "),
+               getProtonResourceLimitsXml("      "),
                "    </proton>",
                "  </engine>");
         if (dispatchXml.isPresent()) {
             xml += dispatchXml.get();
         }
-        return xml + groupXml +
-               "</content>";
+        xml += groupXml;
+        xml += joinLines("  <tuning>",
+               getTuningResourceLimitsXml("    "),
+               "  </tuning>");
+        return xml + "</content>";
     }
 
     private static String getSimpleGroupXml() {
@@ -104,11 +119,19 @@ public class ContentClusterBuilder {
                 "  </group>");
     }
 
-    private String getResourceLimitsXml(String indent) {
-        if (protonDiskLimit.isPresent() || protonMemoryLimit.isPresent()) {
+    private String getProtonResourceLimitsXml(String indent) {
+        return getResourceLimitsXml(indent, protonDiskLimit, protonMemoryLimit);
+    }
+
+    private String getTuningResourceLimitsXml(String indent) {
+        return getResourceLimitsXml(indent, clusterControllerDiskLimit, clusterControllerMemoryLimit);
+    }
+
+    private String getResourceLimitsXml(String indent, Optional<Double> diskLimit, Optional<Double> memoryLimit) {
+        if (diskLimit.isPresent() || memoryLimit.isPresent()) {
             String xml = joinLines(indent + "<resource-limits>",
-                    getXmlLine("disk", protonDiskLimit, indent + "  "),
-                    getXmlLine("memory", protonMemoryLimit, indent + "  "),
+                    getXmlLine("disk", diskLimit, indent + "  "),
+                    getXmlLine("memory", memoryLimit, indent + "  "),
                     indent + "</resource-limits>");
             return xml;
         }