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.

1 files changed, 73 insertions, 125 deletions

diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/GroupAutoTakedownTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/GroupAutoTakedownTest.java
index be60fba234a..a7307e0180a 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/GroupAutoTakedownTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/GroupAutoTakedownTest.java
@@ -9,19 +9,22 @@ import com.yahoo.vdslib.state.NodeType;
 import com.yahoo.vdslib.state.State;
 import com.yahoo.vespa.clustercontroller.core.database.DatabaseHandler;
 import com.yahoo.vespa.clustercontroller.core.listeners.NodeStateOrHostInfoChangeHandler;
-import com.yahoo.vespa.clustercontroller.core.listeners.SystemStateListener;
+
+import static com.yahoo.vespa.clustercontroller.core.matchers.EventForNode.eventForNode;
+import static com.yahoo.vespa.clustercontroller.core.matchers.NodeEventWithDescription.nodeEventWithDescription;
 import org.junit.Test;
-import org.mockito.ArgumentMatcher;
 
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 
 import static org.hamcrest.core.AllOf.allOf;
+import static org.hamcrest.core.IsCollectionContaining.hasItem;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.argThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -43,26 +46,29 @@ public class GroupAutoTakedownTest {
     }
 
     private static void setSharedFixtureOptions(ClusterFixture fixture, double minNodeRatioPerGroup) {
-        fixture.generator.setMinNodeRatioPerGroup(minNodeRatioPerGroup);
+        fixture.setMinNodeRatioPerGroup(minNodeRatioPerGroup);
         fixture.disableTransientMaintenanceModeOnDown();
         fixture.disableAutoClusterTakedown();
         fixture.bringEntireClusterUp();
     }
 
     private String stateAfterStorageTransition(ClusterFixture fixture, final int index, final State state) {
-        transitionStoreNodeToState(fixture, index, state);
+        transitionStorageNodeToState(fixture, index, state);
         return fixture.generatedClusterState();
     }
 
     private String verboseStateAfterStorageTransition(ClusterFixture fixture, final int index, final State state) {
-        transitionStoreNodeToState(fixture, index, state);
+        transitionStorageNodeToState(fixture, index, state);
         return fixture.verboseGeneratedClusterState();
     }
 
