Stop concurrent CKS operations clobbering each other, recover stuck states

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;
@@ -118,6 +117,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;
@@ -351,6 +352,8 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
     @Inject
     protected VMInstanceDao vmInstanceDao;
     @Inject
+    protected AsyncJobDao asyncJobDao;
+    @Inject
     protected UserVmJoinDao userVmJoinDao;
     @Inject
     protected NetworkOfferingDao networkOfferingDao;
@@ -2636,35 +2639,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()) {
@@ -2684,6 +2679,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

@@ -157,6 +157,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;
@@ -757,17 +761,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);