CKS: fix scaling a cluster with external nodes

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java

@@ -1243,6 +1243,18 @@ protected void addKubernetesClusterDetailIfIsNotEmpty(List<KubernetesClusterDeta
         }
     }
 
+    /**
+     * Scaling removes only nodes that CKS deployed (see KubernetesClusterScaleWorker.scaleDownKubernetesClusterSize),
+     * external nodes are removed with removeNodesFromKubernetesCluster.
+     *
+     * @param nodes the nodes passed in nodeids to scaleKubernetesCluster
+     */
+    protected void validateKubernetesClusterScaleNodesToRemove(final List<KubernetesClusterVmMapVO> 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()) {

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<Long> 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<FirewallRuleVO> 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<LoadBalancerVO> loadBalancerRules = loadBalancerDao.listByIpAddress(publicIp.getId());

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<Long> 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<Long> getNodeVmIdsForSshRules() {
+        List<KubernetesClusterVmMapVO> vmMaps = getKubernetesClusterVMMaps();
+        List<Long> 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<Long> 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<KubernetesClusterVmMapVO> vmMaps) throw
 
         // Scale network rules to update firewall rule
         try {
-            List<Long> 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<Long> externalNodeIds = getKubernetesClusterVMMaps().stream().filter(KubernetesClusterVmMapVO::isExternalNode).map(KubernetesClusterVmMapVO::getVmId).collect(Collectors.toList());
-            List<Long> 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);
         }

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));