Support configurable feed block hysteresis on the cluster controller

Adds an absolute number delta that is subtracted from the feed block limit
when a node has a resource already in feed blocked state. This means that
there's a lower watermark threshold that must be crossed before feeding
can be unblocked. Avoids flip-flopping between block states.

Default is currently 0.0, i.e. effectively disabled. To be modified
later for system tests and trial roll-outs.

A couple of caveats with the current implementation:
* The cluster state is not recomputed automatically when just the hysteresis
  threshold is crossed, so the description will be out of date on the
  content nodes. However, if any other feed block event happens (or the
  hysteresis threshold is crossed), the state will be recomputed as expected.
  This does not affect correctness, since the feed is still to be blocked.
* A node event remove/add pair is emitted for feed block status when the
  hysteresis threshold is crossed and there's a cluster state recomputation.

3 files changed, 96 insertions, 2 deletions

diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterFeedBlockTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterFeedBlockTest.java
index 75a197ec77a..da62aac66a2 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterFeedBlockTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterFeedBlockTest.java
@@ -81,22 +81,27 @@ public class ClusterFeedBlockTest extends FleetControllerTest {
         super.tearDown();
     }
 
-    private static FleetControllerOptions createOptions(Map<String, Double> feedBlockLimits) {
+    private static FleetControllerOptions createOptions(Map<String, Double> feedBlockLimits,
+                                                        double clusterFeedBlockNoiseLevel) {
         FleetControllerOptions options = defaultOptions("mycluster");
         options.setStorageDistribution(DistributionBuilder.forFlatCluster(NODE_COUNT));
         options.nodes = new HashSet<>(DistributionBuilder.buildConfiguredNodes(NODE_COUNT));
         options.clusterFeedBlockEnabled = true;
         options.clusterFeedBlockLimit = Map.copyOf(feedBlockLimits);
+        options.clusterFeedBlockNoiseLevel = clusterFeedBlockNoiseLevel;
         return options;
     }
 
+    private static FleetControllerOptions createOptions(Map<String, Double> feedBlockLimits) {
+        return createOptions(feedBlockLimits, 0.0);
+    }
+
     private void reportResourceUsageFromNode(int nodeIndex, Set<FeedBlockUtil.UsageDetails> resourceUsages) throws Exception {
         String hostInfo = createResourceUsageJson(resourceUsages);
         communicator.setNodeState(new Node(NodeType.STORAGE, nodeIndex), new NodeState(NodeType.STORAGE, State.UP), hostInfo);
         ctrl.tick();
     }
 
-    // TODO some form of hysteresis
     @Test
     public void cluster_feed_can_be_blocked_and_unblocked_by_single_node() throws Exception {
         initialize(createOptions(mapOf(usage("cheese", 0.7), usage("wine", 0.4))));
@@ -168,4 +173,46 @@ public class ClusterFeedBlockTest extends FleetControllerTest {
         assertEquals("cheese on node 1 [unknown hostname] (0.800 > 0.700)", bundle.getFeedBlock().get().getDescription());
     }
 
+    @Test
+    public void cluster_feed_block_state_is_recomputed_when_usage_enters_hysteresis_range() throws Exception {
+        initialize(createOptions(mapOf(usage("cheese", 0.7), usage("wine", 0.4)), 0.1));
+        assertFalse(ctrl.getClusterStateBundle().clusterFeedIsBlocked());
+
+        reportResourceUsageFromNode(1, setOf(usage("cheese", 0.75), usage("wine", 0.3)));
+        var bundle = ctrl.getClusterStateBundle();
+        assertTrue(bundle.clusterFeedIsBlocked());
+        assertEquals("cheese on node 1 [unknown hostname] (0.750 > 0.700)", bundle.getFeedBlock().get().getDescription());
+
+        reportResourceUsageFromNode(1, setOf(usage("cheese", 0.68), usage("wine", 0.3)));
+        bundle = ctrl.getClusterStateBundle();
+        assertTrue(bundle.clusterFeedIsBlocked());
+        // FIXME Effective limit is modified by hysteresis but due to how we check state deltas this
+        // is not discovered here. Still correct in terms of what resources are blocked or not, but
+        // the description is not up to date here.
+        assertEquals("cheese on node 1 [unknown hostname] (0.750 > 0.700)",
+                     bundle.getFeedBlock().get().getDescription());
+
+        // Trigger an explicit recompute by adding a separate resource exhaustion
+        reportResourceUsageFromNode(1, setOf(usage("cheese", 0.67), usage("wine", 0.5)));
+        bundle = ctrl.getClusterStateBundle();
+        assertTrue(bundle.clusterFeedIsBlocked());
+        assertEquals("cheese on node 1 [unknown hostname] (0.670 > 0.600), " +
+                     "wine on node 1 [unknown hostname] (0.500 > 0.400)", // Not under hysteresis
+                     bundle.getFeedBlock().get().getDescription());
+
+        // Wine usage drops beyond hysteresis range, should be unblocked immediately.
+        reportResourceUsageFromNode(1, setOf(usage("cheese", 0.61), usage("wine", 0.2)));
+        bundle = ctrl.getClusterStateBundle();
+        assertTrue(bundle.clusterFeedIsBlocked());
+        assertEquals("cheese on node 1 [unknown hostname] (0.610 > 0.600)",
+                     bundle.getFeedBlock().get().getDescription());
+
+        // Cheese now drops below hysteresis range, should be unblocked as well.
+        reportResourceUsageFromNode(1, setOf(usage("cheese", 0.59), usage("wine", 0.2)));
+        bundle = ctrl.getClusterStateBundle();
+        assertFalse(bundle.clusterFeedIsBlocked());
+    }
+
+    // FIXME implicit changes in limits due to hysteresis adds spurious exhaustion remove+add node event pair
+
 }
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/FeedBlockUtil.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/FeedBlockUtil.java
index 2254435e629..65199aa9957 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/FeedBlockUtil.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/FeedBlockUtil.java
@@ -89,6 +89,10 @@ public class FeedBlockUtil {
         return new NodeResourceExhaustion(new Node(NodeType.STORAGE, index), type, new ResourceUsage(0.8, null), 0.7, "foo");
     }
 
+    static NodeResourceExhaustion exhaustion(int index, String type, double usage) {
+        return new NodeResourceExhaustion(new Node(NodeType.STORAGE, index), type, new ResourceUsage(usage, null), 0.7, "foo");
+    }
+
     static Set<NodeResourceExhaustion> setOf(NodeResourceExhaustion... exhaustions) {
         return Arrays.stream(exhaustions).collect(Collectors.toCollection(LinkedHashSet::new));
     }
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ResourceExhaustionCalculatorTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ResourceExhaustionCalculatorTest.java
index f5f7b4676d8..55cf173aa25 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ResourceExhaustionCalculatorTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ResourceExhaustionCalculatorTest.java
@@ -5,8 +5,10 @@ import org.junit.Test;
 
 import static com.yahoo.vespa.clustercontroller.core.ClusterFixture.storageNode;
 import static com.yahoo.vespa.clustercontroller.core.FeedBlockUtil.createFixtureWithReportedUsages;
+import static com.yahoo.vespa.clustercontroller.core.FeedBlockUtil.exhaustion;
 import static com.yahoo.vespa.clustercontroller.core.FeedBlockUtil.forNode;
 import static com.yahoo.vespa.clustercontroller.core.FeedBlockUtil.mapOf;
+import static com.yahoo.vespa.clustercontroller.core.FeedBlockUtil.setOf;
 import static com.yahoo.vespa.clustercontroller.core.FeedBlockUtil.usage;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -98,4 +100,45 @@ public class ResourceExhaustionCalculatorTest {
         assertNull(feedBlock);
     }
 
+    @Test
+    public void retain_node_feed_block_status_when_within_hysteresis_window_limit_crossed_edge_case() {
+        var curFeedBlock = ClusterStateBundle.FeedBlock.blockedWith("foo", setOf(exhaustion(1, "memory", 0.51)));
+        var calc = new ResourceExhaustionCalculator(true, mapOf(usage("disk", 0.5), usage("memory", 0.5)), curFeedBlock, 0.1);
+        // Node 1 goes from 0.51 to 0.49, crossing the 0.5 threshold. Should still be blocked.
+        // Node 2 is at 0.49 but was not previously blocked and should not be blocked now either.
+        var cf = createFixtureWithReportedUsages(forNode(1, usage("disk", 0.3), usage("memory", 0.49)),
+                                                 forNode(2, usage("disk", 0.3), usage("memory", 0.49)));
+        var feedBlock = calc.inferContentClusterFeedBlockOrNull(cf.cluster().getNodeInfo());
+        assertNotNull(feedBlock);
+        // TODO should we not change the limits themselves? Explicit mention of hysteresis state?
+        assertEquals("memory on node 1 [storage.1.local] (0.490 > 0.400)",
+                     feedBlock.getDescription());
+    }
+
+    @Test
+    public void retain_node_feed_block_status_when_within_hysteresis_window_under_limit_edge_case() {
+        var curFeedBlock = ClusterStateBundle.FeedBlock.blockedWith("foo", setOf(exhaustion(1, "memory", 0.49)));
+        var calc = new ResourceExhaustionCalculator(true, mapOf(usage("disk", 0.5), usage("memory", 0.5)), curFeedBlock, 0.1);
+        // Node 1 goes from 0.49 to 0.48, NOT crossing the 0.5 threshold. Should still be blocked.
+        // Node 2 is at 0.49 but was not previously blocked and should not be blocked now either.
+        var cf = createFixtureWithReportedUsages(forNode(1, usage("disk", 0.3), usage("memory", 0.48)),
+                                                 forNode(2, usage("disk", 0.3), usage("memory", 0.49)));
+        var feedBlock = calc.inferContentClusterFeedBlockOrNull(cf.cluster().getNodeInfo());
+        assertNotNull(feedBlock);
+        assertEquals("memory on node 1 [storage.1.local] (0.480 > 0.400)",
+                feedBlock.getDescription());
+    }
+
+    @Test
+    public void retained_node_feed_block_cleared_once_hysteresis_threshold_is_passed() {
+        var curFeedBlock = ClusterStateBundle.FeedBlock.blockedWith("foo", setOf(exhaustion(1, "memory", 0.48)));
+        var calc = new ResourceExhaustionCalculator(true, mapOf(usage("disk", 0.5), usage("memory", 0.5)), curFeedBlock, 0.1);
+        // Node 1 goes from 0.48 to 0.39. Should be unblocked
+        // Node 2 is at 0.49 but was not previously blocked and should not be blocked now either.
+        var cf = createFixtureWithReportedUsages(forNode(1, usage("disk", 0.3), usage("memory", 0.39)),
+                                                 forNode(2, usage("disk", 0.3), usage("memory", 0.49)));
+        var feedBlock = calc.inferContentClusterFeedBlockOrNull(cf.cluster().getNodeInfo());
+        assertNull(feedBlock);
+    }
+
 }