NodeAgentImpl implements NodeAgent { void doConverge(NodeAgentContext context) { NodeSpec node = context.node(); Optional<Container> container = getContainer(context); if (!node.equals(lastNode)) { logChangesToNodeSpec(context, lastNode, node); if (currentRebootGeneration < node.currentRebootGeneration()) currentRebootGeneration = node.currentRebootGeneration(); if (currentRestartGeneration.isPresent() != node.currentRestartGeneration().isPresent() || currentRestartGeneration.map(current -> current < node.currentRestartGeneration().get()).orElse(false)) currentRestartGeneration = node.currentRestartGeneration(); lastNode = node; } switch (node.state()) { case ready: case reserved: case failed: case inactive: case parked: removeContainerIfNeededUpdateContainerState(context, container); updateNodeRepoWithCurrentAttributes(context); stopServicesIfNeeded(context); break; case active: storageMaintainer.cleanDiskIfFull(context); storageMaintainer.handleCoreDumpsForContainer(context, container); if (downloadImageIfNeeded(node, container)) { context.log(logger, "Waiting for image to download " + context.node().wantedDockerImage().get().asString()); return; } container = removeContainerIfNeededUpdateContainerState(context, container); credentialsMaintainer.ifPresent(maintainer -> maintainer.converge(context)); if (container.isEmpty()) { containerState = STARTING; container = Optional.of(startContainer(context)); containerState = UNKNOWN; } else { container = Optional.of(updateContainerIfNeeded(context, container.get())); } aclMaintainer.ifPresent(maintainer -> maintainer.converge(context)); startServicesIfNeeded(context); resumeNodeIfNeeded(context); if (healthChecker.isPresent()) { healthChecker.get().verifyHealth(context); if (firstSuccessfulHealthCheckInstant.isEmpty()) firstSuccessfulHealthCheckInstant = Optional.of(clock.instant()); Duration timeLeft = Duration.between(clock.instant(), firstSuccessfulHealthCheckInstant.get().plus(warmUpDuration(context.zone()))); if (!container.get().resources.equalsCpu(getContainerResources(context))) throw new ConvergenceException("Refusing to resume until warm up period ends (" + (timeLeft.isNegative() ? "next tick" : "in " + timeLeft) + ")"); } updateNodeRepoWithCurrentAttributes(context); if (suspendedInOrchestrator || node.allowedToBeDown().orElse(false)) { context.log(logger, "Call resume against Orchestrator"); orchestrator.resume(context.hostname().value()); suspendedInOrchestrator = false; } break; case provisioned: nodeRepository.setNodeState(context.hostname().value(), NodeState.dirty); break; case dirty: removeContainerIfNeededUpdateContainerState(context, container); context.log(logger, "State is " + node.state() + ", will delete application storage and mark node as ready"); credentialsMaintainer.ifPresent(maintainer -> maintainer.clearCredentials(context)); storageMaintainer.archiveNodeStorage(context); updateNodeRepoWithCurrentAttributes(context); nodeRepository.setNodeState(context.hostname().value(), NodeState.ready); break; default: throw new ConvergenceException("UNKNOWN STATE " + node.state().name()); } } NodeAgentImpl(NodeAgentContextSupplier contextSupplier, NodeRepository nodeRepository,
                         Orchestrator orchestrator, ContainerOperations containerOperations, StorageMaintainer storageMaintainer,
                         FlagSource flagSource, Optional<CredentialsMaintainer> credentialsMaintainer,
                         Optional<AclMaintainer> aclMaintainer, Optional<HealthChecker> healthChecker, Clock clock);  NodeAgentImpl(NodeAgentContextSupplier contextSupplier, NodeRepository nodeRepository,
                         Orchestrator orchestrator, ContainerOperations containerOperations, StorageMaintainer storageMaintainer,
                         FlagSource flagSource, Optional<CredentialsMaintainer> credentialsMaintainer,
                         Optional<AclMaintainer> aclMaintainer, Optional<HealthChecker> healthChecker, Clock clock, Duration warmUpDuration); @Override void start(NodeAgentContext initialContext); @Override void stopForRemoval(NodeAgentContext context); @Override void stopForHostSuspension(NodeAgentContext context); void suspend(NodeAgentContext context); void converge(NodeAgentContext context); @Override int getAndResetNumberOfUnhandledExceptions();  }
@Test public void verifyRemoveOldFilesIfDiskFull() { final NodeSpec node = nodeBuilder(NodeState.active) .wantedDockerImage(dockerImage).currentDockerImage(dockerImage) .wantedVespaVersion(vespaVersion).currentVespaVersion(vespaVersion) .build(); NodeAgentContext context = createContext(node); NodeAgentImpl nodeAgent = makeNodeAgent(dockerImage, true); when(nodeRepository.getOptionalNode(hostName)).thenReturn(Optional.of(node)); nodeAgent.doConverge(context); verify(storageMaintainer, times(1)).cleanDiskIfFull(eq(context)); }
@Test public void reservedNodeDoesNotUpdateNodeRepoWithVersion() { final NodeSpec node = nodeBuilder(NodeState.reserved) .wantedDockerImage(dockerImage) .wantedVespaVersion(vespaVersion) .build(); NodeAgentContext context = createContext(node); NodeAgentImpl nodeAgent = makeNodeAgent(null, false); when(nodeRepository.getOptionalNode(hostName)).thenReturn(Optional.of(node)); nodeAgent.doConverge(context); verify(nodeRepository, never()).updateNodeAttributes(eq(hostName), any()); }
@Test public void provisionedNodeIsMarkedAsDirty() { final NodeSpec node = nodeBuilder(NodeState.provisioned) .wantedDockerImage(dockerImage) .build(); NodeAgentContext context = createContext(node); NodeAgentImpl nodeAgent = makeNodeAgent(null, false); when(nodeRepository.getOptionalNode(hostName)).thenReturn(Optional.of(node)); nodeAgent.doConverge(context); verify(nodeRepository, times(1)).setNodeState(eq(hostName), eq(NodeState.dirty)); }
@Test public void resumeProgramRunsUntilSuccess() { final NodeSpec node = nodeBuilder(NodeState.active) .wantedDockerImage(dockerImage).currentDockerImage(dockerImage) .currentVespaVersion(vespaVersion) .wantedRestartGeneration(1).currentRestartGeneration(1) .allowedToBeDown(true) .build(); NodeAgentContext context = createContext(node); NodeAgentImpl nodeAgent = makeNodeAgent(dockerImage, true); when(nodeRepository.getOptionalNode(eq(hostName))).thenReturn(Optional.of(node)); final InOrder inOrder = inOrder(orchestrator, containerOperations, nodeRepository); doThrow(new RuntimeException("Failed 1st time")) .doNothing() .when(containerOperations).resumeNode(eq(context)); try { nodeAgent.doConverge(context); fail("Expected to throw an exception"); } catch (RuntimeException ignored) { } inOrder.verify(containerOperations, times(1)).resumeNode(any()); inOrder.verifyNoMoreInteractions(); nodeAgent.doConverge(context); inOrder.verify(containerOperations).resumeNode(any()); inOrder.verify(orchestrator).resume(hostName); inOrder.verifyNoMoreInteractions(); }