Rewrite and refactor core cluster controller state generation logic

Cluster controller will now generate the new cluster state on-demand in a "pure functional" way instead of conditionally patching a working state over time. This makes understanding (and changing) the state generation logic vastly easier than it previously was.

4 files changed, 239 insertions, 18 deletions

diff --git a/vdslib/src/main/java/com/yahoo/vdslib/state/ClusterState.java b/vdslib/src/main/java/com/yahoo/vdslib/state/ClusterState.java
index b3d572e48ae..d70b55c66a2 100644
--- a/vdslib/src/main/java/com/yahoo/vdslib/state/ClusterState.java
+++ b/vdslib/src/main/java/com/yahoo/vdslib/state/ClusterState.java
@@ -11,6 +11,9 @@ import java.util.*;
  */
 public class ClusterState implements Cloneable {
 
+    private static final NodeState DEFAULT_STORAGE_UP_NODE_STATE = new NodeState(NodeType.STORAGE, State.UP);
+    private static final NodeState DEFAULT_DISTRIBUTOR_UP_NODE_STATE = new NodeState(NodeType.DISTRIBUTOR, State.UP);
+
     private int version = 0;
     private State state = State.DOWN;
     // nodeStates maps each of the non-up nodes that have an index <= the node count for its type.
@@ -30,6 +33,22 @@ public class ClusterState implements Cloneable {
         deserialize(serialized);
     }
 
+    /**
+     * Parse a given cluster state string into a returned ClusterState instance, wrapping any
+     * parse exceptions in a RuntimeException.
+     */
+    public static ClusterState stateFromString(final String stateStr) {
+        try {
+            return new ClusterState(stateStr);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public static ClusterState emptyState() {
+        return stateFromString("");
+    }
+
     public ClusterState clone() {
         try{
             ClusterState state = (ClusterState) super.clone();
@@ -61,22 +80,81 @@ public class ClusterState implements Cloneable {
         return true;
     }
 
+    @FunctionalInterface
+    private interface NodeStateCmp {
+        boolean similar(NodeType nodeType, NodeState lhs, NodeState rhs);
+    }
+
     public boolean similarTo(Object o) {
         if (!(o instanceof ClusterState)) { return false; }
-        ClusterState other = (ClusterState) o;
+        final ClusterState other = (ClusterState) o;
 
-        if (state.equals(State.DOWN) && other.state.equals(State.DOWN)) return true; // both down, means equal (why??)
-        if (version != other.version || !state.equals(other.state)) return false;
-        if (distributionBits != other.distributionBits) return false;
-        if ( ! nodeCount.equals(other.nodeCount)) return false;
+        return similarToImpl(other, this::normalizedNodeStateSimilarTo);
+    }
+
+    public boolean similarToIgnoringInitProgress(final ClusterState other) {
+        return similarToImpl(other, this::normalizedNodeStateSimilarToIgnoringInitProgress);
+    }
 
-        for (Map.Entry<Node, NodeState> nodeStateEntry : nodeStates.entrySet()) {
-            NodeState otherNodeState = other.nodeStates.get(nodeStateEntry.getKey());
-            if (otherNodeState == null || ! otherNodeState.similarTo(nodeStateEntry.getValue())) return false;
+    private boolean similarToImpl(final ClusterState other, final NodeStateCmp nodeStateCmp) {
+        // Two cluster states are considered similar if they are both down. When clusters
+        // are down, their individual node states do not matter to ideal state computations
+        // and content nodes therefore do not need to observe them.
+        if (state.equals(State.DOWN) && other.state.equals(State.DOWN)) {
+            return true;
+        }
+        if (!metaInformationSimilarTo(other)) {
+            return false;
+        }
+        // TODO verify behavior of C++ impl against this
+        for (Node node : unionNodeSetWith(other.nodeStates.keySet())) {
+            final NodeState lhs = nodeStates.get(node);
+            final NodeState rhs = other.nodeStates.get(node);
+            if (!nodeStateCmp.similar(node.getType(), lhs, rhs)) {
+                return false;
+            }
         }
         return true;
     }
 
+    private Set<Node> unionNodeSetWith(final Set<Node> otherNodes) {
+        final Set<Node> unionNodeSet = new TreeSet<Node>(nodeStates.keySet());
+        unionNodeSet.addAll(otherNodes);
+        return unionNodeSet;
+    }
+
+    private boolean metaInformationSimilarTo(final ClusterState other) {
+        if (version != other.version || !state.equals(other.state)) {
+            return false;
+        }
+        if (distributionBits != other.distributionBits) {
+            return false;
+        }
+        return nodeCount.equals(other.nodeCount);
+    }
+
+    private boolean normalizedNodeStateSimilarTo(final NodeType nodeType, final NodeState lhs, final NodeState rhs) {
+        final NodeState lhsNormalized = (lhs != null ? lhs : defaultUpNodeState(nodeType));
+        final NodeState rhsNormalized = (rhs != null ? rhs : defaultUpNodeState(nodeType));
+
+        return lhsNormalized.similarTo(rhsNormalized);
+    }
+
+    private boolean normalizedNodeStateSimilarToIgnoringInitProgress(
+            final NodeType nodeType, final NodeState lhs, final NodeState rhs)
+    {
+        final NodeState lhsNormalized = (lhs != null ? lhs : defaultUpNodeState(nodeType));
+        final NodeState rhsNormalized = (rhs != null ? rhs : defaultUpNodeState(nodeType));
+
+        return lhsNormalized.similarToIgnoringInitProgress(rhsNormalized);
+    }
+
+    private static NodeState defaultUpNodeState(final NodeType nodeType) {
+        return nodeType == NodeType.STORAGE
+                ? DEFAULT_STORAGE_UP_NODE_STATE
+                : DEFAULT_DISTRIBUTOR_UP_NODE_STATE;
+    }
+
     /**
      * Fleet controller marks states that are actually sent out to nodes as official states. Only fleetcontroller
      * should set this to official, and only just before sending it out. This state is currently not serialized with
@@ -97,7 +175,7 @@ public class ClusterState implements Cloneable {
         public void addNodeState() throws ParseException {
             if (!empty) {
                 NodeState ns = NodeState.deserialize(node.getType(), sb.toString());
-                if (!ns.equals(new NodeState(node.getType(), State.UP))) {
+                if (!ns.equals(defaultUpNodeState(node.getType()))) {
                     nodeStates.put(node, ns);
                 }
                 if (nodeCount.get(node.getType().ordinal()) <= node.getIndex()) {
diff --git a/vdslib/src/main/java/com/yahoo/vdslib/state/NodeState.java b/vdslib/src/main/java/com/yahoo/vdslib/state/NodeState.java
index 8c31938dfaf..15c929fe49d 100644
--- a/vdslib/src/main/java/com/yahoo/vdslib/state/NodeState.java
+++ b/vdslib/src/main/java/com/yahoo/vdslib/state/NodeState.java
@@ -112,17 +112,27 @@ public class NodeState implements Cloneable {
      * Cluster state will check for that.
      */
     public boolean similarTo(Object o) {
-        if (!(o instanceof NodeState)) { return false; }
-        NodeState other = (NodeState) o;
+        if (!(o instanceof NodeState)) {
+            return false;
+        }
+        return similarToImpl((NodeState)o, true);
+    }
+
+    public boolean similarToIgnoringInitProgress(final NodeState other) {
+        return similarToImpl(other, false);
+    }
 
+    private boolean similarToImpl(final NodeState other, boolean considerInitProgress) {
         if (state != other.state) return false;
         if (Math.abs(capacity - other.capacity) > 0.0000000001) return false;
         if (Math.abs(reliability - other.reliability) > 0.0000000001) return false;
         if (startTimestamp != other.startTimestamp) return false;
 
         // Init progress on different sides of the init progress limit boundary is not similar.
-        if (type.equals(NodeType.STORAGE)
-            && initProgress < getListingBucketsInitProgressLimit() ^ other.initProgress < getListingBucketsInitProgressLimit())
+        if (considerInitProgress
+            && type.equals(NodeType.STORAGE)
+            && (initProgress < getListingBucketsInitProgressLimit()
+                ^ other.initProgress < getListingBucketsInitProgressLimit()))
         {
             return false;
         }
diff --git a/vdslib/src/test/java/com/yahoo/vdslib/state/ClusterStateTestCase.java b/vdslib/src/test/java/com/yahoo/vdslib/state/ClusterStateTestCase.java
index c058a7c9919..0d06fcc6faa 100644
--- a/vdslib/src/test/java/com/yahoo/vdslib/state/ClusterStateTestCase.java
+++ b/vdslib/src/test/java/com/yahoo/vdslib/state/ClusterStateTestCase.java
@@ -1,10 +1,18 @@
 // Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vdslib.state;
 
+import org.junit.Test;
+
 import java.text.ParseException;
+import java.util.function.BiFunction;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
-public class ClusterStateTestCase extends junit.framework.TestCase {
+public class ClusterStateTestCase{
 
+    @Test
     public void testSetNodeState() throws ParseException {
         ClusterState state = new ClusterState("");
         assertEquals("", state.toString());
@@ -22,6 +30,7 @@ public class ClusterStateTestCase extends junit.framework.TestCase {
         assertEquals("distributor:5 .0.s:d .2.s:d .3.s:d storage:1 .0.d:4 .0.d.1.s:d", state.toString());
     }
 
+    @Test
     public void testClone() throws ParseException {
         ClusterState state = new ClusterState("");
         state.setNodeState(new Node(NodeType.DISTRIBUTOR, 1), new NodeState(NodeType.DISTRIBUTOR, State.UP).setDescription("available"));
@@ -31,8 +40,9 @@ public class ClusterStateTestCase extends junit.framework.TestCase {
         assertEquals(state.toString(true), other.toString(true));
         assertEquals(state.toString(false), other.toString(false));
         assertEquals(state, other);
-   }
+    }
 
+    @Test
     public void testEquals() throws ParseException {
         ClusterState state = new ClusterState("");
 
@@ -55,6 +65,7 @@ public class ClusterStateTestCase extends junit.framework.TestCase {
             ClusterState state2 = new ClusterState("distributor:3 .1.s:d .2.s:m storage:3 .1.s:i .2.s:m");
             assertFalse(state1.equals(state2));
             assertFalse(state1.similarTo(state2));
+            assertFalse(state1.similarToIgnoringInitProgress(state2));
         }
 
         {
@@ -62,6 +73,7 @@ public class ClusterStateTestCase extends junit.framework.TestCase {
             ClusterState state2 = new ClusterState("cluster:d version:1 bits:20 distributor:1 storage:1 .0.s:d");
             assertFalse(state1.equals(state2));
             assertTrue(state1.similarTo(state2));
+            assertTrue(state1.similarToIgnoringInitProgress(state2));
         }
 
         {
@@ -69,6 +81,7 @@ public class ClusterStateTestCase extends junit.framework.TestCase {
             ClusterState state2 = new ClusterState("distributor:3 storage:3");
             assertFalse(state1.equals(state2));
             assertFalse(state1.similarTo(state2));
+            assertFalse(state1.similarToIgnoringInitProgress(state2));
         }
 
         assertFalse(state.equals("class not instance of ClusterState"));
@@ -78,6 +91,92 @@ public class ClusterStateTestCase extends junit.framework.TestCase {
         assertTrue(state.similarTo(state));
     }
 
+    private static ClusterState stateFromString(final String stateStr) {
+        try {
+            return new ClusterState(stateStr);
+        } catch (ParseException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private void do_test_differing_storage_node_sets(BiFunction<ClusterState, ClusterState, Boolean> cmp) {
+        final ClusterState a = stateFromString("distributor:3 storage:3 .0.s:d");
+        final ClusterState b = stateFromString("distributor:3 storage:3");
+        assertFalse(cmp.apply(a, b));
+        assertFalse(cmp.apply(b, a));
+        assertTrue(cmp.apply(a, a));
+        assertTrue(cmp.apply(b, b));
+    }
+
+    private void do_test_differing_distributor_node_sets(BiFunction<ClusterState, ClusterState, Boolean> cmp) {
+        final ClusterState a = stateFromString("distributor:3 .0.s:d storage:3");
+        final ClusterState b = stateFromString("distributor:3 storage:3");
+        assertFalse(cmp.apply(a, b));
+        assertFalse(cmp.apply(b, a));
+        assertTrue(cmp.apply(a, a));
+        assertTrue(cmp.apply(b, b));
+    }
+
+    @Test
+    public void similarity_check_considers_differing_distributor_node_state_sets() {
+        do_test_differing_distributor_node_sets((a, b) -> a.similarTo(b));
+    }
+
+    @Test
+    public void similarity_check_considers_differing_storage_node_state_sets() {
+        do_test_differing_storage_node_sets((a, b) -> a.similarTo(b));
+    }
+
+    @Test
+    public void structural_similarity_check_considers_differing_distributor_node_state_sets() {
+        do_test_differing_distributor_node_sets((a, b) -> a.similarToIgnoringInitProgress(b));
+    }
+
+    @Test
+    public void init_progress_ignoring_similarity_check_considers_differing_storage_node_state_sets() {
+        do_test_differing_storage_node_sets((a, b) -> a.similarToIgnoringInitProgress(b));
+    }
+
+    private void do_test_similarity_for_down_cluster_state(BiFunction<ClusterState, ClusterState, Boolean> cmp) {
+        final ClusterState a = stateFromString("cluster:d distributor:3 .0.s:d storage:3 .2:s:d");
+        final ClusterState b = stateFromString("cluster:d distributor:3 storage:3 .1:s:d");
+        assertTrue(cmp.apply(a, b));
+        assertTrue(cmp.apply(b, a));
+    }
+
+    @Test
+    public void similarity_check_considers_differing_down_cluster_states_similar() {
+        do_test_similarity_for_down_cluster_state((a, b) -> a.similarTo(b));
+    }
+
+    @Test
+    public void init_progress_ignoring__similarity_check_considers_differing_down_cluster_states_similar() {
+        do_test_similarity_for_down_cluster_state((a, b) -> a.similarToIgnoringInitProgress(b));
+    }
+
+    // If we naively only look at the NodeState sets in the ClusterState instances to be
+    // compared, we might get false positives. If state A has a NodeState(Up, minBits 15)
+    // while state B has NodeState(Up, minBits 16), the latter will be pruned away from the
+    // NodeState set because it's got a "default" Up state. The two states are still semantically
+    // similar, and should be returned as such. But their state sets technically differ.
+    @Test
+    public void similarity_check_does_not_consider_per_storage_node_min_bits() {
+        final ClusterState a = stateFromString("distributor:4 storage:4");
+        final ClusterState b = stateFromString("distributor:4 storage:4");
+        b.setNodeState(new Node(NodeType.STORAGE, 1), new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(15));
+        assertTrue(a.similarTo(b));
+        assertTrue(b.similarTo(a));
+    }
+
+    @Test
+    public void init_progress_ignoring_similarity_check_does_in_fact_ignore_init_progress() {
+        final ClusterState a = stateFromString("distributor:3 storage:3 .0.i:0.01 .1.i:0.1 .2.i:0.9");
+        final ClusterState b = stateFromString("distributor:3 storage:3 .0.i:0.2 .1.i:0.5 .2.i:0.99");
+        assertTrue(a.similarToIgnoringInitProgress(b));
+        assertTrue(b.similarToIgnoringInitProgress(a));
+    }
+
+    @Test
     public void testTextDiff() throws ParseException {
         ClusterState state1 = new ClusterState("distributor:9 storage:4");
         ClusterState state2 = new ClusterState("distributor:7 storage:6");
@@ -94,6 +193,7 @@ public class ClusterStateTestCase extends junit.framework.TestCase {
         assertEquals("version: 123 => 0, bits: 16 => 21, official: false => true, storage: [2: [Initializing => Up, disks: 2 => 0, description: Booting => ], 4: Down => Up, 5: Down => Up], distributor: [7: Up => Down, 8: Up => Down]", state1.getTextualDifference(state2));
     }
 
+    @Test
     public void testHtmlDiff() throws ParseException {
         ClusterState state1 = new ClusterState("distributor:9 storage:4");
         ClusterState state2 = new ClusterState("distributor:7 storage:6");
@@ -133,7 +233,7 @@ public class ClusterStateTestCase extends junit.framework.TestCase {
                 "]", state1.getHtmlDifference(state2));
     }
 
-
+    @Test
     public void testParser() throws ParseException {
         ClusterState state = new ClusterState("distributor:2 storage:17 .2.s:d .13.s:r m:cluster\\x20message");
         assertEquals("cluster message", state.getDescription());
@@ -191,17 +291,20 @@ public class ClusterStateTestCase extends junit.framework.TestCase {
         } catch (Exception e) {}
     }
 
+    @Test
     public void testCapacityExponential() throws ParseException {
 
         ClusterState state = new ClusterState("distributor:27 storage:170 .2.s:d .13.c:3E-8 .13.s:r");
-        assertEquals(3E-8, state.getNodeState(new Node(NodeType.STORAGE, 13)).getCapacity());
+        assertEquals(3E-8, state.getNodeState(new Node(NodeType.STORAGE, 13)).getCapacity(), 1E-8);
     }
 
+    @Test
     public void testCapacityExponentialCpp() throws ParseException {
         ClusterState state = new ClusterState("distributor:27 storage:170 .2.s:d .13.c:3e-08 .13.s:r");
-        assertEquals(3E-8, state.getNodeState(new Node(NodeType.STORAGE, 13)).getCapacity());
+        assertEquals(3E-8, state.getNodeState(new Node(NodeType.STORAGE, 13)).getCapacity(), 1E-8);
     }
 
+    @Test
     public void testSetState() throws ParseException {
         ClusterState state = new ClusterState("distributor:2 storage:2");
         state.setNodeState(new Node(NodeType.DISTRIBUTOR, 0), new NodeState(NodeType.DISTRIBUTOR, State.DOWN));
@@ -209,6 +312,7 @@ public class ClusterStateTestCase extends junit.framework.TestCase {
         assertEquals("distributor:2 .0.s:d storage:2", state.toString());
     }
 
+    @Test
     public void testVersionAndClusterStates() throws ParseException {
         ClusterState state = new ClusterState("version:4 cluster:i distributor:2 .1.s:i storage:2 .0.s:i .0.i:0.345");
         assertEquals(4, state.getVersion());
@@ -220,6 +324,7 @@ public class ClusterStateTestCase extends junit.framework.TestCase {
         assertEquals("version:5 cluster:d bits:12 distributor:2 .1.s:i .1.i:1.0 storage:2 .0.s:i .0.i:0.345", state.toString());
     }
 
+    @Test
     public void testNotRemovingCommentedDownNodesAtEnd() throws ParseException {
         ClusterState state = new ClusterState("");
         state.setNodeState(new Node(NodeType.DISTRIBUTOR, 0), new NodeState(NodeType.DISTRIBUTOR, State.UP));
@@ -234,6 +339,7 @@ public class ClusterStateTestCase extends junit.framework.TestCase {
         assertEquals("distributor:1 storage:2", state.toString(false));
     }
 
+    @Test
     public void testWhitespace() throws ParseException {
         ClusterState state = new ClusterState("distributor:2\n  .1.t:3\nstorage:2\n\t.0.s:i   \r\f.1.s:m");
         assertEquals(2, state.getNodeCount(NodeType.DISTRIBUTOR));
@@ -243,4 +349,22 @@ public class ClusterStateTestCase extends junit.framework.TestCase {
         assertEquals(new NodeState(NodeType.STORAGE, State.INITIALIZING), state.getNodeState(new Node(NodeType.STORAGE, 0)));
         assertEquals(new NodeState(NodeType.STORAGE, State.MAINTENANCE), state.getNodeState(new Node(NodeType.STORAGE, 1)));
     }
+
+    @Test
+    public void empty_state_factory_method_returns_empty_state() {
+        final ClusterState state = ClusterState.emptyState();
+        assertEquals("", state.toString());
+    }
+
+    @Test
+    public void state_from_string_factory_method_returns_cluster_state_constructed_from_input() {
+        final String stateStr = "version:123 distributor:2 storage:2";
+        final ClusterState state = ClusterState.stateFromString(stateStr);
+        assertEquals(stateStr, state.toString());
+    }
+
+    @Test(expected=RuntimeException.class)
+    public void state_from_string_factory_method_throws_runtime_exception_on_parse_failure() {
+        ClusterState.stateFromString("fraggle rock");
+    }
 }
diff --git a/vdslib/src/test/java/com/yahoo/vdslib/state/NodeStateTestCase.java b/vdslib/src/test/java/com/yahoo/vdslib/state/NodeStateTestCase.java
index 63137a92c7b..9362838b63c 100644
--- a/vdslib/src/test/java/com/yahoo/vdslib/state/NodeStateTestCase.java
+++ b/vdslib/src/test/java/com/yahoo/vdslib/state/NodeStateTestCase.java
@@ -165,6 +165,12 @@ public class NodeStateTestCase extends junit.framework.TestCase {
             assertFalse(ns2.similarTo(ns3));
             assertTrue(ns3.similarTo(ns4));
 
+            assertTrue(ns1.similarToIgnoringInitProgress(ns2));
+            assertTrue(ns1.similarToIgnoringInitProgress(ns3));
+            assertTrue(ns3.similarToIgnoringInitProgress(ns1));
+            assertTrue(ns1.similarToIgnoringInitProgress(ns4));
+            assertTrue(ns2.similarToIgnoringInitProgress(ns4));
+
             assertFalse(ns1.equals(ns2));
             assertFalse(ns2.equals(ns3));
             assertFalse(ns3.equals(ns4));
@@ -176,6 +182,7 @@ public class NodeStateTestCase extends junit.framework.TestCase {
             NodeState ns1 = new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(16);
             NodeState ns2 = new NodeState(NodeType.STORAGE, State.UP).setMinUsedBits(18);
             assertTrue(ns1.similarTo(ns2));
+            assertTrue(ns1.similarToIgnoringInitProgress(ns2));
             assertFalse(ns1.equals(ns2));
         }
         {
@@ -184,12 +191,14 @@ public class NodeStateTestCase extends junit.framework.TestCase {
             assertEquals(ns, ns2Disks);
             assertEquals(ns2Disks, ns);
             assertTrue(ns.similarTo(ns2Disks));
+            assertTrue(ns.similarToIgnoringInitProgress(ns2Disks));
             assertTrue(ns2Disks.similarTo(ns));
 
             ns2Disks.getDiskState(0).setState(State.DOWN);
             assertFalse(ns.equals(ns2Disks));
             assertFalse(ns2Disks.equals(ns));
             assertFalse(ns.similarTo(ns2Disks));
+            assertFalse(ns.similarToIgnoringInitProgress(ns2Disks));
             assertFalse(ns2Disks.similarTo(ns));
         }
     }