Add cluster feed block support to cluster controller

Will push out a new cluster state bundle indicating cluster feed blocked
if one or more nodes in the cluster has one or more resources exhausted.
Similarly, a new state will be pushed out once no nodes have resources
exhausted any more.

The feed block description currently contains up to 3 separate exhausted
resources, possibly across multiple nodes.

A cluster-level event is emitted for both the block and unblock edges.

No hysteresis is present yet, so if a node is oscillating around a block-limit,
so will the cluster state.

2 files changed, 10 insertions, 1 deletions

diff --git a/clustercontroller-apps/src/main/java/com/yahoo/vespa/clustercontroller/apps/clustercontroller/ClusterControllerClusterConfigurer.java b/clustercontroller-apps/src/main/java/com/yahoo/vespa/clustercontroller/apps/clustercontroller/ClusterControllerClusterConfigurer.java
index ad65435c770..9f439cbd992 100644
--- a/clustercontroller-apps/src/main/java/com/yahoo/vespa/clustercontroller/apps/clustercontroller/ClusterControllerClusterConfigurer.java
+++ b/clustercontroller-apps/src/main/java/com/yahoo/vespa/clustercontroller/apps/clustercontroller/ClusterControllerClusterConfigurer.java
@@ -11,6 +11,7 @@ import com.yahoo.vespa.config.content.StorDistributionConfig;
 import com.yahoo.cloud.config.ZookeepersConfig;
 
 import java.time.Duration;
+import java.util.Map;
 
 /**
  * When the cluster controller is reconfigured, a new instance of this is created, which will propagate configured
@@ -75,6 +76,8 @@ public class ClusterControllerClusterConfigurer {
         options.clusterHasGlobalDocumentTypes = config.cluster_has_global_document_types();
         options.minMergeCompletionRatio = config.min_merge_completion_ratio();
         options.enableTwoPhaseClusterStateActivation = config.enable_two_phase_cluster_state_transitions();
+        options.clusterFeedBlockEnabled = config.enable_cluster_feed_block();
+        options.clusterFeedBlockLimit = Map.copyOf(config.cluster_feed_block_limit());
     }
 
     private void configure(SlobroksConfig config) {
diff --git a/clustercontroller-apps/src/test/java/com/yahoo/vespa/clustercontroller/apps/clustercontroller/ClusterControllerClusterConfigurerTest.java b/clustercontroller-apps/src/test/java/com/yahoo/vespa/clustercontroller/apps/clustercontroller/ClusterControllerClusterConfigurerTest.java
index 37131349602..9d2d7610469 100644
--- a/clustercontroller-apps/src/test/java/com/yahoo/vespa/clustercontroller/apps/clustercontroller/ClusterControllerClusterConfigurerTest.java
+++ b/clustercontroller-apps/src/test/java/com/yahoo/vespa/clustercontroller/apps/clustercontroller/ClusterControllerClusterConfigurerTest.java
@@ -31,7 +31,10 @@ public class ClusterControllerClusterConfigurerTest {
                 .cluster_name("storage")
                 .index(0)
                 .zookeeper_server("zoo")
-                .min_node_ratio_per_group(0.123);
+                .min_node_ratio_per_group(0.123)
+                .enable_cluster_feed_block(true)
+                .cluster_feed_block_limit("foo", 0.5)
+                .cluster_feed_block_limit("bar", 0.7);
         SlobroksConfig.Builder slobroksConfig = new SlobroksConfig.Builder();
         SlobroksConfig.Slobrok.Builder slobrok = new SlobroksConfig.Slobrok.Builder();
         slobrok.connectionspec("foo");
@@ -57,6 +60,9 @@ public class ClusterControllerClusterConfigurerTest {
         );
         assertTrue(configurer.getOptions() != null);
         assertEquals(0.123, configurer.getOptions().minNodeRatioPerGroup, 0.01);
+        assertTrue(configurer.getOptions().clusterFeedBlockEnabled);
+        assertEquals(0.5, configurer.getOptions().clusterFeedBlockLimit.get("foo"), 0.01);
+        assertEquals(0.7, configurer.getOptions().clusterFeedBlockLimit.get("bar"), 0.01);
 
         try{
             zookeepersConfig.zookeeperserverlist("");