From ae112c8a68f4becf95076987a91a37fc7d5fce30 Mon Sep 17 00:00:00 2001 From: Marian Fischer Date: Wed, 11 Nov 2026 19:39:36 +0100 Subject: [PATCH] CKS: fix scaling a cluster with external nodes When a cluster on an isolated network is scaled, the SSH rules on the source NAT IP are rebuilt for the new node count, but the rules of the external nodes (added with addNodesToKubernetesCluster) were left in place. On scale-up the new SSH port range overlapped the firewall rule of an external node and the scale failed after the rules of the other nodes had been removed; on scale-down the old rules of the external nodes stayed on their old ports. Revoke the SSH port forwarding rules of the external nodes and the firewall rules for their ports before the rules are rebuilt; they are recreated after the new range as before. Pass the node VMs to the rebuild in the order of their SSH ports on scale-down as well, the external nodes last, as on scale-up. Reject external nodes in the nodeids of scaleKubernetesCluster: scaling never removed them but lowered the node count. External nodes are removed with removeNodesFromKubernetesCluster. The node order and the nodeids check apply to all network types. Fixes #12633 --- .../cluster/KubernetesClusterManagerImpl.java | 13 + ...esClusterResourceModifierActionWorker.java | 53 +++ .../KubernetesClusterScaleWorker.java | 28 +- .../KubernetesClusterManagerImplTest.java | 16 + .../KubernetesClusterScaleWorkerTest.java | 323 ++++++++++++++++++ 5 files changed, 423 insertions(+), 10 deletions(-) diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java index 035554e38237..4c44f898f0d7 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java @@ -1243,6 +1243,18 @@ protected void addKubernetesClusterDetailIfIsNotEmpty(List nodes) { + if (nodes.stream().anyMatch(KubernetesClusterVmMapVO::isExternalNode)) { + throw new InvalidParameterValueException("External nodes can not be removed by scaling a cluster, use removeNodesFromKubernetesCluster"); + } + } + protected void validateKubernetesClusterScaleSize(final KubernetesClusterVO kubernetesCluster, final Long clusterSize, final int maxClusterSize, final DataCenter zone) { if (clusterSize == null) { return; @@ -1353,6 +1365,7 @@ private void validateKubernetesClusterScaleParameters(ScaleKubernetesClusterCmd if (nodes == null || nodes.size() != nodeIds.size()) { throw new InvalidParameterValueException("Invalid node ids"); } + validateKubernetesClusterScaleNodesToRemove(nodes); // Ensure there's always a control node long controleNodesToRemove = nodes.stream().filter(x -> x.isControlNode()).count(); if (controleNodesToRemove >= kubernetesCluster.getControlNodeCount()) { diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorker.java index 573e752ce125..70e6b8c46eac 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorker.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorker.java @@ -80,7 +80,9 @@ import com.cloud.hypervisor.Hypervisor; import com.cloud.kubernetes.cluster.KubernetesCluster; import com.cloud.kubernetes.cluster.KubernetesClusterManagerImpl; +import com.cloud.kubernetes.cluster.KubernetesClusterService; import com.cloud.kubernetes.cluster.KubernetesClusterVO; +import com.cloud.kubernetes.cluster.KubernetesClusterVmMapVO; import com.cloud.network.IpAddress; import com.cloud.network.Network; import com.cloud.network.dao.FirewallRulesDao; @@ -592,6 +594,57 @@ protected void removePortForwardingRules(final IpAddress publicIp, final Network firewallManager.applyRules(pfRules, false, true); } + /** + * Revokes the SSH rules of the external nodes of the cluster (nodes added with addNodesToKubernetesCluster) + * on the given public IP: each port forwarding rule from a node SSH port to port 22 of an external node, + * and the firewall rule for that port. The SSH ports of the external nodes follow the SSH port range of the + * other nodes, so these rules have to be removed before the rules are recreated for a new node count. + * Otherwise the rebuilt range or the recreated rule of an external node conflicts with an old rule of an + * external node. The firewall rule is revoked before the port forwarding rule, so that if a revoke fails, + * the port forwarding rule is left and a retry handles the port again. + * + * @param publicIp the public IP with the SSH rules of the cluster + * @throws ManagementServerException if a rule can't be revoked + */ + protected void removeExternalNodeSshRules(final IpAddress publicIp) throws ManagementServerException { + List externalNodeIds = getKubernetesClusterVMMaps().stream() + .filter(KubernetesClusterVmMapVO::isExternalNode).map(KubernetesClusterVmMapVO::getVmId).collect(Collectors.toList()); + try { + for (Long vmId : externalNodeIds) { + for (PortForwardingRuleVO pfRule : portForwardingRulesDao.listByVm(vmId)) { + if (!isNodeSshPortForwardingRule(pfRule, publicIp)) { + continue; + } + List firewallRules = firewallRulesDao.listByIpPurposePortsProtocolAndNotRevoked(publicIp.getId(), + pfRule.getSourcePortStart(), pfRule.getSourcePortEnd(), NetUtils.TCP_PROTO, FirewallRule.Purpose.Firewall); + for (FirewallRuleVO firewallRule : firewallRules) { + if (!firewallService.revokeIngressFwRule(firewallRule.getId(), true)) { + throw new ManagementServerException(String.format("Failed to revoke the SSH firewall rule %s of the external node with the id %s", + firewallRule, vmId)); + } + logger.debug("The SSH firewall rule {} of the external node with the id {} was revoked", firewallRule, vmId); + } + rulesService.revokePortForwardingRule(pfRule.getId(), true); + logger.debug("The SSH port forwarding rule {} of the external node with the id {} was revoked", pfRule, vmId); + } + } + } catch (CloudRuntimeException e) { + throw new ManagementServerException(String.format("Failed to remove the SSH rules of the external nodes of the Kubernetes cluster : %s", kubernetesCluster.getName()), e); + } + } + + /** + * @return true if the rule forwards a port of the node SSH port range on the given public IP (from + * CLUSTER_NODES_DEFAULT_START_SSH_PORT up to the SSH ports of the etcd nodes) to port 22, as CKS creates + * it for a node + */ + private boolean isNodeSshPortForwardingRule(final PortForwardingRuleVO pfRule, final IpAddress publicIp) { + return Objects.equals(pfRule.getSourceIpAddressId(), publicIp.getId()) + && pfRule.getDestinationPortStart() == DEFAULT_SSH_PORT + && pfRule.getSourcePortStart() >= CLUSTER_NODES_DEFAULT_START_SSH_PORT + && pfRule.getSourcePortStart() < KubernetesClusterService.KubernetesEtcdNodeStartPort.value(); + } + protected void removeLoadBalancingRule(final IpAddress publicIp, final Network network, final Account account) throws ResourceUnavailableException { List loadBalancerRules = loadBalancerDao.listByIpAddress(publicIp.getId()); diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorker.java index 8f8d3aff87eb..ff397cccdb45 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorker.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorker.java @@ -132,12 +132,27 @@ private void logTransitStateToFailedIfNeededAndThrow(final Level logLevel, final logTransitStateToFailedIfNeededAndThrow(logLevel, message, null); } - private void scaleKubernetesClusterIsolatedNetworkRules(final List clusterVMIds) throws ManagementServerException { + /** + * @return the IDs of the node VMs that get an SSH port on the public IP of the cluster, in the order of + * their ports: the control and worker nodes deployed by CKS first, then the external nodes, as in + * addFirewallRulesForNodes and provisionSshPortForwardingRules + */ + protected List getNodeVmIdsForSshRules() { + List vmMaps = getKubernetesClusterVMMaps(); + List vmIds = vmMaps.stream().filter(vm -> !vm.isExternalNode() && !vm.isEtcdNode()).map(KubernetesClusterVmMapVO::getVmId).collect(Collectors.toList()); + vmIds.addAll(vmMaps.stream().filter(KubernetesClusterVmMapVO::isExternalNode).map(KubernetesClusterVmMapVO::getVmId).collect(Collectors.toList())); + return vmIds; + } + + protected void scaleKubernetesClusterIsolatedNetworkRules(final List clusterVMIds) throws ManagementServerException { IpAddress publicIp = getNetworkSourceNatIp(network); if (publicIp == null) { throw new ManagementServerException(String.format("No source NAT IP addresses found for network : %s, Kubernetes cluster : %s", network.getName(), kubernetesCluster.getName())); } + // The SSH ports of the external nodes follow the range below, their rules are recreated after it + removeExternalNodeSshRules(publicIp); + // Remove existing SSH firewall rules FirewallRule firewallRule = removeSshFirewallRule(publicIp, network.getId()); int existingFirewallRuleSourcePortEnd; @@ -444,11 +459,7 @@ private void removeNodesFromCluster(List vmMaps) throw // Scale network rules to update firewall rule try { - List clusterVMIds = getKubernetesClusterVMMaps() - .stream() - .filter(x -> !x.isEtcdNode()) - .map(KubernetesClusterVmMapVO::getVmId).collect(Collectors.toList()); - scaleKubernetesClusterNetworkRules(clusterVMIds); + scaleKubernetesClusterNetworkRules(getNodeVmIdsForSshRules()); } catch (ManagementServerException e) { logTransitStateAndThrow(Level.ERROR, String.format("Scaling failed for Kubernetes " + "cluster %s, unable to update network rules", kubernetesCluster), @@ -504,10 +515,7 @@ private void scaleUpKubernetesClusterSize(final long newVmCount) throws CloudRun logTransitStateToFailedIfNeededAndThrow(Level.ERROR, String.format("Scaling failed for Kubernetes cluster : %s, unable to provision node VM in the cluster", kubernetesCluster.getName()), e); } try { - List externalNodeIds = getKubernetesClusterVMMaps().stream().filter(KubernetesClusterVmMapVO::isExternalNode).map(KubernetesClusterVmMapVO::getVmId).collect(Collectors.toList()); - List clusterVMIds = getKubernetesClusterVMMaps().stream().filter(vm -> !vm.isExternalNode() && !vm.isEtcdNode()).map(KubernetesClusterVmMapVO::getVmId).collect(Collectors.toList()); - clusterVMIds.addAll(externalNodeIds); - scaleKubernetesClusterNetworkRules(clusterVMIds); + scaleKubernetesClusterNetworkRules(getNodeVmIdsForSshRules()); } catch (ManagementServerException e) { logTransitStateToFailedIfNeededAndThrow(Level.ERROR, String.format("Scaling failed for Kubernetes cluster : %s, unable to update network rules", kubernetesCluster.getName()), e); } diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImplTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImplTest.java index 2a381f282de2..2936dca0b11b 100644 --- a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImplTest.java +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImplTest.java @@ -261,6 +261,22 @@ public void testValidateKubernetesClusterScaleSizeUpscaleNoError() { } + private KubernetesClusterVmMapVO createNodeVmMap(long vmId, boolean externalNode) { + KubernetesClusterVmMapVO vmMap = new KubernetesClusterVmMapVO(1L, vmId, false); + vmMap.setExternalNode(externalNode); + return vmMap; + } + + @Test + public void testValidateKubernetesClusterScaleNodesToRemoveWorkerNodes() { + kubernetesClusterManager.validateKubernetesClusterScaleNodesToRemove(List.of(createNodeVmMap(11L, false), createNodeVmMap(12L, false))); + } + + @Test(expected = InvalidParameterValueException.class) + public void testValidateKubernetesClusterScaleNodesToRemoveExternalNode() { + kubernetesClusterManager.validateKubernetesClusterScaleNodesToRemove(List.of(createNodeVmMap(11L, false), createNodeVmMap(12L, true))); + } + @Before public void setUp() throws Exception { CallContext.register(Mockito.mock(User.class), Mockito.mock(Account.class)); diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorkerTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorkerTest.java index c9299bdbaa67..36e47a32742a 100644 --- a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorkerTest.java +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorkerTest.java @@ -16,27 +16,48 @@ // under the License. package com.cloud.kubernetes.cluster.actionworkers; +import com.cloud.exception.ManagementServerException; +import com.cloud.exception.NetworkRuleConflictException; import com.cloud.kubernetes.cluster.KubernetesCluster; import com.cloud.kubernetes.cluster.KubernetesClusterVmMapVO; import com.cloud.kubernetes.cluster.KubernetesClusterManagerImpl; +import com.cloud.kubernetes.cluster.KubernetesClusterService; import com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDao; +import com.cloud.network.IpAddress; +import com.cloud.network.Network; +import com.cloud.network.dao.FirewallRulesDao; +import com.cloud.network.firewall.FirewallService; +import com.cloud.network.rules.FirewallManager; +import com.cloud.network.rules.FirewallRule; +import com.cloud.network.rules.FirewallRuleVO; +import com.cloud.network.rules.PortForwardingRuleVO; +import com.cloud.network.rules.RulesService; +import com.cloud.network.rules.dao.PortForwardingRulesDao; import com.cloud.offering.ServiceOffering; import com.cloud.service.ServiceOfferingVO; import com.cloud.service.dao.ServiceOfferingDao; import com.cloud.utils.Pair; +import com.cloud.utils.exception.CloudRuntimeException; +import com.cloud.utils.net.Ip; +import com.cloud.utils.net.NetUtils; import com.cloud.vm.UserVmVO; import com.cloud.vm.dao.UserVmDao; +import org.apache.cloudstack.context.CallContext; import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; +import org.mockito.MockedStatic; import org.mockito.Mockito; import org.mockito.junit.MockitoJUnitRunner; +import org.springframework.test.util.ReflectionTestUtils; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.stream.Collectors; import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.CONTROL; import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.DEFAULT; @@ -187,4 +208,306 @@ public void testGetWorkerNodesToRemoveForDownsize_noRemoval() { Assert.assertTrue(toRemove.isEmpty()); } + + private static final long CLUSTER_ID = 1L; + private static final long NETWORK_ID = 10L; + private static final long PUBLIC_IP_ID = 20L; + private static final long OTHER_PUBLIC_IP_ID = 21L; + private static final long CONTROL_VM_ID = 100L; + private static final long WORKER_VM_ID = 101L; + private static final long EXTERNAL_VM_ID = 102L; + private static final long NEW_WORKER_VM_ID = 103L; + private static final long SECOND_EXTERNAL_VM_ID = 104L; + private static final long ETCD_VM_ID = 105L; + private static final long THIRD_WORKER_VM_ID = 106L; + private static final int API_PORT = KubernetesClusterActionWorker.CLUSTER_API_PORT; + private static final int SSH_PORT = KubernetesClusterActionWorker.DEFAULT_SSH_PORT; + private static final int ETCD_SSH_PORT = KubernetesClusterService.KubernetesEtcdNodeStartPort.value(); + + // VM map of the cluster and the rules of its network; all firewall rules are on the source NAT IP + private final List vmMaps = new ArrayList<>(); + private final List firewallRules = new ArrayList<>(); + private final List portForwardingRules = new ArrayList<>(); + private long lastRuleId = 0; + private IpAddress publicIp; + + private void addVm(long vmId, boolean controlNode, boolean externalNode, boolean etcdNode) { + KubernetesClusterVmMapVO vmMap = new KubernetesClusterVmMapVO(CLUSTER_ID, vmId, controlNode); + vmMap.setExternalNode(externalNode); + vmMap.setEtcdNode(etcdNode); + vmMaps.add(vmMap); + } + + // As when the VM of a removed node is expunged + private void removeVm(long vmId) { + vmMaps.removeIf(vmMap -> vmMap.getVmId() == vmId); + portForwardingRules.removeIf(rule -> rule.getVirtualMachineId() == vmId); + } + + private void addFirewallRule(int startPort, int endPort) { + FirewallRuleVO rule = new FirewallRuleVO(null, PUBLIC_IP_ID, startPort, endPort, NetUtils.TCP_PROTO, NETWORK_ID, 2L, 3L, + FirewallRule.Purpose.Firewall, List.of("0.0.0.0/0"), null, null, null, FirewallRule.TrafficType.Ingress); + ReflectionTestUtils.setField(rule, "id", ++lastRuleId); + firewallRules.add(rule); + } + + private long addPortForwardingRule(long publicIpId, int publicPort, long vmId, int privatePort) { + PortForwardingRuleVO rule = new PortForwardingRuleVO(null, publicIpId, publicPort, publicPort, new Ip("10.1.1.10"), privatePort, privatePort, + NetUtils.TCP_PROTO, NETWORK_ID, 2L, 3L, vmId); + ReflectionTestUtils.setField(rule, "id", ++lastRuleId); + portForwardingRules.add(rule); + return rule.getId(); + } + + private List getFirewallRules() { + return firewallRules.stream().map(rule -> rule.getSourcePortStart() + "-" + rule.getSourcePortEnd()).sorted().collect(Collectors.toList()); + } + + private List getPortForwardingRules() { + return portForwardingRules.stream().map(rule -> String.format("%d:%d->%d:%d", rule.getSourceIpAddressId(), rule.getSourcePortStart(), + rule.getVirtualMachineId(), rule.getDestinationPortStart())).sorted().collect(Collectors.toList()); + } + + /** + * Returns a worker for the cluster in vmMaps on an isolated network, whose firewall and port forwarding rules + * are kept in firewallRules and portForwardingRules. A new firewall rule is refused if its ports overlap an + * existing one, as FirewallManagerImpl.detectRulesConflict does for two TCP rules with the source CIDR 0.0.0.0/0. + */ + private KubernetesClusterScaleWorker createWorkerForNetworkRules() throws Exception { + PortForwardingRulesDao portForwardingRulesDao = Mockito.mock(PortForwardingRulesDao.class); + FirewallRulesDao firewallRulesDao = Mockito.mock(FirewallRulesDao.class); + FirewallService firewallService = Mockito.mock(FirewallService.class); + RulesService rulesService = Mockito.mock(RulesService.class); + Network network = Mockito.mock(Network.class); + publicIp = Mockito.mock(IpAddress.class); + + Mockito.lenient().when(kubernetesCluster.getId()).thenReturn(CLUSTER_ID); + Mockito.lenient().when(kubernetesCluster.getEtcdNodeCount()).thenAnswer(invocation -> vmMaps.stream().filter(KubernetesClusterVmMapVO::isEtcdNode).count()); + Mockito.lenient().when(kubernetesClusterVmMapDao.listByClusterId(CLUSTER_ID)).thenAnswer(invocation -> new ArrayList<>(vmMaps)); + Mockito.lenient().when(network.getId()).thenReturn(NETWORK_ID); + Mockito.lenient().when(publicIp.getId()).thenReturn(PUBLIC_IP_ID); + Mockito.lenient().when(publicIp.getAddress()).thenReturn(new Ip("192.0.2.10")); + Mockito.lenient().when(firewallRulesDao.listByIpPurposeProtocolAndNotRevoked(PUBLIC_IP_ID, FirewallRule.Purpose.Firewall, NetUtils.TCP_PROTO)) + .thenAnswer(invocation -> new ArrayList<>(firewallRules)); + Mockito.lenient().when(firewallRulesDao.listByIpPurposePortsProtocolAndNotRevoked(Mockito.eq(PUBLIC_IP_ID), Mockito.any(), Mockito.any(), + Mockito.eq(NetUtils.TCP_PROTO), Mockito.eq(FirewallRule.Purpose.Firewall))).thenAnswer(invocation -> firewallRules.stream() + .filter(rule -> rule.getSourcePortStart().equals(invocation.getArgument(1)) && rule.getSourcePortEnd().equals(invocation.getArgument(2))) + .collect(Collectors.toList())); + Mockito.lenient().when(firewallService.revokeIngressFwRule(Mockito.anyLong(), Mockito.eq(true))) + .thenAnswer(invocation -> firewallRules.removeIf(rule -> rule.getId() == (long) invocation.getArgument(0))); + Mockito.lenient().when(portForwardingRulesDao.listByNetwork(NETWORK_ID)).thenAnswer(invocation -> new ArrayList<>(portForwardingRules)); + Mockito.lenient().when(portForwardingRulesDao.listByVm(Mockito.anyLong())).thenAnswer(invocation -> portForwardingRules.stream() + .filter(rule -> rule.getVirtualMachineId() == (long) invocation.getArgument(0)).collect(Collectors.toList())); + Mockito.lenient().when(portForwardingRulesDao.findByNetworkAndPorts(Mockito.eq(NETWORK_ID), Mockito.anyInt(), Mockito.anyInt())) + .thenAnswer(invocation -> portForwardingRules.stream().filter(rule -> rule.getSourcePortStart() == (int) invocation.getArgument(1) + && rule.getSourcePortEnd() == (int) invocation.getArgument(2)).findFirst().orElse(null)); + Mockito.lenient().when(portForwardingRulesDao.remove(Mockito.anyLong())) + .thenAnswer(invocation -> portForwardingRules.removeIf(rule -> rule.getId() == (long) invocation.getArgument(0))); + Mockito.lenient().when(rulesService.revokePortForwardingRule(Mockito.anyLong(), Mockito.eq(true))) + .thenAnswer(invocation -> portForwardingRules.removeIf(rule -> rule.getId() == (long) invocation.getArgument(0))); + + KubernetesClusterScaleWorker scaleWorker = Mockito.spy(new KubernetesClusterScaleWorker(kubernetesCluster, clusterManager)); + scaleWorker.kubernetesClusterVmMapDao = kubernetesClusterVmMapDao; + scaleWorker.portForwardingRulesDao = portForwardingRulesDao; + scaleWorker.firewallRulesDao = firewallRulesDao; + scaleWorker.firewallService = firewallService; + scaleWorker.rulesService = rulesService; + scaleWorker.firewallManager = Mockito.mock(FirewallManager.class); + ReflectionTestUtils.setField(scaleWorker, "network", network); + Mockito.lenient().doReturn(publicIp).when(scaleWorker).getNetworkSourceNatIp(network); + Mockito.lenient().doAnswer(invocation -> { + int startPort = invocation.getArgument(2); + int endPort = invocation.getArgument(3); + for (FirewallRuleVO rule : firewallRules) { + if (startPort <= rule.getSourcePortEnd() && rule.getSourcePortStart() <= endPort) { + throw new NetworkRuleConflictException(String.format("The range specified, %d-%d, conflicts with rule %d-%d", + startPort, endPort, rule.getSourcePortStart(), rule.getSourcePortEnd())); + } + } + addFirewallRule(startPort, endPort); + return null; + }).when(scaleWorker).provisionFirewallRules(Mockito.eq(publicIp), Mockito.any(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.lenient().doAnswer(invocation -> { + addPortForwardingRule(PUBLIC_IP_ID, invocation.getArgument(4), invocation.getArgument(3), invocation.getArgument(5)); + return null; + }).when(scaleWorker).provisionPublicIpPortForwardingRule(Mockito.eq(publicIp), Mockito.eq(network), Mockito.any(), + Mockito.anyLong(), Mockito.anyInt(), Mockito.anyInt()); + return scaleWorker; + } + + // With the node VMs as passed by scaleUpKubernetesClusterSize and removeNodesFromCluster + private void scaleIsolatedNetworkRules(KubernetesClusterScaleWorker scaleWorker) throws ManagementServerException { + try (MockedStatic ignored = Mockito.mockStatic(CallContext.class)) { + scaleWorker.scaleKubernetesClusterIsolatedNetworkRules(scaleWorker.getNodeVmIdsForSshRules()); + } + } + + @Test + public void testGetNodeVmIdsForSshRulesPutsExternalNodesLast() { + // A worker added by a scale-up after an external node, and an etcd node + addVm(CONTROL_VM_ID, true, false, false); + addVm(WORKER_VM_ID, false, false, false); + addVm(EXTERNAL_VM_ID, false, true, false); + addVm(NEW_WORKER_VM_ID, false, false, false); + addVm(ETCD_VM_ID, false, false, true); + Mockito.when(kubernetesCluster.getId()).thenReturn(CLUSTER_ID); + Mockito.when(kubernetesClusterVmMapDao.listByClusterId(CLUSTER_ID)).thenReturn(vmMaps); + + Assert.assertEquals(List.of(CONTROL_VM_ID, WORKER_VM_ID, NEW_WORKER_VM_ID, EXTERNAL_VM_ID), worker.getNodeVmIdsForSshRules()); + } + + @Test + public void testScaleKubernetesClusterIsolatedNetworkRulesScaleUpWithExternalNodes() throws Exception { + // 1 control node and 1 worker (2222-2223), two external nodes (2224, 2225), an etcd node and the worker added by the scale-up + addVm(CONTROL_VM_ID, true, false, false); + addVm(WORKER_VM_ID, false, false, false); + addVm(EXTERNAL_VM_ID, false, true, false); + addVm(SECOND_EXTERNAL_VM_ID, false, true, false); + addVm(ETCD_VM_ID, false, false, true); + addVm(NEW_WORKER_VM_ID, false, false, false); + KubernetesClusterScaleWorker scaleWorker = createWorkerForNetworkRules(); + // The rules of the external nodes are listed first, removeSshFirewallRule must still find the range + addFirewallRule(2224, 2224); + addFirewallRule(2225, 2225); + addFirewallRule(2222, 2223); + addFirewallRule(API_PORT, API_PORT); + addFirewallRule(ETCD_SSH_PORT, ETCD_SSH_PORT); + addPortForwardingRule(PUBLIC_IP_ID, 2222, CONTROL_VM_ID, SSH_PORT); + addPortForwardingRule(PUBLIC_IP_ID, 2223, WORKER_VM_ID, SSH_PORT); + addPortForwardingRule(PUBLIC_IP_ID, 2224, EXTERNAL_VM_ID, SSH_PORT); + addPortForwardingRule(PUBLIC_IP_ID, 2225, SECOND_EXTERNAL_VM_ID, SSH_PORT); + addPortForwardingRule(PUBLIC_IP_ID, ETCD_SSH_PORT, ETCD_VM_ID, SSH_PORT); + + scaleIsolatedNetworkRules(scaleWorker); + + Assert.assertEquals(List.of("2222-2224", "2225-2225", "2226-2226", ETCD_SSH_PORT + "-" + ETCD_SSH_PORT, "6443-6443"), getFirewallRules()); + Assert.assertEquals(List.of("20:2222->100:22", "20:2223->101:22", "20:2224->103:22", "20:2225->102:22", "20:2226->104:22", + "20:" + ETCD_SSH_PORT + "->105:22"), getPortForwardingRules()); + } + + @Test + public void testScaleKubernetesClusterIsolatedNetworkRulesScaleUpThenDownWithExternalNode() throws Exception { + // 1 control node and 1 worker (2222-2223) and an external node (2224) + addVm(CONTROL_VM_ID, true, false, false); + addVm(WORKER_VM_ID, false, false, false); + addVm(EXTERNAL_VM_ID, false, true, false); + KubernetesClusterScaleWorker scaleWorker = createWorkerForNetworkRules(); + addFirewallRule(2222, 2223); + addFirewallRule(API_PORT, API_PORT); + addFirewallRule(2224, 2224); + addPortForwardingRule(PUBLIC_IP_ID, 2222, CONTROL_VM_ID, SSH_PORT); + addPortForwardingRule(PUBLIC_IP_ID, 2223, WORKER_VM_ID, SSH_PORT); + addPortForwardingRule(PUBLIC_IP_ID, 2224, EXTERNAL_VM_ID, SSH_PORT); + + // Scale-up by one worker, which comes after the external node in the VM map + addVm(NEW_WORKER_VM_ID, false, false, false); + scaleIsolatedNetworkRules(scaleWorker); + Assert.assertEquals(List.of("2222-2224", "2225-2225", "6443-6443"), getFirewallRules()); + Assert.assertEquals(List.of("20:2222->100:22", "20:2223->101:22", "20:2224->103:22", "20:2225->102:22"), getPortForwardingRules()); + + // Scale-down removing the first worker by its ID + removeVm(WORKER_VM_ID); + scaleIsolatedNetworkRules(scaleWorker); + Assert.assertEquals(List.of("2222-2223", "2224-2224", "6443-6443"), getFirewallRules()); + Assert.assertEquals(List.of("20:2222->100:22", "20:2223->103:22", "20:2224->102:22"), getPortForwardingRules()); + + // Scale-up again + addVm(THIRD_WORKER_VM_ID, false, false, false); + scaleIsolatedNetworkRules(scaleWorker); + Assert.assertEquals(List.of("2222-2224", "2225-2225", "6443-6443"), getFirewallRules()); + Assert.assertEquals(List.of("20:2222->100:22", "20:2223->103:22", "20:2224->106:22", "20:2225->102:22"), getPortForwardingRules()); + } + + @Test + public void testScaleKubernetesClusterIsolatedNetworkRulesScaleUpWithoutExternalNode() throws Exception { + addVm(CONTROL_VM_ID, true, false, false); + addVm(WORKER_VM_ID, false, false, false); + addVm(NEW_WORKER_VM_ID, false, false, false); + KubernetesClusterScaleWorker scaleWorker = createWorkerForNetworkRules(); + addFirewallRule(2222, 2223); + addFirewallRule(API_PORT, API_PORT); + addPortForwardingRule(PUBLIC_IP_ID, 2222, CONTROL_VM_ID, SSH_PORT); + addPortForwardingRule(PUBLIC_IP_ID, 2223, WORKER_VM_ID, SSH_PORT); + + scaleIsolatedNetworkRules(scaleWorker); + + Assert.assertEquals(List.of("2222-2224", "6443-6443"), getFirewallRules()); + Assert.assertEquals(List.of("20:2222->100:22", "20:2223->101:22", "20:2224->103:22"), getPortForwardingRules()); + Mockito.verify(scaleWorker.portForwardingRulesDao, Mockito.never()).listByVm(Mockito.anyLong()); + Mockito.verify(scaleWorker.rulesService, Mockito.never()).revokePortForwardingRule(Mockito.anyLong(), Mockito.anyBoolean()); + } + + @Test + public void testRemoveExternalNodeSshRulesKeepsOtherRules() throws Exception { + addVm(CONTROL_VM_ID, true, false, false); + addVm(WORKER_VM_ID, false, false, false); + addVm(EXTERNAL_VM_ID, false, true, false); + KubernetesClusterScaleWorker scaleWorker = createWorkerForNetworkRules(); + addFirewallRule(2222, 2223); + addFirewallRule(API_PORT, API_PORT); + addFirewallRule(2224, 2224); + addPortForwardingRule(PUBLIC_IP_ID, 2222, CONTROL_VM_ID, SSH_PORT); + addPortForwardingRule(PUBLIC_IP_ID, 2223, WORKER_VM_ID, SSH_PORT); + addPortForwardingRule(PUBLIC_IP_ID, 2224, EXTERNAL_VM_ID, SSH_PORT); + // An old SSH rule of the external node, left by an earlier scale-down + addFirewallRule(2226, 2226); + addPortForwardingRule(PUBLIC_IP_ID, 2226, EXTERNAL_VM_ID, SSH_PORT); + // Other rules of the external node VM: to another port, from outside the node SSH port range, and on another IP + addFirewallRule(8080, 8080); + addPortForwardingRule(PUBLIC_IP_ID, 8080, EXTERNAL_VM_ID, 80); + addFirewallRule(2022, 2022); + addPortForwardingRule(PUBLIC_IP_ID, 2022, EXTERNAL_VM_ID, SSH_PORT); + addFirewallRule(ETCD_SSH_PORT + 22, ETCD_SSH_PORT + 22); + addPortForwardingRule(PUBLIC_IP_ID, ETCD_SSH_PORT + 22, EXTERNAL_VM_ID, SSH_PORT); + addPortForwardingRule(OTHER_PUBLIC_IP_ID, 2224, EXTERNAL_VM_ID, SSH_PORT); + + scaleWorker.removeExternalNodeSshRules(publicIp); + + String etcdRangePort = String.valueOf(ETCD_SSH_PORT + 22); + Assert.assertEquals(List.of("2022-2022", "2222-2223", etcdRangePort + "-" + etcdRangePort, "6443-6443", "8080-8080"), getFirewallRules()); + Assert.assertEquals(List.of("20:2022->102:22", "20:2222->100:22", "20:2223->101:22", "20:" + etcdRangePort + "->102:22", "20:8080->102:80", + "21:2224->102:22"), getPortForwardingRules()); + } + + @Test + public void testRemoveExternalNodeSshRulesCanBeRetriedAfterAFailure() throws Exception { + addVm(CONTROL_VM_ID, true, false, false); + addVm(EXTERNAL_VM_ID, false, true, false); + addVm(SECOND_EXTERNAL_VM_ID, false, true, false); + KubernetesClusterScaleWorker scaleWorker = createWorkerForNetworkRules(); + addFirewallRule(2222, 2222); + addFirewallRule(2223, 2223); + addFirewallRule(2224, 2224); + addPortForwardingRule(PUBLIC_IP_ID, 2222, CONTROL_VM_ID, SSH_PORT); + addPortForwardingRule(PUBLIC_IP_ID, 2223, EXTERNAL_VM_ID, SSH_PORT); + long failingRuleId = addPortForwardingRule(PUBLIC_IP_ID, 2224, SECOND_EXTERNAL_VM_ID, SSH_PORT); + // Revoking the port forwarding rule of the second external node fails once + Mockito.doThrow(new CloudRuntimeException("Failed to delete port forwarding rule")) + .doAnswer(invocation -> portForwardingRules.removeIf(rule -> rule.getId() == failingRuleId)) + .when(scaleWorker.rulesService).revokePortForwardingRule(failingRuleId, true); + + Assert.assertThrows(ManagementServerException.class, () -> scaleWorker.removeExternalNodeSshRules(publicIp)); + Assert.assertEquals(List.of("2222-2222"), getFirewallRules()); + Assert.assertEquals(List.of("20:2222->100:22", "20:2224->104:22"), getPortForwardingRules()); + + scaleWorker.removeExternalNodeSshRules(publicIp); + Assert.assertEquals(List.of("2222-2222"), getFirewallRules()); + Assert.assertEquals(List.of("20:2222->100:22"), getPortForwardingRules()); + } + + @Test + public void testRemoveExternalNodeSshRulesReportsAFirewallRuleThatIsNotRevoked() throws Exception { + addVm(CONTROL_VM_ID, true, false, false); + addVm(EXTERNAL_VM_ID, false, true, false); + KubernetesClusterScaleWorker scaleWorker = createWorkerForNetworkRules(); + addFirewallRule(2222, 2222); + addFirewallRule(2223, 2223); + addPortForwardingRule(PUBLIC_IP_ID, 2222, CONTROL_VM_ID, SSH_PORT); + addPortForwardingRule(PUBLIC_IP_ID, 2223, EXTERNAL_VM_ID, SSH_PORT); + Mockito.doReturn(false).when(scaleWorker.firewallService).revokeIngressFwRule(Mockito.anyLong(), Mockito.eq(true)); + + Assert.assertThrows(ManagementServerException.class, () -> scaleWorker.removeExternalNodeSshRules(publicIp)); + Assert.assertEquals(List.of("20:2222->100:22", "20:2223->102:22"), getPortForwardingRules()); + Mockito.verify(scaleWorker.rulesService, Mockito.never()).revokePortForwardingRule(Mockito.anyLong(), Mockito.anyBoolean()); + } }