-    private void transitionStoreNodeToState(ClusterFixture fixture, int index, State state) {
+    private void transitionStorageNodeToState(ClusterFixture fixture, int index, State state) {
         fixture.reportStorageNodeState(index, state);
-        SystemStateListener listener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
+    }
+
+    private AnnotatedClusterState annotatedStateAfterStorageTransition(ClusterFixture fixture, final int index, final State state) {
+        transitionStorageNodeToState(fixture, index, state);
+        return fixture.annotatedGeneratedClusterState();
     }
 
     /**
@@ -74,12 +80,9 @@ public class GroupAutoTakedownTest {
     public void config_does_not_apply_to_flat_hierarchy_clusters() {
         ClusterFixture fixture = createFixtureForAllUpFlatCluster(5, 0.99);
 
-        SystemStateListener listener = mock(SystemStateListener.class);
-        // First invocation; generates initial state and clears "new state" flag
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
-        assertEquals("version:1 distributor:5 storage:5", fixture.generatedClusterState());
+        assertEquals("distributor:5 storage:5", fixture.generatedClusterState());
 
-        assertEquals("version:2 distributor:5 storage:5 .1.s:d",
+        assertEquals("distributor:5 storage:5 .1.s:d",
                      stateAfterStorageTransition(fixture, 1, State.DOWN));
     }
 
@@ -88,15 +91,13 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
 
-        SystemStateListener listener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
-        assertEquals("version:1 distributor:6 storage:6", fixture.generatedClusterState());
+        assertEquals("distributor:6 storage:6", fixture.generatedClusterState());
 
         // Same group as node 4
-        assertEquals("version:2 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                      stateAfterStorageTransition(fixture, 5, State.DOWN));
         // Same group as node 1
-        assertEquals("version:3 distributor:6 storage:4 .0.s:d .1.s:d",
+        assertEquals("distributor:6 storage:4 .0.s:d .1.s:d",
                      stateAfterStorageTransition(fixture, 0, State.DOWN));
     }
 
@@ -106,11 +107,11 @@ public class GroupAutoTakedownTest {
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
 
         // Group #2 -> down
-        assertEquals("version:1 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                      stateAfterStorageTransition(fixture, 5, State.DOWN));
 
         // Group #2 -> back up again
-        assertEquals("version:2 distributor:6 storage:6",
+        assertEquals("distributor:6 storage:6",
                      stateAfterStorageTransition(fixture, 5, State.UP));
     }
 
@@ -119,16 +120,12 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
 
-        assertEquals("version:1 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
 
         // 4, 5 in same group; this should not cause a new state since it's already implicitly down
         fixture.reportStorageNodeState(4, State.DOWN);
-
-        SystemStateListener listener = mock(SystemStateListener.class);
-        assertFalse(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
-
-        assertEquals("version:1 distributor:6 storage:4", fixture.generatedClusterState());
+        assertEquals("distributor:6 storage:4", fixture.generatedClusterState());
     }
 
     @Test
@@ -139,7 +136,7 @@ public class GroupAutoTakedownTest {
         // Nodes 6 and 7 are taken down implicitly and should have a message reflecting this.
         // Node 8 is taken down by the fixture and gets a fixture-assigned message that
         // we should _not_ lose/overwrite.
-        assertEquals("version:1 distributor:9 storage:9 .6.s:d " +
+        assertEquals("distributor:9 storage:9 .6.s:d " +
                         ".6.m:group\\x20node\\x20availability\\x20below\\x20configured\\x20threshold " +
                         ".7.s:d " +
                         ".7.m:group\\x20node\\x20availability\\x20below\\x20configured\\x20threshold " +
@@ -151,12 +148,12 @@ public class GroupAutoTakedownTest {
     public void legacy_cluster_wide_availabilty_ratio_is_computed_after_group_takedowns() {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
-        fixture.generator.setMinNodesUp(5, 5, 0.51, 0.51);
+        fixture.setMinNodesUp(5, 5, 0.51, 0.51);
 
         // Taking down a node in a group forces the entire group down, which leaves us with
         // only 4 content nodes (vs. minimum of 5 as specified above). The entire cluster
         // should be marked as down in this case.
-        assertEquals("version:1 cluster:d distributor:6 storage:4",
+        assertEquals("cluster:d distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
     }
 
@@ -165,16 +162,12 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(3), 0.99);
 
-        NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 5));
-        fixture.generator.proposeNewNodeState(nodeInfo, new NodeState(NodeType.STORAGE, State.MAINTENANCE));
-        SystemStateListener listener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
-
+        fixture.proposeStorageNodeWantedState(5, State.MAINTENANCE);
         // Maintenance not counted as down, so group still up
-        assertEquals("version:1 distributor:9 storage:9 .5.s:m", fixture.generatedClusterState());
+        assertEquals("distributor:9 storage:9 .5.s:m", fixture.generatedClusterState());
 
         // Group goes down, but maintenance node should still be in maintenance
-        assertEquals("version:2 distributor:9 storage:9 .3.s:d .4.s:d .5.s:m",
+        assertEquals("distributor:9 storage:9 .3.s:d .4.s:d .5.s:m",
                 stateAfterStorageTransition(fixture, 4, State.DOWN));
     }
 
@@ -186,51 +179,16 @@ public class GroupAutoTakedownTest {
 
         // Our timers are mocked, so taking down node 4 will deterministically transition to
         // a transient maintenance mode. Group should not be taken down here.
-        assertEquals("version:1 distributor:9 storage:9 .4.s:m",
+        assertEquals("distributor:9 storage:9 .4.s:m",
                 stateAfterStorageTransition(fixture, 4, State.DOWN));
 
         // However, once grace period expires the group should be taken down.
         fixture.timer.advanceTime(1001);
         NodeStateOrHostInfoChangeHandler changeListener = mock(NodeStateOrHostInfoChangeHandler.class);
-        fixture.generator.watchTimers(fixture.cluster, changeListener);
-        SystemStateListener stateListener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, stateListener));
-
-        assertEquals("version:2 distributor:9 storage:9 .3.s:d .4.s:d .5.s:d", fixture.generatedClusterState());
-    }
-
-    private static class NodeEventWithDescription extends ArgumentMatcher<NodeEvent> {
-        private final String expected;
-
-        NodeEventWithDescription(String expected) {
-            this.expected = expected;
-        }
-
-        @Override
-        public boolean matches(Object o) {
-            return expected.equals(((NodeEvent)o).getDescription());
-        }
-    }
+        fixture.nodeStateChangeHandler.watchTimers(
+                fixture.cluster, fixture.annotatedGeneratedClusterState().getClusterState(), changeListener);
 
-    private static NodeEventWithDescription nodeEventWithDescription(String description) {
-        return new NodeEventWithDescription(description);
-    }
-
-    private static class EventForNode extends ArgumentMatcher<NodeEvent> {
-        private final Node expected;
-
-        EventForNode(Node expected) {
-            this.expected = expected;
-        }
-
-        @Override
-        public boolean matches(Object o) {
-            return ((NodeEvent)o).getNode().getNode().equals(expected);
-        }
-    }
-
-    private static EventForNode eventForNode(Node expected) {
-        return new EventForNode(expected);
+        assertEquals("distributor:9 storage:9 .3.s:d .4.s:d .5.s:d", fixture.generatedClusterState());
     }
 
     private static Node contentNode(int index) {
@@ -242,13 +200,14 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
 
-        assertEquals("version:1 distributor:6 storage:4",
-                stateAfterStorageTransition(fixture, 5, State.DOWN));
+        final List<Event> events = EventDiffCalculator.computeEventDiff(EventDiffCalculator.params()
+                .cluster(fixture.cluster)
+                .fromState(fixture.annotatedGeneratedClusterState())
+                .toState(annotatedStateAfterStorageTransition(fixture, 5, State.DOWN)));
 
-        verify(fixture.eventLog).addNodeOnlyEvent(argThat(allOf(
-                nodeEventWithDescription("Setting node down as the total availability of its group is " +
-                        "below the configured threshold"),
-                eventForNode(contentNode(4)))), any());
+        assertThat(events, hasItem(allOf(
+                nodeEventWithDescription("Group node availability is below configured threshold"),
+                eventForNode(contentNode(4)))));
     }
 
     @Test
@@ -256,30 +215,31 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
 
-        assertEquals("version:1 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
-        assertEquals("version:2 distributor:6 storage:6",
-                stateAfterStorageTransition(fixture, 5, State.UP));
 
-        verify(fixture.eventLog).addNodeOnlyEvent(argThat(allOf(
-                nodeEventWithDescription("Group availability restored; taking node back up"),
-                eventForNode(contentNode(4)))), any());
+        final List<Event> events = EventDiffCalculator.computeEventDiff(EventDiffCalculator.params()
+                .cluster(fixture.cluster)
+                .fromState(fixture.annotatedGeneratedClusterState())
+                .toState(annotatedStateAfterStorageTransition(fixture, 5, State.UP)));
+
+        assertThat(events, hasItem(allOf(
+                nodeEventWithDescription("Group node availability has been restored"),
+                eventForNode(contentNode(4)))));
     }
 
     @Test
-    public void wanted_state_retired_implicitly_down_node_transitioned_it_to_retired_mode_immediately() {
+    public void wanted_state_retired_implicitly_down_node_is_transitioned_to_retired_mode_immediately() {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(3), 0.99);
 
-        assertEquals("version:1 distributor:9 storage:6",
+        assertEquals("distributor:9 storage:6",
                 stateAfterStorageTransition(fixture, 6, State.DOWN));
         // Node 7 is implicitly down. Mark wanted state as retired. It should now be Retired
         // but not Down.
         fixture.proposeStorageNodeWantedState(7, State.RETIRED);
 
-        SystemStateListener stateListener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, stateListener));
-        assertEquals("version:2 distributor:9 storage:8 .6.s:d .7.s:r", fixture.generatedClusterState());
+        assertEquals("distributor:9 storage:8 .6.s:d .7.s:r", fixture.generatedClusterState());
     }
 
     @Test
@@ -287,9 +247,9 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.49);
 
-        assertEquals("version:1 distributor:6 storage:6 .4.s:d",
+        assertEquals("distributor:6 storage:6 .4.s:d",
                 stateAfterStorageTransition(fixture, 4, State.DOWN));
-        assertEquals("version:2 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
 
         // Node 5 gets config-retired under our feet.
@@ -299,9 +259,8 @@ public class GroupAutoTakedownTest {
         // TODO this should ideally also set the retired flag in the distribution
         // config, but only the ConfiguredNodes are actually looked at currently.
         fixture.cluster.setNodes(nodes);
-        fixture.generator.setNodes(fixture.cluster.clusterInfo());
 
-        assertEquals("version:3 distributor:6 storage:6 .4.s:d .5.s:r",
+        assertEquals("distributor:6 storage:6 .4.s:d .5.s:r",
                 stateAfterStorageTransition(fixture, 5, State.UP));
     }
 
@@ -314,14 +273,12 @@ public class GroupAutoTakedownTest {
         newState.setInitProgress(0.5);
 
         fixture.reportStorageNodeState(4, newState);
-        SystemStateListener stateListener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, stateListener));
 
-        assertEquals("version:1 distributor:6 storage:6 .4.s:i .4.i:0.5", fixture.generatedClusterState());
+        assertEquals("distributor:6 storage:6 .4.s:i .4.i:0.5", fixture.generatedClusterState());
 
-        assertEquals("version:2 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
-        assertEquals("version:3 distributor:6 storage:6 .4.s:i .4.i:0.5",
+        assertEquals("distributor:6 storage:6 .4.s:i .4.i:0.5",
                 stateAfterStorageTransition(fixture, 5, State.UP));
     }
 
@@ -330,20 +287,17 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(2), 0.51);
 
-        final Node node = new Node(NodeType.STORAGE, 4);
         final NodeState newState = new NodeState(NodeType.STORAGE, State.UP);
         newState.setDiskCount(7);
         newState.setDiskState(5, new DiskState(State.DOWN));
 
         fixture.reportStorageNodeState(4, newState);
-        SystemStateListener stateListener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, stateListener));
 
-        assertEquals("version:1 distributor:6 storage:6 .4.d:7 .4.d.5.s:d", fixture.generatedClusterState());
+        assertEquals("distributor:6 storage:6 .4.d:7 .4.d.5.s:d", fixture.generatedClusterState());
 
-        assertEquals("version:2 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
-        assertEquals("version:3 distributor:6 storage:6 .4.d:7 .4.d.5.s:d",
+        assertEquals("distributor:6 storage:6 .4.d:7 .4.d.5.s:d",
                 stateAfterStorageTransition(fixture, 5, State.UP));
     }
 
@@ -352,19 +306,15 @@ public class GroupAutoTakedownTest {
         ClusterFixture fixture = createFixtureForAllUpHierarchicCluster(
                 DistributionBuilder.withGroups(3).eachWithNodeCount(3), 0.60);
 
-        NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 5));
-        nodeInfo.setWantedState(new NodeState(NodeType.STORAGE, State.DOWN).setDescription("borkbork"));
-        fixture.generator.proposeNewNodeState(nodeInfo, nodeInfo.getWantedState());
-        SystemStateListener listener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
+        fixture.proposeStorageNodeWantedState(5, State.DOWN, "borkbork");
 
-        assertEquals("version:1 distributor:9 storage:9 .5.s:d .5.m:borkbork", fixture.verboseGeneratedClusterState());
+        assertEquals("distributor:9 storage:9 .5.s:d .5.m:borkbork", fixture.verboseGeneratedClusterState());
 
-        assertEquals("version:2 distributor:9 storage:9 " +
+        assertEquals("distributor:9 storage:9 " +
                 ".3.s:d .3.m:group\\x20node\\x20availability\\x20below\\x20configured\\x20threshold " +
                 ".4.s:d .4.m:mockdesc .5.s:d .5.m:borkbork",
                 verboseStateAfterStorageTransition(fixture, 4, State.DOWN));
-        assertEquals("version:3 distributor:9 storage:9 .5.s:d .5.m:borkbork",
+        assertEquals("distributor:9 storage:9 .5.s:d .5.m:borkbork",
                 verboseStateAfterStorageTransition(fixture, 4, State.UP));
     }
 
@@ -378,25 +328,23 @@ public class GroupAutoTakedownTest {
 
         fixture.reportStorageNodeState(4, newState);
 
-        SystemStateListener listener = mock(SystemStateListener.class);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
-
-        assertEquals("version:1 distributor:6 storage:6 .4.t:123456", fixture.generatedClusterState());
+        assertEquals("distributor:6 storage:6 .4.t:123456", fixture.generatedClusterState());
 
         DatabaseHandler handler = mock(DatabaseHandler.class);
         DatabaseHandler.Context context = mock(DatabaseHandler.Context.class);
         when(context.getCluster()).thenReturn(fixture.cluster);
 
-        fixture.generator.handleAllDistributorsInSync(handler, context);
-        assertTrue(fixture.generator.notifyIfNewSystemState(fixture.cluster, listener));
+        Set<ConfiguredNode> nodes = new HashSet<>(fixture.cluster.clusterInfo().getConfiguredNodes().values());
+        fixture.nodeStateChangeHandler.handleAllDistributorsInSync(
+                fixture.annotatedGeneratedClusterState().getClusterState(), nodes, handler, context);
 
         // Timestamp should now be cleared from state
-        assertEquals("version:2 distributor:6 storage:6", fixture.generatedClusterState());
+        assertEquals("distributor:6 storage:6", fixture.generatedClusterState());
 
         // Trigger a group down+up edge. Timestamp should _not_ be reintroduced since it was previously cleared.
-        assertEquals("version:3 distributor:6 storage:4",
+        assertEquals("distributor:6 storage:4",
                 stateAfterStorageTransition(fixture, 5, State.DOWN));
-        assertEquals("version:4 distributor:6 storage:6",
+        assertEquals("distributor:6 storage:6",
                 stateAfterStorageTransition(fixture, 5, State.UP));
     }