ClusterStateGenerator { static AnnotatedClusterState generatedStateFrom(final Params params) { final ContentCluster cluster = params.cluster; final ClusterState workingState = ClusterState.emptyState(); final Map<Node, NodeStateReason> nodeStateReasons = new HashMap<>(); for (final NodeInfo nodeInfo : cluster.getNodeInfo()) { final NodeState nodeState = computeEffectiveNodeState(nodeInfo, params, nodeStateReasons); workingState.setNodeState(nodeInfo.getNode(), nodeState); } takeDownGroupsWithTooLowAvailability(workingState, nodeStateReasons, params); final Optional<ClusterStateReason> reasonToBeDown = clusterDownReason(workingState, params); if (reasonToBeDown.isPresent()) { workingState.setClusterState(State.DOWN); } workingState.setDistributionBits(inferDistributionBitCount(cluster, workingState, params)); return new AnnotatedClusterState(workingState, reasonToBeDown, nodeStateReasons); }    }
@Test public void distributor_nodes_are_not_implicitly_transitioned_to_maintenance_mode() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(5).bringEntireClusterUp(); final ClusterStateGenerator.Params params = fixture.generatorParams() .currentTimeInMilllis(10_000) .transitionTimes(2000); fixture.reportDistributorNodeState(2, State.DOWN); final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.DISTRIBUTOR, 2)); nodeInfo.setTransitionTime(9000); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:5 .2.s:d storage:5")); assertThat(state.getNodeStateReasons(), not(hasStateReasonForNode(distributorNode(1), NodeStateReason.NODE_NOT_BACK_UP_WITHIN_GRACE_PERIOD))); }
@Test public void transient_maintenance_mode_does_not_override_wanted_down_state() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(5).bringEntireClusterUp(); final ClusterStateGenerator.Params params = fixture.generatorParams() .currentTimeInMilllis(10_000) .transitionTimes(2000); fixture.proposeStorageNodeWantedState(2, State.DOWN); fixture.reportStorageNodeState(2, State.DOWN); final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 2)); nodeInfo.setTransitionTime(9000); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:5 storage:5 .2.s:d")); }
@Test public void reported_down_retired_node_within_transition_time_transitions_to_maintenance() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(5).bringEntireClusterUp(); final ClusterStateGenerator.Params params = fixture.generatorParams() .currentTimeInMilllis(10_000) .transitionTimes(2000); fixture.proposeStorageNodeWantedState(2, State.RETIRED); fixture.reportStorageNodeState(2, State.DOWN); final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 2)); nodeInfo.setTransitionTime(9000); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:5 storage:5 .2.s:m")); }
@Test public void crash_count_exceeding_limit_marks_node_as_down() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(5).bringEntireClusterUp(); final ClusterStateGenerator.Params params = fixture.generatorParams().maxPrematureCrashes(10); final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 3)); nodeInfo.setPrematureCrashCount(11); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:5 storage:5 .3.s:d")); }
@Test public void crash_count_not_exceeding_limit_does_not_mark_node_as_down() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(5).bringEntireClusterUp(); final ClusterStateGenerator.Params params = fixture.generatorParams().maxPrematureCrashes(10); final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 3)); nodeInfo.setPrematureCrashCount(10); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:5 storage:5")); }
@Test public void exceeded_crash_count_does_not_override_wanted_maintenance_state() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(5) .bringEntireClusterUp() .proposeStorageNodeWantedState(1, State.MAINTENANCE); final ClusterStateGenerator.Params params = fixture.generatorParams().maxPrematureCrashes(10); final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 1)); nodeInfo.setPrematureCrashCount(11); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:5 storage:5 .1.s:m")); }
@Test public void cluster_down_if_less_than_min_count_of_storage_nodes_available() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(3) .bringEntireClusterUp() .reportStorageNodeState(0, State.DOWN) .reportStorageNodeState(2, State.DOWN); final ClusterStateGenerator.Params params = fixture.generatorParams().minStorageNodesUp(2); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("cluster:d distributor:3 storage:2 .0.s:d")); assertThat(state.getClusterStateReason(), equalTo(Optional.of(ClusterStateReason.TOO_FEW_STORAGE_NODES_AVAILABLE))); }
@Test public void cluster_not_down_if_more_than_min_count_of_storage_nodes_are_available() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(3) .bringEntireClusterUp() .reportStorageNodeState(0, State.DOWN); final ClusterStateGenerator.Params params = fixture.generatorParams().minStorageNodesUp(2); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:3 storage:3 .0.s:d")); assertThat(state.getClusterStateReason(), equalTo(Optional.empty())); }
@Test public void cluster_down_if_less_than_min_count_of_distributors_available() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(3) .bringEntireClusterUp() .reportDistributorNodeState(0, State.DOWN) .reportDistributorNodeState(2, State.DOWN); final ClusterStateGenerator.Params params = fixture.generatorParams().minDistributorNodesUp(2); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("cluster:d distributor:2 .0.s:d storage:3")); assertThat(state.getClusterStateReason(), equalTo(Optional.of(ClusterStateReason.TOO_FEW_DISTRIBUTOR_NODES_AVAILABLE))); }
@Test public void cluster_not_down_if_more_than_min_count_of_distributors_are_available() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(3) .bringEntireClusterUp() .reportDistributorNodeState(0, State.DOWN); final ClusterStateGenerator.Params params = fixture.generatorParams().minDistributorNodesUp(2); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:3 .0.s:d storage:3")); assertThat(state.getClusterStateReason(), equalTo(Optional.empty())); }
@Test public void maintenance_mode_counted_as_down_for_cluster_availability() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(3) .bringEntireClusterUp() .reportStorageNodeState(0, State.DOWN) .proposeStorageNodeWantedState(2, State.MAINTENANCE); final ClusterStateGenerator.Params params = fixture.generatorParams().minStorageNodesUp(2); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("cluster:d distributor:3 storage:3 .0.s:d .2.s:m")); }
@Test public void init_and_retired_counted_as_up_for_cluster_availability() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(3) .bringEntireClusterUp() .reportStorageNodeState(0, State.INITIALIZING) .proposeStorageNodeWantedState(1, State.RETIRED); final ClusterStateGenerator.Params params = fixture.generatorParams().minStorageNodesUp(3); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:3 storage:3 .0.s:i .0.i:1.0 .1.s:r")); }
@Test public void cluster_down_if_less_than_min_ratio_of_storage_nodes_available() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(3) .bringEntireClusterUp() .reportStorageNodeState(0, State.DOWN) .reportStorageNodeState(2, State.DOWN); final ClusterStateGenerator.Params params = fixture.generatorParams().minRatioOfStorageNodesUp(0.5); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("cluster:d distributor:3 storage:2 .0.s:d")); assertThat(state.getClusterStateReason(), equalTo(Optional.of(ClusterStateReason.TOO_LOW_AVAILABLE_STORAGE_NODE_RATIO))); }
@Test public void cluster_not_down_if_more_than_min_ratio_of_storage_nodes_available() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(3) .bringEntireClusterUp() .reportStorageNodeState(0, State.DOWN); final ClusterStateGenerator.Params params = fixture.generatorParams().minRatioOfStorageNodesUp(0.5); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:3 storage:3 .0.s:d")); assertThat(state.getClusterStateReason(), equalTo(Optional.empty())); }
@Test public void cluster_down_if_less_than_min_ratio_of_distributors_available() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(3) .bringEntireClusterUp() .reportDistributorNodeState(0, State.DOWN) .reportDistributorNodeState(2, State.DOWN); final ClusterStateGenerator.Params params = fixture.generatorParams().minRatioOfDistributorNodesUp(0.5); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("cluster:d distributor:2 .0.s:d storage:3")); assertThat(state.getClusterStateReason(), equalTo(Optional.of(ClusterStateReason.TOO_LOW_AVAILABLE_DISTRIBUTOR_NODE_RATIO))); }
@Test public void cluster_not_down_if_more_than_min_ratio_of_distributors_available() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(3) .bringEntireClusterUp() .reportDistributorNodeState(0, State.DOWN); final ClusterStateGenerator.Params params = fixture.generatorParams().minRatioOfDistributorNodesUp(0.5); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:3 .0.s:d storage:3")); assertThat(state.getClusterStateReason(), equalTo(Optional.empty())); }
@Test public void group_nodes_are_marked_down_if_group_availability_too_low() { final ClusterFixture fixture = ClusterFixture .forHierarchicCluster(DistributionBuilder.withGroups(3).eachWithNodeCount(3)) .bringEntireClusterUp() .reportStorageNodeState(4, State.DOWN); final ClusterStateGenerator.Params params = fixture.generatorParams().minNodeRatioPerGroup(0.68); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:9 storage:9 .3.s:d .4.s:d .5.s:d")); }
@Test public void group_nodes_are_not_marked_down_if_group_availability_sufficiently_high() { final ClusterFixture fixture = ClusterFixture .forHierarchicCluster(DistributionBuilder.withGroups(3).eachWithNodeCount(3)) .bringEntireClusterUp() .reportStorageNodeState(4, State.DOWN); final ClusterStateGenerator.Params params = fixture.generatorParams().minNodeRatioPerGroup(0.65); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:9 storage:9 .4.s:d")); }
@Test public void implicitly_downed_group_nodes_receive_a_state_description() { final ClusterFixture fixture = ClusterFixture .forHierarchicCluster(DistributionBuilder.withGroups(2).eachWithNodeCount(2)) .bringEntireClusterUp() .reportStorageNodeState(3, State.DOWN); final ClusterStateGenerator.Params params = fixture.generatorParams().minNodeRatioPerGroup(0.51); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(true), equalTo("distributor:4 storage:4 " + ".2.s:d .2.m:group\\x20node\\x20availability\\x20below\\x20configured\\x20threshold " + ".3.s:d .3.m:mockdesc")); }
@Test public void maintenance_nodes_in_downed_group_are_not_affected() { final ClusterFixture fixture = ClusterFixture .forHierarchicCluster(DistributionBuilder.withGroups(3).eachWithNodeCount(3)) .bringEntireClusterUp() .proposeStorageNodeWantedState(3, State.MAINTENANCE) .reportStorageNodeState(4, State.DOWN); final ClusterStateGenerator.Params params = fixture.generatorParams().minNodeRatioPerGroup(0.68); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:9 storage:9 .3.s:m .4.s:d .5.s:d")); }
@Test public void distribution_bits_bounded_by_config_parameter() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(3).bringEntireClusterUp(); final ClusterStateGenerator.Params params = fixture.generatorParams().idealDistributionBits(12); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("bits:12 distributor:3 storage:3")); }
@Test public void unstable_retired_node_should_be_marked_down() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(5) .bringEntireClusterUp() .proposeStorageNodeWantedState(3, State.RETIRED); final ClusterStateGenerator.Params params = fixture.generatorParams().maxPrematureCrashes(10); final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 3)); nodeInfo.setPrematureCrashCount(11); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:5 storage:5 .3.s:d")); }
@Test public void configured_zero_init_progress_time_disables_auto_init_to_down_feature() { final ClusterFixture fixture = ClusterFixture.forFlatCluster(3) .bringEntireClusterUp() .reportStorageNodeState(0, new NodeState(NodeType.STORAGE, State.INITIALIZING).setInitProgress(0.5)); final NodeInfo nodeInfo = fixture.cluster.getNodeInfo(new Node(NodeType.STORAGE, 0)); nodeInfo.setInitProgressTime(10_000); final ClusterStateGenerator.Params params = fixture.generatorParams() .maxInitProgressTime(0) .currentTimeInMilllis(11_000); final AnnotatedClusterState state = ClusterStateGenerator.generatedStateFrom(params); assertThat(state.toString(), equalTo("distributor:3 storage:3 .0.s:i .0.i:0.5")); }