Stop concurrent CKS operations clobbering each other, recover stuck states

A Kubernetes cluster's state was written by id alone, so two operations
started on the same cluster at once both got past their first
transition. Each then ran its steps against VMs the other was changing,
and whichever finished last decided the final state. The status scanner
could also overwrite the state of an operation that started after it
read the cluster.

The state is now compared and set in one update, as the VM and volume
states are. Each operation moves the cluster into its own state before
it changes anything, and fails without touching the cluster when another
operation got there first. From then on it only moves the cluster while
the cluster is in a state the operation moved it to. Starting a cluster
in Alert now moves it to Starting like any other start. A start from
Error, a stop from Created or Error, and a retried destroy still go on
without a state change, as they did before; the start and stop then
leave the state alone when they end. From any other state with no
transition for it, an operation is now refused instead of running
alongside the one that holds the cluster. The scanner only completes a
stop or a recovery that it started itself.

A management server that stopped during an operation left the cluster in
that operation's state. Nothing moved it on, and every later operation
on it was refused until the database was edited by hand. The scanner now
fails the operation of a cluster in an operation's state when no job is
in progress for it on any management server. The existing Alert
reconciliation then returns a healthy cluster to Running. To find those
jobs, the start, stop, scale and upgrade jobs now record their cluster,
as the create, delete and node jobs already did. A cluster named by a
job in progress on a management server of an older version, which does
not record it, is left alone. This replaces the scanner's first-run
handling of Starting clusters, which tried a transition that does not
exist.

No schema or API change.

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

Author: bhouse-nexthop

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

@@ -88,6 +88,7 @@ enum State {
             s_fsm.addTransition(State.Stopping, Event.OperationFailed, State.Alert);
 
             s_fsm.addTransition(State.Stopped, Event.StartRequested, State.Starting);
+            s_fsm.addTransition(State.Alert, Event.StartRequested, State.Starting);
             s_fsm.addTransition(State.Stopped, Event.OperationSucceeded, State.Stopped);
             s_fsm.addTransition(State.Running, Event.OperationSucceeded, State.Running);
 

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

@@ -29,7 +29,6 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.Date;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -122,6 +121,8 @@
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.engine.orchestration.service.NetworkOrchestrationService;
 import org.apache.cloudstack.framework.config.ConfigKey;
+import org.apache.cloudstack.framework.jobs.dao.AsyncJobDao;
+import org.apache.cloudstack.framework.jobs.impl.AsyncJobVO;
 import org.apache.cloudstack.managed.context.ManagedContextRunnable;
 import org.apache.cloudstack.network.RoutedIpv4Manager;
 import org.apache.commons.beanutils.BeanUtils;
@@ -360,6 +361,8 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
     @Inject
     protected VMInstanceDao vmInstanceDao;
     @Inject
+    protected AsyncJobDao asyncJobDao;
+    @Inject
     protected UserVmJoinDao userVmJoinDao;
     @Inject
     protected NetworkOfferingDao networkOfferingDao;
@@ -2909,35 +2912,27 @@ public void reallyRun() {
                             startWorker = ComponentContext.inject(startWorker);
                             startWorker.reconcileAlertCluster();
                         } else if (isClusterVMsInDesiredState(kubernetesCluster, VirtualMachine.State.Stopped)) {
-                            stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.StopRequested);
-                            stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.OperationSucceeded);
+                            // move the cluster as listed, so that an operation started on it since then, which may
+                            // also allow a stop, is left alone
+                            if (_stateMachine.transitTo(kubernetesCluster, KubernetesCluster.Event.StopRequested, null, kubernetesClusterDao)) {
+                                _stateMachine.transitTo(kubernetesCluster, KubernetesCluster.Event.OperationSucceeded, null, kubernetesClusterDao);
+                            }
                         }
                     } catch (Exception e) {
                         logger.warn("Failed to run Kubernetes cluster Alert state scanner on Kubernetes cluster: {} status scanner", kubernetesCluster, e);
                     }
                 }
 
 
