EventDiffCalculator { public static List<Event> computeEventDiff(final Params params) { final List<Event> events = new ArrayList<>(); emitPerNodeDiffEvents(createBaselineParams(params), events); emitWholeClusterDiffEvent(createBaselineParams(params), events); emitDerivedBucketSpaceStatesDiffEvents(params, events); return events; }  static Params params(); static List<Event> computeEventDiff(final Params params);  }
@Test public void single_storage_node_state_transition_emits_altered_node_state_event() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .clusterStateAfter("distributor:3 storage:3 .0.s:d"); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(1)); assertThat(events, hasItem(allOf( eventForNode(storageNode(0)), eventTypeIs(NodeEvent.Type.CURRENT), nodeEventWithDescription("Altered node state in cluster state from 'U' to 'D'")))); }
@Test public void single_distributor_node_state_transition_emits_altered_node_state_event() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .clusterStateAfter("distributor:3 .1.s:d storage:3"); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(1)); assertThat(events, hasItem(allOf( eventForNode(distributorNode(1)), eventTypeIs(NodeEvent.Type.CURRENT), nodeEventWithDescription("Altered node state in cluster state from 'U' to 'D'")))); }
@Test public void node_state_change_event_is_tagged_with_given_time() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .clusterStateAfter("distributor:3 storage:3 .0.s:d") .currentTimeMs(123456); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(1)); assertThat(events, hasItem(eventTimeIs(123456))); }
@Test public void multiple_node_state_transitions_emit_multiple_node_state_events() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3 .1.s:d") .clusterStateAfter("distributor:3 .2.s:d storage:3 .0.s:r"); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(3)); assertThat(events, hasItem(allOf( eventForNode(distributorNode(2)), nodeEventWithDescription("Altered node state in cluster state from 'U' to 'D'")))); assertThat(events, hasItem(allOf( eventForNode(storageNode(0)), nodeEventWithDescription("Altered node state in cluster state from 'U' to 'R'")))); assertThat(events, hasItem(allOf( eventForNode(storageNode(1)), nodeEventWithDescription("Altered node state in cluster state from 'D' to 'U'")))); }
@Test public void no_emitted_node_state_event_when_node_state_not_changed() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .clusterStateAfter("distributor:3 storage:3"); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(0)); }
@Test public void node_down_edge_with_group_down_reason_has_separate_event_emitted() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .clusterStateAfter("distributor:3 storage:3 .1.s:d") .storageNodeReasonAfter(1, NodeStateReason.GROUP_IS_DOWN); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(2)); assertThat(events, hasItem(allOf( eventForNode(storageNode(1)), nodeEventWithDescription("Altered node state in cluster state from 'U' to 'D'")))); assertThat(events, hasItem(allOf( eventForNode(storageNode(1)), eventTypeIs(NodeEvent.Type.CURRENT), nodeEventWithDescription("Group node availability is below configured threshold")))); }
@Test public void group_down_to_group_down_does_not_emit_new_event() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3 .1.s:d") .clusterStateAfter("distributor:3 storage:3 .1.s:m") .storageNodeReasonBefore(1, NodeStateReason.GROUP_IS_DOWN) .storageNodeReasonAfter(1, NodeStateReason.GROUP_IS_DOWN); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(1)); assertThat(events, hasItem(allOf( eventForNode(storageNode(1)), nodeEventWithDescription("Altered node state in cluster state from 'D' to 'M'")))); }
@Test public void group_down_to_clear_reason_emits_group_up_event() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3 .2.s:d") .clusterStateAfter("distributor:3 storage:3") .storageNodeReasonBefore(2, NodeStateReason.GROUP_IS_DOWN); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(2)); assertThat(events, hasItem(allOf( eventForNode(storageNode(2)), nodeEventWithDescription("Altered node state in cluster state from 'D' to 'U'")))); assertThat(events, hasItem(allOf( eventForNode(storageNode(2)), eventTypeIs(NodeEvent.Type.CURRENT), nodeEventWithDescription("Group node availability has been restored")))); }
@Test public void cluster_up_edge_emits_sufficient_node_availability_event() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("cluster:d distributor:3 storage:3") .clusterStateAfter("distributor:3 storage:3"); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(1)); assertThat(events, hasItem( clusterEventWithDescription("Enough nodes available for system to become up"))); }
@Test public void cluster_down_event_without_reason_annotation_emits_generic_down_event() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .clusterStateAfter("cluster:d distributor:3 storage:3"); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(1)); assertThat(events, hasItem( clusterEventWithDescription("Cluster is down"))); }
@Test public void cluster_event_is_tagged_with_given_time() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .clusterStateAfter("cluster:d distributor:3 storage:3") .currentTimeMs(56789); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(1)); assertThat(events, hasItem(eventTimeIs(56789))); }
@Test public void no_event_emitted_for_cluster_down_to_down_edge() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("cluster:d distributor:3 storage:3") .clusterStateAfter("cluster:d distributor:3 storage:3"); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(0)); }
@Test public void too_few_storage_nodes_cluster_down_reason_emits_corresponding_event() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .clusterStateAfter("cluster:d distributor:3 storage:3") .clusterReasonAfter(ClusterStateReason.TOO_FEW_STORAGE_NODES_AVAILABLE); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(1)); assertThat(events, hasItem( clusterEventWithDescription("Too few storage nodes available in cluster. Setting cluster state down"))); }
@Test public void too_few_distributor_nodes_cluster_down_reason_emits_corresponding_event() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .clusterStateAfter("cluster:d distributor:3 storage:3") .clusterReasonAfter(ClusterStateReason.TOO_FEW_DISTRIBUTOR_NODES_AVAILABLE); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(1)); assertThat(events, hasItem( clusterEventWithDescription("Too few distributor nodes available in cluster. Setting cluster state down"))); }
@Test public void too_low_storage_node_ratio_cluster_down_reason_emits_corresponding_event() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .clusterStateAfter("cluster:d distributor:3 storage:3") .clusterReasonAfter(ClusterStateReason.TOO_LOW_AVAILABLE_STORAGE_NODE_RATIO); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(1)); assertThat(events, hasItem( clusterEventWithDescription("Too low ratio of available storage nodes. Setting cluster state down"))); }
@Test public void too_low_distributor_node_ratio_cluster_down_reason_emits_corresponding_event() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .clusterStateAfter("cluster:d distributor:3 storage:3") .clusterReasonAfter(ClusterStateReason.TOO_LOW_AVAILABLE_DISTRIBUTOR_NODE_RATIO); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(1)); assertThat(events, hasItem( clusterEventWithDescription("Too low ratio of available distributor nodes. Setting cluster state down"))); }
@Test public void may_have_merges_pending_up_edge_event_emitted_if_derived_bucket_space_state_differs_from_baseline() { EventFixture f = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .derivedClusterStateBefore("default", "distributor:3 storage:3") .clusterStateAfter("distributor:3 storage:3") .derivedClusterStateAfter("default", "distributor:3 storage:3 .1.s:m") .derivedStorageNodeReasonAfter("default", 1, NodeStateReason.MAY_HAVE_MERGES_PENDING); List<Event> events = f.computeEventDiff(); assertThat(events.size(), equalTo(2)); assertThat(events, hasItem(allOf( eventForNode(storageNode(1)), nodeEventForBucketSpace("default"), nodeEventWithDescription("Altered node state in cluster state from 'U' to 'M'")))); assertThat(events, hasItem(allOf( eventForNode(storageNode(1)), nodeEventForBucketSpace("default"), nodeEventWithDescription("Node may have merges pending")))); }
@Test public void may_have_merges_pending_down_edge_event_emitted_if_derived_bucket_space_state_differs_from_baseline() { EventFixture f = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .derivedClusterStateBefore("default", "distributor:3 storage:3 .1.s:m") .derivedStorageNodeReasonBefore("default", 1, NodeStateReason.MAY_HAVE_MERGES_PENDING) .clusterStateAfter("distributor:3 storage:3") .derivedClusterStateAfter("default", "distributor:3 storage:3"); List<Event> events = f.computeEventDiff(); assertThat(events.size(), equalTo(2)); assertThat(events, hasItem(allOf( eventForNode(storageNode(1)), nodeEventForBucketSpace("default"), nodeEventWithDescription("Altered node state in cluster state from 'M' to 'U'")))); assertThat(events, hasItem(allOf( eventForNode(storageNode(1)), nodeEventForBucketSpace("default"), nodeEventWithDescription("Node no longer has merges pending")))); }
@Test public void both_baseline_and_derived_bucket_space_state_events_are_emitted() { EventFixture f = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .derivedClusterStateBefore("default", "distributor:3 storage:3") .clusterStateAfter("distributor:3 storage:3 .0.s:m") .derivedClusterStateAfter("default", "distributor:3 storage:3 .1.s:m"); List<Event> events = f.computeEventDiff(); assertThat(events.size(), equalTo(2)); assertThat(events, hasItem(allOf( eventForNode(storageNode(0)), nodeEventForBaseline(), nodeEventWithDescription("Altered node state in cluster state from 'U' to 'M'")))); assertThat(events, hasItem(allOf( eventForNode(storageNode(1)), nodeEventForBucketSpace("default"), nodeEventWithDescription("Altered node state in cluster state from 'U' to 'M'")))); }
@Test public void derived_bucket_space_state_events_are_not_emitted_if_similar_to_baseline() { EventFixture f = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3") .derivedClusterStateBefore("default", "distributor:3 storage:3") .derivedClusterStateBefore("global", "distributor:3 storage:3") .clusterStateAfter("distributor:3 storage:3 .0.s:m") .derivedClusterStateAfter("default", "distributor:3 storage:3 .0.s:m") .derivedClusterStateAfter("global", "distributor:3 storage:3 .0.s:m"); List<Event> events = f.computeEventDiff(); assertThat(events.size(), equalTo(1)); assertThat(events, hasItem(allOf( eventForNode(storageNode(0)), nodeEventForBaseline(), nodeEventWithDescription("Altered node state in cluster state from 'U' to 'M'")))); }
@Test public void storage_node_passed_maintenance_grace_period_emits_event() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3 .0.s:m") .clusterStateAfter("distributor:3 storage:3 .0.s:d") .maxMaintenanceGracePeriodTimeMs(123_456) .storageNodeReasonAfter(0, NodeStateReason.NODE_NOT_BACK_UP_WITHIN_GRACE_PERIOD); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(2)); assertThat(events, hasItem(allOf( eventForNode(storageNode(0)), nodeEventWithDescription("Exceeded implicit maintenance mode grace period of 123456 milliseconds. Marking node down."), nodeEventForBaseline()))); }
@Test public void storage_node_maintenance_grace_period_event_only_emitted_on_maintenance_to_down_edge() { final EventFixture fixture = EventFixture.createForNodes(3) .clusterStateBefore("distributor:3 storage:3 .0.s:u") .clusterStateAfter("distributor:3 storage:3 .0.s:d") .maxMaintenanceGracePeriodTimeMs(123_456) .storageNodeReasonAfter(0, NodeStateReason.NODE_NOT_BACK_UP_WITHIN_GRACE_PERIOD); final List<Event> events = fixture.computeEventDiff(); assertThat(events.size(), equalTo(1)); assertThat(events, hasItem(allOf( eventForNode(storageNode(0)), nodeEventForBaseline()))); }