CKS: fix node drain on scale-down and upgrade, make a failed upgrade resumable

Scale-down and upgrade both drain a node before touching it, each with its
own copy of the drain loop, and neither works well:

- Scale-down gives up at once when kubectl drain exits non-zero (a
  PodDisruptionBudget that does not allow an eviction yet, say), and when
  it does retry after an exception it counts the attempt twice. The node
  is left cordoned, and the cluster goes to Alert although nothing has
  been removed and every node still works.
- Upgrade retries straight away, with no wait for the eviction to become
  possible, and on the last failure leaves the node cordoned.
- In both, kubectl drain has no --timeout, so it is cut off by a fixed
  60s SSH timeout rather than a deadline of its own.

Both workers now use one drain helper in KubernetesClusterActionWorker:
one count per attempt, a retry (after a wait) on a non-zero exit or an
exception, kubectl drain --timeout from the new setting
cloud.kubernetes.cluster.node.drain.timeout (300s), cut down so that no
attempt runs past the operation's deadline, with the SSH wait derived
from it, and the end of the drain output in the error. The caller
uncordons the node, and only if it cordoned it. Scale-down cordons every node it is going to
remove (leaving alone the ones that were cordoned already), drains them, and if one cannot be drained uncordons them all and
puts the cluster back in the state it was in, with an error that names
the node, instead of Alert; the same happens if the scale times out
before every node is drained. Once they are drained, the removal is not
stopped part way by the timeout any more. If removing a node fails, the
nodes still in the cluster are uncordoned.

The upgrade had two more problems:

- It only waited for the first control node to be Ready, so the next
  node could be drained while the previous one was still coming back.
  It now waits for every node, after it has been uncordoned.
- It reached node i on sshPort + i. The port-forwarding rules are not
  created in the order of the upgrade's VM list (manually upgraded nodes
  are left out of that list, and nodes added later get the next free
  port), so the script could run on the wrong node. It now looks up the
  node's own SSH port-forwarding rule, as the remove worker already does.

A failed upgrade left the cluster in Alert with no way to finish it, as
upgradeKubernetesCluster only accepts a Running cluster. An upgrade now
records the version it is upgrading to as a cluster detail (an existing
key/value table) and removes it once the cluster is Running or Stopped
again. upgradeKubernetesCluster
accepts a cluster in Alert (Alert -> Upgrading) when that detail is the
requested version (or another one with the same semantic version), which
is a real upgrade of the cluster's own version,
and all of the cluster's VMs are running, and refuses
any other Alert cluster with a clear message. For an upgrade that failed
before this change, a cluster still at the old version with nodes
recorded at the target version (kubernetesnodeversion, which is set as
each node is done) is accepted too. Nodes that are already at the target
version, both as recorded and as reported by kubectl get node, and
that are Ready, are skipped. The first control node is still upgraded first; if it is
already done, kubeadm upgrade apply is not run again. The node version
check now compares versions exactly, so 1.28.1 no longer matches
1.28.10, line by line, as the SSH output can hold stderr too, and
ignoring semver build metadata. Failure messages name the node the upgrade stopped on. The upgrade
stops if it can't move the cluster to Upgrading, and the Alert state
scanner no longer marks a cluster Running when it couldn't move it to
Recovering, and a failed upgrade detaches its ISO before going to Alert,
so a retry and the scanner or the failing job don't undo each other's
work.

No schema change.

Signed-off-by: Brad House <bhouse@nexthop.ai>
(cherry picked from commit 0dcaa8f)

Author: bhouse-nexthop

api/src/main/java/com/cloud/kubernetes/cluster/KubernetesCluster.java

@@ -104,6 +104,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);
 

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

@@ -131,6 +131,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;
@@ -175,6 +176,7 @@
 import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterStopWorker;
 import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterUpgradeWorker;
 import com.cloud.kubernetes.cluster.dao.KubernetesClusterAffinityGroupMapDao;
+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;
@@ -1570,6 +1572,40 @@ protected boolean isAnyNodeOfferingEmpty(Map<String, Long> 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<KubernetesClusterVmMapVO> 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();
@@ -1590,7 +1626,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());
@@ -1604,6 +1641,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",
@@ -1640,7 +1687,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;
@@ -3135,6 +3188,7 @@ public ConfigKey<?>[] getConfigKeys() {
             KubernetesClusterScaleTimeout,
             KubernetesClusterUpgradeTimeout,
             KubernetesClusterUpgradeRetries,
+            KubernetesClusterNodeDrainTimeout,
             KubernetesClusterAddNodeTimeout,
             KubernetesClusterRemoveNodeTimeout,
             KubernetesClusterExperimentalFeaturesEnabled,

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

@@ -88,6 +88,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<Long> KubernetesClusterNodeDrainTimeout = new ConfigKey<Long>("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<Long> KubernetesClusterAddNodeTimeout = new ConfigKey<Long>("Advanced", Long.class,
             "cloud.kubernetes.cluster.add.node.timeout",
             "3600",