NodeAdminStateUpdater { public void converge(State wantedState) { if (wantedState == RESUMED) { adjustNodeAgentsToRunFromNodeRepository(); } else if (currentState == TRANSITIONING && nodeAdmin.subsystemFreezeDuration().compareTo(FREEZE_CONVERGENCE_TIMEOUT) > 0) { adjustNodeAgentsToRunFromNodeRepository(); nodeAdmin.setFrozen(false); NodeState currentNodeState = nodeRepository.getNode(hostHostname).state(); if (currentNodeState == NodeState.active) orchestrator.resume(hostHostname); throw new ConvergenceException("Timed out trying to freeze all nodes: will force an unfrozen tick"); } if (currentState == wantedState) return; currentState = TRANSITIONING; boolean wantFrozen = wantedState != RESUMED; if (!nodeAdmin.setFrozen(wantFrozen)) { throw new ConvergenceException("NodeAdmin is not yet " + (wantFrozen ? "frozen" : "unfrozen")); } boolean hostIsActiveInNR = nodeRepository.getNode(hostHostname).state() == NodeState.active; switch (wantedState) { case RESUMED: if (hostIsActiveInNR) orchestrator.resume(hostHostname); break; case SUSPENDED_NODE_ADMIN: if (hostIsActiveInNR) orchestrator.suspend(hostHostname); break; case SUSPENDED: List<String> nodesInActiveState = getNodesInActiveState(); List<String> nodesToSuspend = new ArrayList<>(nodesInActiveState); if (hostIsActiveInNR) nodesToSuspend.add(hostHostname); if (!nodesToSuspend.isEmpty()) { orchestrator.suspend(hostHostname, nodesToSuspend); log.info("Orchestrator allows suspension of " + nodesToSuspend); } nodeAdmin.stopNodeAgentServices(nodesInActiveState); break; default: throw new IllegalStateException("Unknown wanted state " + wantedState); } log.info("State changed from " + currentState + " to " + wantedState); currentState = wantedState; } NodeAdminStateUpdater(
            NodeAgentContextFactory nodeAgentContextFactory,
            NodeRepository nodeRepository,
            Orchestrator orchestrator,
            NodeAdmin nodeAdmin,
            HostName hostHostname,
            Clock clock); void start(); void stop(); void converge(State wantedState);  }
@Test public void state_convergence() { mockNodeRepo(NodeState.active, 4); List<String> activeHostnames = nodeRepository.getNodes(hostHostname.value()).stream() .map(NodeSpec::hostname) .collect(Collectors.toList()); List<String> suspendHostnames = new ArrayList<>(activeHostnames); suspendHostnames.add(hostHostname.value()); when(nodeAdmin.subsystemFreezeDuration()).thenReturn(Duration.ofSeconds(1)); { assertConvergeError(RESUMED, "NodeAdmin is not yet unfrozen"); when(nodeAdmin.setFrozen(eq(false))).thenReturn(true); updater.converge(RESUMED); verify(orchestrator, times(1)).resume(hostHostname.value()); updater.converge(RESUMED); verify(orchestrator, times(1)).resume(hostHostname.value()); verify(nodeAdmin, times(2)).setFrozen(eq(false)); when(nodeAdmin.setFrozen(eq(true))).thenReturn(false); assertConvergeError(SUSPENDED_NODE_ADMIN, "NodeAdmin is not yet frozen"); verify(nodeAdmin, times(2)).setFrozen(eq(false)); } { final String exceptionMessage = "Cannot allow to suspend because some reason"; when(nodeAdmin.setFrozen(eq(true))).thenReturn(true); doThrow(new RuntimeException(exceptionMessage)).doNothing() .when(orchestrator).suspend(eq(hostHostname.value())); assertConvergeError(SUSPENDED_NODE_ADMIN, exceptionMessage); verify(nodeAdmin, times(2)).setFrozen(eq(false)); updater.converge(SUSPENDED_NODE_ADMIN); verify(nodeAdmin, times(2)).setFrozen(eq(false)); } { final String exceptionMessage = "Failed to stop services"; verify(orchestrator, times(0)).suspend(eq(hostHostname.value()), eq(suspendHostnames)); doThrow(new RuntimeException(exceptionMessage)).doNothing().when(nodeAdmin).stopNodeAgentServices(eq(activeHostnames)); assertConvergeError(SUSPENDED, exceptionMessage); verify(orchestrator, times(1)).suspend(eq(hostHostname.value()), eq(suspendHostnames)); verify(nodeAdmin, times(2)).setFrozen(eq(false)); updater.converge(SUSPENDED); } }
@Test public void half_transition_revert() { final String exceptionMsg = "Cannot allow to suspend because some reason"; mockNodeRepo(NodeState.active, 3); when(nodeAdmin.setFrozen(eq(false))).thenReturn(true); updater.converge(RESUMED); verify(nodeAdmin, times(1)).setFrozen(eq(false)); verify(nodeAdmin, times(1)).refreshContainersToRun(any()); when(nodeAdmin.subsystemFreezeDuration()).thenReturn(Duration.ofSeconds(1)); when(nodeAdmin.setFrozen(eq(true))).thenReturn(true); doThrow(new RuntimeException(exceptionMsg)).when(orchestrator).suspend(eq(hostHostname.value())); assertConvergeError(SUSPENDED_NODE_ADMIN, exceptionMsg); verify(nodeAdmin, times(1)).setFrozen(eq(true)); verify(orchestrator, times(1)).suspend(eq(hostHostname.value())); assertConvergeError(SUSPENDED_NODE_ADMIN, exceptionMsg); verify(nodeAdmin, times(2)).setFrozen(eq(true)); verify(orchestrator, times(2)).suspend(eq(hostHostname.value())); assertConvergeError(SUSPENDED_NODE_ADMIN, exceptionMsg); verify(nodeAdmin, times(3)).setFrozen(eq(true)); verify(orchestrator, times(3)).suspend(eq(hostHostname.value())); verify(nodeAdmin, times(1)).setFrozen(eq(false)); verify(nodeAdmin, times(1)).refreshContainersToRun(any()); when(nodeAdmin.subsystemFreezeDuration()).thenReturn(Duration.ofHours(1)); assertConvergeError(SUSPENDED_NODE_ADMIN, "Timed out trying to freeze all nodes: will force an unfrozen tick"); verify(nodeAdmin, times(2)).setFrozen(eq(false)); verify(orchestrator, times(3)).suspend(eq(hostHostname.value())); verify(nodeAdmin, times(2)).refreshContainersToRun(any()); updater.converge(RESUMED); verify(nodeAdmin, times(3)).setFrozen(eq(false)); }
@Test public void do_not_orchestrate_host_when_not_active() { when(nodeAdmin.subsystemFreezeDuration()).thenReturn(Duration.ofHours(1)); when(nodeAdmin.setFrozen(anyBoolean())).thenReturn(true); mockNodeRepo(NodeState.ready, 3); updater.converge(RESUMED); verify(orchestrator, never()).resume(eq(hostHostname.value())); updater.converge(SUSPENDED_NODE_ADMIN); verify(orchestrator, never()).suspend(eq(hostHostname.value())); List<String> activeHostnames = nodeRepository.getNodes(hostHostname.value()).stream() .map(NodeSpec::hostname) .collect(Collectors.toList()); updater.converge(SUSPENDED); verify(orchestrator, times(1)).suspend(eq(hostHostname.value()), eq(activeHostnames)); }