+                // move clusters left in an operation's state by a management server that stopped to a state the
+                // reconciliation above can work from
+                try {
+                    recoverOrphanedOperationStates();
+                } catch (Exception e) {
+                    logger.warn("Failed to recover Kubernetes clusters left in an operation's state", e);
+                }
+
                 if (firstRun) {
-                    // run through Kubernetes clusters in 'Starting' state and reconcile state as 'Alert' or 'Error' if the VM's are running
-                    List<KubernetesClusterVO> startingKubernetesClusters = kubernetesClusterDao.findManagedKubernetesClustersInState(KubernetesCluster.State.Starting);
-                    for (KubernetesCluster kubernetesCluster : startingKubernetesClusters) {
-                        if ((new Date()).getTime() - kubernetesCluster.getCreated().getTime() < 10*60*1000) {
-                            continue;
-                        }
-                        if (logger.isInfoEnabled()) {
-                            logger.info("Running Kubernetes cluster state scanner on Kubernetes cluster: {} for state: {}", kubernetesCluster, KubernetesCluster.State.Starting.toString());
-                        }
-                        try {
-                            if (isClusterVMsInDesiredState(kubernetesCluster, VirtualMachine.State.Running)) {
-                                stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.FaultsDetected);
-                            } else {
-                                stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed);
-                            }
-                        } catch (Exception e) {
-                            logger.warn("Failed to run Kubernetes cluster Starting state scanner on Kubernetes cluster: {} status scanner", kubernetesCluster, e);
-                        }
-                    }
                     List<KubernetesClusterVO> destroyingKubernetesClusters = kubernetesClusterDao.findManagedKubernetesClustersInState(KubernetesCluster.State.Destroying);
                     for (KubernetesCluster kubernetesCluster : destroyingKubernetesClusters) {
                         if (logger.isInfoEnabled()) {
@@ -2957,6 +2952,65 @@ public void reallyRun() {
         }
     }
 
+    /**
+     * States which a cluster holds only while an operation runs on it, and which nothing else moves it out of.
+     */
+    protected static final List<KubernetesCluster.State> OPERATION_STATES = List.of(KubernetesCluster.State.Starting,
+            KubernetesCluster.State.Stopping, KubernetesCluster.State.Scaling, KubernetesCluster.State.ScalingStoppedCluster,
+            KubernetesCluster.State.Upgrading, KubernetesCluster.State.Importing, KubernetesCluster.State.RemovingNodes,
+            KubernetesCluster.State.Recovering);
+
+    /**
+     * Fails the operation of each cluster in an operation's state which has no job in progress on any management
+     * server. Such a cluster was left there by a management server that stopped during the operation: the jobs of a
+     * stopped management server are failed when it starts again or when another one sees it leave, but nothing moved
+     * the cluster on, and every later operation on it was refused.
+     */
+    protected void recoverOrphanedOperationStates() {
+        List<KubernetesClusterVO> clusters = kubernetesClusterDao.findManagedKubernetesClustersInStates(OPERATION_STATES);
+        if (clusters.isEmpty()) {
+            return;
+        }
+        // A management server of an older version does not record the cluster of its start, stop, scale and upgrade
+        // jobs, but their parameters name it by its UUID. A cluster that such a job may be acting on is left alone.
+        List<String> untaggedJobsCmdInfo = new ArrayList<>();
+        for (AsyncJobVO job : asyncJobDao.findInstancePendingAsyncJobs(ApiCommandResourceType.KubernetesCluster.toString(), null)) {
+            if (job.getInstanceId() == null) {
+                untaggedJobsCmdInfo.add(StringUtils.defaultString(job.getCmdInfo()));
+            }
+        }
+        for (KubernetesClusterVO cluster : clusters) {
+            if (untaggedJobsCmdInfo.stream().anyMatch(cmdInfo -> cmdInfo.isEmpty() || cmdInfo.contains(cluster.getUuid()))) {
+                logger.debug("Not recovering Kubernetes cluster {} from state {} while a job of an older management server may be acting on it", cluster, cluster.getState());
+                continue;
+            }
+            try {
+                recoverOrphanedOperationState(cluster.getId());
+            } catch (Exception e) {
+                logger.warn("Failed to recover Kubernetes cluster {} from state {}", cluster, cluster.getState(), e);
+            }
+        }
+    }
+
+    /**
+     * The cluster is read before its jobs. An operation that ends between the two reads moves the cluster to another
+     * state, so the compare-and-set of the state as read misses. For it to hit a live operation instead, one operation
+     * would have to end and another start and reach the same state between the two reads and the update right after.
+     */
+    protected void recoverOrphanedOperationState(long clusterId) throws NoTransitionException {
+        KubernetesClusterVO cluster = kubernetesClusterDao.findById(clusterId);
+        if (cluster == null || !OPERATION_STATES.contains(cluster.getState())) {
+            return;
+        }
+        if (asyncJobDao.findInstancePendingAsyncJob(ApiCommandResourceType.KubernetesCluster.toString(), clusterId) != null) {
+            return;
+        }
+        final KubernetesCluster.State state = cluster.getState();
+        if (_stateMachine.transitTo(cluster, KubernetesCluster.Event.OperationFailed, null, kubernetesClusterDao)) {
+            logger.warn("Kubernetes cluster {} was in state {} with no operation in progress, as its operation ended without moving it on, for example because the management server running it stopped; marked the operation as failed", cluster, state);
+        }
+    }
+
     // checks if Kubernetes cluster is in desired state
     private boolean isClusterVMsInDesiredState(KubernetesCluster kubernetesCluster, VirtualMachine.State state) {
         List<KubernetesClusterVmMapVO> clusterVMs = kubernetesClusterVmMapDao.listByClusterId(kubernetesCluster.getId());

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

@@ -164,6 +164,10 @@ public class KubernetesClusterActionWorker {
     protected final static List<KubernetesClusterNodeType> CLUSTER_NODES_TYPES_LIST = Arrays.asList(WORKER, CONTROL, ETCD);
 
     protected StateMachine2<KubernetesCluster.State, KubernetesCluster.Event, KubernetesCluster> _stateMachine = KubernetesCluster.State.getStateMachine();
+    // whether this worker has started its operation with transitToOperationState, and the state the operation moved
+    // the cluster to and has not moved it out of yet
+    private boolean operationStarted = false;
+    private KubernetesCluster.State ownedState;
 
     @Inject
     protected CAManager caManager;
@@ -770,17 +774,96 @@ protected void updateLoginUserDetails(List<Long> clusterVMs) {
         }
     }
 
+    /**
+     * Moves the cluster on with the given event. Once this worker has started its operation with
+     * {@link #transitToOperationState}, it only moves the cluster out of the state that the operation last moved it
+     * to: a cluster it never moved, or that has moved on since, may be held by another operation by now.
+     */
     protected boolean stateTransitTo(long kubernetesClusterId, KubernetesCluster.Event e) {
         KubernetesClusterVO kubernetesCluster = kubernetesClusterDao.findById(kubernetesClusterId);
+        if (kubernetesCluster == null) {
+            logger.warn("Unable to find Kubernetes cluster with ID {} to apply event {}", kubernetesClusterId, e);
+            return false;
+        }
+        if (operationStarted) {
+            if (ownedState == null || kubernetesCluster.getState() != ownedState) {
+                logger.debug("Not applying event {} to Kubernetes cluster {} in state {}, which its operation does not hold", e, kubernetesCluster, kubernetesCluster.getState());
+                return false;
+            }
+        }
         try {
-            return _stateMachine.transitTo(kubernetesCluster, e, null, kubernetesClusterDao);
+            boolean transitioned = _stateMachine.transitTo(kubernetesCluster, e, null, kubernetesClusterDao);
+            if (operationStarted) {
+                // the operation holds the state it moved the cluster to, until another operation moves it on; on a
+                // miss, another operation already has
+                ownedState = transitioned ? kubernetesCluster.getState() : null;
+            }
+            return transitioned;
         } 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);
             return false;
         }
     }
 
