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..d1f8a1943318 100644 --- a/api/src/main/java/com/cloud/kubernetes/cluster/KubernetesCluster.java +++ b/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); 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..99ed9b52145d 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 @@ -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,8 +2639,11 @@ 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); @@ -2645,26 +2651,15 @@ public void reallyRun() { } + // 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 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 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 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 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 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 clusterVMs = kubernetesClusterVmMapDao.listByClusterId(kubernetesCluster.getId()); 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..4f7b2b36f5bd 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 @@ -157,6 +157,10 @@ public class KubernetesClusterActionWorker { protected final static List CLUSTER_NODES_TYPES_LIST = Arrays.asList(WORKER, CONTROL, ETCD); protected StateMachine2 _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,10 +761,31 @@ protected void updateLoginUserDetails(List 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); @@ -768,6 +793,64 @@ protected boolean stateTransitTo(long kubernetesClusterId, KubernetesCluster.Eve } } + /** + * 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. + *

+ * 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 publicIpSshPort = getKubernetesClusterServerIpSshPort(null); diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterAddWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterAddWorker.java index 8b694adf1cca..53012b6c93fc 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterAddWorker.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterAddWorker.java @@ -86,8 +86,9 @@ public boolean addNodesToCluster(List 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 nodesAddedAndMemory = importNodeToCluster(nodeIds, network, publicIp, controlNodeGuestIp, mountCksIsoOnVr); int nodesAdded = nodesAddedAndMemory.first(); @@ -113,7 +114,14 @@ public boolean addNodesToCluster(List 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); } } diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterDestroyWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterDestroyWorker.java index bdf515a9bf40..06a65fd4355a 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterDestroyWorker.java +++ b/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) { diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterRemoveWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterRemoveWorker.java index 07d062e23a19..275911466ee3 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterRemoveWorker.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterRemoveWorker.java @@ -70,7 +70,7 @@ public boolean removeNodesFromCluster(List 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); diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorker.java index 573e752ce125..9dfeab35507d 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorker.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorker.java @@ -123,6 +123,9 @@ public class KubernetesClusterResourceModifierActionWorker extends KubernetesClusterActionWorker { + private final KubernetesCluster.State initialState; + private boolean scalingStateEntered = false; + @Inject protected CapacityManager capacityManager; @Inject @@ -170,6 +173,7 @@ public class KubernetesClusterResourceModifierActionWorker extends KubernetesClu protected KubernetesClusterResourceModifierActionWorker(final KubernetesCluster kubernetesCluster, final KubernetesClusterManagerImpl clusterManager) { super(kubernetesCluster, clusterManager); + this.initialState = kubernetesCluster.getState(); } protected void init() { @@ -918,10 +922,34 @@ private KubernetesClusterVO updateKubernetesClusterEntry(final Boolean autoscale return kubernetesClusterVO; } - protected boolean autoscaleCluster(boolean enable, Long minSize, Long maxSize) { - if (!kubernetesCluster.getState().equals(KubernetesCluster.State.Scaling)) { - stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.AutoscaleRequested); + /** + * Moves the cluster into its scaling state the first time a scaling step needs it, failing if another operation + * moved the cluster first. Later steps of the same operation find the cluster already in that state. + */ + protected void enterScalingState(KubernetesCluster.Event event) throws CloudRuntimeException { + if (scalingStateEntered) { + return; + } + // a cluster which was never started has no scaling states: only its stored settings change. Created has no + // transition to compare and set, so this only checks that no start has taken the cluster since the request + // was accepted, which leaves a narrow window open. + if (KubernetesCluster.State.Created.equals(initialState)) { + KubernetesClusterVO cluster = kubernetesClusterDao.findById(kubernetesCluster.getId()); + if (cluster == null || !KubernetesCluster.State.Created.equals(cluster.getState())) { + throw operationRaced(initialState, cluster == null ? null : cluster.getState(), event); + } + return; } + transitToOperationState(event); + scalingStateEntered = true; + } + + protected boolean isScalingStateEntered() { + return scalingStateEntered; + } + + protected boolean autoscaleCluster(boolean enable, Long minSize, Long maxSize) { + enterScalingState(KubernetesCluster.Event.AutoscaleRequested); File pkFile = getManagementServerSshPublicKeyFile(); Pair publicIpSshPort = getKubernetesClusterServerIpSshPort(null); @@ -966,10 +994,13 @@ protected boolean autoscaleCluster(boolean enable, Long minSize, Long maxSize) { return true; } catch (Exception e) { String msg = String.format("Failed to autoscale Kubernetes cluster: %s : %s", kubernetesCluster.getName(), e.getMessage()); + // fail the scaling state this operation entered; a cluster never started holds none + if (isScalingStateEntered()) { + logTransitStateAndThrow(Level.ERROR, msg, kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); + } logAndThrow(Level.ERROR, msg); return false; } finally { - // Deploying the autoscaler might fail but it can be deployed manually too, so no need to go to an alert state updateLoginUserDetails(null); } } 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..668ee93e2345 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 @@ -120,8 +120,8 @@ protected void init() { } private void logTransitStateToFailedIfNeededAndThrow(final Level logLevel, final String message, final Exception e) throws CloudRuntimeException { - KubernetesCluster cluster = kubernetesClusterDao.findById(kubernetesCluster.getId()); - if (cluster != null && KubernetesCluster.State.Scaling.equals(cluster.getState())) { + // only fail the scaling this worker started: before it moved the cluster, a scaling state belongs to another one + if (isScalingStateEntered()) { logTransitStateAndThrow(logLevel, message, kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed, e); } else { logAndThrow(logLevel, message, e); @@ -381,10 +381,7 @@ private void validateKubernetesClusterScaleSizeParameters() throws CloudRuntimeE private void scaleKubernetesClusterOffering(KubernetesClusterNodeType nodeType, ServiceOffering serviceOffering, boolean updateNodeOffering, boolean updateClusterOffering) throws CloudRuntimeException { validateKubernetesClusterScaleOfferingParameters(); - List scalingStates = List.of(KubernetesCluster.State.Scaling, KubernetesCluster.State.ScalingStoppedCluster); - if (!scalingStates.contains(kubernetesCluster.getState())) { - stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.ScaleUpRequested); - } + enterScalingState(KubernetesCluster.Event.ScaleUpRequested); if (KubernetesCluster.State.Created.equals(originalState)) { kubernetesCluster = updateKubernetesClusterEntryForNodeType(null, nodeType, serviceOffering, updateNodeOffering, updateClusterOffering); return; @@ -457,9 +454,7 @@ private void removeNodesFromCluster(List vmMaps) throw } private void scaleDownKubernetesClusterSize() throws CloudRuntimeException { - if (!kubernetesCluster.getState().equals(KubernetesCluster.State.Scaling)) { - stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.ScaleDownRequested); - } + enterScalingState(KubernetesCluster.Event.ScaleDownRequested); List vmList; if (this.nodeIds != null) { vmList = getKubernetesClusterVMMapsForNodes(this.nodeIds).stream().filter(vm -> !vm.isExternalNode()).collect(Collectors.toList()); @@ -489,9 +484,7 @@ public List getWorkerNodesToRemove() { } private void scaleUpKubernetesClusterSize(final long newVmCount) throws CloudRuntimeException { - if (!kubernetesCluster.getState().equals(KubernetesCluster.State.Scaling)) { - stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.ScaleUpRequested); - } + enterScalingState(KubernetesCluster.Event.ScaleUpRequested); List clusterVMs = new ArrayList<>(); if (isDefaultTemplateUsed()) { LaunchPermissionVO launchPermission = new LaunchPermissionVO(clusterTemplate.getId(), owner.getId()); @@ -528,9 +521,7 @@ private void scaleKubernetesClusterSize(KubernetesClusterNodeType nodeType) thro final long originalClusterSize = kubernetesCluster.getNodeCount(); final long newVmRequiredCount = clusterSize - originalClusterSize; if (KubernetesCluster.State.Created.equals(originalState)) { - if (!kubernetesCluster.getState().equals(KubernetesCluster.State.Scaling)) { - stateTransitTo(kubernetesCluster.getId(), newVmRequiredCount > 0 ? KubernetesCluster.Event.ScaleUpRequested : KubernetesCluster.Event.ScaleDownRequested); - } + enterScalingState(newVmRequiredCount > 0 ? KubernetesCluster.Event.ScaleUpRequested : KubernetesCluster.Event.ScaleDownRequested); kubernetesCluster = updateKubernetesClusterEntryForNodeType(null, nodeType, serviceOfferingNodeTypeMap.get(nodeType.name()), false, false); return; } @@ -606,7 +597,7 @@ public boolean scaleCluster() throws CloudRuntimeException { if (autoScaled && isNodeOfferingScalingNeeded) { scaleKubernetesClusterOffering(nodeType, scalingServiceOffering, updateNodeOffering, updateClusterOffering); } - stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.OperationSucceeded); + completeScaling(); return autoScaled; } final boolean clusterSizeScalingNeeded = isWorkerNode && clusterSize != null && clusterSize != originalClusterSize; @@ -625,10 +616,17 @@ public boolean scaleCluster() throws CloudRuntimeException { } } - stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.OperationSucceeded); + completeScaling(); return true; } + private void completeScaling() { + // a cluster this worker did not move, such as one never started, may now be held by another operation + if (isScalingStateEntered()) { + stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.OperationSucceeded); + } + } + private ServiceOffering getExistingServiceOfferingForNodeType(KubernetesClusterNodeType nodeType, KubernetesCluster kubernetesCluster) { Long existingOfferingId = getExistingOfferingIdForNodeType(nodeType, kubernetesCluster); if (existingOfferingId == null) { 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..a9dc99b0d842 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 @@ -774,7 +774,7 @@ public boolean startKubernetesClusterOnCreate(Long domainId, Long accountId, Lon logger.info("Starting Kubernetes cluster: {}", kubernetesCluster); } final long startTimeoutTime = System.currentTimeMillis() + KubernetesClusterService.KubernetesClusterStartTimeout.value() * 1000; - stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.StartRequested); + transitToOperationState(KubernetesCluster.Event.StartRequested); DeployDestination dest = null; try { VMTemplateVO clusterTemplate = templateDao.findById(kubernetesCluster.getTemplateId()); @@ -907,7 +907,9 @@ public boolean startStoppedKubernetesCluster(Long domainId, Long accountId) thro logger.info("Starting Kubernetes cluster: {}", kubernetesCluster); } final long startTimeoutTime = System.currentTimeMillis() + KubernetesClusterService.KubernetesClusterStartTimeout.value() * 1000; - stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.StartRequested); + // a cluster in Error, whose create failed, is started without taking a state, so that it stays in Error if the + // start fails too, rather than going to Alert where it could be marked Running without being set up + transitToOperationState(KubernetesCluster.Event.StartRequested, KubernetesCluster.State.Error); startKubernetesClusterVMs(domainId, accountId); try { InetAddress address = InetAddress.getByName(new URL(kubernetesCluster.getEndpoint()).getHost()); @@ -978,8 +980,11 @@ 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 an operation started on it while the checks above ran: the + // operation now owns the cluster's state, and completing it here would end that state early + 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/KubernetesClusterStopWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterStopWorker.java index 59d74751dff9..4a1643d5db4f 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterStopWorker.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterStopWorker.java @@ -40,7 +40,8 @@ public boolean stop() throws CloudRuntimeException { if (logger.isInfoEnabled()) { logger.info("Stopping Kubernetes cluster: {}", kubernetesCluster); } - stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.StopRequested); + // a cluster that was never started or failed to start may still have VMs to stop; it is left in its state + transitToOperationState(KubernetesCluster.Event.StopRequested, KubernetesCluster.State.Created, KubernetesCluster.State.Error); List clusterVMs = getKubernetesClusterVMs(); for (UserVm vm : clusterVMs) { if (vm == null) { 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..aa0f967b7852 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 @@ -174,7 +174,7 @@ public boolean upgradeCluster() throws CloudRuntimeException { } filterOutManualUpgradeNodesFromClusterUpgrade(); retrieveScriptFiles(); - stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.UpgradeRequested); + transitToOperationState(KubernetesCluster.Event.UpgradeRequested); attachIsoKubernetesVMs(clusterVMs, upgradeVersion); upgradeKubernetesClusterNodes(); detachIsoKubernetesVMs(clusterVMs); diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDao.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDao.java index 7df6a6b1dce7..dd23f0f451fc 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDao.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDao.java @@ -29,6 +29,7 @@ public interface KubernetesClusterDao extends GenericDao findKubernetesClustersToGarbageCollect(); List findManagedKubernetesClustersInState(KubernetesCluster.State state); + List findManagedKubernetesClustersInStates(List states); List listByNetworkId(long networkId); List listAllByKubernetesVersion(long kubernetesVersionId); } diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDaoImpl.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDaoImpl.java index 7bec98d5d25b..55304e2ec080 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDaoImpl.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDaoImpl.java @@ -25,7 +25,7 @@ import com.cloud.utils.db.GenericDaoBase; import com.cloud.utils.db.SearchBuilder; import com.cloud.utils.db.SearchCriteria; -import com.cloud.utils.db.TransactionLegacy; +import com.cloud.utils.db.UpdateBuilder; @Component public class KubernetesClusterDaoImpl extends GenericDaoBase implements KubernetesClusterDao { @@ -36,6 +36,8 @@ public class KubernetesClusterDaoImpl extends GenericDaoBase ManagedStateSearch; private final SearchBuilder SameNetworkSearch; private final SearchBuilder KubernetesVersionSearch; + private final SearchBuilder StateChangeSearch; + private final SearchBuilder ManagedStatesSearch; public KubernetesClusterDaoImpl() { CleanupAccountIdSearch = createSearchBuilder(); @@ -61,6 +63,11 @@ public KubernetesClusterDaoImpl() { ManagedStateSearch.and("cluster_type", ManagedStateSearch.entity().getClusterType(), SearchCriteria.Op.EQ); ManagedStateSearch.done(); + ManagedStatesSearch = createSearchBuilder(); + ManagedStatesSearch.and("state", ManagedStatesSearch.entity().getState(), SearchCriteria.Op.IN); + ManagedStatesSearch.and("cluster_type", ManagedStatesSearch.entity().getClusterType(), SearchCriteria.Op.EQ); + ManagedStatesSearch.done(); + SameNetworkSearch = createSearchBuilder(); SameNetworkSearch.and("network_id", SameNetworkSearch.entity().getNetworkId(), SearchCriteria.Op.EQ); SameNetworkSearch.done(); @@ -68,6 +75,11 @@ public KubernetesClusterDaoImpl() { KubernetesVersionSearch = createSearchBuilder(); KubernetesVersionSearch.and("kubernetesVersionId", KubernetesVersionSearch.entity().getKubernetesVersionId(), SearchCriteria.Op.EQ); KubernetesVersionSearch.done(); + + StateChangeSearch = createSearchBuilder(); + StateChangeSearch.and("id", StateChangeSearch.entity().getId(), SearchCriteria.Op.EQ); + StateChangeSearch.and("state", StateChangeSearch.entity().getState(), SearchCriteria.Op.EQ); + StateChangeSearch.done(); } @Override @@ -104,18 +116,42 @@ public List findManagedKubernetesClustersInState(Kubernetes return listBy(sc); } + @Override + public List findManagedKubernetesClustersInStates(List states) { + SearchCriteria sc = ManagedStatesSearch.create(); + sc.setParameters("state", states.toArray()); + sc.setParameters("cluster_type", KubernetesCluster.ClusterType.CloudManaged); + return listBy(sc); + } + + /** + * Moves the cluster to the next state only if it is still in the given current state, so that of two operations + * started on the same cluster at once only one gets past its first transition. + */ @Override public boolean updateState(KubernetesCluster.State currentState, KubernetesCluster.Event event, KubernetesCluster.State nextState, KubernetesCluster vo, Object data) { - // TODO: ensure this update is correct - TransactionLegacy txn = TransactionLegacy.currentTxn(); - txn.start(); - KubernetesClusterVO ccVo = (KubernetesClusterVO)vo; - ccVo.setState(nextState); - super.update(ccVo.getId(), ccVo); - - txn.commit(); + SearchCriteria sc = StateChangeSearch.create(); + sc.setParameters("id", ccVo.getId()); + sc.setParameters("state", currentState); + + UpdateBuilder builder = getUpdateBuilder(ccVo); + builder.set(ccVo, "state", nextState); + int rows = update(builder, sc, null); + if (rows == 0) { + // the builder set the next state on the cluster object too: put back the state the caller read, which + // the database no longer has either, without leaving it as a change for a later update to write + builder.set(ccVo, "state", currentState); + builder.clear(); + if (!logger.isDebugEnabled()) { + return false; + } + KubernetesClusterVO dbCluster = findByIdIncludingRemoved(ccVo.getId()); + logger.debug("Unable to move Kubernetes cluster {} from state {} to {} on event {}, its state is now {}", ccVo, currentState, nextState, event, + dbCluster == null ? null : dbCluster.getState()); + return false; + } return true; } diff --git a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/ScaleKubernetesClusterCmd.java b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/ScaleKubernetesClusterCmd.java index c7ee0b7da92a..28fd446f86f7 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/ScaleKubernetesClusterCmd.java +++ b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/ScaleKubernetesClusterCmd.java @@ -164,6 +164,11 @@ public ApiCommandResourceType getApiResourceType() { return ApiCommandResourceType.KubernetesCluster; } + @Override + public Long getApiResourceId() { + return getId(); + } + ///////////////////////////////////////////////////// /////////////// API Implementation/////////////////// ///////////////////////////////////////////////////// diff --git a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/StartKubernetesClusterCmd.java b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/StartKubernetesClusterCmd.java index dae367548190..8ab6e3b450de 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/StartKubernetesClusterCmd.java +++ b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/StartKubernetesClusterCmd.java @@ -94,6 +94,11 @@ public ApiCommandResourceType getApiResourceType() { return ApiCommandResourceType.KubernetesCluster; } + @Override + public Long getApiResourceId() { + return getId(); + } + ///////////////////////////////////////////////////// /////////////// API Implementation/////////////////// ///////////////////////////////////////////////////// diff --git a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/StopKubernetesClusterCmd.java b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/StopKubernetesClusterCmd.java index 5abeb27f59b1..c14f8d7fefbe 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/StopKubernetesClusterCmd.java +++ b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/StopKubernetesClusterCmd.java @@ -92,6 +92,11 @@ public ApiCommandResourceType getApiResourceType() { return ApiCommandResourceType.KubernetesCluster; } + @Override + public Long getApiResourceId() { + return getId(); + } + ///////////////////////////////////////////////////// /////////////// API Implementation/////////////////// ///////////////////////////////////////////////////// diff --git a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/UpgradeKubernetesClusterCmd.java b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/UpgradeKubernetesClusterCmd.java index 28962c9f2892..c395ec2d65ed 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/UpgradeKubernetesClusterCmd.java +++ b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/UpgradeKubernetesClusterCmd.java @@ -102,6 +102,11 @@ public ApiCommandResourceType getApiResourceType() { return ApiCommandResourceType.KubernetesCluster; } + @Override + public Long getApiResourceId() { + return getId(); + } + ///////////////////////////////////////////////////// /////////////// API Implementation/////////////////// ///////////////////////////////////////////////////// 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..3821569bd0a0 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 @@ -46,11 +46,14 @@ import com.cloud.utils.net.NetUtils; import com.cloud.vm.VMInstanceVO; import com.cloud.vm.dao.VMInstanceDao; +import org.apache.cloudstack.api.ApiCommandResourceType; import org.apache.cloudstack.api.BaseCmd; import org.apache.cloudstack.api.command.user.kubernetes.cluster.AddVirtualMachinesToKubernetesClusterCmd; import org.apache.cloudstack.api.command.user.kubernetes.cluster.RemoveVirtualMachinesFromKubernetesClusterCmd; import org.apache.cloudstack.context.CallContext; 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.commons.collections.MapUtils; import org.junit.After; import org.junit.Assert; @@ -103,6 +106,9 @@ public class KubernetesClusterManagerImplTest { @Mock private ServiceOfferingDao serviceOfferingDao; + @Mock + private AsyncJobDao asyncJobDao; + @Spy @InjectMocks KubernetesClusterManagerImpl kubernetesClusterManager; @@ -441,4 +447,107 @@ public void testGetCksClusterPreferredArchSameArch() { String cksClusterPreferredArch = kubernetesClusterManager.getCksClusterPreferredArch(systemVMArch, cksIso); Assert.assertEquals(CPU.CPUArch.amd64.getType(), cksClusterPreferredArch); } + + private static final String CLUSTER_TYPE = ApiCommandResourceType.KubernetesCluster.toString(); + + private KubernetesClusterVO mockClusterInState(long id, KubernetesCluster.State state) { + KubernetesClusterVO cluster = Mockito.mock(KubernetesClusterVO.class); + Mockito.lenient().when(cluster.getId()).thenReturn(id); + Mockito.lenient().when(cluster.getState()).thenReturn(state); + Mockito.lenient().when(cluster.getUuid()).thenReturn("uuid-" + id); + Mockito.lenient().when(kubernetesClusterDao.findById(id)).thenReturn(cluster); + return cluster; + } + + private AsyncJobVO mockPendingJob(Long instanceId) { + AsyncJobVO job = Mockito.mock(AsyncJobVO.class); + Mockito.when(job.getInstanceId()).thenReturn(instanceId); + return job; + } + + @Test + public void recoverOrphanedOperationStatesFailsOnlyOperationsWithoutJob() { + KubernetesClusterVO orphaned = mockClusterInState(1L, KubernetesCluster.State.Upgrading); + KubernetesClusterVO busy = mockClusterInState(2L, KubernetesCluster.State.Scaling); + Mockito.when(kubernetesClusterDao.findManagedKubernetesClustersInStates(KubernetesClusterManagerImpl.OPERATION_STATES)).thenReturn(List.of(orphaned, busy)); + AsyncJobVO job = mockPendingJob(2L); + Mockito.when(asyncJobDao.findInstancePendingAsyncJobs(CLUSTER_TYPE, null)).thenReturn(List.of(job)); + Mockito.when(asyncJobDao.findInstancePendingAsyncJob(CLUSTER_TYPE, 2L)).thenReturn(job); + Mockito.when(kubernetesClusterDao.updateState(KubernetesCluster.State.Upgrading, KubernetesCluster.Event.OperationFailed, + KubernetesCluster.State.Alert, orphaned, null)).thenReturn(true); + + kubernetesClusterManager.recoverOrphanedOperationStates(); + + Mockito.verify(kubernetesClusterDao, Mockito.times(1)).updateState(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()); + } + + @Test + public void recoverOrphanedOperationStatesLeavesClusterNamedByJobOfOlderManagementServer() { + KubernetesClusterVO named = mockClusterInState(1L, KubernetesCluster.State.Starting); + KubernetesClusterVO orphaned = mockClusterInState(2L, KubernetesCluster.State.Upgrading); + Mockito.when(kubernetesClusterDao.findManagedKubernetesClustersInStates(KubernetesClusterManagerImpl.OPERATION_STATES)).thenReturn(List.of(named, orphaned)); + AsyncJobVO job = mockPendingJob(null); + Mockito.when(job.getCmdInfo()).thenReturn("{\"id\":\"uuid-1\"}"); + Mockito.when(asyncJobDao.findInstancePendingAsyncJobs(CLUSTER_TYPE, null)).thenReturn(List.of(job)); + Mockito.when(kubernetesClusterDao.updateState(KubernetesCluster.State.Upgrading, KubernetesCluster.Event.OperationFailed, + KubernetesCluster.State.Alert, orphaned, null)).thenReturn(true); + + kubernetesClusterManager.recoverOrphanedOperationStates(); + + Mockito.verify(kubernetesClusterDao, Mockito.times(1)).updateState(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()); + Mockito.verify(kubernetesClusterDao).updateState(KubernetesCluster.State.Upgrading, KubernetesCluster.Event.OperationFailed, + KubernetesCluster.State.Alert, orphaned, null); + } + + @Test + public void recoverOrphanedOperationStatesLeavesEveryClusterForJobOfOlderManagementServerWithoutParameters() { + KubernetesClusterVO cluster = mockClusterInState(1L, KubernetesCluster.State.Starting); + Mockito.when(kubernetesClusterDao.findManagedKubernetesClustersInStates(KubernetesClusterManagerImpl.OPERATION_STATES)).thenReturn(List.of(cluster)); + AsyncJobVO job = mockPendingJob(null); + Mockito.when(asyncJobDao.findInstancePendingAsyncJobs(CLUSTER_TYPE, null)).thenReturn(List.of(job)); + + kubernetesClusterManager.recoverOrphanedOperationStates(); + + Mockito.verify(kubernetesClusterDao, Mockito.never()).updateState(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()); + } + + @Test + public void recoverOrphanedOperationStatesSkipsJobLookupWithoutClusters() { + kubernetesClusterManager.recoverOrphanedOperationStates(); + + Mockito.verifyNoInteractions(asyncJobDao); + } + + @Test + public void recoverOrphanedOperationStateLeavesClusterWhoseOperationEnded() throws Exception { + mockClusterInState(1L, KubernetesCluster.State.Running); + + kubernetesClusterManager.recoverOrphanedOperationState(1L); + + Mockito.verifyNoInteractions(asyncJobDao); + Mockito.verify(kubernetesClusterDao, Mockito.never()).updateState(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()); + } + + @Test + public void recoverOrphanedOperationStateLeavesClusterWithJobStartedSinceTheScan() throws Exception { + mockClusterInState(1L, KubernetesCluster.State.Scaling); + Mockito.when(asyncJobDao.findInstancePendingAsyncJob(CLUSTER_TYPE, 1L)).thenReturn(Mockito.mock(AsyncJobVO.class)); + + kubernetesClusterManager.recoverOrphanedOperationState(1L); + + Mockito.verify(kubernetesClusterDao, Mockito.never()).updateState(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()); + } + + @Test + public void recoverOrphanedOperationStateFailsStart() throws Exception { + KubernetesClusterVO cluster = mockClusterInState(1L, KubernetesCluster.State.Starting); + Mockito.when(kubernetesClusterDao.updateState(KubernetesCluster.State.Starting, KubernetesCluster.Event.OperationFailed, + KubernetesCluster.State.Alert, cluster, null)).thenReturn(true); + + kubernetesClusterManager.recoverOrphanedOperationState(1L); + + Mockito.verify(kubernetesClusterDao).updateState(KubernetesCluster.State.Starting, KubernetesCluster.Event.OperationFailed, + KubernetesCluster.State.Alert, cluster, null); + } + } 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..2d2ad52f3aff 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 @@ -30,6 +30,7 @@ import com.cloud.kubernetes.cluster.KubernetesCluster; import com.cloud.kubernetes.cluster.KubernetesClusterDetailsVO; import com.cloud.kubernetes.cluster.KubernetesClusterManagerImpl; +import com.cloud.kubernetes.cluster.KubernetesClusterVO; import com.cloud.kubernetes.cluster.dao.KubernetesClusterDao; import com.cloud.kubernetes.cluster.dao.KubernetesClusterDetailsDao; import com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDao; @@ -38,6 +39,7 @@ import com.cloud.network.Network; import com.cloud.network.dao.IPAddressDao; import com.cloud.network.dao.IPAddressVO; +import com.cloud.utils.exception.CloudRuntimeException; @RunWith(MockitoJUnitRunner.class) public class KubernetesClusterActionWorkerTest { @@ -130,4 +132,109 @@ public void testGetVpcTierKubernetesPublicIpValid() { IpAddress result = actionWorker.getVpcTierKubernetesPublicIp(mockNetworkForGetVpcTierKubernetesPublicIpTest()); Assert.assertNotNull(result); } + + private KubernetesClusterVO mockClusterInState(KubernetesCluster.State state) { + KubernetesClusterVO cluster = Mockito.mock(KubernetesClusterVO.class); + Mockito.when(cluster.getState()).thenReturn(state); + Mockito.when(kubernetesClusterDao.findById(DEFAULT_ID)).thenReturn(cluster); + return cluster; + } + + @Test + public void testTransitToOperationStateMovesCluster() { + KubernetesClusterVO cluster = mockClusterInState(KubernetesCluster.State.Running); + Mockito.when(kubernetesClusterDao.updateState(KubernetesCluster.State.Running, KubernetesCluster.Event.UpgradeRequested, + KubernetesCluster.State.Upgrading, cluster, null)).thenReturn(true); + Assert.assertTrue(actionWorker.transitToOperationState(KubernetesCluster.Event.UpgradeRequested)); + } + + @Test(expected = CloudRuntimeException.class) + public void testTransitToOperationStateFailsWhenAnotherOperationMovedClusterFirst() { + KubernetesClusterVO cluster = mockClusterInState(KubernetesCluster.State.Running); + Mockito.when(kubernetesClusterDao.updateState(KubernetesCluster.State.Running, KubernetesCluster.Event.StopRequested, + KubernetesCluster.State.Stopping, cluster, null)).thenReturn(false); + // the compare-and-set misses even where the operation could go on without a transition: the other operation owns the state now + actionWorker.transitToOperationState(KubernetesCluster.Event.StopRequested, KubernetesCluster.State.Running); + } + + @Test(expected = CloudRuntimeException.class) + public void testTransitToOperationStateFailsWithoutTransitionInOtherStates() { + mockClusterInState(KubernetesCluster.State.Scaling); + actionWorker.transitToOperationState(KubernetesCluster.Event.DestroyRequested, KubernetesCluster.State.Created, KubernetesCluster.State.Destroying); + } + + @Test + public void testTransitToOperationStateGoesOnWithoutTransitionInGivenStates() { + mockClusterInState(KubernetesCluster.State.Destroying); + Assert.assertFalse(actionWorker.transitToOperationState(KubernetesCluster.Event.DestroyRequested, KubernetesCluster.State.Created, KubernetesCluster.State.Destroying)); + Mockito.verify(kubernetesClusterDao, Mockito.never()).updateState(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()); + } + + @Test + public void testStateTransitToFollowsTheStatesTheOperationMovesTheClusterTo() { + KubernetesClusterVO cluster = mockClusterInState(KubernetesCluster.State.Running); + Mockito.when(kubernetesClusterDao.updateState(Mockito.eq(KubernetesCluster.State.Running), Mockito.eq(KubernetesCluster.Event.UpgradeRequested), + Mockito.eq(KubernetesCluster.State.Upgrading), Mockito.eq(cluster), Mockito.isNull())).thenAnswer(invocation -> { + Mockito.when(cluster.getState()).thenReturn(KubernetesCluster.State.Upgrading); + return true; + }); + Mockito.when(kubernetesClusterDao.updateState(Mockito.eq(KubernetesCluster.State.Upgrading), Mockito.eq(KubernetesCluster.Event.OperationFailed), + Mockito.eq(KubernetesCluster.State.Alert), Mockito.eq(cluster), Mockito.isNull())).thenAnswer(invocation -> { + Mockito.when(cluster.getState()).thenReturn(KubernetesCluster.State.Alert); + return true; + }); + Mockito.when(kubernetesClusterDao.updateState(KubernetesCluster.State.Alert, KubernetesCluster.Event.OperationSucceeded, + KubernetesCluster.State.Running, cluster, null)).thenReturn(true); + + actionWorker.transitToOperationState(KubernetesCluster.Event.UpgradeRequested); + Assert.assertTrue(actionWorker.stateTransitTo(DEFAULT_ID, KubernetesCluster.Event.OperationFailed)); + // a step that failed and was retried successfully still completes the operation + Assert.assertTrue(actionWorker.stateTransitTo(DEFAULT_ID, KubernetesCluster.Event.OperationSucceeded)); + } + + @Test + public void testStateTransitToLeavesClusterAnotherOperationMovedOn() { + KubernetesClusterVO cluster = mockClusterInState(KubernetesCluster.State.Running); + Mockito.when(kubernetesClusterDao.updateState(Mockito.eq(KubernetesCluster.State.Running), Mockito.eq(KubernetesCluster.Event.UpgradeRequested), + Mockito.eq(KubernetesCluster.State.Upgrading), Mockito.eq(cluster), Mockito.isNull())).thenAnswer(invocation -> { + Mockito.when(cluster.getState()).thenReturn(KubernetesCluster.State.Upgrading); + return true; + }); + actionWorker.transitToOperationState(KubernetesCluster.Event.UpgradeRequested); + Mockito.when(cluster.getState()).thenReturn(KubernetesCluster.State.Scaling); + + Assert.assertFalse(actionWorker.stateTransitTo(DEFAULT_ID, KubernetesCluster.Event.OperationFailed)); + } + + @Test + public void testStateTransitToLeavesClusterTheOperationWentOnWithoutMoving() { + mockClusterInState(KubernetesCluster.State.Error); + Assert.assertFalse(actionWorker.transitToOperationState(KubernetesCluster.Event.StopRequested, KubernetesCluster.State.Error)); + Assert.assertFalse(actionWorker.stateTransitTo(DEFAULT_ID, KubernetesCluster.Event.OperationSucceeded)); + Mockito.verify(kubernetesClusterDao, Mockito.never()).updateState(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()); + } + + @Test + public void testStartTakesClusterInAlert() { + KubernetesClusterVO cluster = mockClusterInState(KubernetesCluster.State.Alert); + Mockito.when(kubernetesClusterDao.updateState(KubernetesCluster.State.Alert, KubernetesCluster.Event.StartRequested, + KubernetesCluster.State.Starting, cluster, null)).thenReturn(true); + Assert.assertTrue(actionWorker.transitToOperationState(KubernetesCluster.Event.StartRequested)); + } + + @Test + public void testStateTransitToKeepsTheStateAfterAnEventWithoutTransition() { + KubernetesClusterVO cluster = mockClusterInState(KubernetesCluster.State.Running); + Mockito.when(kubernetesClusterDao.updateState(Mockito.eq(KubernetesCluster.State.Running), Mockito.eq(KubernetesCluster.Event.ScaleUpRequested), + Mockito.eq(KubernetesCluster.State.Scaling), Mockito.eq(cluster), Mockito.isNull())).thenAnswer(invocation -> { + Mockito.when(cluster.getState()).thenReturn(KubernetesCluster.State.Scaling); + return true; + }); + Mockito.when(kubernetesClusterDao.updateState(KubernetesCluster.State.Scaling, KubernetesCluster.Event.OperationFailed, + KubernetesCluster.State.Alert, cluster, null)).thenReturn(true); + + actionWorker.transitToOperationState(KubernetesCluster.Event.ScaleUpRequested); + Assert.assertFalse(actionWorker.stateTransitTo(DEFAULT_ID, KubernetesCluster.Event.CreateFailed)); + Assert.assertTrue(actionWorker.stateTransitTo(DEFAULT_ID, KubernetesCluster.Event.OperationFailed)); + } } diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorkerTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorkerTest.java index c220a3468afb..2af43f1ad8ad 100644 --- a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorkerTest.java +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorkerTest.java @@ -19,10 +19,12 @@ import com.cloud.kubernetes.cluster.KubernetesCluster; import com.cloud.kubernetes.cluster.KubernetesClusterManagerImpl; +import com.cloud.kubernetes.cluster.KubernetesClusterVO; 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.dao.KubernetesSupportedVersionDao; +import com.cloud.utils.exception.CloudRuntimeException; import org.junit.Assert; import org.junit.Before; import org.junit.Test; @@ -135,4 +137,58 @@ public void getKubernetesClusterNodeNamePrefixTestNormalizedPrefixShouldNotStart Mockito.when(kubernetesClusterMock.getName()).thenReturn(originalPrefix); Assert.assertEquals(expectedPrefix, kubernetesClusterResourceModifierActionWorker.getKubernetesClusterNodeNamePrefix()); } + + @Test + public void enterScalingStateTransitionsOnlyOnceForAllStepsOfAnOperation() { + KubernetesClusterVO cluster = Mockito.mock(KubernetesClusterVO.class); + Mockito.when(cluster.getState()).thenReturn(KubernetesCluster.State.Running); + Mockito.when(kubernetesClusterMock.getId()).thenReturn(1L); + Mockito.when(kubernetesClusterMock.getState()).thenReturn(KubernetesCluster.State.Running); + kubernetesClusterResourceModifierActionWorker = new KubernetesClusterResourceModifierActionWorker(kubernetesClusterMock, kubernetesClusterManagerMock); + Mockito.when(kubernetesClusterDaoMock.findById(1L)).thenReturn(cluster); + Mockito.when(kubernetesClusterDaoMock.updateState(KubernetesCluster.State.Running, KubernetesCluster.Event.ScaleUpRequested, + KubernetesCluster.State.Scaling, cluster, null)).thenReturn(true); + + kubernetesClusterResourceModifierActionWorker.enterScalingState(KubernetesCluster.Event.ScaleUpRequested); + kubernetesClusterResourceModifierActionWorker.enterScalingState(KubernetesCluster.Event.ScaleDownRequested); + + Mockito.verify(kubernetesClusterDaoMock, Mockito.times(1)).updateState(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()); + } + + @Test(expected = CloudRuntimeException.class) + public void enterScalingStateFailsWhenAnotherOperationMovedTheClusterFirst() { + KubernetesClusterVO cluster = Mockito.mock(KubernetesClusterVO.class); + Mockito.when(cluster.getState()).thenReturn(KubernetesCluster.State.Upgrading); + Mockito.when(kubernetesClusterMock.getId()).thenReturn(1L); + Mockito.when(kubernetesClusterMock.getState()).thenReturn(KubernetesCluster.State.Running); + kubernetesClusterResourceModifierActionWorker = new KubernetesClusterResourceModifierActionWorker(kubernetesClusterMock, kubernetesClusterManagerMock); + Mockito.when(kubernetesClusterDaoMock.findById(1L)).thenReturn(cluster); + + kubernetesClusterResourceModifierActionWorker.enterScalingState(KubernetesCluster.Event.ScaleUpRequested); + } + + private void mockNeverStartedClusterNowIn(KubernetesCluster.State state) { + Mockito.when(kubernetesClusterMock.getId()).thenReturn(1L); + Mockito.when(kubernetesClusterMock.getState()).thenReturn(KubernetesCluster.State.Created); + kubernetesClusterResourceModifierActionWorker = new KubernetesClusterResourceModifierActionWorker(kubernetesClusterMock, kubernetesClusterManagerMock); + KubernetesClusterVO cluster = Mockito.mock(KubernetesClusterVO.class); + Mockito.when(cluster.getState()).thenReturn(state); + Mockito.when(kubernetesClusterDaoMock.findById(1L)).thenReturn(cluster); + } + + @Test + public void enterScalingStateLeavesNeverStartedClusterInItsState() { + mockNeverStartedClusterNowIn(KubernetesCluster.State.Created); + + kubernetesClusterResourceModifierActionWorker.enterScalingState(KubernetesCluster.Event.ScaleUpRequested); + + Mockito.verify(kubernetesClusterDaoMock, Mockito.never()).updateState(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()); + } + + @Test(expected = CloudRuntimeException.class) + public void enterScalingStateFailsForNeverStartedClusterTakenByAStart() { + mockNeverStartedClusterNowIn(KubernetesCluster.State.Starting); + + kubernetesClusterResourceModifierActionWorker.enterScalingState(KubernetesCluster.Event.ScaleUpRequested); + } } diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDaoImplTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDaoImplTest.java new file mode 100644 index 000000000000..6f87cb635e33 --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDaoImplTest.java @@ -0,0 +1,78 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +package com.cloud.kubernetes.cluster.dao; + +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import org.mockito.Spy; +import org.mockito.junit.MockitoJUnitRunner; + +import com.cloud.kubernetes.cluster.KubernetesCluster; +import com.cloud.kubernetes.cluster.KubernetesClusterVO; +import com.cloud.utils.db.GenericDaoBase; +import com.cloud.utils.db.SearchCriteria; +import com.cloud.utils.db.UpdateBuilder; + +@RunWith(MockitoJUnitRunner.class) +public class KubernetesClusterDaoImplTest { + + @Spy + KubernetesClusterDaoImpl kubernetesClusterDao; + + private boolean updateStateWithRowsUpdated(int rows, KubernetesClusterVO cluster, UpdateBuilder builder) { + Mockito.doReturn(rows).when(kubernetesClusterDao).update(Mockito.eq(builder), Mockito.any(SearchCriteria.class), Mockito.isNull()); + try (MockedStatic dao = Mockito.mockStatic(GenericDaoBase.class)) { + dao.when(() -> GenericDaoBase.getUpdateBuilder(cluster)).thenReturn(builder); + return kubernetesClusterDao.updateState(KubernetesCluster.State.Running, KubernetesCluster.Event.ScaleUpRequested, + KubernetesCluster.State.Scaling, cluster, null); + } + } + + @Test + public void updateStateSetsStateOnlyWhereClusterIsStillInCurrentState() { + KubernetesClusterVO cluster = Mockito.mock(KubernetesClusterVO.class); + Mockito.when(cluster.getId()).thenReturn(1L); + UpdateBuilder builder = Mockito.mock(UpdateBuilder.class); + + Assert.assertTrue(updateStateWithRowsUpdated(1, cluster, builder)); + + Mockito.verify(builder).set(cluster, "state", KubernetesCluster.State.Scaling); + Mockito.verify(kubernetesClusterDao).update(Mockito.eq(builder), Mockito.argThat(sc -> { + String where = sc.getWhereClause(); + return where.contains("id") && where.contains("state"); + }), Mockito.isNull()); + } + + @Test + public void updateStateFailsWhenClusterLeftCurrentState() { + KubernetesClusterVO cluster = Mockito.mock(KubernetesClusterVO.class); + Mockito.when(cluster.getId()).thenReturn(1L); + Mockito.lenient().doReturn(null).when(kubernetesClusterDao).findByIdIncludingRemoved(1L); + + UpdateBuilder builder = Mockito.mock(UpdateBuilder.class); + Assert.assertFalse(updateStateWithRowsUpdated(0, cluster, builder)); + + // the cluster object gets back the state it still has, with no change left for a later update + org.mockito.InOrder inOrder = Mockito.inOrder(builder); + inOrder.verify(builder).set(cluster, "state", KubernetesCluster.State.Scaling); + inOrder.verify(builder).set(cluster, "state", KubernetesCluster.State.Running); + inOrder.verify(builder).clear(); + } +}