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.

5 files changed, 75 insertions, 9 deletions

diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/EventDiffCalculator.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/EventDiffCalculator.java
index f4975ee4ee4..900f85be888 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/EventDiffCalculator.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/EventDiffCalculator.java
@@ -177,14 +177,14 @@ public class EventDiffCalculator {
         Set<NodeResourceExhaustion> fromBlockSet = params.feedBlockFrom != null ? params.feedBlockFrom.getConcreteExhaustions() : Collections.emptySet();
         Set<NodeResourceExhaustion> toBlockSet   = params.feedBlockTo   != null ? params.feedBlockTo.getConcreteExhaustions()   : Collections.emptySet();
 
-        for (var ex : setSubtraction(toBlockSet, fromBlockSet)) {
-            var info = cluster.getNodeInfo(ex.node);
-            events.add(createNodeEvent(info, String.format("Added resource exhaustion: %s", ex.toExhaustionAddedDescription()), params));
-        }
         for (var ex : setSubtraction(fromBlockSet, toBlockSet)) {
             var info = cluster.getNodeInfo(ex.node);
             events.add(createNodeEvent(info, String.format("Removed resource exhaustion: %s", ex.toExhaustionRemovedDescription()), params));
         }
+        for (var ex : setSubtraction(toBlockSet, fromBlockSet)) {
+            var info = cluster.getNodeInfo(ex.node);
+            events.add(createNodeEvent(info, String.format("Added resource exhaustion: %s", ex.toExhaustionAddedDescription()), params));
+        }
     }
 
     private static void emitSingleNodeEvents(PerStateParams params, List<Event> events, ContentCluster cluster, ClusterState fromState, ClusterState toState, Node n) {
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java
index 60b14e86f50..83efb5d8ded 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java
@@ -345,7 +345,6 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd
         if (!options.clusterFeedBlockEnabled) {
             return;
         }
-        // TODO hysteresis to prevent oscillations!
         var calc = createResourceExhaustionCalculator();
         // Important: nodeInfo contains the _current_ host info _prior_ to newHostInfo being applied.
         var previouslyExhausted = calc.enumerateNodeResourceExhaustions(nodeInfo);
@@ -953,7 +952,10 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd
     }
 
     private ResourceExhaustionCalculator createResourceExhaustionCalculator() {
-        return new ResourceExhaustionCalculator(options.clusterFeedBlockEnabled, options.clusterFeedBlockLimit);
+        return new ResourceExhaustionCalculator(
+                options.clusterFeedBlockEnabled, options.clusterFeedBlockLimit,
+                stateVersionTracker.getLatestCandidateStateBundle().getFeedBlockOrNull(),
+                options.clusterFeedBlockNoiseLevel);
     }
 
     private static ClusterStateDeriver createIdentityClonedBucketSpaceStateDeriver() {
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetControllerOptions.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetControllerOptions.java
index 9c5aaecd468..e63531229d6 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetControllerOptions.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetControllerOptions.java
@@ -138,6 +138,8 @@ public class FleetControllerOptions implements Cloneable {
     // Resource type -> limit in [0, 1]
     public Map<String, Double> clusterFeedBlockLimit = Collections.emptyMap();
 
+    public double clusterFeedBlockNoiseLevel = 0.01;
+
     public FleetControllerOptions(String clusterName, Collection<ConfiguredNode> nodes) {
         this.clusterName = clusterName;
         maxTransitionTime.put(NodeType.DISTRIBUTOR, 0);
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ResourceExhaustionCalculator.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ResourceExhaustionCalculator.java
index 231d9f95bdb..00edd767ad6 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ResourceExhaustionCalculator.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ResourceExhaustionCalculator.java
@@ -7,6 +7,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedHashSet;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -14,15 +15,67 @@ import java.util.stream.Collectors;
  * Given a mapping of (opaque) resource names and their exclusive limits,
  * this class acts as an utility to easily enumerate all the resources that
  * a given node (or set of nodes) have exhausted.
+ *
+ * In order to support hysteresis, optionally takes in the _current_ feed
+ * block state. This lets the calculator make the decision to emit a resource
+ * exhaustion for a node that is technically below the feed block limit, as
+ * long as it's not yet below the hysteresis threshold.
  */
 public class ResourceExhaustionCalculator {
 
     private final boolean feedBlockEnabled;
     private final Map<String, Double> feedBlockLimits;
+    private final double feedBlockNoiseLevel;
+    private final Set<NodeAndResourceType> previouslyBlockedNodeResources;
+
+    private static class NodeAndResourceType {
+        public final int nodeIndex;
+        public final String resourceType;
+
+        public NodeAndResourceType(int nodeIndex, String resourceType) {
+            this.nodeIndex = nodeIndex;
+            this.resourceType = resourceType;
+        }
+
+        public static NodeAndResourceType of(int nodeIndex, String resourceType) {
+            return new NodeAndResourceType(nodeIndex, resourceType);
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            NodeAndResourceType that = (NodeAndResourceType) o;
+            return nodeIndex == that.nodeIndex &&
+                    Objects.equals(resourceType, that.resourceType);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(nodeIndex, resourceType);
+        }
+    }
 
     public ResourceExhaustionCalculator(boolean feedBlockEnabled, Map<String, Double> feedBlockLimits) {
         this.feedBlockEnabled = feedBlockEnabled;
         this.feedBlockLimits = feedBlockLimits;
+        this.feedBlockNoiseLevel = 0.0;
+        this.previouslyBlockedNodeResources = Collections.emptySet();
+    }
+
+    public ResourceExhaustionCalculator(boolean feedBlockEnabled, Map<String, Double> feedBlockLimits,
+                                        ClusterStateBundle.FeedBlock previousFeedBlock,
+                                        double feedBlockNoiseLevel) {
+        this.feedBlockEnabled = feedBlockEnabled;
+        this.feedBlockLimits = feedBlockLimits;
+        this.feedBlockNoiseLevel = feedBlockNoiseLevel;
+        if (previousFeedBlock != null) {
+            this.previouslyBlockedNodeResources = previousFeedBlock.getConcreteExhaustions().stream()
+                    .map(ex -> NodeAndResourceType.of(ex.node.getIndex(), ex.resourceType))
+                    .collect(Collectors.toSet());
+        } else {
+            this.previouslyBlockedNodeResources = Collections.emptySet();
+        }
     }
 
     public ClusterStateBundle.FeedBlock inferContentClusterFeedBlockOrNull(Collection<NodeInfo> nodeInfos) {
@@ -50,13 +103,18 @@ public class ResourceExhaustionCalculator {
     public Set<NodeResourceExhaustion> resourceExhaustionsFromHostInfo(NodeInfo nodeInfo, HostInfo hostInfo) {
         Set<NodeResourceExhaustion> exceedingLimit = null;
         for (var usage : hostInfo.getContentNode().getResourceUsage().entrySet()) {
-            double limit = feedBlockLimits.getOrDefault(usage.getKey(), 1.0);
-            if (usage.getValue().getUsage() > limit) {
+            double configuredLimit = feedBlockLimits.getOrDefault(usage.getKey(), 1.0);
+            // To enable hysteresis on feed un-block we adjust the effective limit iff the particular
+            // <node, resource> tuple was blocked in the previous state.
+            boolean wasBlocked = previouslyBlockedNodeResources.contains(NodeAndResourceType.of(nodeInfo.getNodeIndex(), usage.getKey()));
+            double effectiveLimit = wasBlocked ? Math.max(configuredLimit - feedBlockNoiseLevel, 0.0)
+                                               : configuredLimit;
+            if (usage.getValue().getUsage() > effectiveLimit) {
                 if (exceedingLimit == null) {
                     exceedingLimit = new LinkedHashSet<>();
                 }
                 exceedingLimit.add(new NodeResourceExhaustion(nodeInfo.getNode(), usage.getKey(), usage.getValue(),
-                                                              limit, nodeInfo.getRpcAddress()));
+                                                              effectiveLimit, nodeInfo.getRpcAddress()));
             }
         }
         return (exceedingLimit != null) ? exceedingLimit : Collections.emptySet();
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateVersionTracker.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateVersionTracker.java
index e2f98cf5492..12338a5bafa 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateVersionTracker.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateVersionTracker.java
@@ -124,6 +124,10 @@ public class StateVersionTracker {
         return latestCandidateState.getBaselineAnnotatedState();
     }
 
+    public ClusterStateBundle getLatestCandidateStateBundle() {
+        return latestCandidateState;
+    }
+
     public List<ClusterStateHistoryEntry> getClusterStateHistory() {
         return Collections.unmodifiableList(clusterStateHistory);
     }