+    /**
+     * Moves the cluster into the state of the operation that is starting, failing if another operation got there first.
+     * The state is compared and set in one update, so of two operations started on the cluster at once only one gets
+     * past this point. The other fails without changing the state, and without undoing anything that the operation in
+     * progress may still need.
+     * <p>
+     * From then on the worker only moves the cluster out of that state, see {@link #stateTransitTo}.
+     *
+     * @param statesToGoOnWithout the states with no transition for the event in which the operation still goes on,
+     *                            without a state change, as it did before; in any other such state it fails. Going on
+     *                            this way, the operation holds no state, and does not move the cluster when it ends.
+     * @return whether the cluster was moved to the operation's state
+     */
+    protected boolean transitToOperationState(final KubernetesCluster.Event event, final KubernetesCluster.State... statesToGoOnWithout) throws CloudRuntimeException {
+        operationStarted = true;
+        KubernetesClusterVO cluster = kubernetesClusterDao.findById(kubernetesCluster.getId());
+        if (cluster == null) {
+            throw new CloudRuntimeException(String.format("Unable to find Kubernetes cluster %s", kubernetesCluster.getName()));
+        }
+        final KubernetesCluster.State state = cluster.getState();
+        boolean transitioned;
+        try {
+            transitioned = _stateMachine.transitTo(cluster, event, null, kubernetesClusterDao);
+        } catch (NoTransitionException e) {
+            if (Arrays.asList(statesToGoOnWithout).contains(state)) {
+                return false;
+            }
+            String message = String.format("Kubernetes cluster %s is in state %s, which does not allow %s", kubernetesCluster.getName(), state, event);
+            logger.warn(message);
+            throw new CloudRuntimeException(message);
+        }
+        if (!transitioned) {
+            KubernetesClusterVO current = kubernetesClusterDao.findById(cluster.getId());
+            throw operationRaced(state, current == null ? null : current.getState(), event);
+        }
+        ownedState = cluster.getState();
+        return true;
+    }
+
+    /**
+     * Makes the operation hold a state it went on in without moving the cluster, so that it moves the cluster out of
+     * it when it ends: a retried destroy completes the Destroying state.
+     */
+    protected void holdOperationState(final KubernetesCluster.State state) {
+        ownedState = state;
+    }
+
+    protected boolean isOperationStarted() {
+        return operationStarted;
+    }
+
+    protected CloudRuntimeException operationRaced(final KubernetesCluster.State from, final KubernetesCluster.State to, final KubernetesCluster.Event event) {
+        String message = String.format("Kubernetes cluster %s changed from state %s to %s before %s could start; another operation may be in progress, please retry once it has finished",
+                kubernetesCluster.getName(), from, to, event);
+        logger.warn(message);
+        return new CloudRuntimeException(message);
+    }
+
     protected boolean createCloudStackSecret(String[] keys) {
         File pkFile = getManagementServerSshPublicKeyFile();
         Pair<String, Integer> publicIpSshPort = getKubernetesClusterServerIpSshPort(null);

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

@@ -86,8 +86,9 @@ public boolean addNodesToCluster(List<Long> nodeIds, boolean mountCksIsoOnVr, bo
             } catch (ManagementServerException e) {
                 throw new CloudRuntimeException(String.format("Failed to retrieve public IP for the network: %s ", network.getName()));
             }
+            // take the cluster before attaching the ISO, so that an add which is refused leaves nothing behind
+            transitToOperationState(KubernetesCluster.Event.AddNodeRequested);
             attachCksIsoForNodesAdditionToCluster(nodeIds, kubernetesCluster.getId(), mountCksIsoOnVr);
-            stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.AddNodeRequested);
             String controlNodeGuestIp = getControlVmPrivateIp();
             Ternary<Integer, Long, Long> nodesAddedAndMemory = importNodeToCluster(nodeIds, network, publicIp, controlNodeGuestIp, mountCksIsoOnVr);
             int nodesAdded = nodesAddedAndMemory.first();
