diff --git a/api/src/main/java/com/cloud/kubernetes/cluster/KubernetesCluster.java b/api/src/main/java/com/cloud/kubernetes/cluster/KubernetesCluster.java index ce905b293ff3..751fb030693f 100644 --- a/api/src/main/java/com/cloud/kubernetes/cluster/KubernetesCluster.java +++ b/api/src/main/java/com/cloud/kubernetes/cluster/KubernetesCluster.java @@ -103,6 +103,7 @@ enum State { s_fsm.addTransition(State.ScalingStoppedCluster, Event.OperationFailed, State.Alert); s_fsm.addTransition(State.Running, Event.UpgradeRequested, State.Upgrading); + s_fsm.addTransition(State.Alert, Event.UpgradeRequested, State.Upgrading); s_fsm.addTransition(State.Upgrading, Event.OperationSucceeded, State.Running); s_fsm.addTransition(State.Upgrading, Event.OperationFailed, State.Alert); 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..d1ff79e90e12 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 @@ -126,6 +126,7 @@ import org.apache.commons.collections.MapUtils; import org.apache.commons.lang3.ObjectUtils; import org.apache.commons.lang3.StringUtils; +import org.apache.commons.lang3.math.NumberUtils; import org.apache.logging.log4j.Level; import com.cloud.api.ApiDBUtils; @@ -169,6 +170,7 @@ import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterStartWorker; import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterStopWorker; import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterUpgradeWorker; +import com.cloud.kubernetes.cluster.utils.KubernetesClusterUtil; import com.cloud.kubernetes.cluster.dao.KubernetesClusterDao; import com.cloud.kubernetes.cluster.dao.KubernetesClusterDetailsDao; import com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDao; @@ -1478,6 +1480,40 @@ protected boolean isAnyNodeOfferingEmpty(Map map) { return map.values().stream().anyMatch(Objects::isNull); } + /** + * Tells whether upgrading a cluster in Alert state to a version retries an upgrade to that version that failed. + * An upgrade records the version it is upgrading to as a detail of the cluster, which is removed once the cluster + * is Running again. + * For an upgrade that failed before that detail was recorded, the nodes' versions are used: the version of the cluster is only changed once every node has been upgraded, while the version recorded for + * each node is changed as soon as that node has been upgraded. So, a cluster that is still at an older version + * but has nodes at the target version is one whose upgrade to that version stopped part way. Nodes that are + * upgraded manually are left out, as they may be at any version. + */ + protected boolean isRetryOfFailedUpgrade(KubernetesCluster kubernetesCluster, KubernetesSupportedVersion upgradeVersion) { + if (Objects.equals(upgradeVersion.getId(), kubernetesCluster.getKubernetesVersionId())) { + return false; + } + // Another version with the same semantic version as the cluster's is not an upgrade either + KubernetesSupportedVersionVO clusterVersion = kubernetesSupportedVersionDao.findById(kubernetesCluster.getKubernetesVersionId()); + if (clusterVersion == null || KubernetesClusterUtil.isSameNodeVersion(clusterVersion.getSemanticVersion(), upgradeVersion.getSemanticVersion())) { + return false; + } + KubernetesClusterDetailsVO upgradeDetail = kubernetesClusterDetailsDao.findDetail(kubernetesCluster.getId(), KubernetesClusterUpgradeWorker.UPGRADE_VERSION_DETAIL); + if (upgradeDetail != null) { + if (String.valueOf(upgradeVersion.getId()).equals(upgradeDetail.getValue())) { + return true; + } + // Another version with the same semantic version, for example one registered with a fixed ISO + KubernetesSupportedVersionVO failedVersion = NumberUtils.isDigits(upgradeDetail.getValue()) ? + kubernetesSupportedVersionDao.findByIdIncludingRemoved(Long.parseLong(upgradeDetail.getValue())) : null; + return failedVersion != null && KubernetesClusterUtil.isSameNodeVersion(failedVersion.getSemanticVersion(), upgradeVersion.getSemanticVersion()); + } + List vmMaps = kubernetesClusterVmMapDao.listByClusterId(kubernetesCluster.getId()); + return vmMaps != null && vmMaps.stream() + .filter(vmMap -> !vmMap.isManualUpgrade()) + .anyMatch(vmMap -> KubernetesClusterUtil.isSameNodeVersion(vmMap.getNodeVersion(), upgradeVersion.getSemanticVersion())); + } + private void validateKubernetesClusterUpgradeParameters(UpgradeKubernetesClusterCmd cmd) { // Validate parameters ApiServiceConfiguration.validateEndpointUrl(); @@ -1498,7 +1534,8 @@ private void validateKubernetesClusterUpgradeParameters(UpgradeKubernetesCluster throw new InvalidParameterValueException(String.format("Upgrade kubernetes cluster is not supported for an externally managed cluster (%s)", kubernetesCluster.getName())); } accountManager.checkAccess(CallContext.current().getCallingAccount(), SecurityChecker.AccessType.OperateEntry, false, kubernetesCluster); - if (!KubernetesCluster.State.Running.equals(kubernetesCluster.getState())) { + boolean inAlert = KubernetesCluster.State.Alert.equals(kubernetesCluster.getState()); + if (!inAlert && !KubernetesCluster.State.Running.equals(kubernetesCluster.getState())) { throw new InvalidParameterValueException(String.format("Kubernetes cluster : %s is not in running state", kubernetesCluster.getName())); } final DataCenter zone = dataCenterDao.findById(kubernetesCluster.getZoneId()); @@ -1512,6 +1549,16 @@ private void validateKubernetesClusterUpgradeParameters(UpgradeKubernetesCluster if (!KubernetesSupportedVersion.State.Enabled.equals(upgradeVersion.getState())) { throw new InvalidParameterValueException(String.format("Kubernetes version ID: %s for upgrade is in %s state", upgradeVersion.getUuid(), upgradeVersion.getState())); } + if (inAlert) { + if (!isRetryOfFailedUpgrade(kubernetesCluster, upgradeVersion)) { + throw new InvalidParameterValueException(String.format("Kubernetes cluster : %s is in Alert state. It can be upgraded in that state only " + + "to retry an upgrade that failed, to the same Kubernetes version as that upgrade", kubernetesCluster.getName())); + } + if (!isClusterVMsInDesiredState(kubernetesCluster, VirtualMachine.State.Running)) { + throw new InvalidParameterValueException(String.format("Kubernetes cluster : %s is in Alert state and not all of its VMs are running. " + + "Start them before retrying the upgrade", kubernetesCluster.getName())); + } + } KubernetesSupportedVersionVO clusterVersion = kubernetesSupportedVersionDao.findById(kubernetesCluster.getKubernetesVersionId()); if (clusterVersion == null || clusterVersion.getRemoved() != null) { throw new InvalidParameterValueException(String.format("Invalid Kubernetes version associated with cluster : %s", @@ -1548,7 +1595,13 @@ private void validateKubernetesClusterUpgradeParameters(UpgradeKubernetesCluster protected boolean stateTransitTo(long kubernetesClusterId, KubernetesCluster.Event e) { KubernetesClusterVO kubernetesCluster = kubernetesClusterDao.findById(kubernetesClusterId); try { - return _stateMachine.transitTo(kubernetesCluster, e, null, kubernetesClusterDao); + boolean transited = _stateMachine.transitTo(kubernetesCluster, e, null, kubernetesClusterDao); + if (transited && (KubernetesCluster.State.Running.equals(kubernetesCluster.getState()) || + KubernetesCluster.State.Stopped.equals(kubernetesCluster.getState()))) { + // As in KubernetesClusterActionWorker.stateTransitTo: no longer in Alert because of a failed upgrade + kubernetesClusterDetailsDao.removeDetail(kubernetesClusterId, KubernetesClusterUpgradeWorker.UPGRADE_VERSION_DETAIL); + } + return transited; } catch (NoTransitionException nte) { logger.warn("Failed to transition state of the Kubernetes cluster: {} in state {} on event {}", kubernetesCluster, kubernetesCluster.getState().toString(), e.toString(), nte); return false; @@ -2808,6 +2861,7 @@ public ConfigKey[] getConfigKeys() { KubernetesClusterScaleTimeout, KubernetesClusterUpgradeTimeout, KubernetesClusterUpgradeRetries, + KubernetesClusterNodeDrainTimeout, KubernetesClusterAddNodeTimeout, KubernetesClusterRemoveNodeTimeout, KubernetesClusterExperimentalFeaturesEnabled, diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterService.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterService.java index 6bdb4265e019..1afcd86fcb22 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterService.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterService.java @@ -87,6 +87,13 @@ public interface KubernetesClusterService extends PluggableService, Configurable "The number of retries if fail to upgrade kubernetes cluster due to some reasons (e.g. drain node, etcdserver leader changed)", true, KubernetesServiceEnabled.key()); + static final ConfigKey KubernetesClusterNodeDrainTimeout = new ConfigKey("Advanced", Long.class, + "cloud.kubernetes.cluster.node.drain.timeout", + "300", + "Timeout interval (in seconds) for draining a Kubernetes node before it is removed from a cluster or upgraded. " + + "A drain that does not finish in time, for example because a PodDisruptionBudget does not allow an eviction, is retried, and the node is uncordoned if it still fails. Must be at least 1, smaller values use the default", + true, + KubernetesServiceEnabled.key()); static final ConfigKey KubernetesClusterAddNodeTimeout = new ConfigKey("Advanced", Long.class, "cloud.kubernetes.cluster.add.node.timeout", "3600", diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorker.java index 7a06b11e3bf4..d1817504b046 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorker.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorker.java @@ -31,6 +31,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; @@ -147,6 +148,12 @@ public class KubernetesClusterActionWorker { public static final int ETCD_NODE_CLIENT_REQUEST_PORT = 2379; public static final int ETCD_NODE_PEER_COMM_PORT = 2380; public static final int CLUSTER_NODES_DEFAULT_SSH_PORT_SG = DEFAULT_SSH_PORT; + /** How long to wait before retrying a failed drain of a node, in milliseconds. */ + protected static final long DRAIN_RETRY_WAIT = 30000; + /** How much of the output of a failed drain to put in the error, in characters. */ + protected static final int MAX_DRAIN_OUTPUT_IN_ERROR = 2000; + /** How much longer than kubectl drain's own timeout the SSH call waits, in seconds. */ + protected static final long DRAIN_SSH_MARGIN = 60; public static final String CKS_CLUSTER_SECURITY_GROUP_NAME = "CKSSecurityGroup"; public static final String CKS_SECURITY_GROUP_DESCRIPTION = "Security group for CKS nodes"; @@ -355,8 +362,9 @@ protected void logMessage(final Level logLevel, final String message, final Exce protected void logTransitStateDetachIsoAndThrow(final Level logLevel, final String message, final KubernetesCluster kubernetesCluster, final List clusterVMs, final KubernetesCluster.Event event, final Exception e) throws CloudRuntimeException { logMessage(logLevel, message, e); - stateTransitTo(kubernetesCluster.getId(), event); + // Detach first: once the cluster is in Alert, a retry of the upgrade may attach the ISO again detachIsoKubernetesVMs(clusterVMs); + stateTransitTo(kubernetesCluster.getId(), event); if (e == null) { throw new CloudRuntimeException(message); } @@ -757,10 +765,156 @@ protected void updateLoginUserDetails(List clusterVMs) { } } + /** Runs a command on a cluster node over SSH, as the cluster's login user with the management server's key. */ + protected Pair executeSshCommand(final String ipAddress, final int port, final String command, + final int connectTimeout, final int kexTimeout, final int waitTimeout) throws Exception { + return SshHelper.sshExecute(ipAddress, port, getControlNodeLoginUser(), getManagementServerSshPublicKeyFile(), null, + command, connectTimeout, kexTimeout, waitTimeout); + } + + /** Waits before a retry. Returns false if the thread was interrupted, in which case the caller should stop retrying. */ + protected boolean sleepBeforeRetry(final long millis) { + try { + Thread.sleep(millis); + return true; + } catch (InterruptedException ie) { + logger.warn("Interrupted while waiting to retry an operation on Kubernetes cluster: {}", kubernetesCluster, ie); + Thread.currentThread().interrupt(); + return false; + } + } + + protected static String getNodeName(final UserVm vm) { + String hostName = vm.getHostName(); + return StringUtils.isNotEmpty(hostName) ? hostName.toLowerCase() : hostName; + } + + /** + * Drains a Kubernetes node through the cluster's API endpoint, so that it can be removed or upgraded. + * Each attempt runs "kubectl drain" with the timeout of the cloud.kubernetes.cluster.node.drain.timeout setting, + * and the SSH call waits a little longer than that. A failed attempt, whether the command fails or the SSH call + * does, is retried after waiting, until the attempts are used up or the deadline has passed. No attempt runs + * past the deadline. A failed drain leaves the node cordoned: the caller uncordons it if it cordoned it, as only + * the caller knows whether the node was cordoned before. + * + * @return whether the node was drained, and the output of the last attempt, which explains a failure + */ + protected Pair drainKubernetesClusterNode(final String ipAddress, final int port, final UserVm vm, + final int attempts, final long retryWait, final long deadline) { + final String nodeName = getNodeName(vm); + Long configuredTimeout = KubernetesClusterService.KubernetesClusterNodeDrainTimeout.value(); + final long drainTimeout = configuredTimeout != null && configuredTimeout >= 1 ? configuredTimeout : + Long.parseLong(KubernetesClusterService.KubernetesClusterNodeDrainTimeout.defaultValue()); + String output = "no time was left to drain the node before the operation timed out"; + for (int attempt = 1; attempt <= attempts; attempt++) { + if (attempt > 1) { + // Only wait if there will still be time for an attempt afterwards + if (System.currentTimeMillis() + retryWait + (DRAIN_SSH_MARGIN + 1) * 1000 > deadline || !sleepBeforeRetry(retryWait)) { + break; + } + } + // Don't let an attempt run past the deadline of the operation, which would evict pods for nothing + final long secondsLeft = (deadline - System.currentTimeMillis()) / 1000 - DRAIN_SSH_MARGIN; + if (secondsLeft < 1) { + break; + } + final long timeout = Math.min(drainTimeout, secondsLeft); + final String command = String.format("sudo /opt/bin/kubectl drain %s --ignore-daemonsets --delete-emptydir-data --timeout=%ds", + nodeName, timeout); + final int sshWaitTimeout = (int) Math.min(Integer.MAX_VALUE, (timeout + DRAIN_SSH_MARGIN) * 1000); + try { + Pair result = executeSshCommand(ipAddress, port, command, 10000, 10000, sshWaitTimeout); + output = result.second(); + if (Boolean.TRUE.equals(result.first())) { + return new Pair<>(true, output); + } + logger.warn("Draining node: {} on VM: {} in Kubernetes cluster: {} failed (attempt {} of {}): {}", + nodeName, vm, kubernetesCluster, attempt, attempts, output); + } catch (Exception e) { + output = e.getMessage(); + logger.warn("Draining node: {} on VM: {} in Kubernetes cluster: {} failed (attempt {} of {})", + nodeName, vm, kubernetesCluster, attempt, attempts, e); + } + } + // A drain blocked by a PodDisruptionBudget logs every retry of every eviction; the end is what explains it + return new Pair<>(false, StringUtils.right(output, MAX_DRAIN_OUTPUT_IN_ERROR)); + } + + /** + * Tells whether a node is cordoned already. If that can't be found out, it is taken as cordoned, so that a + * rollback doesn't uncordon a node that someone else may have cordoned. + */ + protected boolean isNodeCordoned(final UserVm vm) { + try { + Pair result = executeSshCommand(publicIpAddress, sshPort, + String.format("sudo /opt/bin/kubectl get node %s -o jsonpath='{.spec.unschedulable}'", getNodeName(vm)), 10000, 10000, 30000); + if (!Boolean.TRUE.equals(result.first()) || result.second() == null) { + logger.warn("Unable to find out whether node: {} on VM: {} in Kubernetes cluster: {} is cordoned, taking it as cordoned: {}", + getNodeName(vm), vm, kubernetesCluster, result.second()); + return true; + } + // The output also holds anything written to stderr, such as a warning from sudo + return result.second().lines().anyMatch(line -> "true".equals(line.trim())); + } catch (Exception e) { + logger.warn("Unable to find out whether node: {} on VM: {} in Kubernetes cluster: {} is cordoned, taking it as cordoned", getNodeName(vm), vm, kubernetesCluster, e); + return true; + } + } + + /** Uncordons nodes, only logging a warning for the ones that can't be uncordoned. */ + protected void uncordonNodes(final List vms) { + for (UserVm vm : vms) { + if (!uncordonKubernetesClusterNode(publicIpAddress, sshPort, vm)) { + logger.warn("Unable to uncordon node: {} on VM: {} in Kubernetes cluster: {}", getNodeName(vm), vm, kubernetesCluster); + } + } + } + + /** Uncordons a node, trying for up to a minute. */ + protected boolean uncordonKubernetesClusterNode(final String ipAddress, final int port, final UserVm vm) { + return KubernetesClusterUtil.uncordonKubernetesClusterNode(kubernetesCluster, ipAddress, port, getControlNodeLoginUser(), + getManagementServerSshPublicKeyFile(), vm, System.currentTimeMillis() + 60000, 10000); + } + + /** + * Returns the address and port to reach a node over SSH. On a network with port forwarding, the port is the + * one the node's port-forwarding rule actually uses. The rules are not always in the order of the cluster's VMs: + * external nodes and nodes left out of an operation, such as manual-upgrade nodes, shift the order. When the node + * has no such rule, the port is derived from its position in the list of nodes, as before. + */ + protected Pair getNodeSshAddress(final UserVm vm, final int index) { + if (sshPort == DEFAULT_SSH_PORT) { + return new Pair<>(index > 0 ? vm.getPrivateIpAddress() : publicIpAddress, DEFAULT_SSH_PORT); + } + List rules = portForwardingRulesDao.listByVm(vm.getId()); + if (rules != null) { + Optional rule = rules.stream() + .filter(r -> !FirewallRule.State.Revoke.equals(r.getState())) + .filter(r -> r.getDestinationPortStart() == DEFAULT_SSH_PORT && r.getSourcePortStart() >= CLUSTER_NODES_DEFAULT_START_SSH_PORT) + .filter(r -> isRuleOnAddress(r, publicIpAddress)) + .findFirst(); + if (rule.isPresent()) { + return new Pair<>(publicIpAddress, rule.get().getSourcePortStart()); + } + } + return new Pair<>(publicIpAddress, sshPort + index); + } + + protected boolean isRuleOnAddress(final PortForwardingRuleVO rule, final String address) { + IPAddressVO ip = ipAddressDao.findById(rule.getSourceIpAddressId()); + return ip != null && ip.getAddress() != null && ip.getAddress().addr().equals(address); + } + protected boolean stateTransitTo(long kubernetesClusterId, KubernetesCluster.Event e) { KubernetesClusterVO kubernetesCluster = kubernetesClusterDao.findById(kubernetesClusterId); try { - return _stateMachine.transitTo(kubernetesCluster, e, null, kubernetesClusterDao); + boolean transited = _stateMachine.transitTo(kubernetesCluster, e, null, kubernetesClusterDao); + if (transited && (KubernetesCluster.State.Running.equals(kubernetesCluster.getState()) || + KubernetesCluster.State.Stopped.equals(kubernetesCluster.getState()))) { + // A cluster that is Running or Stopped again is no longer in Alert because of a failed upgrade + kubernetesClusterDetailsDao.removeDetail(kubernetesClusterId, KubernetesClusterUpgradeWorker.UPGRADE_VERSION_DETAIL); + } + return transited; } catch (NoTransitionException nte) { logger.warn(String.format("Failed to transition state of the Kubernetes cluster : %s in state %s on event %s", kubernetesCluster.getName(), kubernetesCluster.getState().toString(), e.toString()), nte); 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..1e1275054ccd 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 @@ -17,7 +17,6 @@ package com.cloud.kubernetes.cluster.actionworkers; -import java.io.File; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -40,6 +39,8 @@ import org.apache.commons.lang3.StringUtils; import com.cloud.dc.DataCenter; +import com.cloud.event.ActionEventUtils; +import com.cloud.event.EventVO; import com.cloud.exception.InsufficientCapacityException; import com.cloud.exception.ManagementServerException; import com.cloud.exception.NetworkRuleConflictException; @@ -47,6 +48,7 @@ import com.cloud.exception.VirtualMachineMigrationException; import com.cloud.hypervisor.Hypervisor; import com.cloud.kubernetes.cluster.KubernetesCluster; +import com.cloud.kubernetes.cluster.KubernetesClusterEventTypes; import com.cloud.kubernetes.cluster.KubernetesClusterManagerImpl; import com.cloud.kubernetes.cluster.KubernetesClusterService; import com.cloud.kubernetes.cluster.KubernetesClusterVO; @@ -55,12 +57,12 @@ import com.cloud.network.IpAddress; import com.cloud.network.Network; import com.cloud.network.rules.FirewallRule; +import com.cloud.network.rules.PortForwardingRuleVO; import com.cloud.offering.ServiceOffering; import com.cloud.storage.LaunchPermissionVO; import com.cloud.uservm.UserVm; import com.cloud.utils.Pair; import com.cloud.utils.exception.CloudRuntimeException; -import com.cloud.utils.ssh.SshHelper; import com.cloud.vm.UserVmVO; import com.cloud.vm.VMInstanceVO; import com.cloud.vm.VirtualMachine; @@ -85,7 +87,9 @@ public class KubernetesClusterScaleWorker extends KubernetesClusterResourceModif private Long minSize; private Long maxSize; private Boolean isAutoscalingEnabled; - private long scaleTimeoutTime; + protected long scaleTimeoutTime; + /** How many times to try to drain a node that is being removed, as scale-down did before. */ + private static final int SCALE_DOWN_DRAIN_ATTEMPTS = 3; protected KubernetesClusterScaleWorker(final KubernetesCluster kubernetesCluster, final KubernetesClusterManagerImpl clusterManager) { super(kubernetesCluster, clusterManager); @@ -287,46 +291,89 @@ protected Pair calculateNodesCapacity(ServiceOffering offering, long return new Pair<>(offering.getCpu() * nodeCount, offering.getRamSize() * nodeCount); } + /** + * Deletes the Kubernetes node object of a VM that is being removed from the cluster. The node has been drained + * already, see {@link #drainNodesToRemove(List)}. A failed attempt is retried after waiting. + */ private boolean removeKubernetesClusterNode(final String ipAddress, final int port, final UserVm userVm, final int retries, final int waitDuration) { - File pkFile = getManagementServerSshPublicKeyFile(); - int retryCounter = 0; - String hostName = userVm.getHostName(); - if (StringUtils.isNotEmpty(hostName)) { - hostName = hostName.toLowerCase(); - } - while (retryCounter < retries) { - retryCounter++; + final String hostName = getNodeName(userVm); + for (int attempt = 1; attempt <= retries; attempt++) { + if (attempt > 1 && !sleepBeforeRetry(waitDuration)) { + break; + } try { - Pair result = SshHelper.sshExecute(ipAddress, port, getControlNodeLoginUser(), - pkFile, null, String.format("sudo /opt/bin/kubectl drain %s --ignore-daemonsets --delete-emptydir-data", hostName), - 10000, 10000, 60000); - if (!result.first()) { - logger.warn("Draining node: {} on VM: {} in Kubernetes cluster: {} unsuccessful", hostName, userVm, kubernetesCluster); - } else { - result = SshHelper.sshExecute(ipAddress, port, getControlNodeLoginUser(), - pkFile, null, String.format("sudo /opt/bin/kubectl delete node %s", hostName), - 10000, 10000, 30000); - if (result.first()) { - return true; - } else { - logger.warn("Deleting node: {} on VM: {} in Kubernetes cluster: {} unsuccessful", hostName, userVm, kubernetesCluster); - } + Pair result = executeSshCommand(ipAddress, port, + String.format("sudo /opt/bin/kubectl delete node %s --ignore-not-found", hostName), 10000, 10000, 30000); + if (result.first()) { + return true; } - break; + logger.warn("Deleting node: {} on VM: {} in Kubernetes cluster: {} unsuccessful: {}", hostName, userVm, kubernetesCluster, result.second()); } catch (Exception e) { - String msg = String.format("Failed to remove Kubernetes cluster: %s node: %s on VM: %s", kubernetesCluster, hostName, userVm); - logger.warn(msg, e); + logger.warn("Failed to remove Kubernetes cluster: {} node: {} on VM: {}", kubernetesCluster, hostName, userVm, e); } + } + return false; + } + + /** + * Drains the nodes that are being removed, before any of them is removed. All of them are cordoned first, so + * that the pods evicted from one are not moved to another one that is going away. If a node can't be drained, + * all of them are uncordoned and the scale-down stops with nothing removed: the nodes are still working members + * of the cluster, so the cluster goes back to the state it was in, rather than to Alert. + * + * @return the nodes that were cordoned here, as opposed to the ones that were cordoned already + */ + protected List drainNodesToRemove(final List vms) throws CloudRuntimeException { + List cordoned = new ArrayList<>(); + String failure = null; + for (UserVmVO vm : vms) { + // A node cordoned by someone else has to stay cordoned if the scale-down is rolled back + final boolean cordonedAlready = isNodeCordoned(vm); + String output; try { - Thread.sleep(waitDuration); - } catch (InterruptedException ie) { - logger.error("Error while waiting for Kubernetes cluster: {} node: {} on VM: {} removal", kubernetesCluster, hostName, userVm, ie); + Pair result = executeSshCommand(publicIpAddress, sshPort, + String.format("sudo /opt/bin/kubectl cordon %s", getNodeName(vm)), 10000, 10000, 30000); + if (result.first()) { + if (!cordonedAlready) { + cordoned.add(vm); + } + continue; + } + output = result.second(); + } catch (Exception e) { + output = e.getMessage(); + } + failure = String.format("unable to cordon Kubernetes node: %s on VM: %s: %s", getNodeName(vm), vm, output); + break; + } + if (failure == null) { + for (UserVmVO vm : vms) { + if (System.currentTimeMillis() > scaleTimeoutTime) { + failure = String.format("scaling action timed out before Kubernetes node: %s on VM: %s was drained", getNodeName(vm), vm); + break; + } + Pair drained = drainKubernetesClusterNode(publicIpAddress, sshPort, vm, SCALE_DOWN_DRAIN_ATTEMPTS, DRAIN_RETRY_WAIT, scaleTimeoutTime); + if (!drained.first()) { + failure = String.format("unable to drain Kubernetes node: %s on VM: %s: %s", getNodeName(vm), vm, drained.second()); + break; + } } - retryCounter++; } - return false; + if (failure == null) { + return cordoned; + } + uncordonNodes(cordoned); + String message = String.format("Scaling failed for Kubernetes cluster %s, %s. No node was removed", kubernetesCluster, failure); + logger.error(message); + // Nothing has changed, so the cluster goes back to the state it was in: Scaling goes back to Running (and + // ScalingStoppedCluster to Stopped) on OperationSucceeded. Scaling a cluster in Alert is not allowed today, + // but if it is, such a cluster has to go back to Alert, not to Running. + stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.State.Alert.equals(originalState) ? + KubernetesCluster.Event.OperationFailed : KubernetesCluster.Event.OperationSucceeded); + throw new CloudRuntimeException(message); } + private void validateKubernetesClusterScaleOfferingParameters() throws CloudRuntimeException { if (KubernetesCluster.State.Created.equals(originalState)) { return; @@ -411,48 +458,225 @@ private void scaleKubernetesClusterOffering(KubernetesClusterNodeType nodeType, kubernetesCluster = updateKubernetesClusterEntryForNodeType(null, nodeType, serviceOffering, updateNodeOffering, updateClusterOffering); } - private void removeNodesFromCluster(List vmMaps) throws CloudRuntimeException { - for (KubernetesClusterVmMapVO vmMapVO : vmMaps) { - UserVmVO userVM = userVmDao.findById(vmMapVO.getVmId()); - logger.info("Removing vm {} from cluster {}", userVM, kubernetesCluster); - if (!removeKubernetesClusterNode(publicIpAddress, sshPort, userVM, 3, 30000)) { - logTransitStateAndThrow(Level.ERROR, String.format("Scaling failed for Kubernetes" + - " cluster %s, failed to remove Kubernetes node: %s running on VM : %s", - kubernetesCluster, userVM.getHostName(), userVM), kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); + protected void removeNodesFromCluster(List vmMaps) throws CloudRuntimeException { + List vms = vmMaps.stream().map(vmMapVO -> userVmDao.findById(vmMapVO.getVmId())).collect(Collectors.toList()); + List cordoned = drainNodesToRemove(vms); + // Index of the first VM whose Kubernetes node is still in the cluster + int firstInCluster = 0; + // VMs taken out of the cluster's VM map so far + List removed = new ArrayList<>(); + boolean completed = false; + try { + for (int i = 0; i < vmMaps.size(); i++) { + // No timeout check here: the nodes are drained, and stopping part way would leave the cluster + // with some of them removed and the node count not updated + UserVmVO userVM = vms.get(i); + KubernetesClusterVmMapVO vmMapVO = vmMaps.get(i); + logger.info("Removing vm {} from cluster {}", userVM, kubernetesCluster); + if (!removeKubernetesClusterNode(publicIpAddress, sshPort, userVM, 3, 30000)) { + logTransitStateAndThrow(Level.ERROR, String.format("Scaling failed for Kubernetes" + + " cluster %s, failed to remove Kubernetes node: %s running on VM : %s", + kubernetesCluster, userVM.getHostName(), userVM), kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); + } + firstInCluster = i + 1; + try { + destroyRemovedNodeVm(userVM); + } finally { + // The node has left the Kubernetes cluster, so its VM leaves the cluster's VM map, whatever + // happened to the VM. A map entry kept for it would never match a Kubernetes node again: the + // state scanner would hold the cluster in Alert, which blocks the next scale, and the node count + // and the next scale-down's choice of nodes would be worked out from a VM that is gone. The + // entry is deleted only now as the VM's expunge looks it up (to leave alone the port-forwarding + // rules of a CKS node in an NSX zone). + kubernetesClusterVmMapDao.expunge(vmMapVO.getId()); + removed.add(userVM); + } } - CallContext vmContext = CallContext.register(CallContext.current(), - ApiCommandResourceType.VirtualMachine); - vmContext.setEventResourceId(userVM.getId()); + + // Scale network rules to update firewall rule try { - UserVm vm = userVmService.destroyVm(userVM.getId(), true); - if (!userVmManager.expunge(userVM)) { - logTransitStateAndThrow(Level.ERROR, String.format("Scaling Kubernetes cluster %s failed, unable to expunge VM '%s'." - , kubernetesCluster, vm), kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); + List clusterVMIds = getKubernetesClusterVMMaps() + .stream() + .filter(x -> !x.isEtcdNode()) + .map(KubernetesClusterVmMapVO::getVmId).collect(Collectors.toList()); + scaleKubernetesClusterNetworkRules(clusterVMIds); + } catch (ManagementServerException e) { + logTransitStateAndThrow(Level.ERROR, String.format("Scaling failed for Kubernetes " + + "cluster %s, unable to update network rules", kubernetesCluster), + kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed, e); + } + completed = true; + } finally { + if (!completed) { + // The nodes that are still in the cluster were drained above and would stay unschedulable + updateKubernetesClusterEntryForRemovedNodes(removed); + List inCluster = vms.subList(firstInCluster, vms.size()); + uncordonNodes(cordoned.stream().filter(inCluster::contains).collect(Collectors.toList())); + } + } + } + + /** + * Destroys and expunges the VM of a node that has been deleted from the Kubernetes cluster. The caller then takes + * the VM out of the cluster's VM map, whether this succeeds or not. + *
    + *
  • If the VM can't be destroyed, the scale fails: the VM may still be running, though it is no longer a + * Kubernetes node nor a member of the cluster, and has to be destroyed by hand.
    • + *
  • If the destroyed VM can't be expunged, the scale goes on, with a warning event that names the VM and its + * state. The expunge task retries VMs left Destroyed, Expunging or in Error.
    • + *
+ */ + protected void destroyRemovedNodeVm(final UserVmVO userVM) throws CloudRuntimeException { + CallContext vmContext = CallContext.register(CallContext.current(), + ApiCommandResourceType.VirtualMachine); + vmContext.setEventResourceId(userVM.getId()); + // Whether the VM is left behind, destroyed or not + boolean leftBehind = false; + try { + try { + userVmService.destroyVm(userVM.getId(), true); + } catch (ResourceUnavailableException | CloudRuntimeException e) { + // destroyVm can fail after the VM has been destroyed, on the resource count updates that follow + VMInstanceVO vm = null; + boolean found = false; + try { + vm = vmInstanceDao.findByIdIncludingRemoved(userVM.getId()); + found = true; + } catch (RuntimeException lookupError) { + logger.warn("Unable to look up VM {} after it failed to be destroyed", userVM, lookupError); + } + if (found && (vm == null || vm.getRemoved() != null)) { + logger.warn("Destroying VM {} of the node removed from Kubernetes cluster {} reported an error, but the VM is expunged", + userVM, kubernetesCluster, e); + return; + } + if (found && List.of(VirtualMachine.State.Destroyed, VirtualMachine.State.Expunging).contains(vm.getState())) { + logger.warn("Destroying VM {} of the node removed from Kubernetes cluster {} reported an error, but the VM is {}", + userVM, kubernetesCluster, vm.getState(), e); + } else { + leftBehind = true; + logTransitStateAndThrow(Level.ERROR, String.format("Scaling Kubernetes cluster %s failed, unable to destroy VM %s%s." + + " Its node was removed from the Kubernetes cluster and the VM is no longer part of the cluster;" + + " it may still be running and has to be destroyed manually", kubernetesCluster, userVM, + vm == null ? "" : String.format(" (left in state %s)", vm.getState())), + kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed, e); } - } catch (ResourceUnavailableException e) { - logTransitStateAndThrow(Level.ERROR, String.format("Scaling Kubernetes cluster %s failed, unable to remove VM ID: %s", - kubernetesCluster, userVM), kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed, e); - } finally { - CallContext.unregister(); } - kubernetesClusterVmMapDao.expunge(vmMapVO.getId()); - if (System.currentTimeMillis() > scaleTimeoutTime) { - logTransitStateAndThrow(Level.WARN, String.format("Scaling Kubernetes cluster %s failed, scaling action timed out", - kubernetesCluster), kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); + boolean expunged = false; + Exception expungeError = null; + try { + expunged = userVmManager.expunge(userVM); + } catch (RuntimeException e) { + expungeError = e; + } + if (!expunged) { + leftBehind = warnVmNotExpunged(userVM, expungeError); + } + } finally { + CallContext.unregister(); + if (leftBehind) { + revokeRemovedNodeVmSshRules(userVM); } } + } - // Scale network rules to update firewall rule + /** @return false if the VM turns out to be expunged, true if it is left behind. Doesn't throw. */ + private boolean warnVmNotExpunged(final UserVmVO userVM, final Exception expungeError) { + VMInstanceVO vm; try { - List clusterVMIds = getKubernetesClusterVMMaps() - .stream() - .filter(x -> !x.isEtcdNode()) - .map(KubernetesClusterVmMapVO::getVmId).collect(Collectors.toList()); - scaleKubernetesClusterNetworkRules(clusterVMIds); - } catch (ManagementServerException e) { - logTransitStateAndThrow(Level.ERROR, String.format("Scaling failed for Kubernetes " + - "cluster %s, unable to update network rules", kubernetesCluster), - kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed, e); + vm = vmInstanceDao.findByIdIncludingRemoved(userVM.getId()); + } catch (RuntimeException e) { + logger.warn("Unable to expunge VM {} of the node removed from Kubernetes cluster {}; it is no longer part of the cluster", + userVM, kubernetesCluster, expungeError); + return true; + } + if (vm == null || vm.getRemoved() != null) { + // Expunged meanwhile, by the expunge task for one + logger.info("VM {} of the node removed from Kubernetes cluster {} has been expunged", userVM, kubernetesCluster); + return false; + } + String message = String.format("Unable to expunge VM %s of the node removed from Kubernetes cluster %s. The VM is" + + " no longer part of the cluster and is left in state %s, for the expunge task to retry. Until it is expunged" + + " it keeps the cluster's network in use, and the cluster can't be deleted with its network", + userVM, kubernetesCluster, vm.getState()); + logger.warn(message, expungeError); + try { + // Recorded for the VM's owner, who has to see why the VM still holds the network + ActionEventUtils.onCompletedActionEvent(CallContext.current().getCallingUserId(), userVM.getAccountId(), + EventVO.LEVEL_WARN, KubernetesClusterEventTypes.EVENT_KUBERNETES_CLUSTER_SCALE, + message, userVM.getId(), ApiCommandResourceType.VirtualMachine.toString(), 0); + } catch (RuntimeException e) { + logger.warn("Unable to record an event for VM {} that was not expunged", userVM, e); + } + return true; + } + + /** + * Revokes the SSH port-forwarding rule that the cluster has on a removed node's VM that is left behind. Expunging + * the VM would revoke it; the cluster's network rule updates may not: a VPC tier's only clears the ports of the + * nodes that remain, where the rule would conflict with the rule of a node added later, and none runs if the + * scale-down stops. Left alone are: a rule provided by NSX (the VM's expunge leaves the port-forwarding rules of a + * CKS node alone in an NSX zone); the rule on the port this scale reaches the cluster through, if it is the removed + * VM's, as the nodes that remain are still to be reached through it; and the VM's other rules. + */ + protected void revokeRemovedNodeVmSshRules(final UserVmVO userVM) { + boolean registered = false; + try { + if (network == null || publicIpAddress == null || manager.isDirectAccess(network) || + networkModel.isProviderSupportServiceInNetwork(network.getId(), Network.Service.PortForwarding, Network.Provider.Nsx)) { + return; + } + // A context of its own rather than the VM's or the scale's + CallContext.register(CallContext.current(), null); + registered = true; + for (PortForwardingRuleVO pfRule : portForwardingRulesDao.listByVm(userVM.getId())) { + if (!FirewallRule.State.Revoke.equals(pfRule.getState()) && pfRule.getNetworkId() == network.getId() && + pfRule.getDestinationPortStart() == DEFAULT_SSH_PORT && pfRule.getSourcePortStart() >= CLUSTER_NODES_DEFAULT_START_SSH_PORT && + pfRule.getSourcePortStart() != sshPort && isRuleOnAddress(pfRule, publicIpAddress)) { + try { + rulesService.revokePortForwardingRule(pfRule.getId(), true); + } catch (Exception e) { + logger.warn("Unable to revoke the SSH port-forwarding rule {} of VM {} of the node removed from Kubernetes cluster {}", + pfRule, userVM, kubernetesCluster, e); + } + } + } + } catch (Exception e) { + logger.warn("Unable to revoke the SSH port-forwarding rule of VM {} of the node removed from Kubernetes cluster {}", + userVM, kubernetesCluster, e); + } finally { + if (registered) { + CallContext.unregister(); + } + } + } + + /** + * Takes the VMs that have left the cluster's VM map off the cluster's node count, cores and memory, when a + * scale-down stops part way, so that the cluster matches its VM map. The scale-down updates them itself once every + * node has been removed, and this takes the same view of the node count: every VM removed, including a control + * node removed by ID, comes off the node count, so that the cluster's total node count matches its VM map. Any + * failure here is logged, so that the scale-down's own error is the one reported. + */ + private void updateKubernetesClusterEntryForRemovedNodes(final List removed) { + if (removed.isEmpty()) { + return; + } + try { + long removedCores = 0; + long removedMemory = 0; + for (UserVmVO vm : removed) { + ServiceOfferingVO offering = serviceOfferingDao.findByIdIncludingRemoved(vm.getId(), vm.getServiceOfferingId()); + removedCores += offering.getCpu(); + removedMemory += offering.getRamSize(); + } + KubernetesClusterVO cluster = kubernetesClusterDao.findById(kubernetesCluster.getId()); + updateKubernetesClusterEntry(cluster.getCores() - removedCores, cluster.getMemory() - removedMemory, + cluster.getNodeCount() - removed.size(), null, cluster.getAutoscalingEnabled(), cluster.getMinSize(), + cluster.getMaxSize(), WORKER, false, false); + } catch (RuntimeException e) { + logger.error("Unable to take {} removed nodes off the node count, cores and memory of Kubernetes cluster {};" + + " they no longer match its VMs", removed.size(), kubernetesCluster, e); } } diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterStartWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterStartWorker.java index 4b7f3c672258..bd6441ac2a31 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterStartWorker.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterStartWorker.java @@ -978,8 +978,10 @@ public boolean reconcileAlertCluster() { if (!isKubernetesClusterDashboardServiceRunning(false, startTimeoutTime)) { return false; } - // mark the cluster to be running - stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.RecoveryRequested); + // mark the cluster to be running, unless it has left Alert meanwhile, for example to retry an upgrade + if (!stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.RecoveryRequested)) { + return false; + } stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.OperationSucceeded); return true; } diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterUpgradeWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterUpgradeWorker.java index 4c2725fc2a2f..ca4cf3b73049 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterUpgradeWorker.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterUpgradeWorker.java @@ -19,6 +19,7 @@ import java.io.File; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.Objects; import java.util.stream.Collectors; @@ -43,6 +44,11 @@ public class KubernetesClusterUpgradeWorker extends KubernetesClusterActionWorker { + /** Cluster detail holding the id of the Kubernetes version that an upgrade, not finished yet, is upgrading to. */ + public static final String UPGRADE_VERSION_DETAIL = "upgradeKubernetesVersionId"; + /** Prints the STATUS of a node, such as Ready, NotReady or Ready,SchedulingDisabled. */ + protected static final String NODE_STATUS_COMMAND = "sudo /opt/bin/kubectl get nodes | awk '{if ($1 == \"%s\") print $2}'"; + protected List clusterVMs = new ArrayList<>(); private KubernetesSupportedVersion upgradeVersion; private final String upgradeScriptFilename = "upgrade-kubernetes.sh"; @@ -64,8 +70,9 @@ protected void retrieveScriptFiles() { } private Pair runInstallScriptOnVM(final UserVm vm, final int index) throws Exception { - int nodeSshPort = sshPort == 22 ? sshPort : sshPort + index; - String nodeAddress = (index > 0 && sshPort == 22) ? vm.getPrivateIpAddress() : publicIpAddress; + Pair nodeSshAddress = getNodeSshAddress(vm, index); + String nodeAddress = nodeSshAddress.first(); + int nodeSshPort = nodeSshAddress.second(); SshHelper.scpTo(nodeAddress, nodeSshPort, getControlNodeLoginUser(), sshKeyFile, null, "~/", upgradeScriptFile.getAbsolutePath(), "0755"); String cmdStr = String.format("sudo ./%s %s %s %s %s %s", @@ -82,40 +89,35 @@ private Pair runInstallScriptOnVM(final UserVm vm, final int in private void upgradeKubernetesClusterNodes() { for (int i = 0; i < clusterVMs.size(); ++i) { UserVm vm = clusterVMs.get(i); - String hostName = vm.getHostName(); - if (StringUtils.isNotEmpty(hostName)) { - hostName = hostName.toLowerCase(); - } + String hostName = getNodeName(vm); Pair result; + if (isNodeUpgraded(vm, hostName)) { + // A previous, failed, upgrade to this version got this far. On the first control node this also + // means that kubeadm upgrade apply has run, so the other nodes only need kubeadm upgrade node. + logger.info("Skipping node {} on VM {} in Kubernetes cluster {}, it is already at Kubernetes version {}", hostName, vm, kubernetesCluster, upgradeVersion); + continue; + } if (logger.isInfoEnabled()) { - logger.info("Upgrading node on VM {} in Kubernetes cluster {} with Kubernetes version {}", vm, kubernetesCluster, upgradeVersion); + logger.info("Upgrading node {} on VM {} in Kubernetes cluster {} with Kubernetes version {}", hostName, vm, kubernetesCluster, upgradeVersion); } - String errorMessage = String.format("Failed to upgrade Kubernetes cluster : %s, unable to drain Kubernetes node on VM : %s", kubernetesCluster.getName(), vm.getDisplayName()); - for (int retry = KubernetesClusterService.KubernetesClusterUpgradeRetries.value(); retry >= 0; retry--) { - try { - result = SshHelper.sshExecute(publicIpAddress, sshPort, getControlNodeLoginUser(), sshKeyFile, null, - String.format("sudo /opt/bin/kubectl drain %s --ignore-daemonsets --delete-emptydir-data", hostName), - 10000, 10000, 60000); - if (result.first()) { - break; - } - if (retry > 0) { - logger.error(String.format("%s, retries left: %s", errorMessage, retry)); - } else { - logTransitStateDetachIsoAndThrow(Level.ERROR, errorMessage, kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, null); - } - } catch (Exception e) { - if (retry > 0) { - logger.error(String.format("%s due to %s, retries left: %s", errorMessage, e, retry)); - } else { - logTransitStateDetachIsoAndThrow(Level.ERROR, errorMessage, kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, e); - } - } + if (System.currentTimeMillis() > upgradeTimeoutTime) { + logTransitStateDetachIsoAndThrow(Level.ERROR, String.format("Failed to upgrade Kubernetes cluster : %s, upgrade action timed out on Kubernetes node : %s on VM : %s", kubernetesCluster.getName(), hostName, vm.getDisplayName()), kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, null); + } + // A node cordoned by someone else stays cordoned if the upgrade stops before upgrading it + final List toUncordon = isNodeCordoned(vm) ? List.of() : List.of(vm); + Pair drained = drainKubernetesClusterNode(publicIpAddress, sshPort, vm, + KubernetesClusterService.KubernetesClusterUpgradeRetries.value() + 1, DRAIN_RETRY_WAIT, upgradeTimeoutTime); + if (!drained.first()) { + uncordonNodes(toUncordon); + logTransitStateDetachIsoAndThrow(Level.ERROR, String.format("Failed to upgrade Kubernetes cluster : %s, unable to drain Kubernetes node : %s on VM : %s: %s", + kubernetesCluster.getName(), hostName, vm.getDisplayName(), drained.second()), kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, null); } if (System.currentTimeMillis() > upgradeTimeoutTime) { - logTransitStateDetachIsoAndThrow(Level.ERROR, String.format("Failed to upgrade Kubernetes cluster : %s, upgrade action timed out", kubernetesCluster.getName()), kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, null); + // Nothing has been changed on the node, so don't leave it unschedulable + uncordonNodes(toUncordon); + logTransitStateDetachIsoAndThrow(Level.ERROR, String.format("Failed to upgrade Kubernetes cluster : %s, upgrade action timed out on Kubernetes node : %s on VM : %s", kubernetesCluster.getName(), hostName, vm.getDisplayName()), kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, null); } - errorMessage = String.format("Failed to upgrade Kubernetes cluster : %s, unable to upgrade Kubernetes node on VM : %s", kubernetesCluster.getName(), vm.getDisplayName()); + String errorMessage = String.format("Failed to upgrade Kubernetes cluster : %s, unable to upgrade Kubernetes node : %s on VM : %s", kubernetesCluster.getName(), hostName, vm.getDisplayName()); for (int retry = KubernetesClusterService.KubernetesClusterUpgradeRetries.value(); retry >= 0; retry--) { try { deployProvider(); @@ -137,18 +139,19 @@ private void upgradeKubernetesClusterNodes() { } } if (System.currentTimeMillis() > upgradeTimeoutTime) { - logTransitStateDetachIsoAndThrow(Level.ERROR, String.format("Failed to upgrade Kubernetes cluster : %s, upgrade action timed out", kubernetesCluster.getName()), kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, null); + // The node has been upgraded, so don't leave it unschedulable, unless someone else cordoned it + uncordonNodes(toUncordon); + logTransitStateDetachIsoAndThrow(Level.ERROR, String.format("Failed to upgrade Kubernetes cluster : %s, upgrade action timed out on Kubernetes node : %s on VM : %s", kubernetesCluster.getName(), hostName, vm.getDisplayName()), kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, null); } if (!KubernetesClusterUtil.uncordonKubernetesClusterNode(kubernetesCluster, publicIpAddress, sshPort, getControlNodeLoginUser(), getManagementServerSshPublicKeyFile(), vm, upgradeTimeoutTime, 15000)) { - logTransitStateDetachIsoAndThrow(Level.ERROR, String.format("Failed to upgrade Kubernetes cluster : %s, unable to uncordon Kubernetes node on VM : %s", kubernetesCluster.getName(), vm.getDisplayName()), kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, null); + logTransitStateDetachIsoAndThrow(Level.ERROR, String.format("Failed to upgrade Kubernetes cluster : %s, unable to uncordon Kubernetes node : %s on VM : %s", kubernetesCluster.getName(), hostName, vm.getDisplayName()), kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, null); } - if (i == 0) { // Wait for control node to get in Ready state - if (!KubernetesClusterUtil.isKubernetesClusterNodeReady(kubernetesCluster, publicIpAddress, sshPort, getControlNodeLoginUser(), getManagementServerSshPublicKeyFile(), hostName, upgradeTimeoutTime, 15000)) { - logTransitStateDetachIsoAndThrow(Level.ERROR, String.format("Failed to upgrade Kubernetes cluster : %s, unable to get control Kubernetes node on VM : %s in ready state", kubernetesCluster.getName(), vm.getDisplayName()), kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, null); - } + // Wait for every node, not only the first control node, to be Ready before the next one is drained + if (!KubernetesClusterUtil.isKubernetesClusterNodeReady(kubernetesCluster, publicIpAddress, sshPort, getControlNodeLoginUser(), getManagementServerSshPublicKeyFile(), hostName, upgradeTimeoutTime, 15000)) { + logTransitStateDetachIsoAndThrow(Level.ERROR, String.format("Failed to upgrade Kubernetes cluster : %s, unable to get Kubernetes node : %s on VM : %s in ready state", kubernetesCluster.getName(), hostName, vm.getDisplayName()), kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, null); } if (!KubernetesClusterUtil.clusterNodeVersionMatches(upgradeVersion.getSemanticVersion(), publicIpAddress, sshPort, getControlNodeLoginUser(), getManagementServerSshPublicKeyFile(), hostName, upgradeTimeoutTime, 15000, vm.getId(), kubernetesClusterVmMapDao)) { - logTransitStateDetachIsoAndThrow(Level.ERROR, String.format("Failed to upgrade Kubernetes cluster : %s, unable to get Kubernetes node on VM : %s upgraded to version %s", kubernetesCluster.getName(), vm.getDisplayName(), upgradeVersion.getSemanticVersion()), kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, null); + logTransitStateDetachIsoAndThrow(Level.ERROR, String.format("Failed to upgrade Kubernetes cluster : %s, unable to get Kubernetes node : %s on VM : %s upgraded to version %s", kubernetesCluster.getName(), hostName, vm.getDisplayName(), upgradeVersion.getSemanticVersion()), kubernetesCluster, clusterVMs, KubernetesCluster.Event.OperationFailed, null); } if (logger.isInfoEnabled()) { logger.info("Successfully upgraded node on VM {} in Kubernetes cluster {} with Kubernetes version {}", vm, kubernetesCluster, upgradeVersion); @@ -156,6 +159,42 @@ private void upgradeKubernetesClusterNodes() { } } + /** + * Tells whether a node was upgraded to the target version by an earlier upgrade of the cluster that failed on a + * later node, so that a retry can carry on from where that one stopped. The version recorded for the node is + * only updated once the node has been upgraded, uncordoned and found Ready, but it is not trusted on its own: + * the node must also report the target version to Kubernetes, and still be Ready and schedulable. + */ + protected boolean isNodeUpgraded(final UserVm vm, final String hostName) { + KubernetesClusterVmMapVO vmMapVO = kubernetesClusterVmMapDao.getClusterMapFromVmId(vm.getId()); + String version = upgradeVersion.getSemanticVersion(); + if (vmMapVO == null || !KubernetesClusterUtil.isSameNodeVersion(vmMapVO.getNodeVersion(), version)) { + return false; + } + try { + Pair result = executeSshCommand(publicIpAddress, sshPort, + String.format(KubernetesClusterUtil.CLUSTER_NODE_VERSION_COMMAND, hostName), 10000, 10000, 20000); + if (!Boolean.TRUE.equals(KubernetesClusterUtil.clusterNodeVersionMatches(result, version).first())) { + logger.info("Node {} on VM {} in Kubernetes cluster {} is recorded at Kubernetes version {}, but reports: {}", + hostName, vm, kubernetesCluster, version, result.second()); + return false; + } + // Only a node that is also Ready is done; any other one is upgraded again. A node that is Ready but + // cordoned (Ready,SchedulingDisabled) was cordoned by someone else after its upgrade, and is left as it is. + result = executeSshCommand(publicIpAddress, sshPort, + String.format(NODE_STATUS_COMMAND, hostName), 10000, 10000, 20000); + if (Boolean.TRUE.equals(result.first()) && result.second() != null && + result.second().lines().anyMatch(line -> Arrays.asList(line.trim().split(",")).contains("Ready"))) { + return true; + } + logger.info("Node {} on VM {} in Kubernetes cluster {} is at Kubernetes version {}, but is not Ready, so it is upgraded again", + hostName, vm, kubernetesCluster, version); + } catch (Exception e) { + logger.warn("Unable to get the Kubernetes version of node {} on VM {} in Kubernetes cluster {}", hostName, vm, kubernetesCluster, e); + } + return false; + } + public boolean upgradeCluster() throws CloudRuntimeException { init(); if (logger.isInfoEnabled()) { @@ -174,7 +213,13 @@ public boolean upgradeCluster() throws CloudRuntimeException { } filterOutManualUpgradeNodesFromClusterUpgrade(); retrieveScriptFiles(); - stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.UpgradeRequested); + if (!stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.UpgradeRequested)) { + // For example, the Alert state scanner has just started to recover the cluster + logAndThrow(Level.ERROR, String.format("Upgrade failed for Kubernetes cluster: %s, its state changed, please retry", kubernetesCluster)); + } + // Kept until the cluster is Running again (see stateTransitTo), so that a failed upgrade can be told apart + // from other causes of Alert + kubernetesClusterDetailsDao.addDetail(kubernetesCluster.getId(), UPGRADE_VERSION_DETAIL, String.valueOf(upgradeVersion.getId()), false); attachIsoKubernetesVMs(clusterVMs, upgradeVersion); upgradeKubernetesClusterNodes(); detachIsoKubernetesVMs(clusterVMs); diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtil.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtil.java index 00625f6e076d..1bea4714b549 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtil.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtil.java @@ -368,11 +368,29 @@ public static boolean clusterNodeVersionMatches(final String version, return false; } - protected static Pair clusterNodeVersionMatches(final Pair result, final String version) { + /** + * Compares a node version, as recorded for a cluster VM or as printed by kubectl (with a leading "v"), with a + * semantic version. The SSH output also holds whatever the command wrote to stderr, such as a warning from sudo, + * so any line of it that is exactly the version, ignoring build metadata such as "+vendor.1", matches. + */ + public static boolean isSameNodeVersion(final String nodeVersion, final String version) { + if (StringUtils.isBlank(nodeVersion) || StringUtils.isBlank(version)) { + return false; + } + final String expected = withoutBuildMetadata(StringUtils.removeStart(version.trim(), "v")); + return nodeVersion.lines().anyMatch(line -> withoutBuildMetadata(StringUtils.removeStart(line.trim(), "v")).equals(expected)); + } + + /** Semantic versions that only differ in their build metadata, after a "+", are the same version. */ + private static String withoutBuildMetadata(final String version) { + return StringUtils.substringBefore(version, "+"); + } + + public static Pair clusterNodeVersionMatches(final Pair result, final String version) { if (result == null || Boolean.FALSE.equals(result.first()) || StringUtils.isBlank(result.second())) { return new Pair<>(false, null); } String response = result.second(); - return new Pair<>(response.contains(String.format("v%s", version)), response); + return new Pair<>(isSameNodeVersion(response, version), response); } } 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..30cd1c1f46a5 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 @@ -26,9 +26,13 @@ import com.cloud.exception.InvalidParameterValueException; import com.cloud.exception.PermissionDeniedException; import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterActionWorker; +import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterUpgradeWorker; import com.cloud.kubernetes.cluster.dao.KubernetesClusterDao; +import com.cloud.kubernetes.cluster.dao.KubernetesClusterDetailsDao; import com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDao; import com.cloud.kubernetes.version.KubernetesSupportedVersion; +import com.cloud.kubernetes.version.KubernetesSupportedVersionVO; +import com.cloud.kubernetes.version.dao.KubernetesSupportedVersionDao; import com.cloud.network.Network; import com.cloud.network.dao.FirewallRulesDao; import com.cloud.network.rules.FirewallRule; @@ -94,6 +98,12 @@ public class KubernetesClusterManagerImplTest { @Mock KubernetesClusterVmMapDao kubernetesClusterVmMapDao; + @Mock + KubernetesClusterDetailsDao kubernetesClusterDetailsDao; + + @Mock + KubernetesSupportedVersionDao kubernetesSupportedVersionDao; + @Mock VMInstanceDao vmInstanceDao; @@ -441,4 +451,101 @@ public void testGetCksClusterPreferredArchSameArch() { String cksClusterPreferredArch = kubernetesClusterManager.getCksClusterPreferredArch(systemVMArch, cksIso); Assert.assertEquals(CPU.CPUArch.amd64.getType(), cksClusterPreferredArch); } + + private KubernetesCluster mockClusterForUpgradeRetry(String... nodeVersions) { + KubernetesCluster cluster = Mockito.mock(KubernetesCluster.class); + Mockito.when(cluster.getId()).thenReturn(1L); + Mockito.when(cluster.getKubernetesVersionId()).thenReturn(10L); + List vmMaps = new ArrayList<>(); + for (String nodeVersion : nodeVersions) { + KubernetesClusterVmMapVO vmMap = Mockito.mock(KubernetesClusterVmMapVO.class); + Mockito.when(vmMap.getNodeVersion()).thenReturn(nodeVersion); + vmMaps.add(vmMap); + } + Mockito.when(kubernetesClusterVmMapDao.listByClusterId(1L)).thenReturn(vmMaps); + mockClusterVersion("1.28.9"); + return cluster; + } + + private void mockClusterVersion(String semanticVersion) { + KubernetesSupportedVersionVO clusterVersion = Mockito.mock(KubernetesSupportedVersionVO.class); + Mockito.when(clusterVersion.getSemanticVersion()).thenReturn(semanticVersion); + Mockito.when(kubernetesSupportedVersionDao.findById(10L)).thenReturn(clusterVersion); + } + + private KubernetesSupportedVersion mockUpgradeVersion(long id) { + KubernetesSupportedVersion version = Mockito.mock(KubernetesSupportedVersion.class); + Mockito.when(version.getId()).thenReturn(id); + Mockito.lenient().when(version.getSemanticVersion()).thenReturn("1.29.2"); + return version; + } + + @Test + public void testIsRetryOfFailedUpgradeWithNodeAtTargetVersion() { + KubernetesCluster cluster = mockClusterForUpgradeRetry("v1.29.2", "1.28.9", "1.28.9"); + Assert.assertTrue(kubernetesClusterManager.isRetryOfFailedUpgrade(cluster, mockUpgradeVersion(11L))); + } + + @Test + public void testIsRetryOfFailedUpgradeWithNoNodeAtTargetVersion() { + KubernetesCluster cluster = mockClusterForUpgradeRetry("1.28.9", "v1.28.9"); + Assert.assertFalse(kubernetesClusterManager.isRetryOfFailedUpgrade(cluster, mockUpgradeVersion(11L))); + } + + @Test + public void testIsRetryOfFailedUpgradeIgnoresManuallyUpgradedNodes() { + KubernetesCluster cluster = mockClusterForUpgradeRetry("1.28.9"); + KubernetesClusterVmMapVO manualNode = Mockito.mock(KubernetesClusterVmMapVO.class); + Mockito.when(manualNode.isManualUpgrade()).thenReturn(true); + Mockito.lenient().when(manualNode.getNodeVersion()).thenReturn("v1.29.2"); + List vmMaps = new ArrayList<>(kubernetesClusterVmMapDao.listByClusterId(1L)); + vmMaps.add(manualNode); + Mockito.when(kubernetesClusterVmMapDao.listByClusterId(1L)).thenReturn(vmMaps); + Assert.assertFalse(kubernetesClusterManager.isRetryOfFailedUpgrade(cluster, mockUpgradeVersion(11L))); + } + + @Test + public void testIsRetryOfFailedUpgradeToCurrentVersion() { + KubernetesCluster cluster = Mockito.mock(KubernetesCluster.class); + Mockito.when(cluster.getKubernetesVersionId()).thenReturn(10L); + KubernetesSupportedVersion version = Mockito.mock(KubernetesSupportedVersion.class); + Mockito.when(version.getId()).thenReturn(10L); + Assert.assertFalse(kubernetesClusterManager.isRetryOfFailedUpgrade(cluster, version)); + Mockito.verifyNoInteractions(kubernetesClusterVmMapDao); + } + + @Test + public void testIsRetryOfFailedUpgradeUsesRecordedUpgradeVersion() { + KubernetesCluster cluster = Mockito.mock(KubernetesCluster.class); + Mockito.when(cluster.getId()).thenReturn(1L); + Mockito.when(cluster.getKubernetesVersionId()).thenReturn(10L); + Mockito.when(kubernetesClusterDetailsDao.findDetail(1L, KubernetesClusterUpgradeWorker.UPGRADE_VERSION_DETAIL)) + .thenReturn(new KubernetesClusterDetailsVO(1L, KubernetesClusterUpgradeWorker.UPGRADE_VERSION_DETAIL, "11", false)); + mockClusterVersion("1.28.9"); + KubernetesSupportedVersion sameVersion = Mockito.mock(KubernetesSupportedVersion.class); + Mockito.when(sameVersion.getId()).thenReturn(11L); + Mockito.when(sameVersion.getSemanticVersion()).thenReturn("1.29.2"); + KubernetesSupportedVersion otherVersion = Mockito.mock(KubernetesSupportedVersion.class); + Mockito.when(otherVersion.getId()).thenReturn(12L); + Mockito.when(otherVersion.getSemanticVersion()).thenReturn("1.29.3"); + // No node has to be at the version, for example when the first control node failed + KubernetesSupportedVersionVO failedVersion = Mockito.mock(KubernetesSupportedVersionVO.class); + Mockito.when(failedVersion.getSemanticVersion()).thenReturn("1.29.2"); + Mockito.when(kubernetesSupportedVersionDao.findByIdIncludingRemoved(11L)).thenReturn(failedVersion); + KubernetesSupportedVersion equivalentVersion = Mockito.mock(KubernetesSupportedVersion.class); + Mockito.when(equivalentVersion.getId()).thenReturn(13L); + Mockito.when(equivalentVersion.getSemanticVersion()).thenReturn("1.29.2"); + Assert.assertTrue(kubernetesClusterManager.isRetryOfFailedUpgrade(cluster, sameVersion)); + Assert.assertTrue(kubernetesClusterManager.isRetryOfFailedUpgrade(cluster, equivalentVersion)); + Assert.assertFalse(kubernetesClusterManager.isRetryOfFailedUpgrade(cluster, otherVersion)); + Mockito.verifyNoInteractions(kubernetesClusterVmMapDao); + } + + @Test + public void testIsRetryOfFailedUpgradeToSameSemanticVersion() { + // Another version row with the cluster's own semantic version, with every node recorded at it + KubernetesCluster cluster = mockClusterForUpgradeRetry("1.29.2", "1.29.2"); + mockClusterVersion("1.29.2"); + Assert.assertFalse(kubernetesClusterManager.isRetryOfFailedUpgrade(cluster, mockUpgradeVersion(11L))); + } } diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorkerTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorkerTest.java index 1eb55808e09d..6b8a558c71ad 100644 --- a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorkerTest.java +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorkerTest.java @@ -16,6 +16,8 @@ // under the License. package com.cloud.kubernetes.cluster.actionworkers; +import java.util.Collections; +import java.util.List; import java.util.UUID; import org.apache.cloudstack.api.ApiConstants; @@ -38,6 +40,11 @@ import com.cloud.network.Network; import com.cloud.network.dao.IPAddressDao; import com.cloud.network.dao.IPAddressVO; +import com.cloud.network.rules.PortForwardingRuleVO; +import com.cloud.network.rules.dao.PortForwardingRulesDao; +import com.cloud.uservm.UserVm; +import com.cloud.utils.Pair; +import com.cloud.utils.net.Ip; @RunWith(MockitoJUnitRunner.class) public class KubernetesClusterActionWorkerTest { @@ -60,6 +67,9 @@ public class KubernetesClusterActionWorkerTest { @Mock IPAddressDao ipAddressDao; + @Mock + PortForwardingRulesDao portForwardingRulesDao; + KubernetesClusterActionWorker actionWorker = null; final static Long DEFAULT_ID = 1L; @@ -74,6 +84,7 @@ public void setUp() throws Exception { Mockito.when(kubernetesCluster.getId()).thenReturn(DEFAULT_ID); actionWorker = new KubernetesClusterActionWorker(kubernetesCluster, kubernetesClusterManager); actionWorker.ipAddressDao = ipAddressDao; + actionWorker.portForwardingRulesDao = portForwardingRulesDao; } @Test @@ -130,4 +141,158 @@ public void testGetVpcTierKubernetesPublicIpValid() { IpAddress result = actionWorker.getVpcTierKubernetesPublicIp(mockNetworkForGetVpcTierKubernetesPublicIpTest()); Assert.assertNotNull(result); } + + private UserVm mockNodeVm() { + UserVm vm = Mockito.mock(UserVm.class); + Mockito.when(vm.getHostName()).thenReturn("Node-1"); + return vm; + } + + private static final String DRAIN_COMMAND = "sudo /opt/bin/kubectl drain node-1 --ignore-daemonsets --delete-emptydir-data --timeout=300s"; + + @Test + public void testDrainKubernetesClusterNodeRetriesUntilSuccess() throws Exception { + KubernetesClusterActionWorker worker = Mockito.spy(actionWorker); + UserVm vm = mockNodeVm(); + Mockito.doReturn(new Pair<>(false, "Cannot evict pod as it would violate the pod's disruption budget"), new Pair<>(true, "node/node-1 drained")) + .when(worker).executeSshCommand("10.0.0.1", 2222, DRAIN_COMMAND, 10000, 10000, 360000); + Mockito.doReturn(true).when(worker).sleepBeforeRetry(Mockito.anyLong()); + + Pair result = worker.drainKubernetesClusterNode("10.0.0.1", 2222, vm, 3, 5, Long.MAX_VALUE); + + Assert.assertTrue(result.first()); + Assert.assertEquals("node/node-1 drained", result.second()); + Mockito.verify(worker, Mockito.times(2)).executeSshCommand("10.0.0.1", 2222, DRAIN_COMMAND, 10000, 10000, 360000); + Mockito.verify(worker, Mockito.times(1)).sleepBeforeRetry(5); + } + + @Test + public void testDrainKubernetesClusterNodeUsesEveryAttempt() throws Exception { + KubernetesClusterActionWorker worker = Mockito.spy(actionWorker); + UserVm vm = mockNodeVm(); + Mockito.doThrow(new RuntimeException("connection refused")).doReturn(new Pair<>(false, "pdb")) + .when(worker).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.doReturn(true).when(worker).sleepBeforeRetry(Mockito.anyLong()); + + Pair result = worker.drainKubernetesClusterNode("10.0.0.1", 2222, vm, 3, 5, Long.MAX_VALUE); + + Assert.assertFalse(result.first()); + Assert.assertEquals("pdb", result.second()); + Mockito.verify(worker, Mockito.times(3)).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.verify(worker, Mockito.times(2)).sleepBeforeRetry(5); + // The caller uncordons the node, as only it knows whether it cordoned it + Mockito.verify(worker, Mockito.never()).uncordonKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), Mockito.any()); + } + + @Test + public void testDrainKubernetesClusterNodeStopsRetryingAtDeadline() throws Exception { + KubernetesClusterActionWorker worker = Mockito.spy(actionWorker); + UserVm vm = mockNodeVm(); + Mockito.doReturn(new Pair<>(false, "timed out")) + .when(worker).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + + // 90 seconds left: one attempt, with kubectl's timeout cut to what is left after the SSH margin + Pair result = worker.drainKubernetesClusterNode("10.0.0.1", 2222, vm, 3, 60000, System.currentTimeMillis() + 90500); + + Assert.assertFalse(result.first()); + Mockito.verify(worker, Mockito.times(1)).executeSshCommand(Mockito.eq("10.0.0.1"), Mockito.eq(2222), + Mockito.matches(".*--timeout=(29|30)s"), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.verify(worker, Mockito.never()).sleepBeforeRetry(Mockito.anyLong()); + } + + @Test + public void testDrainKubernetesClusterNodeWithNoTimeLeft() throws Exception { + KubernetesClusterActionWorker worker = Mockito.spy(actionWorker); + UserVm vm = mockNodeVm(); + + Pair result = worker.drainKubernetesClusterNode("10.0.0.1", 2222, vm, 3, 60000, System.currentTimeMillis() + 30000); + + Assert.assertFalse(result.first()); + Assert.assertTrue(result.second().contains("timed out")); + Mockito.verify(worker, Mockito.never()).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + } + + @Test + public void testDrainKubernetesClusterNodeStopsRetryingWhenInterrupted() throws Exception { + KubernetesClusterActionWorker worker = Mockito.spy(actionWorker); + UserVm vm = mockNodeVm(); + Mockito.doReturn(new Pair<>(false, "pdb")) + .when(worker).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.doReturn(false).when(worker).sleepBeforeRetry(Mockito.anyLong()); + + Assert.assertFalse(worker.drainKubernetesClusterNode("10.0.0.1", 2222, vm, 3, 5, Long.MAX_VALUE).first()); + Mockito.verify(worker, Mockito.times(1)).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + } + + @Test + public void testGetNodeSshAddressIgnoresRevokedRule() { + actionWorker.publicIpAddress = "10.0.0.1"; + actionWorker.sshPort = 2222; + UserVm vm = Mockito.mock(UserVm.class); + Mockito.when(vm.getId()).thenReturn(7L); + mockIpAddress(1L, "10.0.0.1"); + PortForwardingRuleVO revoked = Mockito.mock(PortForwardingRuleVO.class); + Mockito.when(revoked.getState()).thenReturn(com.cloud.network.rules.FirewallRule.State.Revoke); + PortForwardingRuleVO active = mockSshRule(1L, 2226); + Mockito.when(portForwardingRulesDao.listByVm(7L)).thenReturn(List.of(revoked, active)); + + Assert.assertEquals(Integer.valueOf(2226), actionWorker.getNodeSshAddress(vm, 1).second()); + } + + private PortForwardingRuleVO mockSshRule(long ipId, int sourcePort) { + PortForwardingRuleVO rule = Mockito.mock(PortForwardingRuleVO.class); + Mockito.when(rule.getDestinationPortStart()).thenReturn(22); + Mockito.when(rule.getSourcePortStart()).thenReturn(sourcePort); + Mockito.when(rule.getSourceIpAddressId()).thenReturn(ipId); + return rule; + } + + private void mockIpAddress(long id, String address) { + IPAddressVO ip = Mockito.mock(IPAddressVO.class); + Mockito.when(ip.getAddress()).thenReturn(new Ip(address)); + Mockito.when(ipAddressDao.findById(id)).thenReturn(ip); + } + + @Test + public void testGetNodeSshAddressUsesPortForwardingRule() { + actionWorker.publicIpAddress = "10.0.0.1"; + actionWorker.sshPort = 2222; + UserVm vm = Mockito.mock(UserVm.class); + Mockito.when(vm.getId()).thenReturn(7L); + mockIpAddress(1L, "10.0.0.1"); + mockIpAddress(2L, "10.0.0.2"); + List rules = List.of(mockSshRule(2L, 2230), mockSshRule(1L, 2225)); + Mockito.when(portForwardingRulesDao.listByVm(7L)).thenReturn(rules); + + Pair address = actionWorker.getNodeSshAddress(vm, 1); + + Assert.assertEquals("10.0.0.1", address.first()); + Assert.assertEquals(Integer.valueOf(2225), address.second()); + } + + @Test + public void testGetNodeSshAddressWithoutRuleFallsBackToIndex() { + actionWorker.publicIpAddress = "10.0.0.1"; + actionWorker.sshPort = 2222; + UserVm vm = Mockito.mock(UserVm.class); + Mockito.when(vm.getId()).thenReturn(7L); + Mockito.when(portForwardingRulesDao.listByVm(7L)).thenReturn(Collections.emptyList()); + + Pair address = actionWorker.getNodeSshAddress(vm, 2); + + Assert.assertEquals("10.0.0.1", address.first()); + Assert.assertEquals(Integer.valueOf(2224), address.second()); + } + + @Test + public void testGetNodeSshAddressOnSharedNetwork() { + actionWorker.publicIpAddress = "10.0.0.1"; + actionWorker.sshPort = 22; + UserVm vm = Mockito.mock(UserVm.class); + Mockito.when(vm.getPrivateIpAddress()).thenReturn("192.168.0.5"); + + Assert.assertEquals(new Pair<>("10.0.0.1", 22), actionWorker.getNodeSshAddress(vm, 0)); + Assert.assertEquals(new Pair<>("192.168.0.5", 22), actionWorker.getNodeSshAddress(vm, 1)); + Mockito.verifyNoInteractions(portForwardingRulesDao); + } } 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..344f131a081b 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,7 +16,13 @@ // under the License. package com.cloud.kubernetes.cluster.actionworkers; +import com.cloud.event.ActionEventUtils; +import com.cloud.event.EventVO; +import com.cloud.exception.ResourceUnavailableException; import com.cloud.kubernetes.cluster.KubernetesCluster; +import com.cloud.kubernetes.cluster.KubernetesClusterEventTypes; +import com.cloud.kubernetes.cluster.KubernetesClusterVO; +import com.cloud.kubernetes.cluster.dao.KubernetesClusterDao; import com.cloud.kubernetes.cluster.KubernetesClusterVmMapVO; import com.cloud.kubernetes.cluster.KubernetesClusterManagerImpl; import com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDao; @@ -24,14 +30,23 @@ 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.vm.UserVmManager; +import com.cloud.vm.UserVmService; import com.cloud.vm.UserVmVO; +import com.cloud.vm.VMInstanceVO; +import com.cloud.vm.VirtualMachine; import com.cloud.vm.dao.UserVmDao; +import com.cloud.vm.dao.VMInstanceDao; +import org.apache.cloudstack.context.CallContext; +import org.junit.After; 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; @@ -40,6 +55,7 @@ import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.CONTROL; import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.DEFAULT; +import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.WORKER; @RunWith(MockitoJUnitRunner.class) public class KubernetesClusterScaleWorkerTest { @@ -54,6 +70,14 @@ public class KubernetesClusterScaleWorkerTest { private KubernetesClusterVmMapDao kubernetesClusterVmMapDao; @Mock private UserVmDao userVmDao; + @Mock + private UserVmService userVmService; + @Mock + private UserVmManager userVmManager; + @Mock + private KubernetesClusterDao kubernetesClusterDao; + @Mock + private VMInstanceDao vmInstanceDao; private KubernetesClusterScaleWorker worker; @@ -187,4 +211,467 @@ public void testGetWorkerNodesToRemoveForDownsize_noRemoval() { Assert.assertTrue(toRemove.isEmpty()); } + + private UserVmVO mockNodeVm(String hostName) { + UserVmVO vm = Mockito.mock(UserVmVO.class); + Mockito.when(vm.getHostName()).thenReturn(hostName); + return vm; + } + + @Test + public void testDrainNodesToRemoveFailureLeavesClusterUnchanged() throws Exception { + Mockito.when(kubernetesCluster.getId()).thenReturn(10L); + worker.publicIpAddress = "10.0.0.1"; + worker.sshPort = 2222; + worker.scaleTimeoutTime = Long.MAX_VALUE; + KubernetesClusterScaleWorker spy = Mockito.spy(worker); + UserVmVO node1 = mockNodeVm("node-1"); + UserVmVO node2 = mockNodeVm("node-2"); + Mockito.doReturn(new Pair<>(true, "")).when(spy).executeSshCommand(Mockito.eq("10.0.0.1"), Mockito.eq(2222), + Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.doReturn(new Pair<>(true, "drained")).when(spy).drainKubernetesClusterNode(Mockito.eq("10.0.0.1"), Mockito.eq(2222), + Mockito.eq(node1), Mockito.anyInt(), Mockito.anyLong(), Mockito.anyLong()); + Mockito.doReturn(new Pair<>(false, "Cannot evict pod")).when(spy).drainKubernetesClusterNode(Mockito.eq("10.0.0.1"), Mockito.eq(2222), + Mockito.eq(node2), Mockito.anyInt(), Mockito.anyLong(), Mockito.anyLong()); + Mockito.doReturn(true).when(spy).uncordonKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), Mockito.any()); + Mockito.doReturn(true).when(spy).stateTransitTo(Mockito.anyLong(), Mockito.any()); + + try { + spy.drainNodesToRemove(Arrays.asList(node1, node2)); + Assert.fail("Expected the scale-down to stop"); + } catch (CloudRuntimeException e) { + Assert.assertTrue(e.getMessage().contains("node-2")); + Assert.assertTrue(e.getMessage().contains("Cannot evict pod")); + } + Mockito.verify(spy).uncordonKubernetesClusterNode("10.0.0.1", 2222, node1); + Mockito.verify(spy).uncordonKubernetesClusterNode("10.0.0.1", 2222, node2); + Mockito.verify(spy).stateTransitTo(10L, KubernetesCluster.Event.OperationSucceeded); + Mockito.verify(spy, Mockito.never()).stateTransitTo(10L, KubernetesCluster.Event.OperationFailed); + } + + @Test + public void testDrainNodesToRemoveCordonsAllBeforeDraining() throws Exception { + worker.publicIpAddress = "10.0.0.1"; + worker.sshPort = 2222; + worker.scaleTimeoutTime = Long.MAX_VALUE; + KubernetesClusterScaleWorker spy = Mockito.spy(worker); + UserVmVO node1 = mockNodeVm("node-1"); + UserVmVO node2 = mockNodeVm("node-2"); + Mockito.doReturn(new Pair<>(true, "")).when(spy).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), + Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.doReturn(new Pair<>(true, "drained")).when(spy).drainKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), + Mockito.any(), Mockito.anyInt(), Mockito.anyLong(), Mockito.anyLong()); + + spy.drainNodesToRemove(Arrays.asList(node1, node2)); + + org.mockito.InOrder inOrder = Mockito.inOrder(spy); + inOrder.verify(spy).executeSshCommand("10.0.0.1", 2222, "sudo /opt/bin/kubectl cordon node-1", 10000, 10000, 30000); + inOrder.verify(spy).executeSshCommand("10.0.0.1", 2222, "sudo /opt/bin/kubectl cordon node-2", 10000, 10000, 30000); + inOrder.verify(spy).drainKubernetesClusterNode(Mockito.eq("10.0.0.1"), Mockito.eq(2222), Mockito.eq(node1), Mockito.anyInt(), Mockito.anyLong(), Mockito.anyLong()); + inOrder.verify(spy).drainKubernetesClusterNode(Mockito.eq("10.0.0.1"), Mockito.eq(2222), Mockito.eq(node2), Mockito.anyInt(), Mockito.anyLong(), Mockito.anyLong()); + Mockito.verify(spy, Mockito.never()).uncordonKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), Mockito.any()); + Mockito.verify(spy, Mockito.never()).stateTransitTo(Mockito.anyLong(), Mockito.any()); + } + + @Test + public void testDrainNodesToRemoveFailureKeepsAlertCluster() throws Exception { + Mockito.when(kubernetesCluster.getId()).thenReturn(10L); + Mockito.when(kubernetesCluster.getState()).thenReturn(KubernetesCluster.State.Alert); + KubernetesClusterScaleWorker alertWorker = new KubernetesClusterScaleWorker(kubernetesCluster, null, 1L, null, null, null, null, clusterManager); + alertWorker.publicIpAddress = "10.0.0.1"; + alertWorker.sshPort = 2222; + KubernetesClusterScaleWorker spy = Mockito.spy(alertWorker); + UserVmVO node1 = mockNodeVm("node-1"); + Mockito.doReturn(new Pair<>(false, "connection refused")).when(spy).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), + Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.doReturn(true).when(spy).stateTransitTo(Mockito.anyLong(), Mockito.any()); + + CloudRuntimeException e = Assert.assertThrows(CloudRuntimeException.class, () -> spy.drainNodesToRemove(List.of(node1))); + Assert.assertTrue(e.getMessage(), e.getMessage().contains("unable to cordon Kubernetes node: node-1")); + Mockito.verify(spy).stateTransitTo(10L, KubernetesCluster.Event.OperationFailed); + Mockito.verify(spy, Mockito.never()).uncordonKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), Mockito.any()); + } + + @Test + public void testDrainNodesToRemoveStopsAtDeadline() throws Exception { + Mockito.when(kubernetesCluster.getId()).thenReturn(10L); + worker.publicIpAddress = "10.0.0.1"; + worker.sshPort = 2222; + worker.scaleTimeoutTime = System.currentTimeMillis() - 1; + KubernetesClusterScaleWorker spy = Mockito.spy(worker); + UserVmVO node1 = mockNodeVm("node-1"); + Mockito.doReturn(new Pair<>(true, "")).when(spy).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), + Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.doReturn(true).when(spy).uncordonKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), Mockito.any()); + Mockito.doReturn(true).when(spy).stateTransitTo(Mockito.anyLong(), Mockito.any()); + + CloudRuntimeException e = Assert.assertThrows(CloudRuntimeException.class, () -> spy.drainNodesToRemove(List.of(node1))); + Assert.assertTrue(e.getMessage(), e.getMessage().contains("timed out")); + Mockito.verify(spy, Mockito.never()).drainKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), Mockito.any(), Mockito.anyInt(), Mockito.anyLong(), Mockito.anyLong()); + Mockito.verify(spy).uncordonKubernetesClusterNode("10.0.0.1", 2222, node1); + Mockito.verify(spy).stateTransitTo(10L, KubernetesCluster.Event.OperationSucceeded); + } + + @Test + public void testDrainNodesToRemoveKeepsNodeCordonedBeforehand() throws Exception { + Mockito.when(kubernetesCluster.getId()).thenReturn(10L); + worker.publicIpAddress = "10.0.0.1"; + worker.sshPort = 2222; + worker.scaleTimeoutTime = Long.MAX_VALUE; + KubernetesClusterScaleWorker spy = Mockito.spy(worker); + UserVmVO node1 = mockNodeVm("node-1"); + UserVmVO node2 = mockNodeVm("node-2"); + Mockito.doReturn(new Pair<>(true, "")).when(spy).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), + Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.doReturn(new Pair<>(true, "sudo: unable to resolve host node-1\ntrue")).when(spy).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), + Mockito.eq("sudo /opt/bin/kubectl get node node-1 -o jsonpath='{.spec.unschedulable}'"), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.doReturn(new Pair<>(false, "Cannot evict pod")).when(spy).drainKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), + Mockito.any(), Mockito.anyInt(), Mockito.anyLong(), Mockito.anyLong()); + Mockito.doReturn(true).when(spy).uncordonKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), Mockito.any()); + Mockito.doReturn(true).when(spy).stateTransitTo(Mockito.anyLong(), Mockito.any()); + + Assert.assertThrows(CloudRuntimeException.class, () -> spy.drainNodesToRemove(Arrays.asList(node1, node2))); + Mockito.verify(spy, Mockito.never()).uncordonKubernetesClusterNode("10.0.0.1", 2222, node1); + Mockito.verify(spy).uncordonKubernetesClusterNode("10.0.0.1", 2222, node2); + } + + @Test + public void testDrainNodesToRemoveTakesUnknownCordonAsCordoned() throws Exception { + Mockito.when(kubernetesCluster.getId()).thenReturn(10L); + worker.publicIpAddress = "10.0.0.1"; + worker.sshPort = 2222; + worker.scaleTimeoutTime = Long.MAX_VALUE; + KubernetesClusterScaleWorker spy = Mockito.spy(worker); + UserVmVO node1 = mockNodeVm("node-1"); + Mockito.doReturn(new Pair<>(true, "")).when(spy).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), + Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.doThrow(new RuntimeException("timed out")).when(spy).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), + Mockito.startsWith("sudo /opt/bin/kubectl get node"), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.doReturn(new Pair<>(false, "Cannot evict pod")).when(spy).drainKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), + Mockito.any(), Mockito.anyInt(), Mockito.anyLong(), Mockito.anyLong()); + Mockito.doReturn(true).when(spy).stateTransitTo(Mockito.anyLong(), Mockito.any()); + + Assert.assertThrows(CloudRuntimeException.class, () -> spy.drainNodesToRemove(List.of(node1))); + Mockito.verify(spy).executeSshCommand("10.0.0.1", 2222, "sudo /opt/bin/kubectl cordon node-1", 10000, 10000, 30000); + Mockito.verify(spy, Mockito.never()).uncordonKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), Mockito.any()); + } + + private MockedStatic callContextMock; + private MockedStatic actionEventUtilsMock; + + @After + public void tearDown() { + if (callContextMock != null) { + callContextMock.close(); + } + if (actionEventUtilsMock != null) { + actionEventUtilsMock.close(); + } + } + + private static final long CLUSTER_ID = 10L; + private static final long WORKER_OFFERING_ID = 5L; + + /** + * Sets up a worker for {@link KubernetesClusterScaleWorker#removeNodesFromCluster(List)} on a cluster of 3 workers + * of 2 cores and 2048 MB each, with control and drain steps and the Kubernetes node deletion stubbed out. + */ + private KubernetesClusterScaleWorker setUpNodeRemoval() throws Exception { + callContextMock = Mockito.mockStatic(CallContext.class); + CallContext callContext = Mockito.mock(CallContext.class); + callContextMock.when(CallContext::current).thenReturn(callContext); + callContextMock.when(() -> CallContext.register(Mockito.any(CallContext.class), Mockito.any())).thenReturn(callContext); + actionEventUtilsMock = Mockito.mockStatic(ActionEventUtils.class); + + Mockito.when(kubernetesCluster.getId()).thenReturn(CLUSTER_ID); + worker.userVmService = userVmService; + worker.userVmManager = userVmManager; + worker.kubernetesClusterDao = kubernetesClusterDao; + worker.vmInstanceDao = vmInstanceDao; + worker.publicIpAddress = "10.0.0.1"; + worker.sshPort = 2222; + worker.scaleTimeoutTime = Long.MAX_VALUE; + + KubernetesClusterVO clusterVO = Mockito.mock(KubernetesClusterVO.class); + Mockito.lenient().when(clusterVO.getCores()).thenReturn(10L); + Mockito.lenient().when(clusterVO.getMemory()).thenReturn(10240L); + Mockito.lenient().when(clusterVO.getNodeCount()).thenReturn(3L); + Mockito.lenient().when(kubernetesClusterDao.findById(CLUSTER_ID)).thenReturn(clusterVO); + ServiceOfferingVO offering = Mockito.mock(ServiceOfferingVO.class); + Mockito.lenient().when(offering.getCpu()).thenReturn(2); + Mockito.lenient().when(offering.getRamSize()).thenReturn(2048); + Mockito.lenient().when(serviceOfferingDao.findByIdIncludingRemoved(Mockito.anyLong(), Mockito.eq(WORKER_OFFERING_ID))).thenReturn(offering); + + KubernetesClusterScaleWorker spy = Mockito.spy(worker); + Mockito.doAnswer(invocation -> new java.util.ArrayList<>((List) invocation.getArgument(0))) + .when(spy).drainNodesToRemove(Mockito.anyList()); + // kubectl delete node + Mockito.doReturn(new Pair<>(true, "")).when(spy).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), + Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + Mockito.lenient().doReturn(true).when(spy).uncordonKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), Mockito.any()); + Mockito.lenient().doReturn(true).when(spy).stateTransitTo(Mockito.anyLong(), Mockito.any()); + Mockito.lenient().doReturn(KubernetesClusterVO.class.cast(null)).when(spy).updateKubernetesClusterEntry(Mockito.any(), Mockito.any(), + Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean()); + // No network rules to update + Mockito.lenient().when(clusterManager.isDirectAccess(Mockito.any())).thenReturn(true); + return spy; + } + + private KubernetesClusterVmMapVO mockNodeVmMap(long mapId, long vmId, String hostName) { + KubernetesClusterVmMapVO vmMap = Mockito.mock(KubernetesClusterVmMapVO.class); + Mockito.lenient().when(vmMap.getId()).thenReturn(mapId); + Mockito.lenient().when(vmMap.getVmId()).thenReturn(vmId); + UserVmVO vm = Mockito.mock(UserVmVO.class); + Mockito.lenient().when(vm.getHostName()).thenReturn(hostName); + Mockito.lenient().when(vm.getId()).thenReturn(vmId); + Mockito.lenient().when(vm.getServiceOfferingId()).thenReturn(WORKER_OFFERING_ID); + Mockito.when(userVmDao.findById(vmId)).thenReturn(vm); + VMInstanceVO vmInstance = Mockito.mock(VMInstanceVO.class); + Mockito.lenient().when(vmInstance.getState()).thenReturn(VirtualMachine.State.Destroyed); + Mockito.lenient().when(vmInstanceDao.findByIdIncludingRemoved(vmId)).thenReturn(vmInstance); + return vmMap; + } + + @Test + public void testRemoveNodesFromClusterRemovesMapEntriesAndExpungesVms() throws Exception { + KubernetesClusterScaleWorker spy = setUpNodeRemoval(); + KubernetesClusterVmMapVO map1 = mockNodeVmMap(101L, 201L, "node-1"); + KubernetesClusterVmMapVO map2 = mockNodeVmMap(102L, 202L, "node-2"); + UserVmVO vm1 = userVmDao.findById(201L); + UserVmVO vm2 = userVmDao.findById(202L); + Mockito.when(userVmManager.expunge(Mockito.any())).thenReturn(true); + + spy.removeNodesFromCluster(Arrays.asList(map1, map2)); + + org.mockito.InOrder inOrder = Mockito.inOrder(kubernetesClusterVmMapDao, userVmService, userVmManager); + // The VM's expunge still finds the map entry, which goes once the VM is done with + inOrder.verify(userVmService).destroyVm(201L, true); + inOrder.verify(userVmManager).expunge(vm1); + inOrder.verify(kubernetesClusterVmMapDao).expunge(101L); + inOrder.verify(userVmService).destroyVm(202L, true); + inOrder.verify(userVmManager).expunge(vm2); + inOrder.verify(kubernetesClusterVmMapDao).expunge(102L); + Mockito.verify(spy, Mockito.never()).stateTransitTo(Mockito.anyLong(), Mockito.any()); + Mockito.verify(spy, Mockito.never()).revokeRemovedNodeVmSshRules(Mockito.any()); + // The caller updates the cluster entry once every node is removed + Mockito.verify(spy, Mockito.never()).updateKubernetesClusterEntry(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), + Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean()); + actionEventUtilsMock.verifyNoInteractions(); + } + + @Test + public void testRemoveNodesFromClusterExpungeFailureRemovesMapEntryAndGoesOn() throws Exception { + KubernetesClusterScaleWorker spy = setUpNodeRemoval(); + KubernetesClusterVmMapVO map1 = mockNodeVmMap(101L, 201L, "node-1"); + KubernetesClusterVmMapVO map2 = mockNodeVmMap(102L, 202L, "node-2"); + UserVmVO vm1 = userVmDao.findById(201L); + UserVmVO vm2 = userVmDao.findById(202L); + Mockito.when(userVmManager.expunge(vm1)).thenReturn(false); + Mockito.when(userVmManager.expunge(vm2)).thenThrow(new CloudRuntimeException("storage unavailable")); + + spy.removeNodesFromCluster(Arrays.asList(map1, map2)); + + Mockito.verify(kubernetesClusterVmMapDao).expunge(101L); + Mockito.verify(kubernetesClusterVmMapDao).expunge(102L); + Mockito.verify(userVmService).destroyVm(201L, true); + Mockito.verify(userVmService).destroyVm(202L, true); + Mockito.verify(spy, Mockito.never()).stateTransitTo(Mockito.anyLong(), Mockito.any()); + Mockito.verify(spy, Mockito.never()).uncordonKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), Mockito.any()); + Mockito.verify(spy).revokeRemovedNodeVmSshRules(vm1); + Mockito.verify(spy).revokeRemovedNodeVmSshRules(vm2); + actionEventUtilsMock.verify(() -> ActionEventUtils.onCompletedActionEvent(Mockito.anyLong(), Mockito.anyLong(), + Mockito.eq(EventVO.LEVEL_WARN), Mockito.eq(KubernetesClusterEventTypes.EVENT_KUBERNETES_CLUSTER_SCALE), + Mockito.contains("is left in state Destroyed"), Mockito.eq(201L), Mockito.anyString(), Mockito.eq(0L))); + actionEventUtilsMock.verify(() -> ActionEventUtils.onCompletedActionEvent(Mockito.anyLong(), Mockito.anyLong(), + Mockito.eq(EventVO.LEVEL_WARN), Mockito.eq(KubernetesClusterEventTypes.EVENT_KUBERNETES_CLUSTER_SCALE), + Mockito.contains("Unable to expunge VM"), Mockito.eq(202L), Mockito.anyString(), Mockito.eq(0L))); + } + + @Test + public void testRemoveNodesFromClusterDestroyFailureRemovesMapEntryAndFails() throws Exception { + KubernetesClusterScaleWorker spy = setUpNodeRemoval(); + KubernetesClusterVmMapVO map1 = mockNodeVmMap(101L, 201L, "node-1"); + KubernetesClusterVmMapVO map2 = mockNodeVmMap(102L, 202L, "node-2"); + KubernetesClusterVmMapVO map3 = mockNodeVmMap(103L, 203L, "node-3"); + UserVmVO vm1 = userVmDao.findById(201L); + UserVmVO vm2 = userVmDao.findById(202L); + UserVmVO vm3 = userVmDao.findById(203L); + Mockito.when(userVmManager.expunge(Mockito.any())).thenReturn(true); + Mockito.when(userVmService.destroyVm(202L, true)).thenThrow(new ResourceUnavailableException("host down", com.cloud.host.Host.class, 1L)); + Mockito.when(vmInstanceDao.findByIdIncludingRemoved(202L).getState()).thenReturn(VirtualMachine.State.Running); + + CloudRuntimeException e = Assert.assertThrows(CloudRuntimeException.class, + () -> spy.removeNodesFromCluster(Arrays.asList(map1, map2, map3))); + + Assert.assertTrue(e.getMessage(), e.getMessage().contains("unable to destroy VM")); + // node-2 has left the Kubernetes cluster, so its VM has left the map although it is not destroyed + Mockito.verify(kubernetesClusterVmMapDao).expunge(101L); + Mockito.verify(kubernetesClusterVmMapDao).expunge(102L); + Mockito.verify(kubernetesClusterVmMapDao, Mockito.never()).expunge(103L); + Mockito.verify(userVmManager, Mockito.never()).expunge(vm2); + Mockito.verify(spy).revokeRemovedNodeVmSshRules(vm2); + Mockito.verify(userVmService, Mockito.never()).destroyVm(203L, true); + Mockito.verify(spy).stateTransitTo(CLUSTER_ID, KubernetesCluster.Event.OperationFailed); + // node-3 is still in the cluster, drained, so it is uncordoned + Mockito.verify(spy).uncordonKubernetesClusterNode("10.0.0.1", 2222, vm3); + Mockito.verify(spy, Mockito.never()).uncordonKubernetesClusterNode("10.0.0.1", 2222, vm1); + Mockito.verify(spy, Mockito.never()).uncordonKubernetesClusterNode("10.0.0.1", 2222, vm2); + // The cluster entry matches the two map entries removed + Mockito.verify(spy).updateKubernetesClusterEntry(10L - 4L, 10240L - 4096L, 1L, null, false, 0L, 0L, WORKER, false, false); + } + + @Test + public void testRemoveNodesFromClusterKubernetesNodeFailureKeepsMapEntry() throws Exception { + KubernetesClusterScaleWorker spy = setUpNodeRemoval(); + KubernetesClusterVmMapVO map1 = mockNodeVmMap(101L, 201L, "node-1"); + KubernetesClusterVmMapVO map2 = mockNodeVmMap(102L, 202L, "node-2"); + UserVmVO vm2 = userVmDao.findById(202L); + Mockito.when(userVmManager.expunge(Mockito.any())).thenReturn(true); + Mockito.doReturn(true).when(spy).sleepBeforeRetry(Mockito.anyLong()); + Mockito.doReturn(new Pair<>(false, "connection refused")).when(spy).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), + Mockito.eq("sudo /opt/bin/kubectl delete node node-2 --ignore-not-found"), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + + Assert.assertThrows(CloudRuntimeException.class, () -> spy.removeNodesFromCluster(Arrays.asList(map1, map2))); + + Mockito.verify(kubernetesClusterVmMapDao).expunge(101L); + Mockito.verify(kubernetesClusterVmMapDao, Mockito.never()).expunge(102L); + Mockito.verify(userVmService, Mockito.never()).destroyVm(202L, true); + Mockito.verify(spy).uncordonKubernetesClusterNode("10.0.0.1", 2222, vm2); + Mockito.verify(spy).updateKubernetesClusterEntry(10L - 2L, 10240L - 2048L, 2L, null, false, 0L, 0L, WORKER, false, false); + } + + @Test + public void testRemoveNodesFromClusterLaterFailureMatchesClusterToMap() throws Exception { + KubernetesClusterScaleWorker spy = setUpNodeRemoval(); + KubernetesClusterVmMapVO controlMap = mockNodeVmMap(101L, 201L, "control-2"); + KubernetesClusterVmMapVO workerMap = mockNodeVmMap(102L, 202L, "node-2"); + Mockito.when(userVmManager.expunge(Mockito.any())).thenReturn(true); + // Not a CloudRuntimeException, after every node has been removed + Mockito.when(clusterManager.isDirectAccess(Mockito.any())).thenThrow(new IllegalStateException("network gone")); + + Assert.assertThrows(IllegalStateException.class, () -> spy.removeNodesFromCluster(Arrays.asList(controlMap, workerMap))); + + Mockito.verify(kubernetesClusterVmMapDao).expunge(101L); + Mockito.verify(kubernetesClusterVmMapDao).expunge(102L); + Mockito.verify(spy, Mockito.never()).uncordonKubernetesClusterNode(Mockito.anyString(), Mockito.anyInt(), Mockito.any()); + // As a full scale-down does, every VM removed comes off the node count, so the total matches the map + Mockito.verify(spy).updateKubernetesClusterEntry(10L - 4L, 10240L - 4096L, 1L, null, false, 0L, 0L, WORKER, false, false); + } + + @Test + public void testRemoveNodesFromClusterDestroyErrorAfterVmDestroyedGoesOn() throws Exception { + KubernetesClusterScaleWorker spy = setUpNodeRemoval(); + KubernetesClusterVmMapVO map1 = mockNodeVmMap(101L, 201L, "node-1"); + UserVmVO vm1 = userVmDao.findById(201L); + Mockito.when(userVmService.destroyVm(201L, true)).thenThrow(new CloudRuntimeException("Failed to destroy vm with specified vmId")); + Mockito.when(userVmManager.expunge(vm1)).thenReturn(true); + + spy.removeNodesFromCluster(List.of(map1)); + + Mockito.verify(kubernetesClusterVmMapDao).expunge(101L); + Mockito.verify(userVmManager).expunge(vm1); + Mockito.verify(spy, Mockito.never()).stateTransitTo(Mockito.anyLong(), Mockito.any()); + } + + @Test + public void testRemoveNodesFromClusterDestroyErrorAfterVmExpungedSkipsExpunge() throws Exception { + KubernetesClusterScaleWorker spy = setUpNodeRemoval(); + KubernetesClusterVmMapVO map1 = mockNodeVmMap(101L, 201L, "node-1"); + UserVmVO vm1 = userVmDao.findById(201L); + Mockito.when(vmInstanceDao.findByIdIncludingRemoved(201L).getRemoved()).thenReturn(new java.util.Date()); + Mockito.when(userVmService.destroyVm(201L, true)).thenThrow(new CloudRuntimeException("Failed to destroy vm with specified vmId")); + + spy.removeNodesFromCluster(List.of(map1)); + + Mockito.verify(kubernetesClusterVmMapDao).expunge(101L); + Mockito.verify(userVmManager, Mockito.never()).expunge(vm1); + Mockito.verify(spy, Mockito.never()).stateTransitTo(Mockito.anyLong(), Mockito.any()); + actionEventUtilsMock.verifyNoInteractions(); + } + + @Test + public void testRemoveNodesFromClusterExpungedMeanwhileRecordsNoWarning() throws Exception { + KubernetesClusterScaleWorker spy = setUpNodeRemoval(); + KubernetesClusterVmMapVO map1 = mockNodeVmMap(101L, 201L, "node-1"); + UserVmVO vm1 = userVmDao.findById(201L); + Mockito.when(userVmManager.expunge(vm1)).thenReturn(false); + Mockito.when(vmInstanceDao.findByIdIncludingRemoved(201L).getRemoved()).thenReturn(new java.util.Date()); + + spy.removeNodesFromCluster(List.of(map1)); + + Mockito.verify(kubernetesClusterVmMapDao).expunge(101L); + Mockito.verify(spy, Mockito.never()).revokeRemovedNodeVmSshRules(Mockito.any()); + actionEventUtilsMock.verifyNoInteractions(); + } + + private com.cloud.network.rules.PortForwardingRuleVO mockPfRule(long id, long ipId, long networkId, int sourcePort, int destinationPort) { + com.cloud.network.rules.PortForwardingRuleVO rule = Mockito.mock(com.cloud.network.rules.PortForwardingRuleVO.class); + Mockito.lenient().when(rule.getSourceIpAddressId()).thenReturn(ipId); + Mockito.lenient().when(rule.getId()).thenReturn(id); + Mockito.lenient().when(rule.getNetworkId()).thenReturn(networkId); + Mockito.lenient().when(rule.getSourcePortStart()).thenReturn(sourcePort); + Mockito.lenient().when(rule.getDestinationPortStart()).thenReturn(destinationPort); + return rule; + } + + @Test + public void testRevokeRemovedNodeVmSshRulesRevokesOnlyTheClusterSshRule() throws Exception { + callContextMock = Mockito.mockStatic(CallContext.class); + CallContext callContext = Mockito.mock(CallContext.class); + callContextMock.when(CallContext::current).thenReturn(callContext); + worker.publicIpAddress = "10.0.0.1"; + worker.sshPort = 2222; + com.cloud.network.dao.IPAddressDao ipAddressDao = Mockito.mock(com.cloud.network.dao.IPAddressDao.class); + worker.ipAddressDao = ipAddressDao; + com.cloud.network.dao.IPAddressVO clusterIp = Mockito.mock(com.cloud.network.dao.IPAddressVO.class); + Mockito.when(clusterIp.getAddress()).thenReturn(new com.cloud.utils.net.Ip("10.0.0.1")); + com.cloud.network.dao.IPAddressVO otherIp = Mockito.mock(com.cloud.network.dao.IPAddressVO.class); + Mockito.when(otherIp.getAddress()).thenReturn(new com.cloud.utils.net.Ip("10.0.0.2")); + Mockito.when(ipAddressDao.findById(7L)).thenReturn(clusterIp); + Mockito.when(ipAddressDao.findById(8L)).thenReturn(otherIp); + com.cloud.network.dao.NetworkVO network = Mockito.mock(com.cloud.network.dao.NetworkVO.class); + Mockito.when(network.getId()).thenReturn(5L); + worker.networkModel = Mockito.mock(com.cloud.network.NetworkModel.class); + java.lang.reflect.Field networkField = KubernetesClusterScaleWorker.class.getDeclaredField("network"); + networkField.setAccessible(true); + networkField.set(worker, network); + com.cloud.network.rules.dao.PortForwardingRulesDao pfRulesDao = Mockito.mock(com.cloud.network.rules.dao.PortForwardingRulesDao.class); + com.cloud.network.rules.RulesService rulesService = Mockito.mock(com.cloud.network.rules.RulesService.class); + worker.portForwardingRulesDao = pfRulesDao; + worker.rulesService = rulesService; + UserVmVO vm = Mockito.mock(UserVmVO.class); + Mockito.when(vm.getId()).thenReturn(201L); + List rules = List.of( + mockPfRule(1L, 7L, 5L, 8080, 80), + mockPfRule(2L, 7L, 6L, 2223, 22), + mockPfRule(3L, 8L, 5L, 3022, 22), + mockPfRule(4L, 7L, 5L, 2223, 22), + // The port the scale reaches the cluster through + mockPfRule(5L, 7L, 5L, 2222, 22)); + Mockito.when(pfRulesDao.listByVm(201L)).thenReturn(rules); + + worker.revokeRemovedNodeVmSshRules(vm); + + Mockito.verify(rulesService).revokePortForwardingRule(4L, true); + Mockito.verifyNoMoreInteractions(rulesService); + } + + @Test + public void testRevokeRemovedNodeVmSshRulesLeavesNsxRulesAlone() throws Exception { + com.cloud.network.dao.NetworkVO network = Mockito.mock(com.cloud.network.dao.NetworkVO.class); + Mockito.when(network.getId()).thenReturn(5L); + java.lang.reflect.Field networkField = KubernetesClusterScaleWorker.class.getDeclaredField("network"); + networkField.setAccessible(true); + networkField.set(worker, network); + worker.publicIpAddress = "10.0.0.1"; + worker.networkModel = Mockito.mock(com.cloud.network.NetworkModel.class); + Mockito.when(worker.networkModel.isProviderSupportServiceInNetwork(5L, com.cloud.network.Network.Service.PortForwarding, + com.cloud.network.Network.Provider.Nsx)).thenReturn(true); + com.cloud.network.rules.dao.PortForwardingRulesDao pfRulesDao = Mockito.mock(com.cloud.network.rules.dao.PortForwardingRulesDao.class); + worker.portForwardingRulesDao = pfRulesDao; + + worker.revokeRemovedNodeVmSshRules(Mockito.mock(UserVmVO.class)); + + Mockito.verifyNoInteractions(pfRulesDao); + } } diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterUpgradeWorkerTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterUpgradeWorkerTest.java index ff8f875d06e1..60a5853d9525 100644 --- a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterUpgradeWorkerTest.java +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterUpgradeWorkerTest.java @@ -21,7 +21,9 @@ import com.cloud.kubernetes.cluster.KubernetesClusterVmMapVO; import com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDao; import com.cloud.kubernetes.version.KubernetesSupportedVersion; +import com.cloud.kubernetes.cluster.utils.KubernetesClusterUtil; import com.cloud.uservm.UserVm; +import com.cloud.utils.Pair; import org.junit.Assert; import org.junit.Before; import org.junit.Test; @@ -80,4 +82,74 @@ public void testFilterOutManualUpgradeNodesFromClusterUpgrade() { Assert.assertTrue(ids.contains(controlNodeId) && ids.contains(workerNode1Id)); Assert.assertFalse(ids.contains(workerNode2Id)); } + + private static final String VERSION_COMMAND = String.format(KubernetesClusterUtil.CLUSTER_NODE_VERSION_COMMAND, "node-1"); + private static final String READY_COMMAND = String.format(KubernetesClusterUpgradeWorker.NODE_STATUS_COMMAND, "node-1"); + + private KubernetesClusterUpgradeWorker mockNodeAtVersion(String storedVersion) { + Mockito.when(kubernetesSupportedVersion.getSemanticVersion()).thenReturn("1.29.2"); + KubernetesClusterVmMapVO vmMap = Mockito.mock(KubernetesClusterVmMapVO.class); + Mockito.when(vmMap.getNodeVersion()).thenReturn(storedVersion); + Mockito.when(kubernetesClusterVmMapDao.getClusterMapFromVmId(5L)).thenReturn(vmMap); + worker.publicIpAddress = "10.0.0.1"; + worker.sshPort = 2222; + return Mockito.spy(worker); + } + + private UserVm mockVm() { + UserVm vm = Mockito.mock(UserVm.class); + Mockito.when(vm.getId()).thenReturn(5L); + return vm; + } + + @Test + public void testIsNodeUpgradedWhenRecordedAndReported() throws Exception { + KubernetesClusterUpgradeWorker spy = mockNodeAtVersion("v1.29.2"); + Mockito.doReturn(new Pair<>(true, "v1.29.2\n")).when(spy).executeSshCommand("10.0.0.1", 2222, VERSION_COMMAND, 10000, 10000, 20000); + Mockito.doReturn(new Pair<>(true, "Ready\n")).when(spy).executeSshCommand("10.0.0.1", 2222, READY_COMMAND, 10000, 10000, 20000); + Assert.assertTrue(spy.isNodeUpgraded(mockVm(), "node-1")); + } + + @Test + public void testIsNodeUpgradedWhenCordonedAfterUpgrade() throws Exception { + KubernetesClusterUpgradeWorker spy = mockNodeAtVersion("v1.29.2"); + Mockito.doReturn(new Pair<>(true, "v1.29.2\n")).when(spy).executeSshCommand("10.0.0.1", 2222, VERSION_COMMAND, 10000, 10000, 20000); + Mockito.doReturn(new Pair<>(true, "Ready,SchedulingDisabled\n")).when(spy).executeSshCommand("10.0.0.1", 2222, READY_COMMAND, 10000, 10000, 20000); + Assert.assertTrue(spy.isNodeUpgraded(mockVm(), "node-1")); + } + + @Test + public void testIsNodeUpgradedNotWhenNodeNotReady() throws Exception { + KubernetesClusterUpgradeWorker spy = mockNodeAtVersion("v1.29.2"); + Mockito.doReturn(new Pair<>(true, "v1.29.2\n")).when(spy).executeSshCommand("10.0.0.1", 2222, VERSION_COMMAND, 10000, 10000, 20000); + Mockito.doReturn(new Pair<>(true, "NotReady")).when(spy).executeSshCommand("10.0.0.1", 2222, READY_COMMAND, 10000, 10000, 20000); + Assert.assertFalse(spy.isNodeUpgraded(mockVm(), "node-1")); + } + + @Test + public void testIsNodeUpgradedNotTrustingRecordedVersionAlone() throws Exception { + KubernetesClusterUpgradeWorker spy = mockNodeAtVersion("v1.29.2"); + Mockito.doReturn(new Pair<>(true, "v1.28.9")).when(spy).executeSshCommand("10.0.0.1", 2222, VERSION_COMMAND, 10000, 10000, 20000); + Assert.assertFalse(spy.isNodeUpgraded(mockVm(), "node-1")); + } + + @Test + public void testIsNodeUpgradedWhenNodeUnreachable() throws Exception { + KubernetesClusterUpgradeWorker spy = mockNodeAtVersion("1.29.2"); + Mockito.doThrow(new RuntimeException("connection refused")).when(spy).executeSshCommand("10.0.0.1", 2222, VERSION_COMMAND, 10000, 10000, 20000); + Assert.assertFalse(spy.isNodeUpgraded(mockVm(), "node-1")); + } + + @Test + public void testIsNodeUpgradedWhenRecordedAtOldVersion() throws Exception { + KubernetesClusterUpgradeWorker spy = mockNodeAtVersion("v1.28.9"); + Assert.assertFalse(spy.isNodeUpgraded(mockVm(), "node-1")); + Mockito.verify(spy, Mockito.never()).executeSshCommand(Mockito.anyString(), Mockito.anyInt(), Mockito.anyString(), Mockito.anyInt(), Mockito.anyInt(), Mockito.anyInt()); + } + + @Test + public void testIsNodeUpgradedComparesWholeVersion() throws Exception { + KubernetesClusterUpgradeWorker spy = mockNodeAtVersion("v1.29.20"); + Assert.assertFalse(spy.isNodeUpgraded(mockVm(), "node-1")); + } } diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtilTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtilTest.java index 329f9b0e42ad..bccf47f0bf35 100644 --- a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtilTest.java +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtilTest.java @@ -46,6 +46,14 @@ public void testClusterNodeVersionMatches() { executeAndTestVersionMatch(true, v1233WorkerNodeOutput, false); + executeAndTestVersionMatch(true, "v1.24.0\n", true); + + executeAndTestVersionMatch(true, "v1.24.01", false); + + executeAndTestVersionMatch(true, "sudo: unable to resolve host node-1\nv1.24.0\n", true); + + executeAndTestVersionMatch(true, "v1.24.0+vendor.1", true); + executeAndTestVersionMatch(false, v1240WorkerNodeOutput, false); executeAndTestVersionMatch(false, v1233WorkerNodeOutput, false); diff --git a/ui/src/config/section/compute.js b/ui/src/config/section/compute.js index 4575cf2a0572..ab7efa27add7 100644 --- a/ui/src/config/section/compute.js +++ b/ui/src/config/section/compute.js @@ -641,7 +641,7 @@ export default { message: 'message.kubernetes.cluster.upgrade', docHelp: 'plugins/cloudstack-kubernetes-service.html#upgrading-kubernetes-cluster', dataView: true, - show: (record) => { return ['Created', 'Running'].includes(record.state) && record.clustertype === 'CloudManaged' }, + show: (record) => { return ['Created', 'Running', 'Alert'].includes(record.state) && record.clustertype === 'CloudManaged' }, popup: true, component: shallowRef(defineAsyncComponent(() => import('@/views/compute/UpgradeKubernetesCluster.vue'))) },