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..71f41878ac3f 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); + } + + private 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..bfbeab41aea7 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; @@ -60,7 +59,6 @@ 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 +83,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 +287,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; @@ -412,34 +455,29 @@ private void scaleKubernetesClusterOffering(KubernetesClusterNodeType nodeType, } 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); - } - CallContext vmContext = CallContext.register(CallContext.current(), - ApiCommandResourceType.VirtualMachine); - vmContext.setEventResourceId(userVM.getId()); - 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 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; + 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); + 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); } - } 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); + firstInCluster = i + 1; + destroyRemovedNodeVm(vmMaps.get(i), userVM); } + } catch (CloudRuntimeException e) { + // The nodes that are still in the cluster were drained above and would stay unschedulable + List inCluster = vms.subList(firstInCluster, vms.size()); + uncordonNodes(cordoned.stream().filter(inCluster::contains).collect(Collectors.toList())); + throw e; } // Scale network rules to update firewall rule @@ -456,6 +494,26 @@ private void removeNodesFromCluster(List vmMaps) throw } } + /** Destroys and expunges the VM of a node that has been deleted from the Kubernetes cluster. */ + private void destroyRemovedNodeVm(final KubernetesClusterVmMapVO vmMapVO, final UserVmVO userVM) throws CloudRuntimeException { + CallContext vmContext = CallContext.register(CallContext.current(), + ApiCommandResourceType.VirtualMachine); + vmContext.setEventResourceId(userVM.getId()); + 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); + } + } 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()); + } + private void scaleDownKubernetesClusterSize() throws CloudRuntimeException { if (!kubernetesCluster.getState().equals(KubernetesCluster.State.Scaling)) { stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.ScaleDownRequested); 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..4c95f4d0fe0e 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 @@ -24,6 +24,7 @@ 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.UserVmVO; import com.cloud.vm.dao.UserVmDao; import org.junit.Assert; @@ -187,4 +188,148 @@ 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()); + } } 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'))) },