@@ -113,7 +114,14 @@ public boolean addNodesToCluster(List<Long> nodeIds, boolean mountCksIsoOnVr, bo
                     description, kubernetesCluster.getId(), ApiCommandResourceType.KubernetesCluster.toString(), 0);
             return true;
         } catch (Exception e) {
-            stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed);
+            // a failure before this operation took the cluster must not end the operation that holds it, and after,
+            // the cluster is only moved while this operation still holds it, see stateTransitTo
+            if (isOperationStarted()) {
+                stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed);
+            }
+            if (e instanceof CloudRuntimeException) {
+                throw (CloudRuntimeException) e;
+            }
             throw new CloudRuntimeException(e);
         }
     }

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

@@ -328,7 +328,10 @@ public boolean destroy() throws CloudRuntimeException {
         if (logger.isInfoEnabled()) {
             logger.info("Destroying Kubernetes cluster : {}", kubernetesCluster);
         }
-        stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.DestroyRequested);
+        // a cluster already being destroyed is having its destroy retried, which completes it
+        if (!transitToOperationState(KubernetesCluster.Event.DestroyRequested, KubernetesCluster.State.Destroying)) {
+            holdOperationState(KubernetesCluster.State.Destroying);
+        }
         boolean vmsDestroyed = destroyClusterVMs();
         // if there are VM's that were not expunged, we can not delete the network
         if (vmsDestroyed) {

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

@@ -70,7 +70,7 @@ public boolean removeNodesFromCluster(List<Long> nodeIds) {
         } catch (ManagementServerException e) {
             throw new CloudRuntimeException(String.format("Failed to retrieve public IP for the network: %s ", network.getName()));
         }
-        stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.RemoveNodeRequested);
+        transitToOperationState(KubernetesCluster.Event.RemoveNodeRequested);
         boolean result = removeNodesFromCluster(nodeIds, network, publicIp);
         if (!result) {
             stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed);