From fee9834af471dc88ed4b3b86e6732a66e063bf6c Mon Sep 17 00:00:00 2001 From: Brad House Date: Sat, 7 Nov 2026 10:47:03 +0000 Subject: [PATCH] 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 --- .../kubernetes/cluster/KubernetesCluster.java | 1 + .../cluster/KubernetesClusterManagerImpl.java | 98 ++++++++++++---- .../KubernetesClusterActionWorker.java | 85 +++++++++++++- .../KubernetesClusterAddWorker.java | 12 +- .../KubernetesClusterDestroyWorker.java | 5 +- .../KubernetesClusterRemoveWorker.java | 2 +- ...esClusterResourceModifierActionWorker.java | 39 ++++++- .../KubernetesClusterScaleWorker.java | 32 +++-- .../KubernetesClusterStartWorker.java | 13 ++- .../KubernetesClusterStopWorker.java | 3 +- .../KubernetesClusterUpgradeWorker.java | 2 +- .../cluster/dao/KubernetesClusterDao.java | 1 + .../cluster/dao/KubernetesClusterDaoImpl.java | 54 +++++++-- .../cluster/ScaleKubernetesClusterCmd.java | 5 + .../cluster/StartKubernetesClusterCmd.java | 5 + .../cluster/StopKubernetesClusterCmd.java | 5 + .../cluster/UpgradeKubernetesClusterCmd.java | 5 + .../KubernetesClusterManagerImplTest.java | 109 ++++++++++++++++++ .../KubernetesClusterActionWorkerTest.java | 107 +++++++++++++++++ ...usterResourceModifierActionWorkerTest.java | 56 +++++++++ .../dao/KubernetesClusterDaoImplTest.java | 78 +++++++++++++ 21 files changed, 654 insertions(+), 63 deletions(-) create mode 100644 plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDaoImplTest.java 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(); + } +}